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 70a71b99c [lake/hudi] Support predicate pushdown (#3501)
70a71b99c is described below
commit 70a71b99cbf66b4b4d843af80df09422ba22832f
Author: fhan <[email protected]>
AuthorDate: Sun Jun 21 08:43:51 2026 +0800
[lake/hudi] Support predicate pushdown (#3501)
---
.../fluss/lake/hudi/source/HudiLakeSource.java | 46 +++-
.../FlussToHudiExpressionPredicateConverter.java | 281 +++++++++++++++++++++
.../fluss/lake/hudi/source/HudiLakeSourceTest.java | 23 +-
...lussToHudiExpressionPredicateConverterTest.java | 163 ++++++++++++
4 files changed, 491 insertions(+), 22 deletions(-)
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
index 87f9ee0de..7b1636862 100644
---
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/source/HudiLakeSource.java
@@ -18,6 +18,8 @@
package org.apache.fluss.lake.hudi.source;
import org.apache.fluss.config.Configuration;
+import
org.apache.fluss.lake.hudi.utils.FlussToHudiExpressionPredicateConverter;
+import org.apache.fluss.lake.hudi.utils.HudiTableInfo;
import org.apache.fluss.lake.serializer.SimpleVersionedSerializer;
import org.apache.fluss.lake.source.LakeSource;
import org.apache.fluss.lake.source.Planner;
@@ -25,12 +27,17 @@ import org.apache.fluss.lake.source.RecordReader;
import org.apache.fluss.metadata.TablePath;
import org.apache.fluss.predicate.Predicate;
+import org.apache.hudi.org.apache.avro.Schema;
+import org.apache.hudi.source.ExpressionPredicates;
+import org.apache.hudi.util.StreamerUtil;
+
import javax.annotation.Nullable;
import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
+import java.util.Optional;
/** Hudi implementation of {@link LakeSource}. */
public class HudiLakeSource implements LakeSource<HudiSplit> {
@@ -40,6 +47,7 @@ public class HudiLakeSource implements LakeSource<HudiSplit> {
private final Configuration hudiConfig;
private final TablePath tablePath;
private @Nullable int[][] project;
+ private List<ExpressionPredicates.Predicate> predicates =
Collections.emptyList();
public HudiLakeSource(Configuration hudiConfig, TablePath tablePath) {
this.hudiConfig = hudiConfig;
@@ -58,7 +66,32 @@ public class HudiLakeSource implements LakeSource<HudiSplit>
{
@Override
public FilterPushDownResult withFilters(List<Predicate> predicates) {
- return FilterPushDownResult.of(Collections.emptyList(), new
ArrayList<>(predicates));
+ if (predicates.isEmpty()) {
+ this.predicates = Collections.emptyList();
+ return FilterPushDownResult.of(Collections.emptyList(),
Collections.emptyList());
+ }
+
+ List<Predicate> remainingPredicates = new ArrayList<>();
+ List<Predicate> acceptedPredicates = new ArrayList<>();
+ List<ExpressionPredicates.Predicate> convertedPredicates = new
ArrayList<>();
+
+ Schema hudiSchema = getHudiSchema();
+ for (Predicate predicate : predicates) {
+ Optional<ExpressionPredicates.Predicate> convertedPredicate =
+
FlussToHudiExpressionPredicateConverter.convert(hudiSchema, predicate);
+ if (convertedPredicate.isPresent()) {
+ acceptedPredicates.add(predicate);
+ convertedPredicates.add(convertedPredicate.get());
+ } else {
+ remainingPredicates.add(predicate);
+ }
+ }
+
+ this.predicates =
+ convertedPredicates.isEmpty()
+ ? Collections.emptyList()
+ : Collections.unmodifiableList(new
ArrayList<>(convertedPredicates));
+ return FilterPushDownResult.of(acceptedPredicates,
remainingPredicates);
}
@Override
@@ -69,7 +102,8 @@ public class HudiLakeSource implements LakeSource<HudiSplit>
{
@Override
public RecordReader createRecordReader(ReaderContext<HudiSplit> context)
throws IOException {
try {
- return new HudiRecordReader(hudiConfig, tablePath,
context.lakeSplit(), project);
+ return new HudiRecordReader(
+ hudiConfig, tablePath, context.lakeSplit(), project,
predicates);
} catch (Exception e) {
throw new IOException("Fail to create Hudi record reader for " +
tablePath + ".", e);
}
@@ -79,4 +113,12 @@ public class HudiLakeSource implements
LakeSource<HudiSplit> {
public SimpleVersionedSerializer<HudiSplit> getSplitSerializer() {
return new HudiSplitSerializer();
}
+
+ private Schema getHudiSchema() {
+ try (HudiTableInfo hudiTableInfo = HudiTableInfo.create(tablePath,
hudiConfig)) {
+ return
StreamerUtil.getTableAvroSchema(hudiTableInfo.getMetaClient(), true);
+ } catch (Exception e) {
+ throw new RuntimeException("Fail to get Hudi schema for " +
tablePath + ".", e);
+ }
+ }
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/FlussToHudiExpressionPredicateConverter.java
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/FlussToHudiExpressionPredicateConverter.java
new file mode 100644
index 000000000..b4f26739d
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/main/java/org/apache/fluss/lake/hudi/utils/FlussToHudiExpressionPredicateConverter.java
@@ -0,0 +1,281 @@
+/*
+ * 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.fluss.lake.hudi.utils;
+
+import org.apache.fluss.predicate.And;
+import org.apache.fluss.predicate.CompoundPredicate;
+import org.apache.fluss.predicate.FieldRef;
+import org.apache.fluss.predicate.FunctionVisitor;
+import org.apache.fluss.predicate.LeafPredicate;
+import org.apache.fluss.predicate.Or;
+import org.apache.fluss.predicate.Predicate;
+import org.apache.fluss.predicate.PredicateVisitor;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.TimestampLtz;
+import org.apache.fluss.row.TimestampNtz;
+
+import org.apache.flink.table.expressions.FieldReferenceExpression;
+import org.apache.flink.table.expressions.ValueLiteralExpression;
+import org.apache.flink.table.types.DataType;
+import org.apache.flink.table.types.logical.LogicalTypeRoot;
+import org.apache.hudi.org.apache.avro.Schema;
+import org.apache.hudi.source.ExpressionPredicates;
+import org.apache.hudi.util.AvroSchemaConverter;
+
+import java.time.ZoneOffset;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+/** Converts Fluss predicates to Hudi expression predicates for Hudi reader
pushdown. */
+public class FlussToHudiExpressionPredicateConverter
+ implements PredicateVisitor<ExpressionPredicates.Predicate> {
+
+ private static final String HUDI_METADATA_COLUMN_PREFIX = "_hoodie_";
+
+ private final Schema hudiSchema;
+ private final int metadataFieldCount;
+ private final LeafFunctionConverter converter = new
LeafFunctionConverter();
+
+ private FlussToHudiExpressionPredicateConverter(Schema hudiSchema) {
+ this.hudiSchema = hudiSchema;
+ this.metadataFieldCount = metadataFieldCount(hudiSchema);
+ }
+
+ public static Optional<ExpressionPredicates.Predicate> convert(
+ Schema hudiSchema, Predicate flussPredicate) {
+ try {
+ return Optional.of(
+ flussPredicate.visit(new
FlussToHudiExpressionPredicateConverter(hudiSchema)));
+ } catch (UnsupportedOperationException e) {
+ return Optional.empty();
+ }
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visit(LeafPredicate predicate) {
+ return predicate.visit(converter);
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visit(CompoundPredicate predicate) {
+ List<ExpressionPredicates.Predicate> children =
+ predicate.children().stream().map(p ->
p.visit(this)).collect(Collectors.toList());
+
+ CompoundPredicate.Function function = predicate.function();
+ if (function instanceof And) {
+ return reducePredicates(children, true);
+ } else if (function instanceof Or) {
+ return reducePredicates(children, false);
+ }
+
+ throw new UnsupportedOperationException(
+ "Unsupported Fluss compound predicate function: " + function);
+ }
+
+ private ExpressionPredicates.Predicate reducePredicates(
+ List<ExpressionPredicates.Predicate> children, boolean and) {
+ if (children.isEmpty()) {
+ throw new UnsupportedOperationException("Empty compound
predicate.");
+ }
+
+ ExpressionPredicates.Predicate result = children.get(0);
+ for (int i = 1; i < children.size(); i++) {
+ result =
+ and
+ ? ExpressionPredicates.And.getInstance()
+ .bindPredicates(result, children.get(i))
+ : ExpressionPredicates.Or.getInstance()
+ .bindPredicates(result, children.get(i));
+ }
+ return result;
+ }
+
+ private class LeafFunctionConverter implements
FunctionVisitor<ExpressionPredicates.Predicate> {
+
+ @Override
+ public ExpressionPredicates.Predicate visitIsNotNull(FieldRef
fieldRef) {
+ throw new UnsupportedOperationException("Hudi does not support
isNotNull pushdown.");
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitIsNull(FieldRef fieldRef) {
+ throw new UnsupportedOperationException("Hudi does not support
isNull pushdown.");
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitStartsWith(FieldRef
fieldRef, Object literal) {
+ throw new UnsupportedOperationException("Hudi does not support
startsWith pushdown.");
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitEndsWith(FieldRef fieldRef,
Object literal) {
+ throw new UnsupportedOperationException("Hudi does not support
endsWith pushdown.");
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitContains(FieldRef fieldRef,
Object literal) {
+ throw new UnsupportedOperationException("Hudi does not support
contains pushdown.");
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitLessThan(FieldRef fieldRef,
Object literal) {
+ return bind(ExpressionPredicates.LessThan.getInstance(), fieldRef,
literal);
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitGreaterOrEqual(
+ FieldRef fieldRef, Object literal) {
+ return bind(ExpressionPredicates.GreaterThanOrEqual.getInstance(),
fieldRef, literal);
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitNotEqual(FieldRef fieldRef,
Object literal) {
+ return bind(ExpressionPredicates.NotEquals.getInstance(),
fieldRef, literal);
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitLessOrEqual(FieldRef
fieldRef, Object literal) {
+ return bind(ExpressionPredicates.LessThanOrEqual.getInstance(),
fieldRef, literal);
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitEqual(FieldRef fieldRef,
Object literal) {
+ return bind(ExpressionPredicates.Equals.getInstance(), fieldRef,
literal);
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitGreaterThan(FieldRef
fieldRef, Object literal) {
+ return bind(ExpressionPredicates.GreaterThan.getInstance(),
fieldRef, literal);
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitIn(FieldRef fieldRef,
List<Object> literals) {
+ Schema.Field field = getFieldByFlussIndex(fieldRef.index());
+ DataType dataType = supportedHudiDataType(field);
+ List<ValueLiteralExpression> literalExpressions =
+ literals.stream()
+ .map(literal -> createValueLiteral(literal,
dataType))
+ .collect(Collectors.toList());
+ return ExpressionPredicates.In.getInstance()
+ .bindValueLiterals(literalExpressions)
+ .bindFieldReference(createFieldReference(field, dataType));
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitNotIn(FieldRef fieldRef,
List<Object> literals) {
+ ExpressionPredicates.Predicate inPredicate = visitIn(fieldRef,
literals);
+ return
ExpressionPredicates.Not.getInstance().bindPredicate(inPredicate);
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitAnd(
+ List<ExpressionPredicates.Predicate> children) {
+ throw new UnsupportedOperationException("Unsupported visitAnd
method.");
+ }
+
+ @Override
+ public ExpressionPredicates.Predicate visitOr(
+ List<ExpressionPredicates.Predicate> children) {
+ throw new UnsupportedOperationException("Unsupported visitOr
method.");
+ }
+
+ private ExpressionPredicates.Predicate bind(
+ ExpressionPredicates.ColumnPredicate predicate, FieldRef
fieldRef, Object literal) {
+ Schema.Field field = getFieldByFlussIndex(fieldRef.index());
+ DataType dataType = supportedHudiDataType(field);
+ return predicate
+ .bindValueLiteral(createValueLiteral(literal, dataType))
+ .bindFieldReference(createFieldReference(field, dataType));
+ }
+
+ private FieldReferenceExpression createFieldReference(
+ Schema.Field field, DataType dataType) {
+ return new FieldReferenceExpression(field.name(), dataType, 0,
field.pos());
+ }
+
+ private ValueLiteralExpression createValueLiteral(Object literal,
DataType dataType) {
+ if (literal == null) {
+ throw new UnsupportedOperationException(
+ "Hudi does not support null literal pushdown.");
+ }
+ return new ValueLiteralExpression(convertValue(literal),
dataType.notNull());
+ }
+
+ private Schema.Field getFieldByFlussIndex(int index) {
+ int hudiFieldIndex = metadataFieldCount + index;
+ if (index < 0 || hudiFieldIndex >= hudiSchema.getFields().size()) {
+ throw new IllegalArgumentException("Field index out of bounds:
" + index);
+ }
+ return hudiSchema.getFields().get(hudiFieldIndex);
+ }
+
+ private DataType supportedHudiDataType(Schema.Field field) {
+ DataType dataType =
AvroSchemaConverter.convertToDataType(field.schema());
+ LogicalTypeRoot typeRoot = dataType.getLogicalType().getTypeRoot();
+ switch (typeRoot) {
+ case BOOLEAN:
+ case TINYINT:
+ case SMALLINT:
+ case INTEGER:
+ case DATE:
+ case TIME_WITHOUT_TIME_ZONE:
+ case BIGINT:
+ case TIMESTAMP_WITHOUT_TIME_ZONE:
+ case FLOAT:
+ case DOUBLE:
+ case BINARY:
+ case VARBINARY:
+ case CHAR:
+ case VARCHAR:
+ return dataType;
+ default:
+ throw new UnsupportedOperationException(
+ "Hudi does not support predicate pushdown for
field "
+ + field.name()
+ + " with type "
+ + typeRoot
+ + ".");
+ }
+ }
+ }
+
+ private static int metadataFieldCount(Schema schema) {
+ int metadataFieldCount = 0;
+ for (Schema.Field field : schema.getFields()) {
+ if (!field.name().startsWith(HUDI_METADATA_COLUMN_PREFIX)) {
+ break;
+ }
+ metadataFieldCount++;
+ }
+ return metadataFieldCount;
+ }
+
+ private static Object convertValue(Object value) {
+ if (value instanceof BinaryString) {
+ return value.toString();
+ }
+ if (value instanceof TimestampNtz) {
+ return ((TimestampNtz) value).toLocalDateTime();
+ }
+ if (value instanceof TimestampLtz) {
+ return ((TimestampLtz)
value).toInstant().atZone(ZoneOffset.UTC).toLocalDateTime();
+ }
+ return value;
+ }
+}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
index e8446ee85..5d8bbdf29 100644
---
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/source/HudiLakeSourceTest.java
@@ -20,14 +20,9 @@ package org.apache.fluss.lake.hudi.source;
import org.apache.fluss.config.Configuration;
import org.apache.fluss.lake.source.LakeSource;
import org.apache.fluss.metadata.TablePath;
-import org.apache.fluss.predicate.Predicate;
-import org.apache.fluss.predicate.PredicateBuilder;
-import org.apache.fluss.types.IntType;
-import org.apache.fluss.types.RowType;
import org.junit.jupiter.api.Test;
-import java.io.IOException;
import java.util.Collections;
import static org.assertj.core.api.Assertions.assertThat;
@@ -53,16 +48,14 @@ class HudiLakeSourceTest {
}
@Test
- void testFiltersRemainWhenPushDownIsNotSupportedYet() {
+ void testWithEmptyFiltersReturnsEmptyPushDownResult() {
HudiLakeSource source =
new HudiLakeSource(new Configuration(), TablePath.of("db1",
"table1"));
- Predicate predicate = new PredicateBuilder(RowType.of(new
IntType())).equal(0, 1);
- LakeSource.FilterPushDownResult result =
- source.withFilters(Collections.singletonList(predicate));
+ LakeSource.FilterPushDownResult result =
source.withFilters(Collections.emptyList());
assertThat(result.acceptedPredicates()).isEmpty();
- assertThat(result.remainingPredicates()).containsExactly(predicate);
+ assertThat(result.remainingPredicates()).isEmpty();
}
@Test
@@ -74,14 +67,4 @@ class HudiLakeSourceTest {
.isInstanceOf(UnsupportedOperationException.class)
.hasMessageContaining("limit");
}
-
- @Test
- void testCreateRecordReaderWrapsReaderErrors() {
- HudiLakeSource source =
- new HudiLakeSource(new Configuration(), TablePath.of("db1",
"table1"));
-
- assertThatThrownBy(() -> source.createRecordReader(() -> null))
- .isInstanceOf(IOException.class)
- .hasMessageContaining("Fail to create Hudi record reader");
- }
}
diff --git
a/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/FlussToHudiExpressionPredicateConverterTest.java
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/FlussToHudiExpressionPredicateConverterTest.java
new file mode 100644
index 000000000..08c7bb779
--- /dev/null
+++
b/fluss-lake/fluss-lake-hudi/src/test/java/org/apache/fluss/lake/hudi/utils/FlussToHudiExpressionPredicateConverterTest.java
@@ -0,0 +1,163 @@
+/*
+ * 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.fluss.lake.hudi.utils;
+
+import org.apache.fluss.predicate.Predicate;
+import org.apache.fluss.predicate.PredicateBuilder;
+import org.apache.fluss.row.BinaryString;
+import org.apache.fluss.row.TimestampNtz;
+import org.apache.fluss.types.DataTypes;
+import org.apache.fluss.types.RowType;
+
+import org.apache.hudi.org.apache.avro.Schema;
+import org.apache.hudi.source.ExpressionPredicates;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+import java.util.stream.Stream;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Test for {@link FlussToHudiExpressionPredicateConverter}. */
+class FlussToHudiExpressionPredicateConverterTest {
+
+ private static final PredicateBuilder PREDICATE_BUILDER =
+ new PredicateBuilder(
+ RowType.builder()
+ .field("id", DataTypes.BIGINT())
+ .field("name", DataTypes.STRING())
+ .field("score", DataTypes.DOUBLE())
+ .field("event_time", DataTypes.TIMESTAMP(3))
+ .field("payload", DataTypes.BYTES())
+ .field("amount", DataTypes.DECIMAL(10, 2))
+ .build());
+
+ public static Stream<Arguments> supportedPredicates() {
+ return Stream.of(
+ Arguments.of(PREDICATE_BUILDER.equal(0, 12L), "id = 12"),
+ Arguments.of(
+ PREDICATE_BUILDER.notEqual(1,
BinaryString.fromString("test")),
+ "name != test"),
+ Arguments.of(PREDICATE_BUILDER.lessThan(2, 10.5D), "score <
10.5"),
+ Arguments.of(PREDICATE_BUILDER.lessOrEqual(0, 50L), "id <=
50"),
+ Arguments.of(PREDICATE_BUILDER.greaterThan(2, 100.5D), "score
> 100.5"),
+ Arguments.of(PREDICATE_BUILDER.greaterOrEqual(0, 100L), "id >=
100"),
+ Arguments.of(
+ PREDICATE_BUILDER.lessOrEqual(3,
TimestampNtz.fromMillis(1000L)),
+ "event_time <= 1000"),
+ Arguments.of(
+ PREDICATE_BUILDER.in(0, longValues(21)),
+ "id IN([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, "
+ + "10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
20])"),
+ Arguments.of(
+ PREDICATE_BUILDER.notIn(0, longValues(21)),
+ "NOT(id IN([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, "
+ + "10, 11, 12, 13, 14, 15, 16, 17, 18, 19,
20]))"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("supportedPredicates")
+ void testSupportedPredicates(Predicate flussPredicate, String
expectedPredicate) {
+ Optional<ExpressionPredicates.Predicate> convertedPredicate =
+ FlussToHudiExpressionPredicateConverter.convert(hudiSchema(),
flussPredicate);
+
+ assertThat(convertedPredicate).isPresent();
+
assertThat(convertedPredicate.get().toString()).isEqualTo(expectedPredicate);
+ }
+
+ public static Stream<Arguments> unsupportedPredicates() {
+ return Stream.of(
+ Arguments.of(PREDICATE_BUILDER.isNull(1)),
+ Arguments.of(PREDICATE_BUILDER.isNotNull(1)),
+ Arguments.of(PREDICATE_BUILDER.startsWith(1,
BinaryString.fromString("prefix"))),
+ Arguments.of(PREDICATE_BUILDER.endsWith(1,
BinaryString.fromString("suffix"))),
+ Arguments.of(PREDICATE_BUILDER.contains(1,
BinaryString.fromString("part"))),
+ Arguments.of(PREDICATE_BUILDER.equal(0, null)),
+ Arguments.of(PREDICATE_BUILDER.equal(5, 12.34D)));
+ }
+
+ @ParameterizedTest
+ @MethodSource("unsupportedPredicates")
+ void testUnsupportedPredicates(Predicate flussPredicate) {
+
assertThat(FlussToHudiExpressionPredicateConverter.convert(hudiSchema(),
flussPredicate))
+ .isEqualTo(Optional.empty());
+ }
+
+ @ParameterizedTest
+ @MethodSource("compoundPredicates")
+ void testCompoundPredicates(Predicate flussPredicate, String[]
expectedReferences) {
+ Optional<ExpressionPredicates.Predicate> convertedPredicate =
+ FlussToHudiExpressionPredicateConverter.convert(hudiSchema(),
flussPredicate);
+
+ assertThat(convertedPredicate).isPresent();
+
assertThat(convertedPredicate.get().references()).containsExactly(expectedReferences);
+ }
+
+ public static Stream<Arguments> compoundPredicates() {
+ return Stream.of(
+ Arguments.of(
+ PredicateBuilder.and(
+ PREDICATE_BUILDER.equal(0, 1L),
+ PREDICATE_BUILDER.equal(1,
BinaryString.fromString("one")),
+ PREDICATE_BUILDER.greaterThan(2, 10.0D)),
+ new String[] {"id", "name", "score"}),
+ Arguments.of(
+ PredicateBuilder.or(
+ PREDICATE_BUILDER.equal(0, 1L),
+ PREDICATE_BUILDER.greaterThan(2, 10.0D)),
+ new String[] {"id", "score"}));
+ }
+
+ private static Schema hudiSchema() {
+ return new Schema.Parser()
+ .parse(
+ "{"
+ + "\"type\":\"record\","
+ + "\"name\":\"hudi_record\","
+ + "\"fields\":["
+ +
"{\"name\":\"_hoodie_commit_time\",\"type\":\"string\"},"
+ +
"{\"name\":\"_hoodie_commit_seqno\",\"type\":\"string\"},"
+ +
"{\"name\":\"_hoodie_record_key\",\"type\":\"string\"},"
+ +
"{\"name\":\"_hoodie_partition_path\",\"type\":\"string\"},"
+ +
"{\"name\":\"_hoodie_file_name\",\"type\":\"string\"},"
+ + "{\"name\":\"id\",\"type\":\"long\"},"
+ + "{\"name\":\"name\",\"type\":\"string\"},"
+ + "{\"name\":\"score\",\"type\":\"double\"},"
+ + "{\"name\":\"event_time\","
+ +
"\"type\":{\"type\":\"long\",\"logicalType\":\"timestamp-millis\"}},"
+ + "{\"name\":\"payload\",\"type\":\"bytes\"},"
+ + "{\"name\":\"amount\","
+ + "\"type\":{\"type\":\"bytes\","
+ +
"\"logicalType\":\"decimal\",\"precision\":10,\"scale\":2}},"
+ + "{\"name\":\"__bucket\",\"type\":\"int\"},"
+ + "{\"name\":\"__offset\",\"type\":\"long\"},"
+ +
"{\"name\":\"__timestamp\",\"type\":\"long\"}"
+ + "]"
+ + "}");
+ }
+
+ private static java.util.List<Object> longValues(int size) {
+ return IntStream.range(0, size)
+ .mapToObj(i -> (Object) (long) i)
+ .collect(Collectors.toList());
+ }
+}
