ahmedabu98 commented on code in PR #38837:
URL: https://github.com/apache/beam/pull/38837#discussion_r3565660616
##########
sdks/java/io/iceberg/src/main/java/org/apache/beam/sdk/io/iceberg/IcebergUtils.java:
##########
@@ -469,120 +470,141 @@ private static Object getIcebergTimestampValue(Object
beamValue, boolean shouldA
}
}
+ /** Converts a {@link StructLike} to a Beam {@link Row}. */
+ public static Row structToRow(Schema schema, StructLike struct) {
+ checkState(
+ schema.getFieldCount() == struct.size(),
+ "Struct of size %s does not match expected schema size %s",
+ struct.size(),
+ schema.getFieldCount());
+ Row.Builder rowBuilder = Row.withSchema(schema);
+ for (int i = 0; i < schema.getFieldCount(); i++) {
+ Schema.Field field = schema.getField(i);
+ @Nullable Object icebergValue = struct.get(i, Object.class);
+ addIcebergValue(rowBuilder, field, icebergValue);
+ }
+ return rowBuilder.build();
+ }
+
/** Converts an Iceberg {@link Record} to a Beam {@link Row}. */
public static Row icebergRecordToBeamRow(Schema schema, Record record) {
Row.Builder rowBuilder = Row.withSchema(schema);
for (Schema.Field field : schema.getFields()) {
- boolean isNullable = field.getType().getNullable();
@Nullable Object icebergValue = record.getField(field.getName());
- if (icebergValue == null) {
- if (isNullable) {
- rowBuilder.addValue(null);
- continue;
- }
- throw new RuntimeException(
- String.format("Received null value for required field '%s'.",
field.getName()));
+ addIcebergValue(rowBuilder, field, icebergValue);
+ }
+ return rowBuilder.build();
+ }
+
+ private static void addIcebergValue(
+ Row.Builder rowBuilder, Schema.Field field, @Nullable Object
icebergValue) {
+ boolean isNullable = field.getType().getNullable();
+ if (icebergValue == null) {
+ if (isNullable) {
+ rowBuilder.addValue(null);
+ return;
}
- switch (field.getType().getTypeName()) {
- case BYTE:
- case INT16:
- case INT32:
- case INT64:
- case DECIMAL: // Iceberg and Beam both use BigDecimal
- case FLOAT: // Iceberg and Beam both use float
- case DOUBLE: // Iceberg and Beam both use double
- case STRING: // Iceberg and Beam both use String
- case BOOLEAN: // Iceberg and Beam both use boolean
- rowBuilder.addValue(icebergValue);
- break;
- case ARRAY:
- checkState(
- icebergValue instanceof List,
- "Expected List type for field '%s' but received %s",
- field.getName(),
- icebergValue.getClass());
- List<@NonNull ?> beamList = (List<@NonNull ?>) icebergValue;
- Schema.FieldType collectionType =
- checkStateNotNull(field.getType().getCollectionElementType());
- // recurse on struct types
- if (collectionType.getTypeName().isCompositeType()) {
- Schema innerSchema =
checkStateNotNull(collectionType.getRowSchema());
- beamList =
- beamList.stream()
- .map(v -> icebergRecordToBeamRow(innerSchema, (Record) v))
- .collect(Collectors.toList());
- }
- rowBuilder.addValue(beamList);
- break;
- case ITERABLE:
- checkState(
- icebergValue instanceof Iterable,
- "Expected Iterable type for field '%s' but received %s",
- field.getName(),
- icebergValue.getClass());
- Iterable<@NonNull ?> beamIterable = (Iterable<@NonNull ?>)
icebergValue;
- Schema.FieldType iterableCollectionType =
- checkStateNotNull(field.getType().getCollectionElementType());
- // recurse on struct types
- if (iterableCollectionType.getTypeName().isCompositeType()) {
- Schema innerSchema =
checkStateNotNull(iterableCollectionType.getRowSchema());
- ImmutableList.Builder<Row> builder = ImmutableList.builder();
- for (Record v : (Iterable<@NonNull Record>) icebergValue) {
- builder.add(icebergRecordToBeamRow(innerSchema, v));
- }
- beamIterable = builder.build();
+ throw new RuntimeException(
+ String.format("Received null value for required field '%s'.",
field.getName()));
+ }
+ switch (field.getType().getTypeName()) {
+ case BYTE:
+ case INT16:
+ case INT32:
+ case INT64:
+ case DECIMAL: // Iceberg and Beam both use BigDecimal
+ case FLOAT: // Iceberg and Beam both use float
+ case DOUBLE: // Iceberg and Beam both use double
+ case STRING: // Iceberg and Beam both use String
+ case BOOLEAN: // Iceberg and Beam both use boolean
+ rowBuilder.addValue(icebergValue);
+ break;
+ case ARRAY:
+ checkState(
+ icebergValue instanceof List,
+ "Expected List type for field '%s' but received %s",
+ field.getName(),
+ icebergValue.getClass());
+ List<@NonNull ?> beamList = (List<@NonNull ?>) icebergValue;
+ Schema.FieldType collectionType =
+ checkStateNotNull(field.getType().getCollectionElementType());
+ // recurse on struct types
+ if (collectionType.getTypeName().isCompositeType()) {
+ Schema innerSchema =
checkStateNotNull(collectionType.getRowSchema());
+ beamList =
+ beamList.stream()
+ .map(v -> icebergRecordToBeamRow(innerSchema, (Record) v))
+ .collect(Collectors.toList());
+ }
+ rowBuilder.addValue(beamList);
+ break;
+ case ITERABLE:
+ checkState(
+ icebergValue instanceof Iterable,
+ "Expected Iterable type for field '%s' but received %s",
+ field.getName(),
+ icebergValue.getClass());
+ Iterable<@NonNull ?> beamIterable = (Iterable<@NonNull ?>)
icebergValue;
+ Schema.FieldType iterableCollectionType =
+ checkStateNotNull(field.getType().getCollectionElementType());
+ // recurse on struct types
+ if (iterableCollectionType.getTypeName().isCompositeType()) {
+ Schema innerSchema =
checkStateNotNull(iterableCollectionType.getRowSchema());
+ ImmutableList.Builder<Row> builder = ImmutableList.builder();
+ for (Record v : (Iterable<@NonNull Record>) icebergValue) {
+ builder.add(icebergRecordToBeamRow(innerSchema, v));
}
- rowBuilder.addValue(beamIterable);
- break;
- case MAP:
- checkState(
- icebergValue instanceof Map,
- "Expected Map type for field '%s' but received %s",
- field.getName(),
- icebergValue.getClass());
- Map<?, ?> beamMap = (Map<?, ?>) icebergValue;
- Schema.FieldType valueType =
checkStateNotNull(field.getType().getMapValueType());
- // recurse on struct types
- if (valueType.getTypeName().isCompositeType()) {
- Schema innerSchema = checkStateNotNull(valueType.getRowSchema());
- ImmutableMap.Builder<Object, Row> newMap = ImmutableMap.builder();
- for (Map.Entry<?, ?> entry : ((Map<?, ?>)
icebergValue).entrySet()) {
- Record rec = ((Record) entry.getValue());
- newMap.put(
- checkStateNotNull(entry.getKey()),
- icebergRecordToBeamRow(innerSchema, checkStateNotNull(rec)));
- }
- beamMap = newMap.build();
+ beamIterable = builder.build();
+ }
+ rowBuilder.addValue(beamIterable);
+ break;
+ case MAP:
+ checkState(
+ icebergValue instanceof Map,
+ "Expected Map type for field '%s' but received %s",
+ field.getName(),
+ icebergValue.getClass());
+ Map<?, ?> beamMap = (Map<?, ?>) icebergValue;
+ Schema.FieldType valueType =
checkStateNotNull(field.getType().getMapValueType());
+ // recurse on struct types
+ if (valueType.getTypeName().isCompositeType()) {
+ Schema innerSchema = checkStateNotNull(valueType.getRowSchema());
+ ImmutableMap.Builder<Object, Row> newMap = ImmutableMap.builder();
+ for (Map.Entry<?, ?> entry : ((Map<?, ?>) icebergValue).entrySet()) {
+ Record rec = ((Record) entry.getValue());
+ newMap.put(
+ checkStateNotNull(entry.getKey()),
+ icebergRecordToBeamRow(innerSchema, checkStateNotNull(rec)));
}
- rowBuilder.addValue(beamMap);
- break;
- case DATETIME:
- // Iceberg uses a long for micros.
- // Beam DATETIME uses joda's DateTime, which only supports millis,
- // so we do lose some precision here
- rowBuilder.addValue(getBeamDateTimeValue(icebergValue));
- break;
- case BYTES:
- // Iceberg uses ByteBuffer; Beam uses byte[]
- rowBuilder.addValue(((ByteBuffer) icebergValue).array());
- break;
- case ROW:
- Record nestedRecord = (Record) icebergValue;
- Schema nestedSchema =
- checkArgumentNotNull(
- field.getType().getRowSchema(),
- "Corrupted schema: Row type did not have associated nested
schema.");
- rowBuilder.addValue(icebergRecordToBeamRow(nestedSchema,
nestedRecord));
- break;
- case LOGICAL_TYPE:
- rowBuilder.addValue(getLogicalTypeValue(icebergValue,
field.getType()));
- break;
- default:
- throw new UnsupportedOperationException(
- "Unsupported Beam type: " + field.getType().getTypeName());
- }
+ beamMap = newMap.build();
+ }
+ rowBuilder.addValue(beamMap);
+ break;
+ case DATETIME:
+ // Iceberg uses a long for micros.
+ // Beam DATETIME uses joda's DateTime, which only supports millis,
+ // so we do lose some precision here
+ rowBuilder.addValue(getBeamDateTimeValue(icebergValue));
+ break;
+ case BYTES:
+ // Iceberg uses ByteBuffer; Beam uses byte[]
+ rowBuilder.addValue(((ByteBuffer) icebergValue).array());
+ break;
+ case ROW:
+ Record nestedRecord = (Record) icebergValue;
+ Schema nestedSchema =
+ checkArgumentNotNull(
+ field.getType().getRowSchema(),
+ "Corrupted schema: Row type did not have associated nested
schema.");
+ rowBuilder.addValue(icebergRecordToBeamRow(nestedSchema,
nestedRecord));
Review Comment:
Nice
--
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]
