openinx commented on a change in pull request #3240: URL: https://github.com/apache/iceberg/pull/3240#discussion_r736328720
########## File path: flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java ########## @@ -0,0 +1,210 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.flink.data; + +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.RawValueData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; +import org.apache.iceberg.Schema; +import org.apache.iceberg.flink.FlinkSchemaUtil; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.types.Types; + +public class RowDataProjection implements RowData { + + private final RowData.FieldGetter[] getters; + private RowData rowData; + + public static RowDataProjection create(Schema schema, Schema projectSchema) { + return new RowDataProjection(FlinkSchemaUtil.convert(schema), schema.asStruct(), projectSchema.asStruct()); + } + + private RowDataProjection(RowType rowType, Types.StructType rowStruct, Types.StructType projectType) { + this.getters = new RowData.FieldGetter[projectType.fields().size()]; + for (int i = 0; i < getters.length; i++) { + getters[i] = createFieldGetter(rowType, rowStruct, projectType.fields().get(i)); + } + } + + private static RowData.FieldGetter createFieldGetter(RowType rowType, + Types.StructType rowStruct, + Types.NestedField projectField) { + for (int i = 0; i < rowStruct.fields().size(); i++) { Review comment: > I have a little performance concern of n^2 complexity for table with a lot of columns (like thousands or more) I'm fine with either. Because the complexity is actually `n*m`, let's say the `n` is the table's field number and `m` is the projection fields number. If both @stevenzwu and @kbendick think it's necessary to do, I'm okay with it. ########## File path: flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java ########## @@ -0,0 +1,219 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.flink.data; + +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.RawValueData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.flink.FlinkSchemaUtil; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.types.Types; + +public class RowDataProjection implements RowData { + /** + * Creates a projecting wrapper for {@link RowData} rows. + * <p> + * This projection does not work with repeated types like lists and maps. + * + * @param schema schema of rows wrapped by this projection + * @param projectedSchema result schema of the projected rows + * @return a wrapper to project rows + */ + public static RowDataProjection create(Schema schema, Schema projectedSchema) { + return new RowDataProjection(FlinkSchemaUtil.convert(schema), schema.asStruct(), projectedSchema.asStruct()); + } + + private final RowData.FieldGetter[] getters; + private RowData rowData; + + private RowDataProjection(RowType rowType, Types.StructType rowStruct, Types.StructType projectType) { + this.getters = new RowData.FieldGetter[projectType.fields().size()]; + for (int i = 0; i < getters.length; i++) { + getters[i] = createFieldGetter(rowType, rowStruct, projectType.fields().get(i)); + } + } + + private static RowData.FieldGetter createFieldGetter(RowType rowType, + Types.StructType rowStruct, + Types.NestedField projectField) { + for (int i = 0; i < rowStruct.fields().size(); i++) { + Types.NestedField rowField = rowStruct.fields().get(i); + if (rowField.fieldId() == projectField.fieldId()) { + Preconditions.checkArgument(rowField.type().typeId() == projectField.type().typeId(), + String.format("Different iceberg type between row field <%s> and project field <%s>", + rowField, projectField)); + + switch (projectField.type().typeId()) { + case STRUCT: + RowType nestedRowType = (RowType) rowType.getTypeAt(i); + int rowPos = i; + return row -> { + RowData nestedRow = row.isNullAt(rowPos) ? null : row.getRow(rowPos, nestedRowType.getFieldCount()); + return new RowDataProjection(nestedRowType, rowField.type().asStructType(), + projectField.type().asStructType()).wrap(nestedRow); + }; + + case MAP: + Types.MapType projectedMap = projectField.type().asMapType(); + Types.MapType originalMap = rowField.type().asMapType(); + + boolean keyProjectable = !projectedMap.keyType().isNestedType() || + projectedMap.keyType().equals(originalMap.keyType()); + boolean valueProjectable = !projectedMap.valueType().isNestedType() || + projectedMap.valueType().equals(originalMap.valueType()); + Preconditions.checkArgument(keyProjectable && valueProjectable, + "Cannot project a partial map key or value RowData. Trying to project %s out of %s", + projectField, rowField); + + return RowData.createFieldGetter(rowType.getTypeAt(i), i); + + case LIST: + Types.ListType projectedList = projectField.type().asListType(); + Types.ListType originalList = rowField.type().asListType(); + + boolean elementProjectable = !projectedList.elementType().isNestedType() || + projectedList.elementType().equals(originalList.elementType()); + Preconditions.checkArgument(elementProjectable, + "Cannot project a partial list element RowData. Trying to project %s out of %s", + projectField, rowField); + + return RowData.createFieldGetter(rowType.getTypeAt(i), i); + + default: + return RowData.createFieldGetter(rowType.getTypeAt(i), i); + } + } + } + throw new IllegalArgumentException(String.format("Cannot find field %s in %s", projectField, rowStruct)); + } + + public RowData wrap(RowData row) { + this.rowData = row; + return this; + } + + public Object getValue(int pos) { Review comment: Nit: this can be a private method, right ? ########## File path: flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java ########## @@ -0,0 +1,210 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.flink.data; + +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.RawValueData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; +import org.apache.iceberg.Schema; +import org.apache.iceberg.flink.FlinkSchemaUtil; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.types.Types; + +public class RowDataProjection implements RowData { + + private final RowData.FieldGetter[] getters; + private RowData rowData; + + public static RowDataProjection create(Schema schema, Schema projectSchema) { + return new RowDataProjection(FlinkSchemaUtil.convert(schema), schema.asStruct(), projectSchema.asStruct()); + } + + private RowDataProjection(RowType rowType, Types.StructType rowStruct, Types.StructType projectType) { + this.getters = new RowData.FieldGetter[projectType.fields().size()]; + for (int i = 0; i < getters.length; i++) { + getters[i] = createFieldGetter(rowType, rowStruct, projectType.fields().get(i)); + } + } + + private static RowData.FieldGetter createFieldGetter(RowType rowType, + Types.StructType rowStruct, + Types.NestedField projectField) { + for (int i = 0; i < rowStruct.fields().size(); i++) { Review comment: > I have a little performance concern of n^2 complexity for table with a lot of columns (like thousands or more) I'm fine with either. Because the complexity is actually `n*m`, let's say the `n` is the table's field number and `m` is the projection fields number. If both @stevenzwu and @kbendick think it's necessary to do, I'm okay with it. ########## File path: flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java ########## @@ -0,0 +1,219 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.flink.data; + +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.RawValueData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.flink.FlinkSchemaUtil; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.types.Types; + +public class RowDataProjection implements RowData { + /** + * Creates a projecting wrapper for {@link RowData} rows. + * <p> + * This projection does not work with repeated types like lists and maps. + * + * @param schema schema of rows wrapped by this projection + * @param projectedSchema result schema of the projected rows + * @return a wrapper to project rows + */ + public static RowDataProjection create(Schema schema, Schema projectedSchema) { + return new RowDataProjection(FlinkSchemaUtil.convert(schema), schema.asStruct(), projectedSchema.asStruct()); + } + + private final RowData.FieldGetter[] getters; + private RowData rowData; + + private RowDataProjection(RowType rowType, Types.StructType rowStruct, Types.StructType projectType) { + this.getters = new RowData.FieldGetter[projectType.fields().size()]; + for (int i = 0; i < getters.length; i++) { + getters[i] = createFieldGetter(rowType, rowStruct, projectType.fields().get(i)); + } + } + + private static RowData.FieldGetter createFieldGetter(RowType rowType, + Types.StructType rowStruct, + Types.NestedField projectField) { + for (int i = 0; i < rowStruct.fields().size(); i++) { + Types.NestedField rowField = rowStruct.fields().get(i); + if (rowField.fieldId() == projectField.fieldId()) { + Preconditions.checkArgument(rowField.type().typeId() == projectField.type().typeId(), + String.format("Different iceberg type between row field <%s> and project field <%s>", + rowField, projectField)); + + switch (projectField.type().typeId()) { + case STRUCT: + RowType nestedRowType = (RowType) rowType.getTypeAt(i); + int rowPos = i; + return row -> { + RowData nestedRow = row.isNullAt(rowPos) ? null : row.getRow(rowPos, nestedRowType.getFieldCount()); Review comment: I had a small patch for this: ```diff diff --git a/flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java b/flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java index 9d1e8ea67..25a5b3ab3 100644 --- a/flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java +++ b/flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java @@ -45,7 +45,11 @@ public class RowDataProjection implements RowData { * @return a wrapper to project rows */ public static RowDataProjection create(Schema schema, Schema projectedSchema) { - return new RowDataProjection(FlinkSchemaUtil.convert(schema), schema.asStruct(), projectedSchema.asStruct()); + return RowDataProjection.create(FlinkSchemaUtil.convert(schema), schema.asStruct(), projectedSchema.asStruct()); + } + + public static RowDataProjection create(RowType rowType, Types.StructType schema, Types.StructType projectedSchema) { + return new RowDataProjection(rowType, schema, projectedSchema); } private final RowData.FieldGetter[] getters; @@ -73,9 +77,14 @@ public class RowDataProjection implements RowData { RowType nestedRowType = (RowType) rowType.getTypeAt(i); int rowPos = i; return row -> { - RowData nestedRow = row.isNullAt(rowPos) ? null : row.getRow(rowPos, nestedRowType.getFieldCount()); - return new RowDataProjection(nestedRowType, rowField.type().asStructType(), - projectField.type().asStructType()).wrap(nestedRow); + if (row.isNullAt(rowPos)) { + return null; + } else { + RowData nestedRow = row.getRow(rowPos, nestedRowType.getFieldCount()); + return RowDataProjection + .create(nestedRowType, rowField.type().asStructType(), projectField.type().asStructType()) + .wrap(nestedRow); + } }; case MAP: ``` ########## File path: flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java ########## @@ -0,0 +1,219 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.flink.data; + +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.RawValueData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.flink.FlinkSchemaUtil; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.types.Types; + +public class RowDataProjection implements RowData { + /** + * Creates a projecting wrapper for {@link RowData} rows. + * <p> + * This projection does not work with repeated types like lists and maps. + * + * @param schema schema of rows wrapped by this projection + * @param projectedSchema result schema of the projected rows + * @return a wrapper to project rows + */ + public static RowDataProjection create(Schema schema, Schema projectedSchema) { + return new RowDataProjection(FlinkSchemaUtil.convert(schema), schema.asStruct(), projectedSchema.asStruct()); + } + + private final RowData.FieldGetter[] getters; + private RowData rowData; + + private RowDataProjection(RowType rowType, Types.StructType rowStruct, Types.StructType projectType) { + this.getters = new RowData.FieldGetter[projectType.fields().size()]; + for (int i = 0; i < getters.length; i++) { + getters[i] = createFieldGetter(rowType, rowStruct, projectType.fields().get(i)); + } + } + + private static RowData.FieldGetter createFieldGetter(RowType rowType, + Types.StructType rowStruct, + Types.NestedField projectField) { + for (int i = 0; i < rowStruct.fields().size(); i++) { + Types.NestedField rowField = rowStruct.fields().get(i); + if (rowField.fieldId() == projectField.fieldId()) { + Preconditions.checkArgument(rowField.type().typeId() == projectField.type().typeId(), + String.format("Different iceberg type between row field <%s> and project field <%s>", + rowField, projectField)); + + switch (projectField.type().typeId()) { + case STRUCT: + RowType nestedRowType = (RowType) rowType.getTypeAt(i); + int rowPos = i; + return row -> { + RowData nestedRow = row.isNullAt(rowPos) ? null : row.getRow(rowPos, nestedRowType.getFieldCount()); + return new RowDataProjection(nestedRowType, rowField.type().asStructType(), + projectField.type().asStructType()).wrap(nestedRow); + }; + + case MAP: + Types.MapType projectedMap = projectField.type().asMapType(); + Types.MapType originalMap = rowField.type().asMapType(); + + boolean keyProjectable = !projectedMap.keyType().isNestedType() || + projectedMap.keyType().equals(originalMap.keyType()); + boolean valueProjectable = !projectedMap.valueType().isNestedType() || + projectedMap.valueType().equals(originalMap.valueType()); + Preconditions.checkArgument(keyProjectable && valueProjectable, + "Cannot project a partial map key or value RowData. Trying to project %s out of %s", + projectField, rowField); + + return RowData.createFieldGetter(rowType.getTypeAt(i), i); + + case LIST: + Types.ListType projectedList = projectField.type().asListType(); + Types.ListType originalList = rowField.type().asListType(); + + boolean elementProjectable = !projectedList.elementType().isNestedType() || + projectedList.elementType().equals(originalList.elementType()); + Preconditions.checkArgument(elementProjectable, + "Cannot project a partial list element RowData. Trying to project %s out of %s", + projectField, rowField); + + return RowData.createFieldGetter(rowType.getTypeAt(i), i); + + default: + return RowData.createFieldGetter(rowType.getTypeAt(i), i); + } + } + } + throw new IllegalArgumentException(String.format("Cannot find field %s in %s", projectField, rowStruct)); Review comment: Nit: I think we need a more clear message for this exception: `Cannot locate the project field <%s> in the iceberg struct <%s>` ########## File path: flink/src/main/java/org/apache/iceberg/flink/data/RowDataProjection.java ########## @@ -0,0 +1,219 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iceberg.flink.data; + +import org.apache.flink.table.data.ArrayData; +import org.apache.flink.table.data.DecimalData; +import org.apache.flink.table.data.MapData; +import org.apache.flink.table.data.RawValueData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.data.StringData; +import org.apache.flink.table.data.TimestampData; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.types.RowKind; +import org.apache.iceberg.Schema; +import org.apache.iceberg.StructLike; +import org.apache.iceberg.flink.FlinkSchemaUtil; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; +import org.apache.iceberg.types.Types; + +public class RowDataProjection implements RowData { + /** + * Creates a projecting wrapper for {@link RowData} rows. + * <p> + * This projection does not work with repeated types like lists and maps. + * + * @param schema schema of rows wrapped by this projection + * @param projectedSchema result schema of the projected rows + * @return a wrapper to project rows + */ + public static RowDataProjection create(Schema schema, Schema projectedSchema) { + return new RowDataProjection(FlinkSchemaUtil.convert(schema), schema.asStruct(), projectedSchema.asStruct()); + } + + private final RowData.FieldGetter[] getters; + private RowData rowData; + + private RowDataProjection(RowType rowType, Types.StructType rowStruct, Types.StructType projectType) { + this.getters = new RowData.FieldGetter[projectType.fields().size()]; + for (int i = 0; i < getters.length; i++) { + getters[i] = createFieldGetter(rowType, rowStruct, projectType.fields().get(i)); + } + } + + private static RowData.FieldGetter createFieldGetter(RowType rowType, + Types.StructType rowStruct, + Types.NestedField projectField) { + for (int i = 0; i < rowStruct.fields().size(); i++) { + Types.NestedField rowField = rowStruct.fields().get(i); + if (rowField.fieldId() == projectField.fieldId()) { + Preconditions.checkArgument(rowField.type().typeId() == projectField.type().typeId(), Review comment: Nit: this can be simplified by the following lines as the `Preconditions.checkArgument` can format the error message directly. ```java Preconditions.checkArgument(rowField.type().typeId() == projectField.type().typeId(), "Different iceberg type between row field <%s> and project field <%s>", rowField, projectField); ``` -- 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]
