Flink: LogicalTypeWriterBuilder does not support UUID

[rlokugamage]

Apache Iceberg version

1.10.0 (latest release)

Query engine

Flink

Please describe the bug 🐞

flink: 2.0
iceberg-flink: 1.10.0 (2.0 runtime jar)
parquet: 1.16.0

Trying to write to an iceberg table with a UUID column. However when flink tries writing I get this error:

Caused by: java.lang.UnsupportedOperationException: Unsupported logical type: null
	at org.apache.iceberg.flink.data.FlinkParquetWriters$WriteBuilder.primitive(FlinkParquetWriters.java:163)
	at org.apache.iceberg.flink.data.FlinkParquetWriters$WriteBuilder.primitive(FlinkParquetWriters.java:75)
	at org.apache.iceberg.flink.data.ParquetWithFlinkSchemaVisitor.visit(ParquetWithFlinkSchemaVisitor.java:51)
	at org.apache.iceberg.flink.data.ParquetWithFlinkSchemaVisitor.visitField(ParquetWithFlinkSchemaVisitor.java:162)
	at org.apache.iceberg.flink.data.ParquetWithFlinkSchemaVisitor.visitFields(ParquetWithFlinkSchemaVisitor.java:186)
	at org.apache.iceberg.flink.data.ParquetWithFlinkSchemaVisitor.visit(ParquetWithFlinkSchemaVisitor.java:49)
	at org.apache.iceberg.flink.data.FlinkParquetWriters.buildWriter(FlinkParquetWriters.java:72)
	at org.apache.iceberg.flink.sink.FlinkAppenderFactory.lambda$newAppender$2(FlinkAppenderFactory.java:126)

doing some debugging I found that it's failing on the UUID column. I see that in FlinkParquetWriters.java it doesn't support the UUID Logical Type Annotation. Flink and Parquet support UUID so I don't understand why this wasn't included.

Since UUIDs in parquet are fixed length byte arrays, should be implemented like this:

import org.apache.parquet.schema.LogicalTypeAnnotation.UUIDLogicalTypeAnnotation;
...
private static class LogicalTypeWriterBuilder
      implements LogicalTypeAnnotationVisitor<ParquetValueWriter<?>> {
    private final LogicalType flinkType;
    private final ColumnDescriptor desc;

    private LogicalTypeWriterBuilder(LogicalType flinkType, ColumnDescriptor desc) {
      this.flinkType = flinkType;
      this.desc = desc;
    }
...
    @Override
    public Optional<ParquetValueWriter<?>> visit(UUIDLogicalTypeAnnotation uuid) {
        return Optional.of(byteArrays(desc));
    }
}

Willingness to contribute

• I can contribute a fix for this bug independently
• I would be willing to contribute a fix for this bug with guidance from the Iceberg community
• I cannot contribute a fix for this bug at this time

[pvary]

@rlokugamage: What would be the corresponding type in Flink RowData?

[rlokugamage]

@pvary Should be binary as well. Though I see that in the flink docs the UUID function returns a string, so maybe a string/varchar?

https://nightlies.apache.org/flink/flink-docs-release-2.0/docs/dev/table/functions/systemfunctions/#arithmetic-functions

Returns an UUID (Universally Unique Identifier) string (e.g., “3d3c68f7-f608-473f-b60c-b0c44ad4cc4e”)
according to RFC 4122 type 4 (pseudo randomly generated) UUID. 
The UUID is generated using a cryptographically strong pseudo random number generator.

[rlokugamage]

okay, looks like we already convert Iceberg's UUID type to the flink binary type
https://github.com/apache/iceberg/blob/apache-iceberg-1.10.0/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/TypeToFlinkType.java#L132

however if someone is trying to write an existing flink table to iceberg, that would get converted to String type.
https://github.com/apache/iceberg/blob/main/flink/v2.0/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java#L74

hmmm IDK how'd we support reading a flink table and convert string types/binary types to UUID explicitly. Flink doesn't seem to be in any rush to add a UUID type

[rlokugamage]

I think we can still update the parquet writer to write UUID since parquet supports it, and that should work so long as the pipeline is not trying the source to a flink table. For example what I'm trying to do:

flowchart TD
    A[Kafka Source] --> B(Read Table Schema)
    B --> C(Transform to RowData)
    C --> D[Iceberg Sink]

Loading

[rlokugamage]

can flink just copy what spark is doing? Since Spark and Flink both treat UUID as strings
https://github.com/apache/iceberg/blob/main/spark/v4.0/spark/src/main/java/org/apache/iceberg/spark/data/SparkParquetWriters.java

[pvary]

@rlokugamage: So you are suggesting to treat UUID as a String type in Flink, and convert the string to UUID and write the bytes when writing to a Parquet file?

What I see in the tests, that we expect UUID to stored in bytes in the RowData:

iceberg/flink/v2.0/flink/src/test/java/org/apache/iceberg/flink/DataGenerators.java

Lines 196 to 199 in f7e6a27

	byte[] uuidBytes = new byte[16];
	for (int i = 0; i < 16; ++i) {
	uuidBytes[i] = (byte) i;
	}

We might have issues with writing out the UUID to Parquet/ORC/Avro files, but the internal representation seems to be decided on earlier.

[rlokugamage]

@pvary oh cool, if it's already determined to be a byte array then that works better for Parquet since UUIDs are fixed length byte arrays

I think it might be weird to break from the convention that's already in Spark, but using bytes is more correct I guess.

[BryanEdwardsTech7]

@rlokugamage @pvary I was running into an error while trying to write UUID types via avro to Iceberg and ran into this issue. Are there any work arounds to writing UUID values? With parquets, I was able to use DataTypes.FIELD("_uuid", DataTypes.VARBINARY(16)) with the Flink's DataStream API, but with Avros I ran into a casting error even if I try String, Bytes, or UUID types.

[Sbaia]

Is there any progress on this issue? It seems very strange that Flink can read UUIDs from icebergs but not write them.

UUID is an alias of a binary(16), rather than treating UUID as a string it could not be handled as binary(16), rather than treating it as a string. takes up half the space in memory as well, and would treat it as an alias, which is so