pvary commented on code in PR #14245:
URL: https://github.com/apache/iceberg/pull/14245#discussion_r2414538991
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/sink/AvroGenericRecordToRowDataMapper.java:
##########
@@ -41,13 +45,114 @@
public class AvroGenericRecordToRowDataMapper implements
MapFunction<GenericRecord, RowData> {
private final AvroToRowDataConverters.AvroToRowDataConverter converter;
+ private final Schema avroSchema;
+ private final Function<GenericRecord, RowData> customConverter;
- AvroGenericRecordToRowDataMapper(RowType rowType) {
+ AvroGenericRecordToRowDataMapper(RowType rowType, Schema avroSchema) {
+ this.avroSchema = avroSchema;
this.converter = AvroToRowDataConverters.createRowConverter(rowType);
+ // Check if we need custom conversion for timestamp-nanos
+ this.customConverter = needsCustomConversion(avroSchema) ?
this::convertWithNanos : null;
+ }
+
+ private boolean needsCustomConversion(Schema schema) {
+ for (Schema.Field field : schema.getFields()) {
+ if (field.schema().getLogicalType() instanceof
LogicalTypes.TimestampNanos) {
+ return true;
+ }
+ }
+ return false;
+ }
+
+ private RowData convertWithNanos(GenericRecord genericRecord) {
+ RowData baseRowData = (RowData) converter.convert(genericRecord);
+ GenericRowData result = new GenericRowData(baseRowData.getArity());
+ result.setRowKind(baseRowData.getRowKind());
+
+ for (int i = 0; i < avroSchema.getFields().size(); i++) {
+ Schema.Field field = avroSchema.getFields().get(i);
+ if (field.schema().getLogicalType() instanceof
LogicalTypes.TimestampNanos) {
+ // Handle timestamp-nanos by converting from long nanos to
TimestampData
+ Long nanos = (Long) genericRecord.get(i);
+ if (nanos == null) {
+ result.setField(i, null);
+ } else {
+ long millis = Math.floorDiv(nanos, 1_000_000L);
+ int nanoOfMillis = (int) Math.floorMod(nanos, 1_000_000L);
+ result.setField(i, TimestampData.fromEpochMillis(millis,
nanoOfMillis));
+ }
+ } else {
+ // Copy from base conversion
+ if (baseRowData.isNullAt(i)) {
+ result.setField(i, null);
+ } else {
+ // Get the value from baseRowData using the appropriate getter
+ result.setField(i, getFieldValue(baseRowData, i, field));
+ }
+ }
+ }
+ return result;
+ }
+
+ private Object getFieldValue(RowData rowData, int pos, Schema.Field field) {
+ // This is a simplified version - you may need to handle more types
+ Schema fieldSchema = field.schema();
+ if (fieldSchema.getType() == Schema.Type.UNION) {
+ // Handle nullable fields
+ for (Schema unionSchema : fieldSchema.getTypes()) {
+ if (unionSchema.getType() != Schema.Type.NULL) {
+ fieldSchema = unionSchema;
+ break;
+ }
+ }
+ }
+
+ // Check for logical types first
+ if (fieldSchema.getLogicalType() != null) {
+ if (fieldSchema.getLogicalType() instanceof LogicalTypes.TimestampMillis
+ || fieldSchema.getLogicalType() instanceof
LogicalTypes.TimestampMicros) {
+ return rowData.getTimestamp(pos, 6);
+ } else if (fieldSchema.getLogicalType() instanceof LogicalTypes.Decimal)
{
+ LogicalTypes.Decimal decimal = (LogicalTypes.Decimal)
fieldSchema.getLogicalType();
+ return rowData.getDecimal(pos, decimal.getPrecision(),
decimal.getScale());
+ }
+ // timestamp-nanos is handled separately in convertWithNanos
+ }
+
+ switch (fieldSchema.getType()) {
+ case BOOLEAN:
+ return rowData.getBoolean(pos);
+ case INT:
+ return rowData.getInt(pos);
+ case LONG:
+ return rowData.getLong(pos);
+ case FLOAT:
+ return rowData.getFloat(pos);
+ case DOUBLE:
+ return rowData.getDouble(pos);
+ case STRING:
+ return rowData.getString(pos);
+ case BYTES:
+ return rowData.getBinary(pos);
+ case FIXED:
+ return rowData.getBinary(pos);
+ case RECORD:
+ return rowData.getRow(pos, fieldSchema.getFields().size());
+ case ARRAY:
+ return rowData.getArray(pos);
+ case MAP:
+ return rowData.getMap(pos);
+ default:
+ // For complex types, just return as-is
+ return ((GenericRowData) rowData).getField(pos);
Review Comment:
What if the complex type contains a nano timestamp?
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
