You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "fqaiser94 (via GitHub)" <gi...@apache.org> on 2026/07/27 21:38:44 UTC
[PR] Flink: Nested projection pushdow [iceberg]
fqaiser94 opened a new pull request, #17390:
URL: https://github.com/apache/iceberg/pull/17390
Enables **_nested_** projection pushdown in the Flink v2.1 Iceberg source. Previously `supportsNestedProjection()` returned `false` so a query like `SELECT user.name FROM t` read the entire `user` struct. This PR pushes the nested field paths down to the reader so only the selected leaf columns are scanned.
--
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: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Flink: Nested projection pushdown [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #17390: Flink: Nested projection pushdown
URL: https://github.com/apache/iceberg/pull/17390
--
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: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Flink: Nested projection pushdown [iceberg]
Posted by "Guosmilesmile (via GitHub)" <gi...@apache.org>.
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: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Flink: Nested projection pushdown [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17390:
URL: https://github.com/apache/iceberg/pull/17390#issuecomment-5627465960
This pull request has been closed due to lack of activity. This is not a judgement on the merit of the PR in any way. It is just a way of keeping the PR queue manageable. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.
--
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: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Flink: Nested projection pushdown [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #17390:
URL: https://github.com/apache/iceberg/pull/17390#issuecomment-5518438552
This pull request has been marked as stale due to 30 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@iceberg.apache.org list. Thank you for your contributions.
--
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: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Flink: Nested projection pushdown [iceberg]
Posted by "uros-b (via GitHub)" <gi...@apache.org>.
uros-b commented on code in PR #17390:
URL: https://github.com/apache/iceberg/pull/17390#discussion_r3683871562
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
Review Comment:
How about fixing v1.20/v2.0 as well?
--
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: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Flink: Nested projection pushdown [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #17390:
URL: https://github.com/apache/iceberg/pull/17390#discussion_r3694643811
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
Review Comment:
+1 for fixing 2.1 first.
--
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: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Flink: Nested projection pushdown [iceberg]
Posted by "fqaiser94 (via GitHub)" <gi...@apache.org>.
fqaiser94 commented on code in PR #17390:
URL: https://github.com/apache/iceberg/pull/17390#discussion_r3685159050
##########
flink/v2.1/flink/src/main/java/org/apache/iceberg/flink/source/IcebergTableSource.java:
##########
Review Comment:
I'd prefer to keep this PR focused on v2.1 to keep the diff reviewable.
Happy to consider opening follow-up PRs for v1.20 and v2.0 once this one is approved and merged. Should be straight-forward in theory.
--
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: issues-unsubscribe@iceberg.apache.org
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
Re: [PR] Flink: Nested projection pushdown [iceberg]
Posted by "mxm (via GitHub)" <gi...@apache.org>.
mxm commented on code in PR #17390:
URL: https://github.com/apache/iceberg/pull/17390#discussion_r3702634449
##########
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;
+ }
+
+ private static RowType prune(RowType rowType, int[][] projectedFields) {
+ List<RowType.RowField> fields =
+ Arrays.stream(projectedFields)
+ .collect(
+ Collectors.groupingBy(
+ path -> path[0],
+ TreeMap::new,
+ Collectors.mapping(
+ path -> Arrays.copyOfRange(path, 1, path.length), Collectors.toList())))
+ .entrySet()
+ .stream()
+ .map(
+ entry -> {
+ int fieldIndex = entry.getKey();
+ int[][] nestedFieldPaths = entry.getValue().toArray(new int[0][]);
+ RowType.RowField field = rowType.getFields().get(fieldIndex);
+ return prune(field, nestedFieldPaths);
+ })
+ .collect(Collectors.toList());
+
+ return new RowType(rowType.isNullable(), fields);
+ }
+
+ private static RowType.RowField prune(RowType.RowField field, int[][] projectedFields) {
+ boolean selectedWholeField =
+ Arrays.stream(projectedFields).anyMatch(nestedFieldPath -> nestedFieldPath.length == 0);
+
+ final LogicalType type;
+ if (selectedWholeField) {
+ type = field.getType();
+ } else {
+ Preconditions.checkArgument(
+ field.getType() instanceof RowType,
+ "Cannot project subfields of non-struct field <%s> of type <%s>",
+ field.getName(),
+ field.getType());
+ type = prune((RowType) field.getType(), projectedFields);
+ }
+
+ return new RowType.RowField(field.getName(), type, field.getDescription().orElse(null));
+ }
+
+ private static RowData.FieldGetter getter(
+ RowType originalRowType, RowType prunedRowType, int[] originalFieldPath) {
+ int originalFieldIndex = originalFieldPath[0];
+ String fieldName = originalRowType.getFieldNames().get(originalFieldIndex);
+ int prunedFieldIndex = prunedRowType.getFieldNames().indexOf(fieldName);
+ LogicalType prunedFieldType = prunedRowType.getTypeAt(prunedFieldIndex);
+
+ if (originalFieldPath.length == 1) {
+ return FlinkRowData.createFieldGetter(prunedFieldType, prunedFieldIndex);
+ } else {
+ LogicalType originalChildType = originalRowType.getTypeAt(originalFieldIndex);
+ Preconditions.checkArgument(
+ originalChildType instanceof RowType,
+ "Cannot descend into non-struct field <%s> of type <%s> for nested projection",
+ fieldName,
+ originalChildType);
+
+ RowType originalChildRowType = (RowType) originalChildType;
+ RowType prunedChildRowType = (RowType) prunedFieldType;
+ int childArity = prunedChildRowType.getFieldCount();
+ RowData.FieldGetter childGetter =
+ getter(
+ originalChildRowType,
+ prunedChildRowType,
+ Arrays.copyOfRange(originalFieldPath, 1, originalFieldPath.length));
+ return row ->
+ row.isNullAt(prunedFieldIndex)
+ ? null
+ : childGetter.getFieldOrNull(row.getRow(prunedFieldIndex, childArity));
+ }
+ }
+
+ private static final class RowProjection implements MapFunction<RowData, RowData> {
+
+ private final RowData.FieldGetter[] getters;
+
+ RowProjection(RowData.FieldGetter[] getters) {
+ this.getters = getters;
+ }
+
+ @Override
+ public RowData map(RowData row) {
+ GenericRowData output = new GenericRowData(getters.length);
+ for (int col = 0; col < getters.length; col++) {
+ output.setField(col, getters[col].getFieldOrNull(row));
+ }
+
+ return output;
Review Comment:
This will make a copy of all rows read. Even worse, Flink also copies the RowData in between the reader and the map step.
Ideally, we would push the projection into the reader itself. Have you considered that?
##########
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())
Review Comment:
This will apply the parallelism on the map, not the source itself. Probably not what we want.
##########
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;
+ }
+
+ private static RowType prune(RowType rowType, int[][] projectedFields) {
+ List<RowType.RowField> fields =
+ Arrays.stream(projectedFields)
+ .collect(
+ Collectors.groupingBy(
+ path -> path[0],
+ TreeMap::new,
Review Comment:
Why are we using a TreeMap here?
--
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: issues-unsubscribe@iceberg.apache.org
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