Skip to content

FlinkTypeVisitor: Use LogicalTypeVisitor and supports MultisetType #1173

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rdblue merged 2 commits into from
Jul 14, 2020
Merged

FlinkTypeVisitor: Use LogicalTypeVisitor and supports MultisetType #1173

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
4af5bf4
FlinkTypeVisitor: Use LogicalTypeVisitor and supports MultisetType
JingsongLi Jul 7, 2020
47beb33
Public to package
JingsongLi Jul 7, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 6 additions & 4 deletions flink/src/main/java/org/apache/iceberg/flink/FlinkSchemaUtil.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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());
}
Expand Down
185 changes: 117 additions & 68 deletions flink/src/main/java/org/apache/iceberg/flink/FlinkTypeToType.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -34,27 +32,28 @@
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;
import org.apache.flink.table.types.logical.TimestampType;
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<Type> {
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() {
Expand All @@ -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());

@openinx openinx Jul 7, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sounds good that we've extended support the flink multiset data type .

}

@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()

@openinx openinx Jul 9, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Seems here we don't need to loop twice ( the first loop to get List<Type> and the next loop to get List<Types.NestedField> ). Could be simplified like the following:

  @Override
  public Type visit(RowType rowType) {
    List<Types.NestedField> newFields = Lists.newArrayListWithExpectedSize(rowType.getFieldCount());
    boolean isRoot = root == rowType;

    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);
      Type type = field.getType().accept(this);

      if (field.getType().isNullable()) {
        newFields.add(Types.NestedField.optional(id, name, type, comment));
      } else {
        newFields.add(Types.NestedField.required(id, name, type, comment));
      }
    }

    return Types.StructType.of(newFields);
  }

@openinx openinx Jul 9, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

One thing is : we may adjust the place to generate field Id for nested types, then we may need to adjust the unit test ..

@JingsongLi JingsongLi Jul 13, 2020

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd prefer to keep the loop twice. If we need change the generation ID for nested types, I think it is better to change Spark too.

@openinx openinx Jul 13, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm OK about the current twice loop here now, let's just keep the consistence id generation with spark here.

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

@rdblue rdblue Jul 14, 2020

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is also a factory method that accepts a nullability boolean, NestedField.of.

} else {
newFields.add(Types.NestedField.required(id, name, types.get(i), comment));
}
}

return Types.StructType.of(newFields);
}
}
94 changes: 46 additions & 48 deletions flink/src/main/java/org/apache/iceberg/flink/FlinkTypeVisitor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<T> {
public abstract class FlinkTypeVisitor<T> implements LogicalTypeVisitor<T> {

@openinx openinx Jul 7, 2020 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@JingsongLi I'm curious that what's the difference between the flink style LogicalTypeVisitor and iceberg style visitor... Currently, all of the visitor are iceberg style, I'm not quite sure that what's the benifits to convert it to flink style visitor ...

Update: OK, I read the background in this issues here (#1173 (comment)), sounds reasonable.

@openinx openinx Jul 7, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

BTW, seems this FlinkTypeVisitor can be package access (I forget to check the access before).


static <T> T visit(DataType dataType, FlinkTypeVisitor<T> visitor) {
if (dataType instanceof FieldsDataType) {
FieldsDataType fieldsType = (FieldsDataType) dataType;
Map<String, DataType> fields = fieldsType.getFieldDataTypes();
List<T> fieldResults = Lists.newArrayList();
// ------------------------- Unsupported types ------------------------------

Preconditions.checkArgument(dataType.getLogicalType() instanceof RowType, "The logical type must be RowType");
List<RowType.RowField> 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<T> 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);
}
}
Loading