GH-3211: Implement Variant parquet reader

[cashmand]

Rationale for this change

Provide a reference implementation for reading Variant values from Parquet based on the Variant shredding spec in https://github.com/apache/parquet-format/blob/master/VariantShredding.md.

What changes are included in this PR?

Adds VariantConverter classes to convert shredded Variant values in Parquet into a single metadata/value pair, representing the full value.

The PR adds a few new methods to the existing Variant and VariantBuilder classes to simplify the shredding reader, including a couple that are public to simplify construction of the Avro tests.

The PR also adds support in parquet-avro for using this interface: when a parquet schema element is annotated with the Variant logical type, the Avro reader will produce a record containing only the reconstructed metadata/value pair, rather than the full shredded result.

Are these changes tested?

Added unit tests, mostly based on similar tests in https://github.com/apache/iceberg/blob/main/parquet/src/test/java/org/apache/iceberg/parquet/TestVariantReaders.java.

Are there any user-facing changes?

Connectors can now read Variant values, and the parquet-avro connector will infer a Variant schema if the logical type annotation is present.

Closes#3211

[rdblue]

I wonder if we should make metadata part of this API or if we should pass in a metadata instance that does the lookup to resolve a name. Then metadata would be a separate builder that implements that API.

[rdblue]

I think this should use ByteBuffer rather than Binary.

Should this also check that fixed metadata is used?

[cashmand]

We could check if fixed metadata is used, although it is a bit messy because the nested object/array builders would need to bubble the question up to the root VariantBuilder every time.

I could make it a bit cleaner by changing the builder to store a pointer to the root builder (possibly itself). I think this is better than the current approach of storing the parent in the object/array builders, since those builder actually always want the root builder, not the parent.

[cashmand]

I switched to storing a rootBuilder field in VariantBuilder instead of a parent in each child, so that I can add this check. I think it's a mild improvement regardless of this check, but let me know what you think: dffa73c. cc @gene-db

[rdblue]

As long as we are changing the structure, would it make sense to separate out the metadata?

Right now, we assume that the builder can add keys. But that's a very simple interface with just int addDictionaryKey(String). And we also have to add a boolean for fixed metadata and a new setFixedMetadata method. It seems like it may be easier to change the builder so that it accepts some MetadataBuilder:

interfaceMetadata {
intresolve(Stringkey); // maybe a better name?ByteBuffergetEncodedBuffer();
}

Then you could pass fixed metadata using a simple wrapper class, EncodedMetadata that throws an exception if you're resolving a string that is missing. And when building metadata from scratch, resolve would assign the new ID and produce encoded metadata when getEncodedBuffer is called.

I think that would be simpler than having so much metadata-related state in the builder itself.

[cashmand]

Good idea, I refactored along the lines you suggested, with VariantBuilder using MetadataBuilder by default, but ImmutableMetadata when created from the shredding reader.

[rdblue]

If you passed in metadata, then this would not be needed. And it would separate the concerns of building metadata and building values.

[cashmand]

Done.

[rdblue]

I think we should also add a variant logical type name to the record. Probably not blocking though.

[rdblue]

Yeah, I think that adding the logical type here would allow the caller to convert this record into a Variant representation using Avro logical type converters. Not a high priority, but it would be a good addition.

[rdblue]

It appears that this is the top-level Variant. I think that should be called out more clearly than just "VariantColumn". I see why that makes sense, but "column" is overloaded so we should have a name that signals to the reader a little more clearly. Using VariantConverter would make sense to me if it weren't already used for the interface. Maybe VariantValueConverter could be used for the interface instead since those are dealing specifically with values and this is both metadata and value?

[cashmand]

I already use VariantValueConverter for the converter for the value field :(. Maybe VariantConverterInterface? Naming is the hardest thing.

[rdblue]

It looks like the loop and the check above this line could be replaced with:

this.topLevelMetadataIdx = variantSchema.getFieldIndex("metadata");

That will throw an exception if metadata isn't found.

That said, I'm wondering if this should convert all fields in order and just have handling for value, typed_value, and metadata fields. (And ignore or fail on any other names)

[cashmand]

I'll try to replace this with a setBinaryValue() method to push the value to the parent, to be more consistent with how setMetadata() is being called.

[rdblue]

These can (and should) both be final.

[rdblue]

It's usually a good idea to add any context that is helpful, like which key is missing: "Key does not exist in metadata: " + key

[rdblue]

It would probably make sense to refactor these methods into the base BinaryConverter and add a convert method that does the implementation-specific part:

abstractclassBinaryConverter ... {
protectedabstractvoidconvert(Binaryvalue);
publicvoidaddBinary(Binaryvalue) {
convert(value);
}
publicvoidaddValueFromDictionary(intdictionaryId) {
convert(dict[dictionaryId]);
}
}
classVariantMetadataConverterextendsBinaryConverter {
protectedvoidconvert(Binaryvalue) {
getParent().setMetadata(value);
}
}

Also, why use Binary instead of ByteBuffer for the internal API?

[cashmand]

Sounds good. There was no reason to use Binary, it was just what the addBinary method in PrimitiveConverter provides. I can convert it to ByteBuffer immediately.

[cashmand]

Oh, actually, I think there was a reason to use Binary. When I set the metadata, I check for reference equality with the previous value, so that if the metadata was the same dictionary-encoded value in every row, we don't need to rebuild it. If I immediately convert it to a ByteBuffer with toByteBuffer, I don't think it would compare equal to the previous ByteBuffer.

[rdblue]

Does this need to be a VariantConverter? I don't think anything here is specific to Variant. I'm not sure that the children need to be either, or at least that the metadata converter doesn't need to be. That could be passed a Consumer<ByteBuffer> instead of a parent and pass the buffer to it.

That avoids needing to pass null here, which violates the contract of VariantConverter.

[cashmand]

Good point, it doesn't.

[rdblue]

As I mentioned above, changing this so that setMetadata is passed as a Consumer<ByteBuffer>:

abstractstaticclassBinaryConverterextendsPrimitiveConverter {
Binary[] dict;
BinaryConverter() {
dict = null;
}
protectedabstractvoidhandleBinary(Binaryvalue);
@OverridepublicbooleanhasDictionarySupport() {
returntrue;
}
@OverridepublicvoidsetDictionary(Dictionarydictionary) {
dict = newBinary[dictionary.getMaxId() + 1];
for (inti = 0; i <= dictionary.getMaxId(); i++) {
dict[i] = dictionary.decodeToBinary(i);
}
}
@OverridepublicvoidaddBinary(Binaryvalue) {
handleBinary(value);
}
@OverridepublicvoidaddValueFromDictionary(intdictionaryId) {
handleBinary(dict[dictionaryId]);
}
}
staticclassVariantMetadataConverterextendsBinaryConverter {
privatefinalConsumer<ByteBuffer> parent;
publicVariantMetadataConverter(Consumer<ByteBuffer> parent) {
super();
this.parent = parent;
}
@OverrideprotectedvoidhandleBinary(Binaryvalue) {
parent.accept(value.toByteBuffer());
}
}

Then that converter is instantiated like this:

converters[topLevelMetadataIdx] = newVariantConverters.VariantMetadataConverter(this::setMetadata);

[rdblue]

If there is no else case, then I think this is going to fail at read time when Parquet tries to call methods on a null converter. To ignore a field we would need to add a noop converter. For now, I'd recommend just failing with a good exception message, like throw new UnsupportedOperationException("Cannot read variant with unexpected field: " + fieldName);

You probably won't want to throw the exception for "metadata" though, since this is currently reused for groups with metadata.

[rdblue]

Doesn't ByteBuffer guarantee that its position and limit are valid?

[cashmand]

I believe it's possible for position to equal limit (i.e. no byte can be read).

[rdblue]

This looks good to me. I'll merge when tests are passing.

[rdblue]

Thanks for fixing the tests, @cashmand! I merged this.