[GitHub] [iceberg] rdblue commented on a change in pull request #1173: FlinkTypeVisitor: Use LogicalTypeVisitor and supports MultisetType

[GitBox <gi...@apache.org>]

rdblue commented on a change in pull request #1173:
URL: https://github.com/apache/iceberg/pull/1173#discussion_r454503006



##########
File path: flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java
##########
@@ -64,86 +63,136 @@ private int getNextId() {
   }
 
   @Override
-  public Type fields(FieldsDataType fields, List<Type> types) {
-    List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(types.size());
-    boolean isRoot = root == fields;
+  public Type visit(CharType charType) {
+    return Types.StringType.get();
+  }
 
-    List<RowType.RowField> rowFields = ((RowType) fields.getLogicalType()).getFields();
-    Preconditions.checkArgument(rowFields.size() == types.size(), "fields list and types list should have same size.");
+  @Override
+  public Type visit(VarCharType varCharType) {
+    return Types.StringType.get();
+  }
 
-    for (int i = 0; i < rowFields.size(); i++) {
-      int id = isRoot ? i : getNextId();
+  @Override
+  public Type visit(BooleanType booleanType) {
+    return Types.BooleanType.get();
+  }
 
-      RowType.RowField field = rowFields.get(i);
-      String name = field.getName();
-      String comment = field.getDescription().orElse(null);
+  @Override
+  public Type visit(BinaryType binaryType) {
+    return Types.FixedType.ofLength(binaryType.getLength());
+  }
 
-      if (field.getType().isNullable()) {
-        newFields.add(Types.NestedField.optional(id, name, types.get(i), comment));
-      } else {
-        newFields.add(Types.NestedField.required(id, name, types.get(i), comment));
-      }
-    }
+  @Override
+  public Type visit(VarBinaryType varBinaryType) {
+    return Types.BinaryType.get();
+  }
 
-    return Types.StructType.of(newFields);
+  @Override
+  public Type visit(DecimalType decimalType) {
+    return Types.DecimalType.of(decimalType.getPrecision(), decimalType.getScale());
+  }
+
+  @Override
+  public Type visit(TinyIntType tinyIntType) {
+    return Types.IntegerType.get();
+  }
+
+  @Override
+  public Type visit(SmallIntType smallIntType) {
+    return Types.IntegerType.get();
+  }
+
+  @Override
+  public Type visit(IntType intType) {
+    return Types.IntegerType.get();
+  }
+
+  @Override
+  public Type visit(BigIntType bigIntType) {
+    return Types.LongType.get();
+  }
+
+  @Override
+  public Type visit(FloatType floatType) {
+    return Types.FloatType.get();
+  }
+
+  @Override
+  public Type visit(DoubleType doubleType) {
+    return Types.DoubleType.get();
+  }
+
+  @Override
+  public Type visit(DateType dateType) {
+    return Types.DateType.get();
+  }
+
+  @Override
+  public Type visit(TimeType timeType) {
+    return Types.TimeType.get();
   }
 
   @Override
-  public Type collection(CollectionDataType collection, Type elementType) {
-    if (collection.getElementDataType().getLogicalType().isNullable()) {
+  public Type visit(TimestampType timestampType) {
+    return Types.TimestampType.withoutZone();
+  }
+
+  @Override
+  public Type visit(LocalZonedTimestampType localZonedTimestampType) {
+    return Types.TimestampType.withZone();
+  }
+
+  @Override
+  public Type visit(ArrayType arrayType) {
+    Type elementType = arrayType.getElementType().accept(this);
+    if (arrayType.getElementType().isNullable()) {
       return Types.ListType.ofOptional(getNextId(), elementType);
     } else {
       return Types.ListType.ofRequired(getNextId(), elementType);
     }
   }
 
   @Override
-  public Type map(KeyValueDataType map, Type keyType, Type valueType) {
+  public Type visit(MultisetType multisetType) {
+    Type elementType = multisetType.getElementType().accept(this);
+    return Types.MapType.ofRequired(getNextId(), getNextId(), elementType, Types.IntegerType.get());
+  }
+
+  @Override
+  public Type visit(MapType mapType) {
     // keys in map are not allowed to be null.
-    if (map.getValueDataType().getLogicalType().isNullable()) {
+    Type keyType = mapType.getKeyType().accept(this);
+    Type valueType = mapType.getValueType().accept(this);
+    if (mapType.getValueType().isNullable()) {
       return Types.MapType.ofOptional(getNextId(), getNextId(), keyType, valueType);
     } else {
       return Types.MapType.ofRequired(getNextId(), getNextId(), keyType, valueType);
     }
   }
 
-  @SuppressWarnings("checkstyle:CyclomaticComplexity")
-  @Override
-  public Type atomic(AtomicDataType type) {
-    LogicalType inner = type.getLogicalType();
-    if (inner instanceof VarCharType ||
-        inner instanceof CharType) {
-      return Types.StringType.get();
-    } else if (inner instanceof BooleanType) {
-      return Types.BooleanType.get();
-    } else if (inner instanceof IntType ||
-        inner instanceof SmallIntType ||
-        inner instanceof TinyIntType) {
-      return Types.IntegerType.get();
-    } else if (inner instanceof BigIntType) {
-      return Types.LongType.get();
-    } else if (inner instanceof VarBinaryType) {
-      return Types.BinaryType.get();
-    } else if (inner instanceof BinaryType) {
-      BinaryType binaryType = (BinaryType) inner;
-      return Types.FixedType.ofLength(binaryType.getLength());
-    } else if (inner instanceof FloatType) {
-      return Types.FloatType.get();
-    } else if (inner instanceof DoubleType) {
-      return Types.DoubleType.get();
-    } else if (inner instanceof DateType) {
-      return Types.DateType.get();
-    } else if (inner instanceof TimeType) {
-      return Types.TimeType.get();
-    } else if (inner instanceof TimestampType) {
-      return Types.TimestampType.withoutZone();
-    } else if (inner instanceof LocalZonedTimestampType) {
-      return Types.TimestampType.withZone();
-    } else if (inner instanceof DecimalType) {
-      DecimalType decimalType = (DecimalType) inner;
-      return Types.DecimalType.of(decimalType.getPrecision(), decimalType.getScale());
-    } else {
-      throw new UnsupportedOperationException("Not a supported type: " + type.toString());
+  @Override
+  public Type visit(RowType rowType) {
+    List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(rowType.getFieldCount());
+    boolean isRoot = root == rowType;
+
+    List<Type> types = rowType.getFields().stream()
+        .map(f -> f.getType().accept(this))
+        .collect(Collectors.toList());
+
+    for (int i = 0; i < rowType.getFieldCount(); i++) {
+      int id = isRoot ? i : getNextId();
+
+      RowType.RowField field = rowType.getFields().get(i);
+      String name = field.getName();
+      String comment = field.getDescription().orElse(null);
+
+      if (field.getType().isNullable()) {
+        newFields.add(Types.NestedField.optional(id, name, types.get(i), comment));

Review comment:
       There is also a factory method that accepts a nullability boolean, `NestedField.of`.




----------------------------------------------------------------
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.

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