This is an automated email from the ASF dual-hosted git repository.
lzljs3620320 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6d15bd3ea [core] Remove Predicate.test(Object[]) to use InternalRow
(#2796)
6d15bd3ea is described below
commit 6d15bd3ea0de105503f5faa03b57caa5afcc1c23
Author: Jingsong Lee <[email protected]>
AuthorDate: Fri Jan 26 16:45:02 2024 +0800
[core] Remove Predicate.test(Object[]) to use InternalRow (#2796)
---
.../main/java/org/apache/paimon/predicate/And.java | 10 --
.../apache/paimon/predicate/CompoundPredicate.java | 7 -
.../org/apache/paimon/predicate/LeafPredicate.java | 5 -
.../main/java/org/apache/paimon/predicate/Or.java | 10 --
.../org/apache/paimon/predicate/Predicate.java | 7 -
.../apache/paimon/predicate/PredicateFilter.java | 49 -------
.../paimon/predicate/PredicateBuilderTest.java | 21 +--
.../org/apache/paimon/predicate/PredicateTest.java | 157 +++++++++++----------
.../paimon/operation/AbstractFileStoreScan.java | 2 +-
.../paimon/operation/FileStoreCommitImpl.java | 8 +-
.../paimon/partition/PartitionPredicate.java | 12 +-
.../java/org/apache/paimon/table/TableUtils.java | 4 +-
.../paimon/operation/FileStoreCommitTest.java | 5 +-
.../paimon/partition/PartitionPredicateTest.java | 2 +-
.../paimon/schema/SchemaEvolutionUtilTest.java | 4 +-
.../flink/lookup/FileStoreLookupFunction.java | 10 +-
.../paimon/flink/lookup/FullCacheLookupTable.java | 23 +--
.../flink/lookup/NoPrimaryKeyLookupTable.java | 4 +-
.../paimon/flink/lookup/PrimaryKeyLookupTable.java | 4 +-
.../flink/lookup/SecondaryIndexLookupTable.java | 4 +-
.../paimon/flink/lookup/TableStreamingReader.java | 15 +-
.../paimon/flink/PredicateConverterTest.java | 3 +-
.../paimon/flink/lookup/LookupTableTest.java | 27 ++--
23 files changed, 150 insertions(+), 243 deletions(-)
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/And.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/And.java
index 98e53947d..c4ec83525 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/And.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/And.java
@@ -34,16 +34,6 @@ public class And extends CompoundPredicate.Function {
private And() {}
- @Override
- public boolean test(Object[] values, List<Predicate> children) {
- for (Predicate child : children) {
- if (!child.test(values)) {
- return false;
- }
- }
- return true;
- }
-
@Override
public boolean test(InternalRow row, List<Predicate> children) {
for (Predicate child : children) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/CompoundPredicate.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/CompoundPredicate.java
index a162d491a..b8540722f 100644
---
a/paimon-common/src/main/java/org/apache/paimon/predicate/CompoundPredicate.java
+++
b/paimon-common/src/main/java/org/apache/paimon/predicate/CompoundPredicate.java
@@ -48,11 +48,6 @@ public class CompoundPredicate implements Predicate {
return children;
}
- @Override
- public boolean test(Object[] values) {
- return function.test(values, children);
- }
-
@Override
public boolean test(InternalRow row) {
return function.test(row, children);
@@ -95,8 +90,6 @@ public class CompoundPredicate implements Predicate {
/** Evaluate the predicate result based on multiple {@link Predicate}s. */
public abstract static class Function implements Serializable {
- public abstract boolean test(Object[] values, List<Predicate>
children);
-
public abstract boolean test(InternalRow row, List<Predicate>
children);
public abstract boolean test(
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java
index ed291a15e..d2a2461ee 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/LeafPredicate.java
@@ -88,11 +88,6 @@ public class LeafPredicate implements Predicate {
return new LeafPredicate(function, type, fieldIndex, fieldName,
literals);
}
- @Override
- public boolean test(Object[] values) {
- return function.test(type, values[fieldIndex], literals);
- }
-
@Override
public boolean test(InternalRow row) {
return function.test(type, InternalRowUtils.get(row, fieldIndex,
type), literals);
diff --git a/paimon-common/src/main/java/org/apache/paimon/predicate/Or.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/Or.java
index e5beb0ddf..c145878a2 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/Or.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/Or.java
@@ -34,16 +34,6 @@ public class Or extends CompoundPredicate.Function {
private Or() {}
- @Override
- public boolean test(Object[] values, List<Predicate> children) {
- for (Predicate child : children) {
- if (child.test(values)) {
- return true;
- }
- }
- return false;
- }
-
@Override
public boolean test(InternalRow row, List<Predicate> children) {
for (Predicate child : children) {
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/Predicate.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/Predicate.java
index 3c9d04b18..c3d0cd8eb 100644
--- a/paimon-common/src/main/java/org/apache/paimon/predicate/Predicate.java
+++ b/paimon-common/src/main/java/org/apache/paimon/predicate/Predicate.java
@@ -34,13 +34,6 @@ import java.util.Optional;
@Public
public interface Predicate extends Serializable {
- /**
- * Test based on the specific input column values.
- *
- * @return return true when hit, false when not hit.
- */
- boolean test(Object[] values);
-
/**
* Test based on the specific input row.
*
diff --git
a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateFilter.java
b/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateFilter.java
deleted file mode 100644
index f88bc1445..000000000
---
a/paimon-common/src/main/java/org/apache/paimon/predicate/PredicateFilter.java
+++ /dev/null
@@ -1,49 +0,0 @@
-/*
- * 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.paimon.predicate;
-
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.Filter;
-import org.apache.paimon.utils.RowDataToObjectArrayConverter;
-
-import javax.annotation.Nullable;
-
-import java.util.List;
-
-/** A {@link Filter} to filter {@link InternalRow}. */
-public class PredicateFilter implements Filter<InternalRow> {
-
- private final RowDataToObjectArrayConverter arrayConverter;
- @Nullable private final Predicate predicate;
-
- public PredicateFilter(RowType rowType, List<Predicate> predicates) {
- this(rowType, predicates.isEmpty() ? null :
PredicateBuilder.and(predicates));
- }
-
- public PredicateFilter(RowType rowType, @Nullable Predicate predicate) {
- this.arrayConverter = new RowDataToObjectArrayConverter(rowType);
- this.predicate = predicate;
- }
-
- @Override
- public boolean test(InternalRow rowData) {
- return predicate == null ||
predicate.test(arrayConverter.convert(rowData));
- }
-}
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 3f7d821ed..b88240333 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
@@ -18,6 +18,7 @@
package org.apache.paimon.predicate;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.format.FieldStats;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
@@ -36,11 +37,11 @@ public class PredicateBuilderTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.between(0, 1, 3);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(true);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(2, 5,
0L)})).isEqualTo(true);
@@ -55,11 +56,11 @@ public class PredicateBuilderTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.between(0, 1, null);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(2, 5,
0L)})).isEqualTo(false);
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 58cc61dd8..ceac81b38 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
@@ -18,6 +18,7 @@
package org.apache.paimon.predicate;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.format.FieldStats;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowType;
@@ -38,9 +39,9 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.equal(0, 5);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {5})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(5))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(true);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 6,
0L)})).isEqualTo(true);
@@ -56,8 +57,8 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.equal(0, null);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(false);
assertThat(predicate.test(1, new FieldStats[] {new FieldStats(null,
null, 1L)}))
@@ -69,9 +70,9 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.notEqual(0, 5);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {5})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(5))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(true);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 6,
0L)})).isEqualTo(true);
@@ -88,8 +89,8 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.notEqual(0, null);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(false);
assertThat(predicate.test(1, new FieldStats[] {new FieldStats(null,
null, 1L)}))
@@ -101,10 +102,10 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.greaterThan(0, 5);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {5})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {6})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(5))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(6))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 4,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(false);
@@ -121,8 +122,8 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.greaterThan(0, null);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 4,
0L)})).isEqualTo(false);
assertThat(predicate.test(1, new FieldStats[] {new FieldStats(null,
null, 1L)}))
@@ -134,10 +135,10 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.greaterOrEqual(0, 5);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {5})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {6})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(5))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(6))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 4,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(true);
@@ -154,8 +155,8 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.greaterOrEqual(0, null);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 4,
0L)})).isEqualTo(false);
assertThat(predicate.test(1, new FieldStats[] {new FieldStats(null,
null, 1L)}))
@@ -167,10 +168,10 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.lessThan(0, 5);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {5})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {6})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(5))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(6))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(5, 7,
0L)})).isEqualTo(false);
@@ -186,8 +187,8 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.lessThan(0, null);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(false);
assertThat(predicate.test(1, new FieldStats[] {new FieldStats(null,
null, 1L)}))
@@ -199,10 +200,10 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.lessOrEqual(0, 5);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {5})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {6})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(5))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(6))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(5, 7,
0L)})).isEqualTo(true);
@@ -218,8 +219,8 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.lessOrEqual(0, null);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(false);
assertThat(predicate.test(1, new FieldStats[] {new FieldStats(null,
null, 1L)}))
@@ -231,8 +232,8 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.isNull(0);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(true);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(5, 7,
1L)})).isEqualTo(true);
@@ -245,8 +246,8 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType()));
Predicate predicate = builder.isNotNull(0);
- assertThat(predicate.test(new Object[] {4})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(true);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(5, 7,
1L)})).isEqualTo(true);
@@ -262,10 +263,10 @@ public class PredicateTest {
Predicate predicate = builder.in(0, Arrays.asList(1, 3));
assertThat(predicate).isInstanceOf(CompoundPredicate.class);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(true);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(false);
@@ -279,10 +280,10 @@ public class PredicateTest {
Predicate predicate = builder.in(0, Arrays.asList(1, null, 3));
assertThat(predicate).isInstanceOf(CompoundPredicate.class);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(true);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(false);
@@ -296,10 +297,10 @@ public class PredicateTest {
Predicate predicate = builder.notIn(0, Arrays.asList(1, 3));
assertThat(predicate).isInstanceOf(CompoundPredicate.class);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(1, 1,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(3, 3,
0L)})).isEqualTo(false);
@@ -316,10 +317,10 @@ public class PredicateTest {
Predicate predicate = builder.notIn(0, Arrays.asList(1, null, 3));
assertThat(predicate).isInstanceOf(CompoundPredicate.class);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(1, 1,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(3, 3,
0L)})).isEqualTo(false);
@@ -342,10 +343,10 @@ public class PredicateTest {
Predicate predicate = builder.in(0, literals);
assertThat(predicate).isInstanceOf(LeafPredicate.class);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(true);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(false);
@@ -368,10 +369,10 @@ public class PredicateTest {
Predicate predicate = builder.in(0, literals);
assertThat(predicate).isInstanceOf(LeafPredicate.class);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(0, 5,
0L)})).isEqualTo(true);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0L)})).isEqualTo(false);
@@ -393,10 +394,10 @@ public class PredicateTest {
Predicate predicate = builder.notIn(0, literals);
assertThat(predicate).isInstanceOf(LeafPredicate.class);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(1, 1,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(3, 3,
0L)})).isEqualTo(false);
@@ -422,10 +423,10 @@ public class PredicateTest {
Predicate predicate = builder.notIn(0, literals);
assertThat(predicate).isInstanceOf(LeafPredicate.class);
- assertThat(predicate.test(new Object[] {1})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {2})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {null})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(1))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(2))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of((Object)
null))).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(1, 1,
0L)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(3, 3,
0L)})).isEqualTo(false);
@@ -443,10 +444,10 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType(), new IntType()));
Predicate predicate = PredicateBuilder.and(builder.equal(0, 3),
builder.equal(1, 5));
- assertThat(predicate.test(new Object[] {4, 5})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3, 6})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3, 5})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null, 5})).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(4, 5))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3, 6))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3, 5))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(null, 5))).isEqualTo(false);
assertThat(
predicate.test(
@@ -479,10 +480,10 @@ public class PredicateTest {
PredicateBuilder builder = new PredicateBuilder(RowType.of(new
IntType(), new IntType()));
Predicate predicate = PredicateBuilder.or(builder.equal(0, 3),
builder.equal(1, 5));
- assertThat(predicate.test(new Object[] {4, 6})).isEqualTo(false);
- assertThat(predicate.test(new Object[] {3, 6})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {3, 5})).isEqualTo(true);
- assertThat(predicate.test(new Object[] {null, 5})).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(4, 6))).isEqualTo(false);
+ assertThat(predicate.test(GenericRow.of(3, 6))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(3, 5))).isEqualTo(true);
+ assertThat(predicate.test(GenericRow.of(null, 5))).isEqualTo(true);
assertThat(
predicate.test(
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
index 766e17417..94a383f95 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/AbstractFileStoreScan.java
@@ -112,7 +112,7 @@ public abstract class AbstractFileStoreScan implements
FileStoreScan {
@Override
public FileStoreScan withPartitionFilter(Predicate predicate) {
if (partitionType.getFieldCount() > 0 && predicate != null) {
- this.partitionFilter =
PartitionPredicate.fromPredicate(partitionType, predicate);
+ this.partitionFilter = PartitionPredicate.fromPredicate(predicate);
} else {
this.partitionFilter = null;
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
index 1962edcff..044eb7f6e 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/FileStoreCommitImpl.java
@@ -392,8 +392,7 @@ public class FileStoreCommitImpl implements FileStoreCommit
{
// sanity check, all changes must be done within the given
partition
if (partitionFilter != null) {
for (ManifestEntry entry : appendTableFiles) {
- if (!partitionFilter.test(
-
partitionObjectConverter.convert(entry.partition()))) {
+ if (!partitionFilter.test(entry.partition())) {
throw new IllegalArgumentException(
"Trying to overwrite partition "
+ partition
@@ -651,13 +650,10 @@ public class FileStoreCommitImpl implements
FileStoreCommit {
// collect index files
if (latestSnapshot.indexManifest() != null) {
- RowDataToObjectArrayConverter converter =
- new RowDataToObjectArrayConverter(partitionType);
List<IndexManifestEntry> entries =
indexManifestFile.read(latestSnapshot.indexManifest());
for (IndexManifestEntry entry : entries) {
- if (partitionFilter == null
- ||
partitionFilter.test(converter.convert(entry.partition()))) {
+ if (partitionFilter == null ||
partitionFilter.test(entry.partition())) {
indexChangesWithOverwrite.add(entry.toDeleteEntry());
}
}
diff --git
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
index 0a9b5df97..2748ef40b 100644
---
a/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
+++
b/paimon-core/src/main/java/org/apache/paimon/partition/PartitionPredicate.java
@@ -39,9 +39,8 @@ public interface PartitionPredicate {
boolean test(long rowCount, FieldStats[] fieldStats);
- static PartitionPredicate fromPredicate(RowType partitionType, Predicate
predicate) {
- return new DefaultPartitionPredicate(
- new RowDataToObjectArrayConverter(partitionType), predicate);
+ static PartitionPredicate fromPredicate(Predicate predicate) {
+ return new DefaultPartitionPredicate(predicate);
}
static PartitionPredicate fromMultiple(RowType partitionType,
List<BinaryRow> partitions) {
@@ -52,18 +51,15 @@ public interface PartitionPredicate {
/** A {@link PartitionPredicate} using {@link Predicate}. */
class DefaultPartitionPredicate implements PartitionPredicate {
- private final RowDataToObjectArrayConverter converter;
private final Predicate predicate;
- private DefaultPartitionPredicate(
- RowDataToObjectArrayConverter converter, Predicate predicate) {
- this.converter = converter;
+ private DefaultPartitionPredicate(Predicate predicate) {
this.predicate = predicate;
}
@Override
public boolean test(BinaryRow part) {
- return predicate.test(converter.convert(part));
+ return predicate.test(part);
}
@Override
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/TableUtils.java
b/paimon-core/src/main/java/org/apache/paimon/table/TableUtils.java
index 91460c1bd..9b9db719a 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/TableUtils.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/TableUtils.java
@@ -22,7 +22,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.disk.IOManagerImpl;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateFilter;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.sink.BatchTableCommit;
import org.apache.paimon.table.sink.BatchTableWrite;
@@ -58,7 +58,7 @@ public class TableUtils {
BatchTableCommit commit = writeBuilder.newCommit()) {
write.withIOManager(ioManager);
CloseableIterator<InternalRow> iterator =
reader.toCloseableIterator();
- PredicateFilter filter = new PredicateFilter(table.rowType(),
filters);
+ Predicate filter = PredicateBuilder.and(filters);
while (iterator.hasNext()) {
InternalRow row = iterator.next();
if (filter.test(row)) {
diff --git
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
index 40526b156..8d18abc16 100644
---
a/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/operation/FileStoreCommitTest.java
@@ -44,7 +44,6 @@ import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.FailingFileIO;
-import org.apache.paimon.utils.RowDataToObjectArrayConverter;
import org.apache.paimon.utils.SnapshotManager;
import org.apache.paimon.utils.TraceableFileIO;
@@ -650,8 +649,6 @@ public class FileStoreCommitTest {
assertThat(snapshot.commitKind()).isEqualTo(Snapshot.CommitKind.OVERWRITE);
// check data
- RowDataToObjectArrayConverter partitionConverter =
- new
RowDataToObjectArrayConverter(TestKeyValueGenerator.DEFAULT_PART_TYPE);
org.apache.paimon.predicate.Predicate partitionFilter =
partitions.stream()
.map(
@@ -663,7 +660,7 @@ public class FileStoreCommitTest {
List<KeyValue> expectedKvs = new ArrayList<>();
for (Map.Entry<BinaryRow, List<KeyValue>> entry : data.entrySet()) {
- if
(partitionFilter.test(partitionConverter.convert(entry.getKey()))) {
+ if (partitionFilter.test(entry.getKey())) {
continue;
}
expectedKvs.addAll(entry.getValue());
diff --git
a/paimon-core/src/test/java/org/apache/paimon/partition/PartitionPredicateTest.java
b/paimon-core/src/test/java/org/apache/paimon/partition/PartitionPredicateTest.java
index 78ee1377f..68a1a0265 100644
---
a/paimon-core/src/test/java/org/apache/paimon/partition/PartitionPredicateTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/partition/PartitionPredicateTest.java
@@ -57,7 +57,7 @@ public class PartitionPredicateTest {
and(builder.equal(0, 3), builder.equal(1, 5)),
and(builder.equal(0, 4), builder.equal(1, 6)));
- PartitionPredicate p1 = PartitionPredicate.fromPredicate(type,
predicate);
+ PartitionPredicate p1 = PartitionPredicate.fromPredicate(predicate);
PartitionPredicate p2 =
PartitionPredicate.fromMultiple(
type, Arrays.asList(createPart(3, 5), createPart(4,
6)));
diff --git
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaEvolutionUtilTest.java
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaEvolutionUtilTest.java
index 532fb0063..9d947f76d 100644
---
a/paimon-core/src/test/java/org/apache/paimon/schema/SchemaEvolutionUtilTest.java
+++
b/paimon-core/src/test/java/org/apache/paimon/schema/SchemaEvolutionUtilTest.java
@@ -306,8 +306,8 @@ public class SchemaEvolutionUtilTest {
LeafPredicate child = (LeafPredicate) filters.get(0);
// Validate value 1 with index 1
- assertThat(child.test(new Integer[] {0, 1})).isTrue();
+ assertThat(child.test(GenericRow.of(0, 1))).isTrue();
// Validate value 2 with index 1
- assertThat(child.test(new Integer[] {1, 2})).isFalse();
+ assertThat(child.test(GenericRow.of(1, 2))).isFalse();
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
index 5685cd1cd..77ea719a2 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FileStoreLookupFunction.java
@@ -26,12 +26,10 @@ import org.apache.paimon.flink.FlinkRowWrapper;
import org.apache.paimon.flink.utils.TableScanUtils;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateFilter;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.OutOfRangeException;
import org.apache.paimon.utils.FileIOUtils;
-import org.apache.paimon.utils.TypeUtils;
import org.apache.paimon.shade.guava30.com.google.common.primitives.Ints;
@@ -161,8 +159,8 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
storeTable,
projection,
predicate,
+ createProjectedPredicate(projection),
path,
- createRecordFilter(projection),
joinKeys);
this.lookupTable = FullCacheLookupTable.create(context,
options.get(LOOKUP_CACHE_ROWS));
}
@@ -170,7 +168,8 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
lookupTable.open();
}
- private PredicateFilter createRecordFilter(int[] projection) {
+ @Nullable
+ private Predicate createProjectedPredicate(int[] projection) {
Predicate adjustedPredicate = null;
if (predicate != null) {
// adjust to projection index
@@ -182,8 +181,7 @@ public class FileStoreLookupFunction implements
Serializable, Closeable {
.toArray())
.orElse(null);
}
- return new PredicateFilter(
- TypeUtils.project(table.rowType(), projection),
adjustedPredicate);
+ return adjustedPredicate;
}
public Collection<RowData> lookup(RowData keyRow) {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
index fb6e2e3b2..d0f817f79 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/FullCacheLookupTable.java
@@ -34,7 +34,6 @@ import org.apache.paimon.sort.BinaryExternalSortBuffer;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.RowType;
-import org.apache.paimon.utils.Filter;
import org.apache.paimon.utils.MutableObjectIterator;
import org.apache.paimon.utils.PartialRow;
import org.apache.paimon.utils.TypeUtils;
@@ -67,7 +66,7 @@ public abstract class FullCacheLookupTable implements
LookupTable {
context.table.coreOptions().toConfiguration(),
null);
FileStoreTable table = context.table;
- this.reader = new TableStreamingReader(table, context.projection,
context.predicate);
+ this.reader = new TableStreamingReader(table, context.projection,
context.tablePredicate);
this.sequenceFieldEnabled =
table.primaryKeys().size() > 0
&& new
CoreOptions(table.options()).sequenceField().isPresent();
@@ -83,11 +82,12 @@ public abstract class FullCacheLookupTable implements
LookupTable {
BinaryExternalSortBuffer bulkLoadSorter =
RocksDBState.createBulkLoadSorter(
IOManager.create(context.tempPath.toString()),
context.table.coreOptions());
+ Predicate predicate = projectedPredicate();
try (RecordReaderIterator<InternalRow> batch =
new RecordReaderIterator<>(reader.nextBatch(true,
sequenceFieldEnabled))) {
while (batch.hasNext()) {
InternalRow row = batch.next();
- if (recordFilter().test(row)) {
+ if (predicate == null || predicate.test(row)) {
bulkLoadSorter.write(GenericRow.of(toKeyBytes(row),
toValueBytes(row)));
}
}
@@ -143,8 +143,9 @@ public abstract class FullCacheLookupTable implements
LookupTable {
public abstract void refresh(Iterator<InternalRow> input, boolean
orderByLastField)
throws IOException;
- public Filter<InternalRow> recordFilter() {
- return context.recordFilter;
+ @Nullable
+ public Predicate projectedPredicate() {
+ return context.projectedPredicate;
}
public abstract byte[] toKeyBytes(InternalRow row) throws IOException;
@@ -184,23 +185,23 @@ public abstract class FullCacheLookupTable implements
LookupTable {
public final FileStoreTable table;
public final int[] projection;
- public final @Nullable Predicate predicate;
+ @Nullable public final Predicate tablePredicate;
+ @Nullable public final Predicate projectedPredicate;
public final File tempPath;
- public final Filter<InternalRow> recordFilter;
public final List<String> joinKey;
public Context(
FileStoreTable table,
int[] projection,
- @Nullable Predicate predicate,
+ @Nullable Predicate tablePredicate,
+ @Nullable Predicate projectedPredicate,
File tempPath,
- Filter<InternalRow> recordFilter,
List<String> joinKey) {
this.table = table;
this.projection = projection;
- this.predicate = predicate;
+ this.tablePredicate = tablePredicate;
+ this.projectedPredicate = projectedPredicate;
this.tempPath = tempPath;
- this.recordFilter = recordFilter;
this.joinKey = joinKey;
}
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
index 329cd0d6a..9acb39567 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/NoPrimaryKeyLookupTable.java
@@ -24,6 +24,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalSerializers;
import org.apache.paimon.lookup.BulkLoader;
import org.apache.paimon.lookup.RocksDBListState;
+import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.KeyProjectedRow;
import org.apache.paimon.utils.TypeUtils;
@@ -68,11 +69,12 @@ public class NoPrimaryKeyLookupTable extends
FullCacheLookupTable {
"Append table does not support order by last field.");
}
+ Predicate predicate = projectedPredicate();
while (incremental.hasNext()) {
InternalRow row = incremental.next();
joinKeyRow.replaceRow(row);
if (row.getRowKind() == RowKind.INSERT || row.getRowKind() ==
RowKind.UPDATE_AFTER) {
- if (recordFilter().test(row)) {
+ if (predicate == null || predicate.test(row)) {
state.add(joinKeyRow, row);
}
} else {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
index 8757effa3..fbec4d443 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/PrimaryKeyLookupTable.java
@@ -22,6 +22,7 @@ import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalSerializers;
import org.apache.paimon.lookup.BulkLoader;
import org.apache.paimon.lookup.RocksDBValueState;
+import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.KeyProjectedRow;
@@ -87,6 +88,7 @@ public class PrimaryKeyLookupTable extends
FullCacheLookupTable {
@Override
public void refresh(Iterator<InternalRow> incremental, boolean
orderByLastField)
throws IOException {
+ Predicate predicate = projectedPredicate();
while (incremental.hasNext()) {
InternalRow row = incremental.next();
primaryKeyRow.replaceRow(row);
@@ -99,7 +101,7 @@ public class PrimaryKeyLookupTable extends
FullCacheLookupTable {
}
if (row.getRowKind() == RowKind.INSERT || row.getRowKind() ==
RowKind.UPDATE_AFTER) {
- if (recordFilter().test(row)) {
+ if (predicate == null || predicate.test(row)) {
tableState.put(primaryKeyRow, row);
} else {
// The new record under primary key is filtered
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
index 2f973bfd5..d95481a32 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/SecondaryIndexLookupTable.java
@@ -21,6 +21,7 @@ package org.apache.paimon.flink.lookup;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.data.serializer.InternalSerializers;
import org.apache.paimon.lookup.RocksDBSetState;
+import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.utils.KeyProjectedRow;
import org.apache.paimon.utils.TypeUtils;
@@ -67,6 +68,7 @@ public class SecondaryIndexLookupTable extends
PrimaryKeyLookupTable {
@Override
public void refresh(Iterator<InternalRow> incremental, boolean
orderByLastField)
throws IOException {
+ Predicate predicate = projectedPredicate();
while (incremental.hasNext()) {
InternalRow row = incremental.next();
primaryKeyRow.replaceRow(row);
@@ -90,7 +92,7 @@ public class SecondaryIndexLookupTable extends
PrimaryKeyLookupTable {
indexState.retract(secKeyRow.replaceRow(previous),
primaryKeyRow);
}
- if (recordFilter().test(row)) {
+ if (predicate == null || predicate.test(row)) {
tableState.put(primaryKeyRow, row);
indexState.add(secKeyRow.replaceRow(row), primaryKeyRow);
} else {
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/TableStreamingReader.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/TableStreamingReader.java
index 7ff1493cc..c96eab450 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/TableStreamingReader.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/lookup/TableStreamingReader.java
@@ -26,7 +26,6 @@ import org.apache.paimon.io.SplitsParallelReadUtil;
import org.apache.paimon.mergetree.compact.ConcatRecordReader;
import org.apache.paimon.options.Options;
import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.predicate.PredicateFilter;
import org.apache.paimon.reader.RecordReader;
import org.apache.paimon.table.Table;
import org.apache.paimon.table.source.KeyValueTableRead;
@@ -59,7 +58,7 @@ public class TableStreamingReader {
private final Table table;
private final int[] projection;
private final ReadBuilder readBuilder;
- @Nullable private final PredicateFilter recordFilter;
+ @Nullable private final Predicate projectedPredicate;
private final StreamTableScan scan;
public TableStreamingReader(Table table, int[] projection, @Nullable
Predicate predicate) {
@@ -94,12 +93,10 @@ public class TableStreamingReader {
int[] fieldIdxToProjectionIdx =
IntStream.range(0,
table.rowType().getFieldCount()).map(operator).toArray();
- this.recordFilter =
- new PredicateFilter(
- TypeUtils.project(table.rowType(), projection),
- transformFieldMapping(predicate,
fieldIdxToProjectionIdx).orElse(null));
+ this.projectedPredicate =
+ transformFieldMapping(predicate,
fieldIdxToProjectionIdx).orElse(null);
} else {
- recordFilter = null;
+ this.projectedPredicate = null;
}
}
@@ -134,8 +131,8 @@ public class TableStreamingReader {
reader = ConcatRecordReader.create(readers);
}
- if (recordFilter != null) {
- reader = reader.filter(recordFilter::test);
+ if (projectedPredicate != null) {
+ reader = reader.filter(projectedPredicate::test);
}
return reader;
}
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 66d757b09..e5df44849 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
@@ -19,6 +19,7 @@
package org.apache.paimon.flink;
import org.apache.paimon.data.BinaryString;
+import org.apache.paimon.data.GenericRow;
import org.apache.paimon.format.FieldStats;
import org.apache.paimon.predicate.Predicate;
import org.apache.paimon.predicate.PredicateBuilder;
@@ -259,7 +260,7 @@ public class PredicateConverterTest {
IntStream.range(0, valuesList.size())
.forEach(
i ->
- assertThat(predicate.test(valuesList.get(i)))
+
assertThat(predicate.test(GenericRow.of(valuesList.get(i))))
.isEqualTo(expectedForValues.get(i)));
IntStream.range(0, rowCountList.size())
.forEach(
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
index 0e5d0792b..9af88502a 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/lookup/LookupTableTest.java
@@ -29,6 +29,7 @@ import
org.apache.paimon.flink.lookup.FullCacheLookupTable.TableBulkLoader;
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.local.LocalFileIO;
import org.apache.paimon.options.Options;
+import org.apache.paimon.predicate.PredicateBuilder;
import org.apache.paimon.schema.Schema;
import org.apache.paimon.schema.SchemaManager;
import org.apache.paimon.schema.SchemaUtils;
@@ -36,7 +37,6 @@ import org.apache.paimon.schema.TableSchema;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.FileStoreTableFactory;
import org.apache.paimon.table.TableTestBase;
-import org.apache.paimon.types.DataTypes;
import org.apache.paimon.types.IntType;
import org.apache.paimon.types.RowKind;
import org.apache.paimon.types.RowType;
@@ -63,6 +63,7 @@ import java.util.concurrent.ThreadLocalRandom;
import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
+import static org.apache.paimon.types.DataTypes.INT;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -111,8 +112,8 @@ public class LookupTableTest extends TableTestBase {
storeTable,
new int[] {0, 1, 2},
null,
+ null,
tempDir.toFile(),
- r -> true,
singletonList("f0"));
table = FullCacheLookupTable.create(context,
ThreadLocalRandom.current().nextInt(2) * 10);
@@ -169,8 +170,8 @@ public class LookupTableTest extends TableTestBase {
storeTable,
new int[] {0, 1, 2},
null,
+ null,
tempDir.toFile(),
- r -> true,
singletonList("f0"));
table = FullCacheLookupTable.create(context,
ThreadLocalRandom.current().nextInt(2) * 10);
@@ -213,8 +214,8 @@ public class LookupTableTest extends TableTestBase {
storeTable,
new int[] {0, 1, 2},
null,
+ new PredicateBuilder(RowType.of(INT())).lessThan(0, 3),
tempDir.toFile(),
- r -> r.getInt(0) < 3,
singletonList("f0"));
table = FullCacheLookupTable.create(context,
ThreadLocalRandom.current().nextInt(2) * 10);
@@ -244,8 +245,8 @@ public class LookupTableTest extends TableTestBase {
storeTable,
new int[] {0, 1, 2},
null,
+ new PredicateBuilder(RowType.of(INT(),
INT())).lessThan(1, 22),
tempDir.toFile(),
- r -> r.getInt(1) < 22,
singletonList("f0"));
table = FullCacheLookupTable.create(context,
ThreadLocalRandom.current().nextInt(2) * 10);
@@ -267,8 +268,8 @@ public class LookupTableTest extends TableTestBase {
storeTable,
new int[] {0, 1, 2},
null,
+ null,
tempDir.toFile(),
- r -> true,
singletonList("f1"));
table = FullCacheLookupTable.create(context,
ThreadLocalRandom.current().nextInt(2) * 10);
@@ -311,8 +312,8 @@ public class LookupTableTest extends TableTestBase {
storeTable,
new int[] {0, 1, 2},
null,
+ null,
tempDir.toFile(),
- r -> true,
singletonList("f1"));
table = FullCacheLookupTable.create(context,
ThreadLocalRandom.current().nextInt(2) * 10);
@@ -365,8 +366,8 @@ public class LookupTableTest extends TableTestBase {
storeTable,
new int[] {0, 1, 2},
null,
+ new PredicateBuilder(RowType.of(INT())).lessThan(0, 3),
tempDir.toFile(),
- r -> r.getInt(0) < 3,
singletonList("f1"));
table = FullCacheLookupTable.create(context,
ThreadLocalRandom.current().nextInt(2) * 10);
@@ -405,8 +406,8 @@ public class LookupTableTest extends TableTestBase {
storeTable,
new int[] {0, 1, 2},
null,
+ null,
tempDir.toFile(),
- r -> true,
singletonList("f1"));
table = FullCacheLookupTable.create(context,
ThreadLocalRandom.current().nextInt(2) * 10);
@@ -447,8 +448,8 @@ public class LookupTableTest extends TableTestBase {
storeTable,
new int[] {0, 1, 2},
null,
+ new PredicateBuilder(RowType.of(INT(), INT(),
INT())).lessThan(2, 222),
tempDir.toFile(),
- r -> r.getInt(2) < 222,
singletonList("f1"));
table = FullCacheLookupTable.create(context,
ThreadLocalRandom.current().nextInt(2) * 10);
@@ -555,9 +556,9 @@ public class LookupTableTest extends TableTestBase {
String.format("%s/%s.db/%s", warehouse, database,
"T"));
Schema schema =
Schema.newBuilder()
- .column("pk1", DataTypes.INT())
- .column("pk2", DataTypes.INT())
- .column("col2", DataTypes.INT())
+ .column("pk1", INT())
+ .column("pk2", INT())
+ .column("col2", INT())
.primaryKey("pk1", "pk2")
.option(CoreOptions.BUCKET.key(), "2")
.option(CoreOptions.BUCKET_KEY.key(), "pk2")
