This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 4ac91ffa57 [flink] Build balanced binary tree for OR/AND predicates to
avoid StackOverflowError (#8696)
4ac91ffa57 is described below
commit 4ac91ffa578c238ec25a5d4b17f45dc1cbc8d7b8
Author: yunfengzhou-hub <[email protected]>
AuthorDate: Thu Jul 16 22:08:29 2026 +0800
[flink] Build balanced binary tree for OR/AND predicates to avoid
StackOverflowError (#8696)
---
.../apache/paimon/predicate/PredicateBuilder.java | 23 ++-
.../paimon/predicate/PredicateBuilderTest.java | 78 +++++++++
.../paimon/predicate/PredicateJsonSerdeTest.java | 6 +-
.../org/apache/paimon/predicate/PredicateTest.java | 4 +-
.../apache/paimon/flink/PredicateConverter.java | 41 ++++-
.../paimon/flink/PredicateConverterTest.java | 66 ++++++++
.../paimon/flink/source/FlinkTableSourceTest.java | 185 +++++++++++++++++++++
.../format/orc/filter/OrcFilterConverterTest.java | 32 ++--
.../paimon/format/parquet/ParquetFiltersTest.java | 38 ++---
9 files changed, 424 insertions(+), 49 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java
index 05acce1729..b26baa2511 100644
---
a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateBuilder.java
@@ -304,9 +304,7 @@ public class PredicateBuilder {
return optimized.get(0);
}
- return optimized.stream()
- .reduce((a, b) -> new CompoundPredicate(And.INSTANCE,
Arrays.asList(a, b)))
- .get();
+ return buildBinaryTree(And.INSTANCE, optimized);
}
@Nullable
@@ -359,9 +357,22 @@ public class PredicateBuilder {
return noFalsePredicates.get(0);
}
- return noFalsePredicates.stream()
- .reduce((a, b) -> new CompoundPredicate(Or.INSTANCE,
Arrays.asList(a, b)))
- .get();
+ return buildBinaryTree(Or.INSTANCE, noFalsePredicates);
+ }
+
+ private static Predicate buildBinaryTree(CompoundFunction func,
List<Predicate> predicates) {
+ if (predicates.size() == 1) {
+ return predicates.get(0);
+ }
+ if (predicates.size() == 2) {
+ return new CompoundPredicate(func,
Arrays.asList(predicates.get(0), predicates.get(1)));
+ }
+ int mid = predicates.size() / 2;
+ return new CompoundPredicate(
+ func,
+ Arrays.asList(
+ buildBinaryTree(func, predicates.subList(0, mid)),
+ buildBinaryTree(func, predicates.subList(mid,
predicates.size()))));
}
private static boolean isAlwaysFalse(Predicate predicate) {
diff --git
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
index 4b6ac01626..fe66f95241 100644
---
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateBuilderTest.java
@@ -39,6 +39,9 @@ import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
import static org.apache.paimon.predicate.SimpleColStatsTestUtils.test;
import static org.assertj.core.api.Assertions.assertThat;
@@ -217,6 +220,81 @@ public class PredicateBuilderTest {
assertThat(predicate.test(GenericRow.of(10))).isEqualTo(false);
}
+ // ---- or()/and() binary tree structure tests ----
+
+ @Test
+ public void testOrBinaryTree() {
+ PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
+ List<Predicate> predicates =
+ IntStream.range(0, 25)
+ .mapToObj(i -> builder.equal(0, i))
+ .collect(Collectors.toList());
+
+ Predicate result = PredicateBuilder.or(predicates);
+
+ assertThat(result).isInstanceOf(CompoundPredicate.class);
+ CompoundPredicate root = (CompoundPredicate) result;
+ assertThat(root.function()).isEqualTo(Or.INSTANCE);
+ assertThat(root.children()).hasSize(2);
+ assertThat(countLeaves(root)).isEqualTo(25);
+ assertThat(maxDepth(root)).isLessThanOrEqualTo(5);
+ }
+
+ @Test
+ public void testOrBinaryTreeEvaluation() {
+ PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
+ List<Predicate> predicates =
+ IntStream.range(0, 25)
+ .mapToObj(i -> builder.equal(0, i))
+ .collect(Collectors.toList());
+
+ Predicate result = PredicateBuilder.or(predicates);
+
+ assertThat(result.test(GenericRow.of(0))).isTrue();
+ assertThat(result.test(GenericRow.of(24))).isTrue();
+ assertThat(result.test(GenericRow.of(25))).isFalse();
+ }
+
+ @Test
+ public void testAndBinaryTree() {
+ PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
+ List<Predicate> predicates =
+ IntStream.range(0, 25)
+ .mapToObj(i -> builder.greaterThan(0, i))
+ .collect(Collectors.toList());
+
+ Predicate result = PredicateBuilder.and(predicates);
+
+ assertThat(result).isInstanceOf(CompoundPredicate.class);
+ CompoundPredicate root = (CompoundPredicate) result;
+ assertThat(root.function()).isEqualTo(And.INSTANCE);
+ assertThat(root.children()).hasSize(2);
+ assertThat(countLeaves(root)).isEqualTo(25);
+ assertThat(maxDepth(root)).isLessThanOrEqualTo(5);
+ }
+
+ private static int countLeaves(Predicate predicate) {
+ if (!(predicate instanceof CompoundPredicate)) {
+ return 1;
+ }
+ int count = 0;
+ for (Predicate child : ((CompoundPredicate) predicate).children()) {
+ count += countLeaves(child);
+ }
+ return count;
+ }
+
+ private static int maxDepth(Predicate predicate) {
+ if (!(predicate instanceof CompoundPredicate)) {
+ return 0;
+ }
+ int max = 0;
+ for (Predicate child : ((CompoundPredicate) predicate).children()) {
+ max = Math.max(max, maxDepth(child));
+ }
+ return 1 + max;
+ }
+
@Test
public void testConvertToJavaObjectRoundTrip() {
// VARCHAR
diff --git
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java
index 66f0b1f4d1..4837b9863f 100644
---
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java
+++
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateJsonSerdeTest.java
@@ -87,12 +87,12 @@ class PredicateJsonSerdeTest {
// LeafPredicate - In
TestSpec.forPredicate(builder.in(0, Arrays.asList(1, 2, 3)))
.expectJson(
-
"{\"kind\":\"COMPOUND\",\"function\":\"OR\",\"children\":[{\"kind\":\"COMPOUND\",\"function\":\"OR\",\"children\":[{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"EQUAL\",\"literals\":[1]},{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"EQUAL\",\"literals\":[2]}]},{\"kind\":\"LEAF\",\"tran
[...]
+
"{\"kind\":\"COMPOUND\",\"function\":\"OR\",\"children\":[{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"EQUAL\",\"literals\":[1]},{\"kind\":\"COMPOUND\",\"function\":\"OR\",\"children\":[{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"EQUAL\",\"literals\":[2]},{\"kind\":\"LEAF\",\"transf
[...]
// LeafPredicate - NotIn
TestSpec.forPredicate(builder.notIn(0, Arrays.asList(1, 2, 3)))
.expectJson(
-
"{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"NOT_EQUAL\",\"literals\":[1]},{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"NOT_EQUAL\",\"literals\":[2]}]},{\"kind\":\"LEA
[...]
+
"{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"NOT_EQUAL\",\"literals\":[1]},{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"NOT_EQUAL\",\"literals\":[2]},{\"kind\":\"LEAF\
[...]
// LeafPredicate - CastTransform
TestSpec.forPredicate(
@@ -211,7 +211,7 @@ class PredicateJsonSerdeTest {
PredicateBuilder.or(
builder.equal(0, 7),
builder.isNotNull(2))))
.expectJson(
-
"{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"EQUAL\",\"literals\":[1]},{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\
[...]
+
"{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"COMPOUND\",\"function\":\"AND\",\"children\":[{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":0,\"name\":\"f0\",\"type\":\"INT\"}},\"function\":\"EQUAL\",\"literals\":[1]},{\"kind\":\"LEAF\",\"transform\":{\"name\":\"FIELD_REF\",\"fieldRef\":{\"index\":3,\"name\":\"f3\",\"type\":\"INT\"}},\"function\":\"IN\",\"literals\":[0,1,2,3,4,5,6,7,8,9,10,11,12,13,
[...]
// error message testing
TestSpec.forJson("{\"kind\":\"invalid\"}")
diff --git
a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateTest.java
b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateTest.java
index a33112d340..f09c263429 100644
--- a/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateTest.java
+++ b/paimon-common/src/test/java/org/apache/paimon/predicate/PredicateTest.java
@@ -736,13 +736,13 @@ public class PredicateTest {
Predicate p6 = builder6.in(0, Arrays.asList(1, null, 3, 4));
assertThat(p6.toString())
.isEqualTo(
- "Or([Or([Or([Equal(f0, 1), Equal(f0, null)]),
Equal(f0, 3)]), Equal(f0, 4)])");
+ "Or([Or([Equal(f0, 1), Equal(f0, null)]),
Or([Equal(f0, 3), Equal(f0, 4)])])");
PredicateBuilder builder7 = new PredicateBuilder(RowType.of(new
IntType()));
Predicate p7 = builder7.notIn(0, Arrays.asList(1, null, 3, 4));
assertThat(p7.toString())
.isEqualTo(
- "And([And([And([NotEqual(f0, 1), NotEqual(f0, null)]),
NotEqual(f0, 3)]), NotEqual(f0, 4)])");
+ "And([And([NotEqual(f0, 1), NotEqual(f0, null)]),
And([NotEqual(f0, 3), NotEqual(f0, 4)])])");
PredicateBuilder builder8 = new PredicateBuilder(RowType.of(new
IntType()));
List<Object> literals = new ArrayList<>();
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
index a91b2720a3..d5413f0859 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PredicateConverter.java
@@ -38,7 +38,9 @@ import org.apache.flink.table.types.logical.LogicalType;
import org.apache.flink.table.types.logical.LogicalTypeFamily;
import org.apache.flink.table.types.logical.RowType;
+import java.util.ArrayDeque;
import java.util.ArrayList;
+import java.util.Deque;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
@@ -76,9 +78,9 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
List<Expression> children = call.getChildren();
if (func == BuiltInFunctionDefinitions.AND) {
- return PredicateBuilder.and(children.get(0).accept(this),
children.get(1).accept(this));
+ return PredicateBuilder.and(flattenAndConvert(children, func));
} else if (func == BuiltInFunctionDefinitions.OR) {
- return PredicateBuilder.or(children.get(0).accept(this),
children.get(1).accept(this));
+ return PredicateBuilder.or(flattenAndConvert(children, func));
} else if (func == BuiltInFunctionDefinitions.EQUALS) {
return visitBiFunction(children, builder::equal, builder::equal);
} else if (func == BuiltInFunctionDefinitions.NOT_EQUALS) {
@@ -201,6 +203,41 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
throw new UnsupportedExpression();
}
+ /**
+ * Iteratively flattens a nested AND/OR expression tree into a flat list
of child predicates,
+ * avoiding stack overflow caused by recursive {@code accept} calls on
deeply nested trees (e.g.
+ * when Flink expands a large IN clause into nested OR expressions).
+ *
+ * @param children the children of the top-level AND/OR {@link
CallExpression}
+ * @param targetFunc the function definition to flatten ({@code AND} or
{@code OR})
+ * @return a flat list of converted child predicates in original order
+ */
+ private List<Predicate> flattenAndConvert(
+ List<Expression> children, FunctionDefinition targetFunc) {
+ List<Predicate> result = new ArrayList<>();
+ Deque<Expression> stack = new ArrayDeque<>();
+ for (int i = children.size() - 1; i >= 0; i--) {
+ stack.push(children.get(i));
+ }
+ while (!stack.isEmpty()) {
+ Expression expr = stack.pop();
+ if (expr instanceof CallExpression) {
+ CallExpression ce = (CallExpression) expr;
+ if (ce.getFunctionDefinition() == targetFunc) {
+ List<Expression> ceChildren = ce.getChildren();
+ for (int i = ceChildren.size() - 1; i >= 0; i--) {
+ stack.push(ceChildren.get(i));
+ }
+ } else {
+ result.add(ce.accept(this));
+ }
+ } else {
+ result.add(expr.accept(this));
+ }
+ }
+ return result;
+ }
+
private Predicate visitBiFunction(
List<Expression> children,
BiFunction<Integer, Object, Predicate> visit1,
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PredicateConverterTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PredicateConverterTest.java
index c304a70434..d86293277e 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PredicateConverterTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/PredicateConverterTest.java
@@ -21,6 +21,8 @@ package org.apache.paimon.flink;
import org.apache.paimon.data.BinaryString;
import org.apache.paimon.data.GenericRow;
import org.apache.paimon.format.SimpleColStats;
+import org.apache.paimon.predicate.CompoundPredicate;
+import org.apache.paimon.predicate.Or;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.predicate.SimpleColStatsTestUtils;
@@ -808,6 +810,70 @@ public class PredicateConverterTest {
.isInstanceOf(PredicateConverter.UnsupportedExpression.class);
}
+ // ==================== Nested OR Conversion Tests ====================
+ //
+ // When Flink expands IN(v1,...,vN) it produces a deeply nested binary OR
tree:
+ // OR(=(f,v1), OR(=(f,v2), OR(..., =(f,vN))))
+ // PredicateConverter iteratively flattens this tree into a list of
predicates,
+ // which PredicateBuilder.or() combines into a binary tree.
+
+ @Test
+ public void testNestedOrOfEquals() {
+ // Build OR(=(long1, 0), OR(=(long1, 1), ...)) with 25 values
+ FieldReferenceExpression longRef =
+ new FieldReferenceExpression(
+ "long1", DataTypes.BIGINT(), Integer.MAX_VALUE,
Integer.MAX_VALUE);
+
+ ResolvedExpression orTree = null;
+ for (int i = 24; i >= 0; i--) {
+ CallExpression equal =
+ call(BuiltInFunctionDefinitions.EQUALS, longRef, new
ValueLiteralExpression(i));
+ if (orTree == null) {
+ orTree = equal;
+ } else {
+ orTree = call(BuiltInFunctionDefinitions.OR, equal, orTree);
+ }
+ }
+
+ Predicate result = CONVERTER.visit((CallExpression) orTree);
+
+ // OR-of-equals → flattened → PredicateBuilder.or() → binary tree
+ assertThat(result).isInstanceOf(CompoundPredicate.class);
+ CompoundPredicate compound = (CompoundPredicate) result;
+ assertThat(compound.function()).isEqualTo(Or.INSTANCE);
+ assertThat(compound.children()).hasSize(2);
+ }
+
+ @Test
+ public void testNestedOrOfDifferentPredicates() {
+ // Build OR(>(long1, 0), OR(>(long1, 1), ...)) with 25 values
+ FieldReferenceExpression longRef =
+ new FieldReferenceExpression(
+ "long1", DataTypes.BIGINT(), Integer.MAX_VALUE,
Integer.MAX_VALUE);
+
+ ResolvedExpression orTree = null;
+ for (int i = 24; i >= 0; i--) {
+ CallExpression greater =
+ call(
+ BuiltInFunctionDefinitions.GREATER_THAN,
+ longRef,
+ new ValueLiteralExpression(i));
+ if (orTree == null) {
+ orTree = greater;
+ } else {
+ orTree = call(BuiltInFunctionDefinitions.OR, greater, orTree);
+ }
+ }
+
+ Predicate result = CONVERTER.visit((CallExpression) orTree);
+
+ // General OR → binary tree (root has 2 children)
+ assertThat(result).isInstanceOf(CompoundPredicate.class);
+ CompoundPredicate compound = (CompoundPredicate) result;
+ assertThat(compound.function()).isEqualTo(Or.INSTANCE);
+ assertThat(compound.children()).hasSize(2);
+ }
+
private static FieldReferenceExpression field(int i, DataType type) {
return new FieldReferenceExpression("f" + i, type, Integer.MAX_VALUE,
Integer.MAX_VALUE);
}
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceTest.java
index 4965cc4135..ba7dbd0893 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/FlinkTableSourceTest.java
@@ -21,6 +21,8 @@ package org.apache.paimon.flink.source;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
import org.apache.paimon.fs.local.LocalFileIO;
+import org.apache.paimon.predicate.CompoundPredicate;
+import org.apache.paimon.predicate.Or;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.TableSchema;
@@ -40,6 +42,7 @@ import
org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.assertj.core.api.Assertions;
import org.junit.jupiter.api.Test;
+import java.util.ArrayList;
import java.util.List;
/** Test for {@link FlinkTableSource}. */
@@ -136,6 +139,188 @@ public class FlinkTableSourceTest extends TableTestBase {
.isEqualTo(ImmutableList.of(filters.get(1)));
}
+ // ==================== Nested OR Tree Tests ====================
+ //
+ // These tests construct OR trees in various shapes — mimicking what
Flink's
+ // SQL Planner may produce when expanding IN(v1,...,vN) — and pass them
directly
+ // to applyFilters, bypassing Flink's ExpressionResolver.
+ //
+ // They verify that PredicateConverter flattens any nesting shape into a
flat list
+ // of predicates, which PredicateBuilder.or() combines into a binary tree.
+
+ @Test
+ public void testApplyFiltersLargeNestedOr() throws Exception {
+ Table table = createStringTable();
+
+ // 10000 values: the nested OR tree is flattened and combined into a
+ // binary tree (depth ~14), preventing StackOverflowError.
+ int size = 10000;
+ DataTableSource tableSource =
+ new DataTableSource(
+ ObjectIdentifier.of("catalog1", "db1", "T"), table,
false, null);
+ ResolvedExpression orTree = buildNestedOrTree(size);
+
+ tableSource.applyFilters(ImmutableList.of(orTree));
+
+ Assertions.assertThat(tableSource.predicate).isNotNull();
+
Assertions.assertThat(tableSource.predicate).isInstanceOf(CompoundPredicate.class);
+ CompoundPredicate compound = (CompoundPredicate) tableSource.predicate;
+ Assertions.assertThat(compound.function()).isEqualTo(Or.INSTANCE);
+ Assertions.assertThat(compound.children()).hasSize(2);
+ }
+
+ @Test
+ public void testApplyFiltersRightFoldOrTree() throws Exception {
+ Table table = createStringTable();
+ DataTableSource tableSource =
+ new DataTableSource(
+ ObjectIdentifier.of("catalog1", "db1", "T"), table,
false, null);
+
+ // Right-fold tree: OR(OR(OR(=(f,0), =(f,1)), =(f,2)), ...) with 25
values
+ ResolvedExpression orTree = buildRightFoldOrTree(25);
+
+ tableSource.applyFilters(ImmutableList.of(orTree));
+
+ // Regardless of tree shape → flattened → binary tree
+ Assertions.assertThat(tableSource.predicate).isNotNull();
+
Assertions.assertThat(tableSource.predicate).isInstanceOf(CompoundPredicate.class);
+ CompoundPredicate compound = (CompoundPredicate) tableSource.predicate;
+ Assertions.assertThat(compound.function()).isEqualTo(Or.INSTANCE);
+ Assertions.assertThat(compound.children()).hasSize(2);
+ }
+
+ @Test
+ public void testApplyFiltersBalancedOrTree() throws Exception {
+ Table table = createStringTable();
+ DataTableSource tableSource =
+ new DataTableSource(
+ ObjectIdentifier.of("catalog1", "db1", "T"), table,
false, null);
+
+ // Balanced tree: OR(OR(=(f,0), =(f,1)), OR(=(f,2), =(f,3)), ...) with
25 values
+ ResolvedExpression orTree = buildBalancedOrTree(25);
+
+ tableSource.applyFilters(ImmutableList.of(orTree));
+
+ Assertions.assertThat(tableSource.predicate).isNotNull();
+
Assertions.assertThat(tableSource.predicate).isInstanceOf(CompoundPredicate.class);
+ CompoundPredicate compound = (CompoundPredicate) tableSource.predicate;
+ Assertions.assertThat(compound.function()).isEqualTo(Or.INSTANCE);
+ Assertions.assertThat(compound.children()).hasSize(2);
+ }
+
+ @Test
+ public void testApplyFiltersFlatOrWithMultipleChildren() throws Exception {
+ Table table = createStringTable();
+ DataTableSource tableSource =
+ new DataTableSource(
+ ObjectIdentifier.of("catalog1", "db1", "T"), table,
false, null);
+
+ // Flat OR with >2 children in a single CallExpression
+ ResolvedExpression orExpr = buildFlatOrExpression(25);
+
+ tableSource.applyFilters(ImmutableList.of(orExpr));
+
+ Assertions.assertThat(tableSource.predicate).isNotNull();
+
Assertions.assertThat(tableSource.predicate).isInstanceOf(CompoundPredicate.class);
+ CompoundPredicate compound = (CompoundPredicate) tableSource.predicate;
+ Assertions.assertThat(compound.function()).isEqualTo(Or.INSTANCE);
+ Assertions.assertThat(compound.children()).hasSize(2);
+ }
+
+ private Table createStringTable() throws Exception {
+ FileIO fileIO = LocalFileIO.create();
+ Path tablePath = new Path(String.format("%s/%s.db/%s", warehouse,
database, "T"));
+ Schema schema = Schema.newBuilder().column("contract_address",
DataTypes.STRING()).build();
+ TableSchema tableSchema = new SchemaManager(fileIO,
tablePath).createTable(schema);
+ return FileStoreTableFactory.create(LocalFileIO.create(), tablePath,
tableSchema);
+ }
+
+ /**
+ * Build a nested binary OR tree mimicking Flink's IN-to-OR expansion:
OR(=(f, v1), OR(=(f, v2),
+ * OR(..., OR(=(f, vN-1), =(f, vN)))))
+ *
+ * <p>Built iteratively (inside-out) to avoid StackOverflow during
construction.
+ */
+ private ResolvedExpression buildNestedOrTree(int count) {
+ FieldReferenceExpression field =
+ new FieldReferenceExpression(
+ "contract_address",
org.apache.flink.table.api.DataTypes.STRING(), 0, 0);
+
+ // Start with innermost: =(contract_address, addr_{count-1})
+ ResolvedExpression result = equalExpr(field, count - 1);
+
+ // Wrap outward: OR(=(field, addr_i), result) for i = count-2 down to 0
+ for (int i = count - 2; i >= 0; i--) {
+ result = or(equalExpr(field, i), result);
+ }
+ return result;
+ }
+
+ /**
+ * Build a right-fold binary OR tree: OR(OR(OR(=(f, v0), =(f, v1)), =(f,
v2)), =(f, v3), ...).
+ */
+ private ResolvedExpression buildRightFoldOrTree(int count) {
+ FieldReferenceExpression field =
+ new FieldReferenceExpression(
+ "contract_address",
org.apache.flink.table.api.DataTypes.STRING(), 0, 0);
+ ResolvedExpression result = equalExpr(field, 0);
+ for (int i = 1; i < count; i++) {
+ result = or(result, equalExpr(field, i));
+ }
+ return result;
+ }
+
+ /** Build a balanced binary OR tree: OR(OR(=(f, v0), =(f, v1)), OR(=(f,
v2), =(f, v3)), ...). */
+ private ResolvedExpression buildBalancedOrTree(int count) {
+ FieldReferenceExpression field =
+ new FieldReferenceExpression(
+ "contract_address",
org.apache.flink.table.api.DataTypes.STRING(), 0, 0);
+ List<ResolvedExpression> leaves = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ leaves.add(equalExpr(field, i));
+ }
+ while (leaves.size() > 1) {
+ List<ResolvedExpression> next = new ArrayList<>();
+ for (int i = 0; i < leaves.size(); i += 2) {
+ if (i + 1 < leaves.size()) {
+ next.add(or(leaves.get(i), leaves.get(i + 1)));
+ } else {
+ next.add(leaves.get(i));
+ }
+ }
+ leaves = next;
+ }
+ return leaves.get(0);
+ }
+
+ /** Build a flat OR CallExpression with more than 2 children. */
+ private ResolvedExpression buildFlatOrExpression(int count) {
+ FieldReferenceExpression field =
+ new FieldReferenceExpression(
+ "contract_address",
org.apache.flink.table.api.DataTypes.STRING(), 0, 0);
+ List<ResolvedExpression> children = new ArrayList<>();
+ for (int i = 0; i < count; i++) {
+ children.add(equalExpr(field, i));
+ }
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.OR,
+ children,
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private ResolvedExpression equalExpr(FieldReferenceExpression field, int
i) {
+ return CallExpression.anonymous(
+ BuiltInFunctionDefinitions.EQUALS,
+ ImmutableList.of(field, addressLiteral(i)),
+ org.apache.flink.table.api.DataTypes.BOOLEAN());
+ }
+
+ private ValueLiteralExpression addressLiteral(int i) {
+ return new ValueLiteralExpression(
+ String.format("0x%040x", i),
+ org.apache.flink.table.api.DataTypes.STRING().notNull());
+ }
+
private ResolvedExpression col1Equal1() {
return CallExpression.anonymous(
BuiltInFunctionDefinitions.EQUALS,
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/orc/filter/OrcFilterConverterTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/orc/filter/OrcFilterConverterTest.java
index 28f2dfac75..a32a84304a 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/orc/filter/OrcFilterConverterTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/orc/filter/OrcFilterConverterTest.java
@@ -75,10 +75,10 @@ public class OrcFilterConverterTest {
test(
builder.in(0, Arrays.asList(1L, 2L, 3L)),
new OrcFilters.Or(
+ new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 1),
new OrcFilters.Or(
- new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 1),
- new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 2)),
- new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 3)),
+ new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 2),
+ new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 3))),
true);
test(
@@ -92,14 +92,14 @@ public class OrcFilterConverterTest {
test(
builder.notIn(0, Arrays.asList(1L, 2L, 3L)),
new OrcFilters.And(
+ new OrcFilters.Not(
+ new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 1)),
new OrcFilters.And(
new OrcFilters.Not(
- new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 1)),
+ new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 2)),
new OrcFilters.Not(
new OrcFilters.Equals(
- "long1",
PredicateLeaf.Type.LONG, 2))),
- new OrcFilters.Not(
- new OrcFilters.Equals("long1",
PredicateLeaf.Type.LONG, 3))),
+ "long1",
PredicateLeaf.Type.LONG, 3)))),
true);
assertThat(
@@ -186,29 +186,29 @@ public class OrcFilterConverterTest {
Collections.singletonList(
new DataField(0, "testField", new
BigIntType()))));
- // Test IN with multiple values (≤20 values should be converted to OR
of EQUALS)
+ // Test IN with multiple values
test(
builder.in(0, Arrays.asList(1L, 2L, 3L)),
new OrcFilters.Or(
+ new OrcFilters.Equals("testField",
PredicateLeaf.Type.LONG, 1L),
new OrcFilters.Or(
- new OrcFilters.Equals("testField",
PredicateLeaf.Type.LONG, 1L),
- new OrcFilters.Equals("testField",
PredicateLeaf.Type.LONG, 2L)),
- new OrcFilters.Equals("testField",
PredicateLeaf.Type.LONG, 3L)),
+ new OrcFilters.Equals("testField",
PredicateLeaf.Type.LONG, 2L),
+ new OrcFilters.Equals("testField",
PredicateLeaf.Type.LONG, 3L))),
true);
- // Test NOT IN with multiple values (should be converted to AND of NOT
EQUALS)
+ // Test NOT IN with multiple values
test(
builder.notIn(0, Arrays.asList(1L, 2L, 3L)),
new OrcFilters.And(
+ new OrcFilters.Not(
+ new OrcFilters.Equals("testField",
PredicateLeaf.Type.LONG, 1L)),
new OrcFilters.And(
new OrcFilters.Not(
new OrcFilters.Equals(
- "testField",
PredicateLeaf.Type.LONG, 1L)),
+ "testField",
PredicateLeaf.Type.LONG, 2L)),
new OrcFilters.Not(
new OrcFilters.Equals(
- "testField",
PredicateLeaf.Type.LONG, 2L))),
- new OrcFilters.Not(
- new OrcFilters.Equals("testField",
PredicateLeaf.Type.LONG, 3L))),
+ "testField",
PredicateLeaf.Type.LONG, 3L)))),
true);
}
diff --git
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java
index 4fdd1e3927..a10efa2cf2 100644
---
a/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java
+++
b/paimon-format/src/test/java/org/apache/paimon/format/parquet/ParquetFiltersTest.java
@@ -68,7 +68,7 @@ class ParquetFiltersTest {
test(
builder.in(0, Arrays.asList(1L, 2L, 3L)),
- "or(or(eq(long1, 1), eq(long1, 2)), eq(long1, 3))",
+ "or(eq(long1, 1), or(eq(long1, 2), eq(long1, 3)))",
true);
test(builder.between(0, 1L, 3L), "and(gteq(long1, 1), lteq(long1,
3))", true);
@@ -92,7 +92,7 @@ class ParquetFiltersTest {
test(
builder.in(0, Arrays.asList("1", "2", "3")),
- "or(or(eq(string1, Binary{\"1\"}), eq(string1,
Binary{\"2\"})), eq(string1, Binary{\"3\"}))",
+ "or(eq(string1, Binary{\"1\"}), or(eq(string1, Binary{\"2\"}),
eq(string1, Binary{\"3\"})))",
true);
test(
builder.notIn(0, Arrays.asList("1", "2", "3")),
@@ -380,16 +380,15 @@ class ParquetFiltersTest {
Decimal v2 = Decimal.fromBigDecimal(new BigDecimal("200.00"),
precision, scale);
Decimal v3 = Decimal.fromBigDecimal(new BigDecimal("300.00"),
precision, scale);
- // For less than 21 elements, it expands to or(eq, eq, eq)
test(
builder.in(0, Arrays.asList(v1, v2, v3)),
- "or(or(eq(decimal1, "
+ "or(eq(decimal1, "
+ (int) v1.toUnscaledLong()
- + "), eq(decimal1, "
+ + "), or(eq(decimal1, "
+ (int) v2.toUnscaledLong()
- + ")), eq(decimal1, "
+ + "), eq(decimal1, "
+ (int) v3.toUnscaledLong()
- + "))",
+ + ")))",
true);
test(
@@ -421,16 +420,15 @@ class ParquetFiltersTest {
Decimal v2 = Decimal.fromBigDecimal(new
BigDecimal("20000000000.0000"), precision, scale);
Decimal v3 = Decimal.fromBigDecimal(new
BigDecimal("30000000000.0000"), precision, scale);
- // For less than 21 elements, it expands to or(eq, eq, eq)
test(
builder.in(0, Arrays.asList(v1, v2, v3)),
- "or(or(eq(decimal1, "
+ "or(eq(decimal1, "
+ v1.toUnscaledLong()
- + "), eq(decimal1, "
+ + "), or(eq(decimal1, "
+ v2.toUnscaledLong()
- + ")), eq(decimal1, "
+ + "), eq(decimal1, "
+ v3.toUnscaledLong()
- + "))",
+ + ")))",
true);
test(
@@ -554,13 +552,13 @@ class ParquetFiltersTest {
test(
builder.in(0, Arrays.asList(v1, v2, v3)),
- "or(or(eq(ts1, "
+ "or(eq(ts1, "
+ v1.getMillisecond()
- + "), eq(ts1, "
+ + "), or(eq(ts1, "
+ v2.getMillisecond()
- + ")), eq(ts1, "
+ + "), eq(ts1, "
+ v3.getMillisecond()
- + "))",
+ + ")))",
true);
test(
@@ -594,13 +592,13 @@ class ParquetFiltersTest {
test(
builder.in(0, Arrays.asList(v1, v2, v3)),
- "or(or(eq(ts1, "
+ "or(eq(ts1, "
+ micros1
- + "), eq(ts1, "
+ + "), or(eq(ts1, "
+ micros2
- + ")), eq(ts1, "
+ + "), eq(ts1, "
+ micros3
- + "))",
+ + ")))",
true);
test(
