Guosmilesmile commented on code in PR #17390: URL: https://github.com/apache/iceberg/pull/17390#discussion_r3705236223
########## flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/TestProjector.java: ########## @@ -0,0 +1,138 @@ +/* + * 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.source; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.ByteArrayInputStream; +import java.io.ByteArrayOutputStream; +import java.io.ObjectInputStream; +import java.io.ObjectOutputStream; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; +import org.apache.flink.table.connector.Projection; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.IntType; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.NullType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.flink.table.types.logical.VarCharType; +import org.junit.jupiter.api.Test; + +/** + * Unit tests for {@link Projector} to cover behaviour that can't be tested in {@code + * TestFlinkTableSource}. + */ +public class TestProjector { + + // id INT, person ROW<name STRING, age INT>, address ROW<city STRING, geo ROW<lat INT, lng INT>> + private static final RowType PERSON = + row(new String[] {"name", "age"}, VarCharType.STRING_TYPE, new IntType()); + private static final RowType GEO = row(new String[] {"lat", "lng"}, new IntType(), new IntType()); + private static final RowType ADDRESS = + row(new String[] {"city", "geo"}, VarCharType.STRING_TYPE, GEO); + private static final RowType ORIGINAL = + row(new String[] {"id", "person", "address"}, new IntType(), PERSON, ADDRESS); + + private static RowType row(String[] names, LogicalType... types) { + return RowType.of(types, names); + } + + private static Projector projector(int[][] projectedFields) { + RowType producedRowType = (RowType) Projection.of(projectedFields).project(ORIGINAL); + return Projector.of(ORIGINAL, projectedFields, producedRowType); + } + + /** A source stream shaped like the reader output for the given projection (never executed). */ + private static DataStream<RowData> readerStream(int[][] projectedFields) { + RowType readSchema = projector(projectedFields).readSchema(); + return StreamExecutionEnvironment.getExecutionEnvironment() + .fromData(InternalTypeInfo.of(readSchema), new GenericRowData(readSchema.getFieldCount())); + } + + @SuppressWarnings("unchecked") + private static <T> T roundTripSerialize(T instance) throws Exception { + ByteArrayOutputStream bytes = new ByteArrayOutputStream(); + try (ObjectOutputStream out = new ObjectOutputStream(bytes)) { + out.writeObject(instance); + } + try (ObjectInputStream in = + new ObjectInputStream(new ByteArrayInputStream(bytes.toByteArray()))) { + return (T) in.readObject(); + } + } + + @Test + public void addsNoMapForInOrderTopLevelProjection() { + // A non-nested, in-order projection already matches the produced shape, so the reader stream is + // returned unchanged (no map operator). + for (int[][] inOrder : + new int[][][] { + {{0}}, // select only first column + {{1}}, // select only second column + {{2}}, // select only third column + {{0}, {1}}, // select first two columns (in order) + {{1}, {2}}, // select last two columns (in order) + {{0}, {1}, {2}} // select all three columns (in order) + }) { + DataStream<RowData> source = readerStream(inOrder); + assertThat(projector(inOrder).project(source)).isSameAs(source); + } + } + + @Test + public void buildsFieldGetterForUnknownTypeLeaf() { Review Comment: Should we add a test for buildsFieldGetterForTopLevelUnknownType？ ########## flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/Projector.java: ########## @@ -0,0 +1,206 @@ +/* + * 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.source; + +import java.io.Serializable; +import java.util.Arrays; +import java.util.List; +import java.util.TreeMap; +import java.util.stream.Collectors; +import org.apache.flink.api.common.functions.MapFunction; +import org.apache.flink.streaming.api.datastream.DataStream; +import org.apache.flink.table.data.GenericRowData; +import org.apache.flink.table.data.RowData; +import org.apache.flink.table.runtime.typeutils.InternalTypeInfo; +import org.apache.flink.table.types.logical.LogicalType; +import org.apache.flink.table.types.logical.RowType; +import org.apache.iceberg.flink.FlinkRowData; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Applies a Flink (possibly nested) projection pushed down into the Iceberg source. + * + * <p>A single projection has two sides, both derived here from the source {@link RowType} and + * Flink's {@code int[][]} projection paths: + * + * <ul> + * <li>{@link #readSchema()} — the schema handed to the reader: pruned to the projected fields, in + * table-schema order with nested structs left intact. + * <li>{@link #project(DataStream)} — projects the reader's rows into the produced (SELECT-list) + * order, extracting nested leaves into top-level columns, by adding a map step to the stream. + * The map is added only when the projection is nested or reorders top-level fields; an + * in-order projection is returned unchanged. + * </ul> + * + * <p>Projection paths descend only through structs ({@link RowType}); they never descend through + * the element of a list or the key/value of a map. Flink guarantees this: a subfield reference into + * a list/map element (for example {@code SELECT people[1].name}) reads the whole element struct + * rather than pushing a path into it, so a repeated type is always projected in full. + */ +final class Projector implements Serializable { + + private final RowType readSchema; + private final boolean projectionNeeded; + private final RowType producedRowType; + private final RowProjection rowProjection; + + static Projector of(RowType sourceRowType, int[][] projectedFields, RowType producedRowType) { + return new Projector(sourceRowType, projectedFields, producedRowType); + } + + private Projector(RowType sourceRowType, int[][] projectedFields, RowType producedRowType) { + this.readSchema = prune(sourceRowType, projectedFields); + this.projectionNeeded = isProjectionNeeded(projectedFields); + this.producedRowType = producedRowType; + + RowData.FieldGetter[] getters = new RowData.FieldGetter[projectedFields.length]; + for (int col = 0; col < projectedFields.length; col++) { + getters[col] = getter(sourceRowType, readSchema, projectedFields[col]); + } + this.rowProjection = new RowProjection(getters); + } + + /** + * The schema handed to the reader: pruned to the projected fields, preserving original field + * names, nesting, and table-schema order. + */ + RowType readSchema() { + return readSchema; + } + + /** + * Adds a map step projecting the reader's rows into the produced (SELECT-list) shape, or returns + * the stream unchanged when no projection is needed (a non-nested, in-order projection). + */ + DataStream<RowData> project(DataStream<RowData> stream) { + if (!projectionNeeded) { + return stream; + } + + return stream + .map(rowProjection) + .setParallelism(stream.getParallelism()) + .returns(InternalTypeInfo.of(producedRowType)); + } + + private static boolean isProjectionNeeded(int[][] projectedFields) { + int previousFieldIndex = -1; + for (int[] path : projectedFields) { + if (path.length > 1 || path[0] <= previousFieldIndex) { + return true; + } + previousFieldIndex = path[0]; + } + + return false; + } Review Comment: If the query only accesses base columns, but the projection order is different from the table schema, we'll end up introducing an extra map operation. This adds some overhead and changes the current behavior. Is it worth introducing this extra layer? ########## flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkTableSource.java: ########## @@ -558,4 +561,167 @@ public void testFilterPushDown2Literal() { public void testSqlParseNaN() { // todo add some test case to test NaN } + + private void createNestedTable() { + createNestedTable("(1, ROW('a', 10), ROW('nyc', ROW(40, -74)), ARRAY[ROW('a', 10)])"); + } + + private void createNestedTable(String values) { + sql( + "CREATE TABLE %s (" + + "id INT, " + + "person ROW<name STRING, age INT>, " + + "address ROW<city STRING, geo ROW<lat INT, lng INT>>, " + + "people ARRAY<ROW<name STRING, age INT>>" + + ") WITH ('write.format.default'='%s')", + NESTED_TABLE_NAME, format.name()); + + sql("INSERT INTO %s VALUES %s", NESTED_TABLE_NAME, values); + } + + private void createMapTable() { + sql( + "CREATE TABLE %s (" + + "id INT, " + + "attrs MAP<STRING, ROW<code STRING, label STRING>>" + + ") WITH ('write.format.default'='%s')", + MAP_TABLE_NAME, format.name()); + + sql("INSERT INTO %s VALUES (1, MAP['x', ROW('c1', 'l1')])", MAP_TABLE_NAME); + } + + private void assertProjection(String expected) { + assertThat(scanEventCount).isEqualTo(1); + assertThat(lastScanEvent.projection().asStruct()).hasToString(expected); + } + + @TestTemplate + public void testNestedFieldProjectionPushedDown() { Review Comment: Could we add a couple of tests for the following cases? Selecting both the whole struct and a nested field, for example: * SELECT person, person.name * SELECT person.name, person Duplicate projection paths, for example: * SELECT person.name, person.name ########## flink/v2.1/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkTableSource.java: ########## @@ -558,4 +561,167 @@ public void testFilterPushDown2Literal() { public void testSqlParseNaN() { // todo add some test case to test NaN } + + private void createNestedTable() { Review Comment: These private methods should be moved to the end of the class. -- 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]
