This is an automated email from the ASF dual-hosted git repository.
luoyuxia pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/fluss.git
The following commit(s) were added to refs/heads/main by this push:
new 549a739fa [lake/iceberg] Fix predicate push-down for TINYINT /
SMALLINT into Iceberg (#3565)
549a739fa is described below
commit 549a739fa828ac418995488869244d22b3c48699
Author: GuoYu <[email protected]>
AuthorDate: Fri Jul 3 14:14:37 2026 +0800
[lake/iceberg] Fix predicate push-down for TINYINT / SMALLINT into Iceberg
(#3565)
---
.../utils/FlussToIcebergPredicateConverter.java | 27 ++++++-----
.../lake/iceberg/utils/IcebergConversions.java | 55 ++--------------------
.../lake/iceberg/source/IcebergLakeSourceTest.java | 54 ++++++++++++++++++++-
3 files changed, 71 insertions(+), 65 deletions(-)
diff --git
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/FlussToIcebergPredicateConverter.java
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/FlussToIcebergPredicateConverter.java
index 77d50bb96..634ee0f40 100644
---
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/FlussToIcebergPredicateConverter.java
+++
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/FlussToIcebergPredicateConverter.java
@@ -107,7 +107,7 @@ public class FlussToIcebergPredicateConverter implements
PredicateVisitor<Expres
public Expression visitStartsWith(FieldRef fieldRef, Object literal) {
String fieldName = getField(fieldRef.index()).name();
return Expressions.startsWith(
- fieldName, convertToIcebergLiteral(fieldRef.index(),
literal).toString());
+ fieldName, convertToIcebergLiteral(fieldRef,
literal).toString());
}
@Override
@@ -125,41 +125,39 @@ public class FlussToIcebergPredicateConverter implements
PredicateVisitor<Expres
@Override
public Expression visitLessThan(FieldRef fieldRef, Object literal) {
String fieldName = getField(fieldRef.index()).name();
- return Expressions.lessThan(
- fieldName, convertToIcebergLiteral(fieldRef.index(),
literal));
+ return Expressions.lessThan(fieldName,
convertToIcebergLiteral(fieldRef, literal));
}
@Override
public Expression visitGreaterOrEqual(FieldRef fieldRef, Object
literal) {
String fieldName = getField(fieldRef.index()).name();
return Expressions.greaterThanOrEqual(
- fieldName, convertToIcebergLiteral(fieldRef.index(),
literal));
+ fieldName, convertToIcebergLiteral(fieldRef, literal));
}
@Override
public Expression visitNotEqual(FieldRef fieldRef, Object literal) {
String fieldName = getField(fieldRef.index()).name();
- return Expressions.notEqual(
- fieldName, convertToIcebergLiteral(fieldRef.index(),
literal));
+ return Expressions.notEqual(fieldName,
convertToIcebergLiteral(fieldRef, literal));
}
@Override
public Expression visitLessOrEqual(FieldRef fieldRef, Object literal) {
String fieldName = getField(fieldRef.index()).name();
return Expressions.lessThanOrEqual(
- fieldName, convertToIcebergLiteral(fieldRef.index(),
literal));
+ fieldName, convertToIcebergLiteral(fieldRef, literal));
}
@Override
public Expression visitEqual(FieldRef fieldRef, Object literal) {
String fieldName = getField(fieldRef.index()).name();
- return Expressions.equal(fieldName,
convertToIcebergLiteral(fieldRef.index(), literal));
+ return Expressions.equal(fieldName,
convertToIcebergLiteral(fieldRef, literal));
}
@Override
public Expression visitGreaterThan(FieldRef fieldRef, Object literal) {
String fieldName = getField(fieldRef.index()).name();
- Object icebergLiteral = convertToIcebergLiteral(fieldRef.index(),
literal);
+ Object icebergLiteral = convertToIcebergLiteral(fieldRef, literal);
return Expressions.greaterThan(fieldName, icebergLiteral);
}
@@ -168,7 +166,7 @@ public class FlussToIcebergPredicateConverter implements
PredicateVisitor<Expres
String fieldName = getField(fieldRef.index()).name();
List<Object> icebergLiterals =
literals.stream()
- .map(literal ->
convertToIcebergLiteral(fieldRef.index(), literal))
+ .map(literal -> convertToIcebergLiteral(fieldRef,
literal))
.collect(Collectors.toList());
return Expressions.in(fieldName, icebergLiterals);
}
@@ -178,7 +176,7 @@ public class FlussToIcebergPredicateConverter implements
PredicateVisitor<Expres
String fieldName = getField(fieldRef.index()).name();
List<Object> icebergLiterals =
literals.stream()
- .map(literal ->
convertToIcebergLiteral(fieldRef.index(), literal))
+ .map(literal -> convertToIcebergLiteral(fieldRef,
literal))
.collect(Collectors.toList());
return Expressions.notIn(fieldName, icebergLiterals);
}
@@ -199,8 +197,11 @@ public class FlussToIcebergPredicateConverter implements
PredicateVisitor<Expres
return icebergSchema.columns().get(fieldIndex);
}
- private Object convertToIcebergLiteral(int fieldIndex, Object
flussLiteral) {
- return toIcebergLiteral(getField(fieldIndex), flussLiteral);
+ private Object convertToIcebergLiteral(FieldRef fieldRef, Object
flussLiteral) {
+ // Pass the original Fluss field type through so
IcebergConversions can reuse
+ // FlussRowAsIcebergRecord's per-Fluss-type converter directly,
avoiding the lossy
+ // reverse mapping from Iceberg types back to Fluss types.
+ return toIcebergLiteral(getField(fieldRef.index()),
fieldRef.type(), flussLiteral);
}
}
}
diff --git
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java
index b75ce8ad4..111df385f 100644
---
a/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java
+++
b/fluss-lake/fluss-lake-iceberg/src/main/java/org/apache/fluss/lake/iceberg/utils/IcebergConversions.java
@@ -21,9 +21,7 @@ import
org.apache.fluss.lake.iceberg.source.FlussRowAsIcebergRecord;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.row.GenericRow;
import org.apache.fluss.row.InternalRow;
-import org.apache.fluss.types.DataField;
import org.apache.fluss.types.DataType;
-import org.apache.fluss.types.DataTypes;
import org.apache.fluss.types.RowType;
import org.apache.iceberg.PartitionField;
@@ -34,12 +32,10 @@ import org.apache.iceberg.Table;
import org.apache.iceberg.catalog.TableIdentifier;
import org.apache.iceberg.expressions.Expression;
import org.apache.iceberg.expressions.Expressions;
-import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;
import javax.annotation.Nullable;
-import java.util.ArrayList;
import java.util.List;
import static
org.apache.fluss.metadata.ResolvedPartitionSpec.PARTITION_SPEC_SEPARATOR;
@@ -93,55 +89,12 @@ public class IcebergConversions {
return expression;
}
- public static Object toIcebergLiteral(Types.NestedField field, Object
flussLiteral) {
+ public static Object toIcebergLiteral(
+ Types.NestedField icebergField, DataType flussFieldType, Object
flussLiteral) {
InternalRow flussRow = GenericRow.of(flussLiteral);
FlussRowAsIcebergRecord flussRowAsIcebergRecord =
new FlussRowAsIcebergRecord(
- Types.StructType.of(field),
-
RowType.of(convertIcebergTypeToFlussType(field.type())),
- flussRow);
- return flussRowAsIcebergRecord.get(0,
field.type().typeId().javaClass());
- }
-
- /** Converts Iceberg data types to Fluss data types. */
- private static DataType convertIcebergTypeToFlussType(Type icebergType) {
- if (icebergType instanceof Types.BooleanType) {
- return DataTypes.BOOLEAN();
- } else if (icebergType instanceof Types.IntegerType) {
- return DataTypes.INT();
- } else if (icebergType instanceof Types.LongType) {
- return DataTypes.BIGINT();
- } else if (icebergType instanceof Types.DoubleType) {
- return DataTypes.DOUBLE();
- } else if (icebergType instanceof Types.TimeType) {
- return DataTypes.TIME();
- } else if (icebergType instanceof Types.TimestampType) {
- Types.TimestampType timestampType = (Types.TimestampType)
icebergType;
- if (timestampType.shouldAdjustToUTC()) {
- return DataTypes.TIMESTAMP_LTZ();
- } else {
- return DataTypes.TIMESTAMP();
- }
- } else if (icebergType instanceof Types.StringType) {
- return DataTypes.STRING();
- } else if (icebergType instanceof Types.DecimalType) {
- Types.DecimalType decimalType = (Types.DecimalType) icebergType;
- return DataTypes.DECIMAL(decimalType.precision(),
decimalType.scale());
- } else if (icebergType instanceof Types.ListType) {
- Types.ListType listType = (Types.ListType) icebergType;
- return
DataTypes.ARRAY(convertIcebergTypeToFlussType(listType.elementType()));
- } else if (icebergType.isStructType()) {
- Types.StructType structType = icebergType.asStructType();
- List<DataField> fields = new ArrayList<>();
- for (Types.NestedField nestedField : structType.fields()) {
- DataType fieldType =
convertIcebergTypeToFlussType(nestedField.type());
- fields.add(new DataField(nestedField.name(), fieldType));
- }
- return DataTypes.ROW(fields.toArray(new DataField[0]));
- }
-
- throw new UnsupportedOperationException(
- "Unsupported data type conversion for Iceberg type: "
- + icebergType.getClass().getName());
+ Types.StructType.of(icebergField),
RowType.of(flussFieldType), flussRow);
+ return flussRowAsIcebergRecord.get(0,
icebergField.type().typeId().javaClass());
}
}
diff --git
a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergLakeSourceTest.java
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergLakeSourceTest.java
index d69def535..d136e910b 100644
---
a/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergLakeSourceTest.java
+++
b/fluss-lake/fluss-lake-iceberg/src/test/java/org/apache/fluss/lake/iceberg/source/IcebergLakeSourceTest.java
@@ -66,7 +66,10 @@ class IcebergLakeSourceTest extends IcebergSourceTestBase {
PartitionSpec.builderFor(SCHEMA).bucket("id",
DEFAULT_BUCKET_NUM).build();
private static final PredicateBuilder FLUSS_BUILDER =
- new PredicateBuilder(RowType.of(DataTypes.BIGINT(),
DataTypes.STRING()));
+ new PredicateBuilder(RowType.of(DataTypes.INT(),
DataTypes.STRING()));
+
+ private static final PredicateBuilder FLUSS_BUILDER_TINYINT =
+ new PredicateBuilder(RowType.of(DataTypes.TINYINT(),
DataTypes.STRING()));
@BeforeAll
protected static void beforeAll() {
@@ -157,4 +160,53 @@ class IcebergLakeSourceTest extends IcebergSourceTestBase {
assertThat(filterPushDownResult.remainingPredicates().toString())
.isEqualTo(allFilters.toString());
}
+
+ @Test
+ void testWithTinyIntFilter() throws Exception {
+ TablePath tablePath = TablePath.of("fluss", "test_filters_tinyint");
+ createTable(tablePath, SCHEMA, PARTITION_SPEC);
+
+ Table table = getTable(tablePath);
+ List<Record> rows = new ArrayList<>();
+ for (int i = 1; i <= 4; i++) {
+ rows.add(
+ createIcebergRecord(
+ SCHEMA,
+ i,
+ "name" + i,
+ 0,
+ (long) i,
+ OffsetDateTime.now(ZoneOffset.UTC)));
+ }
+ writeRecord(table, rows, null, 0);
+ table.refresh();
+
+ // pass Byte literal (not Integer) to exercise the Fluss TINYINT ->
Iceberg int path
+ Predicate filter1 = FLUSS_BUILDER_TINYINT.greaterOrEqual(0, (byte) 2);
+ Predicate filter2 = FLUSS_BUILDER_TINYINT.lessOrEqual(0, (byte) 3);
+ List<Predicate> allFilters = Arrays.asList(filter1, filter2);
+
+ LakeSource<IcebergSplit> lakeSource =
lakeStorage.createLakeSource(tablePath);
+ LakeSource.FilterPushDownResult filterPushDownResult =
lakeSource.withFilters(allFilters);
+
assertThat(filterPushDownResult.acceptedPredicates()).isEqualTo(allFilters);
+ assertThat(filterPushDownResult.remainingPredicates()).isEmpty();
+
+ List<IcebergSplit> icebergSplits =
+ lakeSource.createPlanner(() ->
table.currentSnapshot().snapshotId()).plan();
+ assertThat(icebergSplits).hasSize(1);
+ IcebergSplit icebergSplit = icebergSplits.get(0);
+
+ List<Row> actual = new ArrayList<>();
+ org.apache.fluss.row.InternalRow.FieldGetter[] fieldGetters =
+ org.apache.fluss.row.InternalRow.createFieldGetters(
+ RowType.of(new IntType(), new StringType()));
+ RecordReader recordReader = lakeSource.createRecordReader(() ->
icebergSplit);
+ try (CloseableIterator<LogRecord> iterator = recordReader.read()) {
+ actual.addAll(
+ convertToFlinkRow(
+ fieldGetters,
+ TransformingCloseableIterator.transform(iterator,
LogRecord::getRow)));
+ }
+ assertThat(actual.toString()).isEqualTo("[+I[2, name2], +I[3,
name3]]");
+ }
}
