stevenzwu commented on code in PR #7109:
URL: https://github.com/apache/iceberg/pull/7109#discussion_r1162268348
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkInputFormat.java:
##########
@@ -140,6 +148,90 @@ public void testBasicProjection() throws IOException {
TestHelpers.assertRows(result, expected);
}
+ @Test
+ public void testBasicFormatFiltering() throws IOException {
+ Schema writeSchema = new Schema(Types.NestedField.required(0, "id",
Types.LongType.get()));
+
+ Table sourceTable =
+ catalogResource.catalog().createTable(TableIdentifier.of("default",
"t"), writeSchema);
+
+ List<Record> writeRecords = RandomGenericData.generate(writeSchema, 22,
0L);
+ new GenericAppenderHelper(sourceTable, fileFormat, TEMPORARY_FOLDER)
+ .appendToTable(writeRecords);
+
+ List<Row> result =
+ runFormat(
+ FlinkSource.forRowData()
+ .tableLoader(tableLoader())
+
.filters(Collections.singletonList(Expressions.greaterThanOrEqual("id", 0)))
+ .buildFormat());
+
+ List<Row> expected =
Review Comment:
it is not intuitive to interpret expected records because we are using
`RandomGenericData` above. I can just deterministic generate the
`GenericRecord` in a for loop and set the id value to the index number. Then it
is easier to understand the expected value.
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java:
##########
@@ -120,6 +130,20 @@ private CloseableIterable<RowData> newIterable(
}
}
+ if (this.filters != null && !this.filters.isEmpty()) {
+ Expression combinedExpression = null;
Review Comment:
nit: should we move the `FlinkSourceFilter` construction to the constructor.
If it is not null, we apply the additional filter on the iterable.
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/reader/AvroGenericRecordReaderFunction.java:
##########
@@ -72,7 +73,8 @@ public AvroGenericRecordReaderFunction(
this.io = io;
this.encryption = encryption;
this.rowDataReader =
- new RowDataFileScanTaskReader(tableSchema, readSchema, nameMapping,
caseSensitive);
+ new RowDataFileScanTaskReader(
+ tableSchema, readSchema, nameMapping, caseSensitive,
Collections.emptyList());
Review Comment:
this constructor needs to take the filters and pass the arg here.
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestStreamingReaderOperator.java:
##########
@@ -109,6 +111,49 @@ public void testProcessAllRecords() throws Exception {
}
}
+ @Test
+ public void testProcessAllRecordsFilter() throws Exception {
Review Comment:
this test seems redundant to what `TestFlinkInputFormat` already has
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/FlinkSourceFilter.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * 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;
+
+import org.apache.flink.api.common.functions.FilterFunction;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.types.logical.RowType;
+import org.apache.iceberg.Schema;
+import org.apache.iceberg.expressions.Evaluator;
+import org.apache.iceberg.expressions.Expression;
+import org.apache.iceberg.types.Types;
+
+public class FlinkSourceFilter implements FilterFunction<RowData> {
+
+ private final RowType rowType;
+ private final Evaluator evaluator;
+ private final Types.StructType struct;
+ private volatile RowDataWrapper wrapper;
+
+ public FlinkSourceFilter(Schema schema, Expression expr, boolean
caseSensitive) {
+ rowType = FlinkSchemaUtil.convert(schema);
+ struct = schema.asStruct();
+ this.evaluator = new Evaluator(struct, expr, caseSensitive);
+ }
+
+ @Override
+ public boolean filter(RowData value) {
+ if (this.wrapper == null) {
Review Comment:
nit: I thought the style is only to use `this.` for setting a new value
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkInputFormat.java:
##########
@@ -140,6 +148,90 @@ public void testBasicProjection() throws IOException {
TestHelpers.assertRows(result, expected);
}
+ @Test
+ public void testBasicFormatFiltering() throws IOException {
+ Schema writeSchema = new Schema(Types.NestedField.required(0, "id",
Types.LongType.get()));
+
+ Table sourceTable =
+ catalogResource.catalog().createTable(TableIdentifier.of("default",
"t"), writeSchema);
+
+ List<Record> writeRecords = RandomGenericData.generate(writeSchema, 22,
0L);
+ new GenericAppenderHelper(sourceTable, fileFormat, TEMPORARY_FOLDER)
+ .appendToTable(writeRecords);
+
+ List<Row> result =
+ runFormat(
+ FlinkSource.forRowData()
+ .tableLoader(tableLoader())
+
.filters(Collections.singletonList(Expressions.greaterThanOrEqual("id", 0)))
+ .buildFormat());
+
+ List<Row> expected =
+ Lists.newArrayList(
+ Row.of(170168523965373507L),
+ Row.of(236240913033168047L),
+ Row.of(6146794652083548235L),
+ Row.of(5072005423257391728L),
+ Row.of(171134583860878546L),
+ Row.of(8730854458729406051L),
+ Row.of(6688467811848818630L),
+ Row.of(428667830982598836L),
+ Row.of(9223372036854775807L),
+ Row.of(4922475540349336432L),
+ Row.of(7138230367502298321L),
+ Row.of(6787954838522539928L),
+ Row.of(9223372036854775807L),
+ Row.of(2440897930508784356L),
+ Row.of(282712201594543727L));
+
+ TestHelpers.assertRows(result, expected);
+ }
+
+ @Test
+ public void testBasicRowDataFiltering() throws Exception {
Review Comment:
this method won't be necessary if we add the new test method above to
`TestFlinkScan`. Currently, `TestFlinkScan#testFilterExp` only covers filter
with partition column. we can add the above test method as `testResidualFilter`
where filter is constructed with non-partition column.
`TestFlinkScan` is the base class for both the old `FlinkSource`(covered by
`TestFlinkInputFormat` and others) and the new FLIP-27 `IcebergSource` (covered
by `TestIcebergSourceBounded` and others)
--
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]
