From 4af5bf4bcdfd43aa4dc42d7064702936918fd8db Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Tue, 7 Jul 2020 11:36:43 +0800 Subject: [PATCH 1/2] FlinkTypeVisitor: Use LogicalTypeVisitor and supports MultisetType --- .../apache/iceberg/flink/FlinkSchemaUtil.java | 10 +- .../apache/iceberg/flink/FlinkTypeToType.java | 185 +++++++++++------- .../iceberg/flink/FlinkTypeVisitor.java | 94 +++++---- .../iceberg/flink/TestFlinkSchemaUtil.java | 16 +- 4 files changed, 179 insertions(+), 126 deletions(-) diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java index aa934c0f3133..d4dd61d1057e 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java +++ b/flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java @@ -20,7 +20,8 @@ package org.apache.iceberg.flink; import org.apache.flink.table.api.TableSchema; -import org.apache.flink.table.types.FieldsDataType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; import org.apache.iceberg.Schema; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.types.Type; @@ -34,10 +35,11 @@ private FlinkSchemaUtil() { * Convert the flink table schema to apache iceberg schema. */ public static Schema convert(TableSchema schema) { - Preconditions.checkArgument(schema.toRowDataType() instanceof FieldsDataType, "Should be FieldsDataType"); + LogicalType schemaType = schema.toRowDataType().getLogicalType(); + Preconditions.checkArgument(schemaType instanceof RowType, "Schema logical type should be RowType."); - FieldsDataType root = (FieldsDataType) schema.toRowDataType(); - Type converted = FlinkTypeVisitor.visit(root, new FlinkTypeToType(root)); + RowType root = (RowType) schemaType; + Type converted = root.accept(new FlinkTypeToType(root)); return new Schema(converted.asStructType().fields()); } diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java index 9595bd30f9c4..30563293f3a6 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java +++ b/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java @@ -20,10 +20,8 @@ package org.apache.iceberg.flink; import java.util.List; -import org.apache.flink.table.types.AtomicDataType; -import org.apache.flink.table.types.CollectionDataType; -import org.apache.flink.table.types.FieldsDataType; -import org.apache.flink.table.types.KeyValueDataType; +import java.util.stream.Collectors; +import org.apache.flink.table.types.logical.ArrayType; import org.apache.flink.table.types.logical.BigIntType; import org.apache.flink.table.types.logical.BinaryType; import org.apache.flink.table.types.logical.BooleanType; @@ -34,7 +32,8 @@ import org.apache.flink.table.types.logical.FloatType; import org.apache.flink.table.types.logical.IntType; import org.apache.flink.table.types.logical.LocalZonedTimestampType; -import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.MapType; +import org.apache.flink.table.types.logical.MultisetType; import org.apache.flink.table.types.logical.RowType; import org.apache.flink.table.types.logical.SmallIntType; import org.apache.flink.table.types.logical.TimeType; @@ -42,19 +41,19 @@ import org.apache.flink.table.types.logical.TinyIntType; import org.apache.flink.table.types.logical.VarBinaryType; import org.apache.flink.table.types.logical.VarCharType; -import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Lists; import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; public class FlinkTypeToType extends FlinkTypeVisitor { - private final FieldsDataType root; - private int nextId = 0; - FlinkTypeToType(FieldsDataType root) { + private final RowType root; + private int nextId; + + FlinkTypeToType(RowType root) { this.root = root; // the root struct's fields use the first ids - this.nextId = root.getFieldDataTypes().size(); + this.nextId = root.getFieldCount(); } private int getNextId() { @@ -64,33 +63,89 @@ private int getNextId() { } @Override - public Type fields(FieldsDataType fields, List types) { - List newFields = Lists.newArrayListWithExpectedSize(types.size()); - boolean isRoot = root == fields; + public Type visit(CharType charType) { + return Types.StringType.get(); + } - List 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); @@ -98,52 +153,46 @@ public Type collection(CollectionDataType collection, Type 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 newFields = Lists.newArrayListWithExpectedSize(rowType.getFieldCount()); + boolean isRoot = root == rowType; + + List 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)); + } else { + newFields.add(Types.NestedField.required(id, name, types.get(i), comment)); + } } + + return Types.StructType.of(newFields); } } diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java index 58f193366329..9d1a3c492cd7 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java +++ b/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java @@ -19,65 +19,63 @@ package org.apache.iceberg.flink; -import java.util.List; -import java.util.Map; -import org.apache.flink.table.types.AtomicDataType; -import org.apache.flink.table.types.CollectionDataType; -import org.apache.flink.table.types.DataType; -import org.apache.flink.table.types.FieldsDataType; -import org.apache.flink.table.types.KeyValueDataType; -import org.apache.flink.table.types.logical.RowType; -import org.apache.iceberg.relocated.com.google.common.base.Preconditions; -import org.apache.iceberg.relocated.com.google.common.collect.Lists; +import org.apache.flink.table.types.logical.DayTimeIntervalType; +import org.apache.flink.table.types.logical.DistinctType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.LogicalTypeVisitor; +import org.apache.flink.table.types.logical.NullType; +import org.apache.flink.table.types.logical.RawType; +import org.apache.flink.table.types.logical.StructuredType; +import org.apache.flink.table.types.logical.SymbolType; +import org.apache.flink.table.types.logical.YearMonthIntervalType; +import org.apache.flink.table.types.logical.ZonedTimestampType; -public class FlinkTypeVisitor { +public abstract class FlinkTypeVisitor implements LogicalTypeVisitor { - static T visit(DataType dataType, FlinkTypeVisitor visitor) { - if (dataType instanceof FieldsDataType) { - FieldsDataType fieldsType = (FieldsDataType) dataType; - Map fields = fieldsType.getFieldDataTypes(); - List fieldResults = Lists.newArrayList(); + // ------------------------- Unsupported types ------------------------------ - Preconditions.checkArgument(dataType.getLogicalType() instanceof RowType, "The logical type must be RowType"); - List rowFields = ((RowType) dataType.getLogicalType()).getFields(); - // Make sure that we're traveling in the same order as the RowFields because the implementation of - // FlinkTypeVisitor#fields may depends on the visit order, please see FlinkTypeToType#fields. - for (RowType.RowField rowField : rowFields) { - String name = rowField.getName(); - fieldResults.add(visit(fields.get(name), visitor)); - } + @Override + public T visit(ZonedTimestampType zonedTimestampType) { + throw new UnsupportedOperationException("Unsupported ZonedTimestampType."); + } + + @Override + public T visit(YearMonthIntervalType yearMonthIntervalType) { + throw new UnsupportedOperationException("Unsupported YearMonthIntervalType."); + } + + @Override + public T visit(DayTimeIntervalType dayTimeIntervalType) { + throw new UnsupportedOperationException("Unsupported DayTimeIntervalType."); + } + + @Override + public T visit(DistinctType distinctType) { + throw new UnsupportedOperationException("Unsupported DistinctType."); + } - return visitor.fields(fieldsType, fieldResults); - } else if (dataType instanceof CollectionDataType) { - CollectionDataType collectionType = (CollectionDataType) dataType; - return visitor.collection(collectionType, - visit(collectionType.getElementDataType(), visitor)); - } else if (dataType instanceof KeyValueDataType) { - KeyValueDataType mapType = (KeyValueDataType) dataType; - return visitor.map(mapType, - visit(mapType.getKeyDataType(), visitor), - visit(mapType.getValueDataType(), visitor)); - } else if (dataType instanceof AtomicDataType) { - AtomicDataType atomic = (AtomicDataType) dataType; - return visitor.atomic(atomic); - } else { - throw new UnsupportedOperationException("Unsupported data type: " + dataType); - } + @Override + public T visit(StructuredType structuredType) { + throw new UnsupportedOperationException("Unsupported StructuredType."); } - public T fields(FieldsDataType type, List fieldResults) { - return null; + @Override + public T visit(NullType nullType) { + throw new UnsupportedOperationException("Unsupported NullType."); } - public T collection(CollectionDataType type, T elementResult) { - return null; + @Override + public T visit(RawType rawType) { + throw new UnsupportedOperationException("Unsupported RawType."); } - public T map(KeyValueDataType type, T keyResult, T valueResult) { - return null; + @Override + public T visit(SymbolType symbolType) { + throw new UnsupportedOperationException("Unsupported SymbolType."); } - public T atomic(AtomicDataType type) { - return null; + @Override + public T visit(LogicalType other) { + throw new UnsupportedOperationException("Unsupported type: " + other); } } diff --git a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java index de0d39f69a15..c7c00c12e33c 100644 --- a/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java +++ b/flink/src/test/java/org/apache/iceberg/flink/TestFlinkSchemaUtil.java @@ -54,6 +54,7 @@ public void testConvertFlinkSchemaToIcebergSchema() { .field("decimal", DataTypes.DECIMAL(2, 2)) .field("decimal2", DataTypes.DECIMAL(38, 2)) .field("decimal3", DataTypes.DECIMAL(10, 1)) + .field("multiset", DataTypes.MULTISET(DataTypes.STRING().notNull())) .build(); Schema actualSchema = FlinkSchemaUtil.convert(flinkSchema); @@ -61,14 +62,14 @@ public void testConvertFlinkSchemaToIcebergSchema() { Types.NestedField.required(0, "id", Types.IntegerType.get(), null), Types.NestedField.optional(1, "name", Types.StringType.get(), null), Types.NestedField.required(2, "salary", Types.DoubleType.get(), null), - Types.NestedField.optional(3, "locations", Types.MapType.ofOptional(23, 24, + Types.NestedField.optional(3, "locations", Types.MapType.ofOptional(24, 25, Types.StringType.get(), Types.StructType.of( - Types.NestedField.required(21, "posX", Types.DoubleType.get(), "X field"), - Types.NestedField.required(22, "posY", Types.DoubleType.get(), "Y field") + Types.NestedField.required(22, "posX", Types.DoubleType.get(), "X field"), + Types.NestedField.required(23, "posY", Types.DoubleType.get(), "Y field") ))), - Types.NestedField.optional(4, "strArray", Types.ListType.ofOptional(25, Types.StringType.get())), - Types.NestedField.optional(5, "intArray", Types.ListType.ofOptional(26, Types.IntegerType.get())), + Types.NestedField.optional(4, "strArray", Types.ListType.ofOptional(26, Types.StringType.get())), + Types.NestedField.optional(5, "intArray", Types.ListType.ofOptional(27, Types.IntegerType.get())), Types.NestedField.required(6, "char", Types.StringType.get()), Types.NestedField.required(7, "varchar", Types.StringType.get()), Types.NestedField.optional(8, "boolean", Types.BooleanType.get()), @@ -83,7 +84,10 @@ public void testConvertFlinkSchemaToIcebergSchema() { Types.NestedField.optional(17, "date", Types.DateType.get()), Types.NestedField.optional(18, "decimal", Types.DecimalType.of(2, 2)), Types.NestedField.optional(19, "decimal2", Types.DecimalType.of(38, 2)), - Types.NestedField.optional(20, "decimal3", Types.DecimalType.of(10, 1)) + Types.NestedField.optional(20, "decimal3", Types.DecimalType.of(10, 1)), + Types.NestedField.optional(21, "multiset", Types.MapType.ofRequired(28, 29, + Types.StringType.get(), + Types.IntegerType.get())) ); Assert.assertEquals(expectedSchema.asStruct(), actualSchema.asStruct()); From 47beb33eda1b024280129f2422a9f32d6a0d79e9 Mon Sep 17 00:00:00 2001 From: JingsongLi Date: Tue, 7 Jul 2020 17:32:16 +0800 Subject: [PATCH 2/2] Public to package --- .../src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java | 2 +- .../main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java index 30563293f3a6..2cc9771c1969 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java +++ b/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java @@ -45,7 +45,7 @@ import org.apache.iceberg.types.Type; import org.apache.iceberg.types.Types; -public class FlinkTypeToType extends FlinkTypeVisitor { +class FlinkTypeToType extends FlinkTypeVisitor { private final RowType root; private int nextId; diff --git a/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java b/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java index 9d1a3c492cd7..e7956e2c5bb2 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java +++ b/flink/src/main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java @@ -30,7 +30,7 @@ import org.apache.flink.table.types.logical.YearMonthIntervalType; import org.apache.flink.table.types.logical.ZonedTimestampType; -public abstract class FlinkTypeVisitor implements LogicalTypeVisitor { +abstract class FlinkTypeVisitor implements LogicalTypeVisitor { // ------------------------- Unsupported types ------------------------------