stevenzwu commented on code in PR #7109:
URL: https://github.com/apache/iceberg/pull/7109#discussion_r1164477096
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkScan.java:
##########
@@ -433,6 +433,24 @@ public void testFilterExp() throws Exception {
TestFixtures.SCHEMA);
}
+ @Test
+ public void testFilterExp() throws Exception {
+ Table table =
+ catalogResource.catalog().createTable(TestFixtures.TABLE_IDENTIFIER,
TestFixtures.SCHEMA);
+
+ List<Record> expectedRecords =
RandomGenericData.generate(TestFixtures.SCHEMA, 3, 0L);
+ expectedRecords.get(0).set(2, "a");
+ expectedRecords.get(1).set(2, "b");
+ expectedRecords.get(2).set(2, "c");
+
+ GenericAppenderHelper helper = new GenericAppenderHelper(table,
fileFormat, TEMPORARY_FOLDER);
+ DataFile dataFile = helper.writeFile(expectedRecords);
+ helper.appendToTable(dataFile);
+ List<Row> actual =
+ runWithFilter(Expressions.greaterThanOrEqual("data", "b"), "where
data>='b'");
+ TestHelpers.assertRecords(actual, expectedRecords, TestFixtures.SCHEMA);
Review Comment:
shouldn't the filter match only 2 rows (not all 3 rows)?
##########
flink/v1.17/flink/src/test/java/org/apache/iceberg/flink/source/TestFlinkInputFormat.java:
##########
@@ -140,6 +144,43 @@ public void testBasicProjection() throws IOException {
TestHelpers.assertRows(result, expected);
}
+ public static Record createRecord(Schema writeSchema, long val) {
+ Record record = GenericRecord.create(writeSchema);
+ record.set(0, val);
+ return record;
+ }
+
+ @Test
+ public void testBasicFormatFiltering() throws IOException {
Review Comment:
this test is also not necessary, right? `TestFlinkInputFormat` extends from
`TestFlinkScan` too
##########
flink/v1.17/flink/src/main/java/org/apache/iceberg/flink/source/RowDataFileScanTaskReader.java:
##########
@@ -54,13 +58,27 @@ public class RowDataFileScanTaskReader implements
FileScanTaskReader<RowData> {
private final Schema projectedSchema;
private final String nameMapping;
private final boolean caseSensitive;
+ private final FlinkSourceFilter rowFilter;
public RowDataFileScanTaskReader(
- Schema tableSchema, Schema projectedSchema, String nameMapping, boolean
caseSensitive) {
+ Schema tableSchema,
+ Schema projectedSchema,
+ String nameMapping,
+ boolean caseSensitive,
+ List<Expression> filters) {
this.tableSchema = tableSchema;
this.projectedSchema = projectedSchema;
this.nameMapping = nameMapping;
this.caseSensitive = caseSensitive;
+
+ if (filters != null && !filters.isEmpty()) {
+ Expression combinedExpression =
+ filters.stream().reduce(Expressions.alwaysTrue(), Expressions::and);
Review Comment:
FlinkSink should have taken a single expression (not a list) for filter.
Created an issue to track: https://github.com/apache/iceberg/issues/7335
--
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-unsubscr...@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@iceberg.apache.org
For additional commands, e-mail: issues-h...@iceberg.apache.org
