Re: [I] Flink: LogicalTypeWriterBuilder does not support UUID [iceberg]

["TheConen (via GitHub)" <gi...@apache.org>]

TheConen commented on issue #14330:
URL: https://github.com/apache/iceberg/issues/14330#issuecomment-5952773552

   I am still having this issue with Iceberg 1.12.0. Right now, we're forced to use strings instead of UUIDs in our Avro schemas.
   
   ```
   Caused by: java.lang.ClassCastException: class org.apache.flink.table.data.binary.BinaryStringData cannot be cast to class [B (org.apache.flink.table.data.binary.BinaryStringData is in unnamed module of loader 'app'; [B is in module java.base of loader 'bootstrap'). Failed to push OutputTag with id 'dynamic-forward-stream' to operator. This can occur when multiple OutputTags with different types but identical names are being used.
   	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.pushToOperator(CopyingChainingOutput.java:81) ~[flink-runtime-2.1.3.jar:2.1.3]
   	at org.apache.flink.streaming.runtime.tasks.CopyingChainingOutput.collect(CopyingChainingOutput.java:61) ~[flink-runtime-2.1.3.jar:2.1.3]
   	at org.apache.flink.streaming.runtime.tasks.ChainingOutput.collectAndCheckIfChained(ChainingOutput.java:97) ~[flink-runtime-2.1.3.jar:2.1.3]
   	at org.apache.flink.streaming.runtime.tasks.BroadcastingOutputCollector.collect(BroadcastingOutputCollector.java:95) ~[flink-runtime-2.1.3.jar:2.1.3]
   	at org.apache.flink.streaming.api.operators.ProcessOperator$ContextImpl.output(ProcessOperator.java:101) ~[flink-runtime-2.1.3.jar:2.1.3]
   	at org.apache.iceberg.flink.sink.dynamic.DynamicRecordProcessor.emit(DynamicRecordProcessor.java:213) ~[iceberg-flink-2.1-1.12.0.jar:?]
   	at org.apache.iceberg.flink.sink.dynamic.DynamicRecordProcessor.collect(DynamicRecordProcessor.java:184) ~[iceberg-flink-2.1-1.12.0.jar:?]
   	at org.apache.iceberg.flink.sink.dynamic.DynamicRecordProcessor.collect(DynamicRecordProcessor.java:37) ~[iceberg-flink-2.1-1.12.0.jar:?]
   ```
   
   We're using an Avro schema and a DynamicIcebergSink to write objects into Iceberg (with Nessie as the catalog).
   
   Example of an Avro schema affected by this:
   
   ```
   record RawKafkaRecord {
   	uuid? id = null;
   	bytes? key = null;
   	bytes? value = null;
   	union{null, array<Header>} headers = null;
   	string? topic = null;
   	int? partition = null;
   	long? offset = null;
   	long? timestamp = null;
   	string? timestamp_type = null;
   	@adjust-to-utc("true")
   	timestamp_ms? ingested_at = null;
   }
   
   record Header {
   	string? key = null;
   	bytes? value = null;
   }
   ```
   
   POJOs are generated from this schema via the avro-maven-plugin.
   
   DynamicIcebergSink:
   
   ```
   public void append(DataStream<RawKafkaRecord> records) {
           DynamicIcebergSink.forInput(records)
               .generator(new RawKafkaRecordToDynamicRecordGenerator())
               .catalogLoader(getNessieCatalogLoader())
               .uidPrefix("dynamic-sink")
               .toBranch(nessieProperties.branch())
               .immediateTableUpdate(true)
               .append();
       }
   ```
   
   The DynamicRecordGenerator:
   ```
   public class RawKafkaRecordToDynamicRecordGenerator implements DynamicRecordGenerator<RawKafkaRecord> {
   
       private static final org.apache.avro.Schema AVRO_SCHEMA = RawKafkaRecord.getClassSchema();
   
       private static final org.apache.iceberg.Schema ICEBERG_SCHEMA = AvroSchemaUtil.toIceberg(AVRO_SCHEMA);
   
       @Serial
       private static final long serialVersionUID = 1L;
   
       @Override
       public void generate(RawKafkaRecord inputRecord, Collector<DynamicRecord> out) throws Exception {
           var genericRecord = (GenericRecord) inputRecord;
           var rowData = AvroGenericRecordToRowDataMapper.forAvroSchema(AVRO_SCHEMA)
               .map(genericRecord);
   
           var nessieProperties = PropertyReader.getNessieProperties();
   
           var tableIdentifier = TableIdentifier.of(Namespace.of(nessieProperties.namespace()),
               inputRecord.getTopic()
                   .replace(".", "-"));
           var partitionSpec = PartitionSpec.unpartitioned(); // TODO Partitioning
   
           var result = new DynamicRecord(tableIdentifier, nessieProperties.branch(), ICEBERG_SCHEMA, rowData, partitionSpec);
           out.collect(result);
       }
   }
   ```
   
   Am I doing something wrong here handling Avro and Iceberg or is this bug still present?


-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org