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/flink-table-store.git
The following commit(s) were added to refs/heads/master by this push:
new ff81dcab [FLINK-27835] Introduce LeafPredicate interface and children
method in Predicate
ff81dcab is described below
commit ff81dcab14c2e1345c35b08ed443c1eeb465a298
Author: tsreaper <[email protected]>
AuthorDate: Wed Jun 1 15:04:04 2022 +0800
[FLINK-27835] Introduce LeafPredicate interface and children method in
Predicate
This closes #143
---
.../store/file/operation/FileStoreScanImpl.java | 3 +-
.../flink/table/store/file/predicate/And.java | 65 +++++++--------
.../{GreaterThan.java => CompoundPredicate.java} | 59 ++++++++------
.../flink/table/store/file/predicate/Equal.java | 43 ++--------
.../table/store/file/predicate/GreaterOrEqual.java | 43 ++--------
.../table/store/file/predicate/GreaterThan.java | 43 ++--------
.../table/store/file/predicate/IsNotNull.java | 38 ++-------
.../flink/table/store/file/predicate/IsNull.java | 38 ++-------
.../{GreaterThan.java => LeafPredicate.java} | 61 ++++++++------
.../table/store/file/predicate/LessOrEqual.java | 43 ++--------
.../flink/table/store/file/predicate/LessThan.java | 43 ++--------
.../flink/table/store/file/predicate/NotEqual.java | 43 ++--------
.../flink/table/store/file/predicate/Or.java | 65 +++++++--------
.../store/file/predicate/PredicateBuilder.java | 66 +++++++++++++--
.../store/file/predicate/PredicateConverter.java | 31 ++++---
.../table/store/file/predicate/StartsWith.java | 23 ++----
.../store/file/operation/FileStoreScanTest.java | 8 +-
.../file/predicate/PredicateConverterTest.java | 24 +++---
.../table/store/file/predicate/PredicateTest.java | 44 ++++++----
.../store/SearchArgumentToPredicateConverter.java | 12 +--
.../SearchArgumentToPredicateConverterTest.java | 95 +++++++++-------------
21 files changed, 370 insertions(+), 520 deletions(-)
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScanImpl.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScanImpl.java
index 172b82fa..6599fe1f 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScanImpl.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/operation/FileStoreScanImpl.java
@@ -25,7 +25,6 @@ import
org.apache.flink.table.store.file.manifest.ManifestEntry;
import org.apache.flink.table.store.file.manifest.ManifestFile;
import org.apache.flink.table.store.file.manifest.ManifestFileMeta;
import org.apache.flink.table.store.file.manifest.ManifestList;
-import org.apache.flink.table.store.file.predicate.Equal;
import org.apache.flink.table.store.file.predicate.Literal;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.predicate.PredicateBuilder;
@@ -125,7 +124,7 @@ public class FileStoreScanImpl implements FileStoreScan {
new Literal(
partitionConverter.rowType().getTypeAt(i),
partitionObjects[i]);
- fieldPredicates.add(new Equal(i, l));
+ fieldPredicates.add(PredicateBuilder.equal(i, l));
}
return PredicateBuilder.and(fieldPredicates);
};
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/And.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/And.java
index cc62b10c..313ffe77 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/And.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/And.java
@@ -20,57 +20,50 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Optional;
-/** A {@link Predicate} to eval and. */
-public class And implements Predicate {
+/** A {@link CompoundPredicate.Function} to eval and. */
+public class And implements CompoundPredicate.Function {
private static final long serialVersionUID = 1L;
- private final Predicate predicate1;
- private final Predicate predicate2;
+ public static final And INSTANCE = new And();
- public And(Predicate predicate1, Predicate predicate2) {
- this.predicate1 = predicate1;
- this.predicate2 = predicate2;
- }
+ private And() {}
@Override
- public boolean test(Object[] values) {
- return predicate1.test(values) && predicate2.test(values);
- }
-
- @Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
- return predicate1.test(rowCount, fieldStats) &&
predicate2.test(rowCount, fieldStats);
- }
-
- @Override
- public Optional<Predicate> negate() {
- Optional<Predicate> negate1 = predicate1.negate();
- Optional<Predicate> negate2 = predicate2.negate();
- if (negate1.isPresent() && negate2.isPresent()) {
- return Optional.of(new Or(negate1.get(), negate2.get()));
- } else {
- return Optional.empty();
+ public boolean test(Object[] values, List<Predicate> children) {
+ for (Predicate child : children) {
+ if (!child.test(values)) {
+ return false;
+ }
}
+ return true;
}
@Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
+ public boolean test(long rowCount, FieldStats[] fieldStats,
List<Predicate> children) {
+ for (Predicate child : children) {
+ if (!child.test(rowCount, fieldStats)) {
+ return false;
+ }
}
- if (!(o instanceof And)) {
- return false;
- }
- And and = (And) o;
- return predicate1.equals(and.predicate1) &&
predicate2.equals(and.predicate2);
+ return true;
}
@Override
- public int hashCode() {
- return Objects.hash(predicate1, predicate2);
+ public Optional<Predicate> negate(List<Predicate> children) {
+ List<Predicate> negatedChildren = new ArrayList<>();
+ for (Predicate child : children) {
+ Optional<Predicate> negatedChild = child.negate();
+ if (negatedChild.isPresent()) {
+ negatedChildren.add(negatedChild.get());
+ } else {
+ return Optional.empty();
+ }
+ }
+ return Optional.of(new CompoundPredicate(Or.INSTANCE,
negatedChildren));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/CompoundPredicate.java
similarity index 50%
copy from
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
copy to
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/CompoundPredicate.java
index 4029c186..4be05b2a 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/CompoundPredicate.java
@@ -20,59 +20,64 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
+import java.io.Serializable;
+import java.util.List;
import java.util.Objects;
import java.util.Optional;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/** A {@link Predicate} to eval greater. */
-public class GreaterThan implements Predicate {
+/**
+ * Non-leaf node in a {@link Predicate} tree. Its evaluation result depends on
the results of its
+ * children.
+ */
+public class CompoundPredicate implements Predicate {
- private static final long serialVersionUID = 1L;
+ private final Function function;
+ private final List<Predicate> children;
- private final int index;
+ public CompoundPredicate(Function function, List<Predicate> children) {
+ this.function = function;
+ this.children = children;
+ }
- private final Literal literal;
+ public Function function() {
+ return function;
+ }
- public GreaterThan(int index, Literal literal) {
- this.index = index;
- this.literal = checkNotNull(literal);
+ public List<Predicate> children() {
+ return children;
}
@Override
public boolean test(Object[] values) {
- Object field = values[index];
- return field != null && literal.compareValueTo(field) < 0;
+ return function.test(values, children);
}
@Override
public boolean test(long rowCount, FieldStats[] fieldStats) {
- FieldStats stats = fieldStats[index];
- if (rowCount == stats.nullCount()) {
- return false;
- }
- return literal.compareValueTo(stats.maxValue()) < 0;
+ return function.test(rowCount, fieldStats, children);
}
@Override
public Optional<Predicate> negate() {
- return Optional.of(new LessOrEqual(index, literal));
+ return function.negate(children);
}
@Override
public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof GreaterThan)) {
+ if (!(o instanceof CompoundPredicate)) {
return false;
}
- GreaterThan that = (GreaterThan) o;
- return index == that.index && literal.equals(that.literal);
+ CompoundPredicate that = (CompoundPredicate) o;
+ return Objects.equals(function, that.function) &&
Objects.equals(children, that.children);
}
- @Override
- public int hashCode() {
- return Objects.hash(index, literal);
+ /** Evaluate the predicate result based on multiple {@link Predicate}s. */
+ public interface Function extends Serializable {
+
+ boolean test(Object[] values, List<Predicate> children);
+
+ boolean test(long rowCount, FieldStats[] fieldStats, List<Predicate>
children);
+
+ Optional<Predicate> negate(List<Predicate> children);
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/Equal.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/Equal.java
index 2ca1c05d..074acf19 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/Equal.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/Equal.java
@@ -20,33 +20,23 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
import java.util.Optional;
-import static org.apache.flink.util.Preconditions.checkNotNull;
+/** A {@link LeafPredicate.Function} to eval equal. */
+public class Equal implements LeafPredicate.Function {
-/** A {@link Predicate} to eval equal. */
-public class Equal implements Predicate {
+ public static final Equal INSTANCE = new Equal();
- private static final long serialVersionUID = 1L;
-
- private final int index;
-
- private final Literal literal;
-
- public Equal(int index, Literal literal) {
- this.index = index;
- this.literal = checkNotNull(literal);
- }
+ private Equal() {}
@Override
- public boolean test(Object[] values) {
+ public boolean test(Object[] values, int index, Literal literal) {
Object field = values[index];
return field != null && literal.compareValueTo(field) == 0;
}
@Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
+ public boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal literal) {
FieldStats stats = fieldStats[index];
if (rowCount == stats.nullCount()) {
return false;
@@ -56,24 +46,7 @@ public class Equal implements Predicate {
}
@Override
- public Optional<Predicate> negate() {
- return Optional.of(new NotEqual(index, literal));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof Equal)) {
- return false;
- }
- Equal equal = (Equal) o;
- return index == equal.index && literal.equals(equal.literal);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(index, literal);
+ public Optional<Predicate> negate(int index, Literal literal) {
+ return Optional.of(new LeafPredicate(NotEqual.INSTANCE, index,
literal));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterOrEqual.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterOrEqual.java
index 22bc54ab..9527d3bf 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterOrEqual.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterOrEqual.java
@@ -20,33 +20,23 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
import java.util.Optional;
-import static org.apache.flink.util.Preconditions.checkNotNull;
+/** A {@link LeafPredicate.Function} to eval greater or equal. */
+public class GreaterOrEqual implements LeafPredicate.Function {
-/** A {@link Predicate} to eval greater or equal. */
-public class GreaterOrEqual implements Predicate {
+ public static final GreaterOrEqual INSTANCE = new GreaterOrEqual();
- private static final long serialVersionUID = 1L;
-
- private final int index;
-
- private final Literal literal;
-
- public GreaterOrEqual(int index, Literal literal) {
- this.index = index;
- this.literal = checkNotNull(literal);
- }
+ private GreaterOrEqual() {}
@Override
- public boolean test(Object[] values) {
+ public boolean test(Object[] values, int index, Literal literal) {
Object field = values[index];
return field != null && literal.compareValueTo(field) <= 0;
}
@Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
+ public boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal literal) {
FieldStats stats = fieldStats[index];
if (rowCount == stats.nullCount()) {
return false;
@@ -55,24 +45,7 @@ public class GreaterOrEqual implements Predicate {
}
@Override
- public Optional<Predicate> negate() {
- return Optional.of(new LessThan(index, literal));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof GreaterOrEqual)) {
- return false;
- }
- GreaterOrEqual that = (GreaterOrEqual) o;
- return index == that.index && literal.equals(that.literal);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(index, literal);
+ public Optional<Predicate> negate(int index, Literal literal) {
+ return Optional.of(new LeafPredicate(LessThan.INSTANCE, index,
literal));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
index 4029c186..d097f888 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
@@ -20,33 +20,23 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
import java.util.Optional;
-import static org.apache.flink.util.Preconditions.checkNotNull;
+/** A {@link LeafPredicate.Function} to eval greater. */
+public class GreaterThan implements LeafPredicate.Function {
-/** A {@link Predicate} to eval greater. */
-public class GreaterThan implements Predicate {
+ public static final GreaterThan INSTANCE = new GreaterThan();
- private static final long serialVersionUID = 1L;
-
- private final int index;
-
- private final Literal literal;
-
- public GreaterThan(int index, Literal literal) {
- this.index = index;
- this.literal = checkNotNull(literal);
- }
+ private GreaterThan() {}
@Override
- public boolean test(Object[] values) {
+ public boolean test(Object[] values, int index, Literal literal) {
Object field = values[index];
return field != null && literal.compareValueTo(field) < 0;
}
@Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
+ public boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal literal) {
FieldStats stats = fieldStats[index];
if (rowCount == stats.nullCount()) {
return false;
@@ -55,24 +45,7 @@ public class GreaterThan implements Predicate {
}
@Override
- public Optional<Predicate> negate() {
- return Optional.of(new LessOrEqual(index, literal));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof GreaterThan)) {
- return false;
- }
- GreaterThan that = (GreaterThan) o;
- return index == that.index && literal.equals(that.literal);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(index, literal);
+ public Optional<Predicate> negate(int index, Literal literal) {
+ return Optional.of(new LeafPredicate(LessOrEqual.INSTANCE, index,
literal));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/IsNotNull.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/IsNotNull.java
index 149a89b2..e4e19c5e 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/IsNotNull.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/IsNotNull.java
@@ -20,49 +20,27 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
import java.util.Optional;
-/** A {@link Predicate} to eval is not null. */
-public class IsNotNull implements Predicate {
+/** A {@link LeafPredicate.Function} to eval is not null. */
+public class IsNotNull implements LeafPredicate.Function {
- private static final long serialVersionUID = 1L;
+ public static final IsNotNull INSTANCE = new IsNotNull();
- private final int index;
-
- public IsNotNull(int index) {
- this.index = index;
- }
+ private IsNotNull() {}
@Override
- public boolean test(Object[] values) {
+ public boolean test(Object[] values, int index, Literal literal) {
return values[index] != null;
}
@Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
+ public boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal literal) {
return fieldStats[index].nullCount() < rowCount;
}
@Override
- public Optional<Predicate> negate() {
- return Optional.of(new IsNull(index));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof IsNotNull)) {
- return false;
- }
- IsNotNull isNotNull = (IsNotNull) o;
- return index == isNotNull.index;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(index);
+ public Optional<Predicate> negate(int index, Literal literal) {
+ return Optional.of(new LeafPredicate(IsNull.INSTANCE, index, literal));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/IsNull.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/IsNull.java
index cd85d9f6..a2279722 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/IsNull.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/IsNull.java
@@ -20,49 +20,27 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
import java.util.Optional;
-/** A {@link Predicate} to eval is null. */
-public class IsNull implements Predicate {
+/** A {@link LeafPredicate.Function} to eval is null. */
+public class IsNull implements LeafPredicate.Function {
- private static final long serialVersionUID = 1L;
+ public static final IsNull INSTANCE = new IsNull();
- private final int index;
-
- public IsNull(int index) {
- this.index = index;
- }
+ private IsNull() {}
@Override
- public boolean test(Object[] values) {
+ public boolean test(Object[] values, int index, Literal literal) {
return values[index] == null;
}
@Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
+ public boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal literal) {
return fieldStats[index].nullCount() > 0;
}
@Override
- public Optional<Predicate> negate() {
- return Optional.of(new IsNotNull(index));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof IsNull)) {
- return false;
- }
- IsNull isNull = (IsNull) o;
- return index == isNull.index;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(index);
+ public Optional<Predicate> negate(int index, Literal literal) {
+ return Optional.of(new LeafPredicate(IsNotNull.INSTANCE, index,
literal));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LeafPredicate.java
similarity index 51%
copy from
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
copy to
flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LeafPredicate.java
index 4029c186..244058e7 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/GreaterThan.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LeafPredicate.java
@@ -20,59 +20,68 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
+import java.io.Serializable;
import java.util.Objects;
import java.util.Optional;
-import static org.apache.flink.util.Preconditions.checkNotNull;
-
-/** A {@link Predicate} to eval greater. */
-public class GreaterThan implements Predicate {
-
- private static final long serialVersionUID = 1L;
+/** Leaf node of a {@link Predicate} tree. Compares a field in the row with an
{@link Literal}. */
+public class LeafPredicate implements Predicate {
+ private final Function function;
private final int index;
-
private final Literal literal;
- public GreaterThan(int index, Literal literal) {
+ public LeafPredicate(Function function, int index, Literal literal) {
+ this.function = function;
this.index = index;
- this.literal = checkNotNull(literal);
+ this.literal = literal;
+ }
+
+ public Function function() {
+ return function;
+ }
+
+ public int index() {
+ return index;
+ }
+
+ public Literal literal() {
+ return literal;
}
@Override
public boolean test(Object[] values) {
- Object field = values[index];
- return field != null && literal.compareValueTo(field) < 0;
+ return function.test(values, index, literal);
}
@Override
public boolean test(long rowCount, FieldStats[] fieldStats) {
- FieldStats stats = fieldStats[index];
- if (rowCount == stats.nullCount()) {
- return false;
- }
- return literal.compareValueTo(stats.maxValue()) < 0;
+ return function.test(rowCount, fieldStats, index, literal);
}
@Override
public Optional<Predicate> negate() {
- return Optional.of(new LessOrEqual(index, literal));
+ return function.negate(index, literal);
}
@Override
public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof GreaterThan)) {
+ if (!(o instanceof LeafPredicate)) {
return false;
}
- GreaterThan that = (GreaterThan) o;
- return index == that.index && literal.equals(that.literal);
+ LeafPredicate that = (LeafPredicate) o;
+ return Objects.equals(function, that.function)
+ && index == that.index
+ && Objects.equals(literal, that.literal);
}
- @Override
- public int hashCode() {
- return Objects.hash(index, literal);
+ /** Function to compare a field in the row with an {@link Literal}. */
+ public interface Function extends Serializable {
+
+ boolean test(Object[] values, int index, Literal literal);
+
+ boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal literal);
+
+ Optional<Predicate> negate(int index, Literal literal);
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LessOrEqual.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LessOrEqual.java
index c09a53f2..5b544eab 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LessOrEqual.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LessOrEqual.java
@@ -20,33 +20,23 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
import java.util.Optional;
-import static org.apache.flink.util.Preconditions.checkNotNull;
+/** A {@link LeafPredicate.Function} to eval less or equal. */
+public class LessOrEqual implements LeafPredicate.Function {
-/** A {@link Predicate} to eval less or equal. */
-public class LessOrEqual implements Predicate {
+ public static final LessOrEqual INSTANCE = new LessOrEqual();
- private static final long serialVersionUID = 1L;
-
- private final int index;
-
- private final Literal literal;
-
- public LessOrEqual(int index, Literal literal) {
- this.index = index;
- this.literal = checkNotNull(literal);
- }
+ private LessOrEqual() {}
@Override
- public boolean test(Object[] values) {
+ public boolean test(Object[] values, int index, Literal literal) {
Object field = values[index];
return field != null && literal.compareValueTo(field) >= 0;
}
@Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
+ public boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal literal) {
FieldStats stats = fieldStats[index];
if (rowCount == stats.nullCount()) {
return false;
@@ -55,24 +45,7 @@ public class LessOrEqual implements Predicate {
}
@Override
- public Optional<Predicate> negate() {
- return Optional.of(new GreaterThan(index, literal));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof LessOrEqual)) {
- return false;
- }
- LessOrEqual that = (LessOrEqual) o;
- return index == that.index && literal.equals(that.literal);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(index, literal);
+ public Optional<Predicate> negate(int index, Literal literal) {
+ return Optional.of(new LeafPredicate(GreaterThan.INSTANCE, index,
literal));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LessThan.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LessThan.java
index 71f06d70..7fc3f2cd 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LessThan.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/LessThan.java
@@ -20,33 +20,23 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
import java.util.Optional;
-import static org.apache.flink.util.Preconditions.checkNotNull;
+/** A {@link LeafPredicate.Function} to eval less. */
+public class LessThan implements LeafPredicate.Function {
-/** A {@link Predicate} to eval less. */
-public class LessThan implements Predicate {
+ public static final LessThan INSTANCE = new LessThan();
- private static final long serialVersionUID = 1L;
-
- private final int index;
-
- private final Literal literal;
-
- public LessThan(int index, Literal literal) {
- this.index = index;
- this.literal = checkNotNull(literal);
- }
+ private LessThan() {}
@Override
- public boolean test(Object[] values) {
+ public boolean test(Object[] values, int index, Literal literal) {
Object field = values[index];
return field != null && literal.compareValueTo(field) > 0;
}
@Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
+ public boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal literal) {
FieldStats stats = fieldStats[index];
if (rowCount == stats.nullCount()) {
return false;
@@ -55,24 +45,7 @@ public class LessThan implements Predicate {
}
@Override
- public Optional<Predicate> negate() {
- return Optional.of(new GreaterOrEqual(index, literal));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof LessThan)) {
- return false;
- }
- LessThan lessThan = (LessThan) o;
- return index == lessThan.index && literal.equals(lessThan.literal);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(index, literal);
+ public Optional<Predicate> negate(int index, Literal literal) {
+ return Optional.of(new LeafPredicate(GreaterOrEqual.INSTANCE, index,
literal));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/NotEqual.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/NotEqual.java
index 404ab837..f3e90662 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/NotEqual.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/NotEqual.java
@@ -20,33 +20,23 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
import java.util.Optional;
-import static org.apache.flink.util.Preconditions.checkNotNull;
+/** A {@link LeafPredicate.Function} to eval not equal. */
+public class NotEqual implements LeafPredicate.Function {
-/** A {@link Predicate} to eval not equal. */
-public class NotEqual implements Predicate {
+ public static final NotEqual INSTANCE = new NotEqual();
- private static final long serialVersionUID = 1L;
-
- private final int index;
-
- private final Literal literal;
-
- public NotEqual(int index, Literal literal) {
- this.index = index;
- this.literal = checkNotNull(literal);
- }
+ private NotEqual() {}
@Override
- public boolean test(Object[] values) {
+ public boolean test(Object[] values, int index, Literal literal) {
Object field = values[index];
return field != null && literal.compareValueTo(field) != 0;
}
@Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
+ public boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal literal) {
FieldStats stats = fieldStats[index];
if (rowCount == stats.nullCount()) {
return false;
@@ -56,24 +46,7 @@ public class NotEqual implements Predicate {
}
@Override
- public Optional<Predicate> negate() {
- return Optional.of(new Equal(index, literal));
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (!(o instanceof NotEqual)) {
- return false;
- }
- NotEqual notEqual = (NotEqual) o;
- return index == notEqual.index && literal.equals(notEqual.literal);
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(index, literal);
+ public Optional<Predicate> negate(int index, Literal literal) {
+ return Optional.of(new LeafPredicate(Equal.INSTANCE, index, literal));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/Or.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/Or.java
index 13234278..5a74bd6a 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/Or.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/Or.java
@@ -20,57 +20,50 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.table.store.file.stats.FieldStats;
-import java.util.Objects;
+import java.util.ArrayList;
+import java.util.List;
import java.util.Optional;
-/** A {@link Predicate} to eval or. */
-public class Or implements Predicate {
+/** A {@link CompoundPredicate.Function} to eval or. */
+public class Or implements CompoundPredicate.Function {
private static final long serialVersionUID = 1L;
- private final Predicate predicate1;
- private final Predicate predicate2;
+ public static final Or INSTANCE = new Or();
- public Or(Predicate predicate1, Predicate predicate2) {
- this.predicate1 = predicate1;
- this.predicate2 = predicate2;
- }
+ private Or() {}
@Override
- public boolean test(Object[] values) {
- return predicate1.test(values) || predicate2.test(values);
- }
-
- @Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
- return predicate1.test(rowCount, fieldStats) ||
predicate2.test(rowCount, fieldStats);
- }
-
- @Override
- public Optional<Predicate> negate() {
- Optional<Predicate> negate1 = predicate1.negate();
- Optional<Predicate> negate2 = predicate2.negate();
- if (negate1.isPresent() && negate2.isPresent()) {
- return Optional.of(new And(negate1.get(), negate2.get()));
- } else {
- return Optional.empty();
+ public boolean test(Object[] values, List<Predicate> children) {
+ for (Predicate child : children) {
+ if (child.test(values)) {
+ return true;
+ }
}
+ return false;
}
@Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
+ public boolean test(long rowCount, FieldStats[] fieldStats,
List<Predicate> children) {
+ for (Predicate child : children) {
+ if (child.test(rowCount, fieldStats)) {
+ return true;
+ }
}
- if (!(o instanceof Or)) {
- return false;
- }
- Or or = (Or) o;
- return predicate1.equals(or.predicate1) &&
predicate2.equals(or.predicate2);
+ return false;
}
@Override
- public int hashCode() {
- return Objects.hash(predicate1, predicate2);
+ public Optional<Predicate> negate(List<Predicate> children) {
+ List<Predicate> negatedChildren = new ArrayList<>();
+ for (Predicate child : children) {
+ Optional<Predicate> negatedChild = child.negate();
+ if (negatedChild.isPresent()) {
+ negatedChildren.add(negatedChild.get());
+ } else {
+ return Optional.empty();
+ }
+ }
+ return Optional.of(new CompoundPredicate(And.INSTANCE,
negatedChildren));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/PredicateBuilder.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/PredicateBuilder.java
index 262dc549..b431d81e 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/PredicateBuilder.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/PredicateBuilder.java
@@ -20,36 +20,90 @@ package org.apache.flink.table.store.file.predicate;
import org.apache.flink.util.Preconditions;
+import java.util.Arrays;
import java.util.List;
/** A utility class to create {@link Predicate} object for common filter
conditions. */
public class PredicateBuilder {
+ public static Predicate equal(int idx, Literal literal) {
+ return new LeafPredicate(Equal.INSTANCE, idx, literal);
+ }
+
+ public static Predicate notEqual(int idx, Literal literal) {
+ return new LeafPredicate(NotEqual.INSTANCE, idx, literal);
+ }
+
+ public static Predicate lessThan(int idx, Literal literal) {
+ return new LeafPredicate(LessThan.INSTANCE, idx, literal);
+ }
+
+ public static Predicate lessOrEqual(int idx, Literal literal) {
+ return new LeafPredicate(LessOrEqual.INSTANCE, idx, literal);
+ }
+
+ public static Predicate greaterThan(int idx, Literal literal) {
+ return new LeafPredicate(GreaterThan.INSTANCE, idx, literal);
+ }
+
+ public static Predicate greaterOrEqual(int idx, Literal literal) {
+ return new LeafPredicate(GreaterOrEqual.INSTANCE, idx, literal);
+ }
+
+ public static Predicate isNull(int idx) {
+ return new LeafPredicate(IsNull.INSTANCE, idx, null);
+ }
+
+ public static Predicate isNotNull(int idx) {
+ return new LeafPredicate(IsNotNull.INSTANCE, idx, null);
+ }
+
+ public static Predicate startsWith(int idx, Literal patternLiteral) {
+ return new LeafPredicate(StartsWith.INSTANCE, idx, patternLiteral);
+ }
+
+ public static Predicate and(Predicate... predicates) {
+ return and(Arrays.asList(predicates));
+ }
+
public static Predicate and(List<Predicate> predicates) {
Preconditions.checkArgument(
predicates.size() > 0,
"There must be at least 1 inner predicate to construct an AND
predicate");
- return predicates.stream().reduce(And::new).get();
+ return predicates.stream()
+ .reduce((a, b) -> new CompoundPredicate(And.INSTANCE,
Arrays.asList(a, b)))
+ .get();
+ }
+
+ public static Predicate or(Predicate... predicates) {
+ return or(Arrays.asList(predicates));
}
public static Predicate or(List<Predicate> predicates) {
Preconditions.checkArgument(
predicates.size() > 0,
"There must be at least 1 inner predicate to construct an OR
predicate");
- return predicates.stream().reduce(Or::new).get();
+ return predicates.stream()
+ .reduce((a, b) -> new CompoundPredicate(Or.INSTANCE,
Arrays.asList(a, b)))
+ .get();
}
public static Predicate in(int idx, List<Literal> literals) {
Preconditions.checkArgument(
literals.size() > 0,
"There must be at least 1 literal to construct an IN
predicate");
- return literals.stream().map(l -> (Predicate) new Equal(idx,
l)).reduce(Or::new).get();
+ return literals.stream()
+ .map(l -> equal(idx, l))
+ .reduce((a, b) -> new CompoundPredicate(Or.INSTANCE,
Arrays.asList(a, b)))
+ .get();
}
public static Predicate between(
int idx, Literal includedLowerBound, Literal includedUpperBound) {
- return new And(
- new GreaterOrEqual(idx, includedLowerBound),
- new LessOrEqual(idx, includedUpperBound));
+ return new CompoundPredicate(
+ And.INSTANCE,
+ Arrays.asList(
+ greaterOrEqual(idx, includedLowerBound),
+ lessOrEqual(idx, includedUpperBound)));
}
}
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/PredicateConverter.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/PredicateConverter.java
index a01b2613..0656e2f6 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/PredicateConverter.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/PredicateConverter.java
@@ -65,9 +65,9 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
LogicalType type = rowType.getTypeAt(idx);
Literal literal = new Literal(type,
TypeUtils.castFromString(entry.getValue(), type));
if (predicate == null) {
- predicate = new Equal(idx, literal);
+ predicate = PredicateBuilder.equal(idx, literal);
} else {
- predicate = new And(predicate, new Equal(idx, literal));
+ predicate = PredicateBuilder.and(predicate,
PredicateBuilder.equal(idx, literal));
}
}
return predicate;
@@ -79,30 +79,35 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
List<Expression> children = call.getChildren();
if (func == BuiltInFunctionDefinitions.AND) {
- return new And(children.get(0).accept(this),
children.get(1).accept(this));
+ return PredicateBuilder.and(children.get(0).accept(this),
children.get(1).accept(this));
} else if (func == BuiltInFunctionDefinitions.OR) {
- return new Or(children.get(0).accept(this),
children.get(1).accept(this));
+ return PredicateBuilder.or(children.get(0).accept(this),
children.get(1).accept(this));
} else if (func == BuiltInFunctionDefinitions.EQUALS) {
- return visitBiFunction(children, Equal::new, Equal::new);
+ return visitBiFunction(children, PredicateBuilder::equal,
PredicateBuilder::equal);
} else if (func == BuiltInFunctionDefinitions.NOT_EQUALS) {
- return visitBiFunction(children, NotEqual::new, NotEqual::new);
+ return visitBiFunction(
+ children, PredicateBuilder::notEqual,
PredicateBuilder::notEqual);
} else if (func == BuiltInFunctionDefinitions.GREATER_THAN) {
- return visitBiFunction(children, GreaterThan::new, LessThan::new);
+ return visitBiFunction(
+ children, PredicateBuilder::greaterThan,
PredicateBuilder::lessThan);
} else if (func == BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL) {
- return visitBiFunction(children, GreaterOrEqual::new,
LessOrEqual::new);
+ return visitBiFunction(
+ children, PredicateBuilder::greaterOrEqual,
PredicateBuilder::lessOrEqual);
} else if (func == BuiltInFunctionDefinitions.LESS_THAN) {
- return visitBiFunction(children, LessThan::new, GreaterThan::new);
+ return visitBiFunction(
+ children, PredicateBuilder::lessThan,
PredicateBuilder::greaterThan);
} else if (func == BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL) {
- return visitBiFunction(children, LessOrEqual::new,
GreaterOrEqual::new);
+ return visitBiFunction(
+ children, PredicateBuilder::lessOrEqual,
PredicateBuilder::greaterOrEqual);
} else if (func == BuiltInFunctionDefinitions.IS_NULL) {
return extractFieldReference(children.get(0))
.map(FieldReferenceExpression::getFieldIndex)
- .map(IsNull::new)
+ .map(PredicateBuilder::isNull)
.orElseThrow(UnsupportedExpression::new);
} else if (func == BuiltInFunctionDefinitions.IS_NOT_NULL) {
return extractFieldReference(children.get(0))
.map(FieldReferenceExpression::getFieldIndex)
- .map(IsNotNull::new)
+ .map(PredicateBuilder::isNotNull)
.orElseThrow(UnsupportedExpression::new);
} else if (func == BuiltInFunctionDefinitions.LIKE) {
FieldReferenceExpression fieldRefExpr =
@@ -165,7 +170,7 @@ public class PredicateConverter implements
ExpressionVisitor<Predicate> {
if (allowQuick) {
Matcher beginMatcher =
BEGIN_PATTERN.matcher(escapedSqlPattern);
if (beginMatcher.matches()) {
- return new StartsWith(
+ return PredicateBuilder.startsWith(
fieldRefExpr.getFieldIndex(),
new Literal(
VarCharType.STRING_TYPE,
diff --git
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/StartsWith.java
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/StartsWith.java
index 5d8f8d77..af031633 100644
---
a/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/StartsWith.java
+++
b/flink-table-store-core/src/main/java/org/apache/flink/table/store/file/predicate/StartsWith.java
@@ -23,28 +23,23 @@ import org.apache.flink.table.store.file.stats.FieldStats;
import java.util.Optional;
-/** A {@link Predicate} to evaluate {@code filter like 'abc%' or filter like
'abc_'}. */
-public class StartsWith implements Predicate {
-
- private static final long serialVersionUID = 1L;
-
- private final int index;
+/**
+ * A {@link LeafPredicate.Function} to evaluate {@code filter like 'abc%' or
filter like 'abc_'}.
+ */
+public class StartsWith implements LeafPredicate.Function {
- private final Literal patternLiteral;
+ public static final StartsWith INSTANCE = new StartsWith();
- public StartsWith(int index, Literal patternLiteral) {
- this.index = index;
- this.patternLiteral = patternLiteral;
- }
+ private StartsWith() {}
@Override
- public boolean test(Object[] values) {
+ public boolean test(Object[] values, int index, Literal patternLiteral) {
BinaryStringData field = (BinaryStringData) values[index];
return field != null && field.startsWith((BinaryStringData)
patternLiteral.value());
}
@Override
- public boolean test(long rowCount, FieldStats[] fieldStats) {
+ public boolean test(long rowCount, FieldStats[] fieldStats, int index,
Literal patternLiteral) {
FieldStats stats = fieldStats[index];
if (rowCount == stats.nullCount()) {
return false;
@@ -57,7 +52,7 @@ public class StartsWith implements Predicate {
}
@Override
- public Optional<Predicate> negate() {
+ public Optional<Predicate> negate(int index, Literal literal) {
return Optional.empty();
}
}
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreScanTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreScanTest.java
index 3e62b351..e590b76b 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreScanTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/operation/FileStoreScanTest.java
@@ -26,8 +26,8 @@ import
org.apache.flink.table.store.file.TestKeyValueGenerator;
import org.apache.flink.table.store.file.manifest.ManifestFileMeta;
import org.apache.flink.table.store.file.manifest.ManifestList;
import
org.apache.flink.table.store.file.mergetree.compact.DeduplicateMergeFunction;
-import org.apache.flink.table.store.file.predicate.Equal;
import org.apache.flink.table.store.file.predicate.Literal;
+import org.apache.flink.table.store.file.predicate.PredicateBuilder;
import org.apache.flink.table.store.file.utils.FileStorePathFactory;
import org.apache.flink.table.types.logical.IntType;
@@ -115,7 +115,8 @@ public class FileStoreScanTest {
FileStoreScan scan = store.newScan();
scan.withSnapshot(snapshot.id());
- scan.withKeyFilter(new Equal(0, new Literal(new IntType(false),
wantedShopId)));
+ scan.withKeyFilter(
+ PredicateBuilder.equal(0, new Literal(new IntType(false),
wantedShopId)));
Map<BinaryRowData, BinaryRowData> expected =
store.toKvMap(
@@ -135,7 +136,8 @@ public class FileStoreScanTest {
FileStoreScan scan = store.newScan();
scan.withSnapshot(snapshot.id());
- scan.withValueFilter(new Equal(2, new Literal(new IntType(false),
wantedShopId)));
+ scan.withValueFilter(
+ PredicateBuilder.equal(2, new Literal(new IntType(false),
wantedShopId)));
Map<BinaryRowData, BinaryRowData> expected =
store.toKvMap(
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/predicate/PredicateConverterTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/predicate/PredicateConverterTest.java
index 8cd57179..c138d2ba 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/predicate/PredicateConverterTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/predicate/PredicateConverterTest.java
@@ -72,50 +72,50 @@ public class PredicateConverterTest {
BuiltInFunctionDefinitions.IS_NULL,
Collections.singletonList(longRefExpr),
DataTypes.BOOLEAN()),
- new IsNull(0)),
+ PredicateBuilder.isNull(0)),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.IS_NOT_NULL,
Collections.singletonList(doubleRefExpr),
DataTypes.BOOLEAN()),
- new IsNotNull(1)),
+ PredicateBuilder.isNotNull(1)),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.EQUALS,
// test literal on left
Arrays.asList(intLitExpr, longRefExpr),
DataTypes.BOOLEAN()),
- new Equal(0, longLit)),
+ PredicateBuilder.equal(0, longLit)),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.NOT_EQUALS,
Arrays.asList(longRefExpr, intLitExpr),
DataTypes.BOOLEAN()),
- new NotEqual(0, longLit)),
+ PredicateBuilder.notEqual(0, longLit)),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.GREATER_THAN,
Arrays.asList(longRefExpr, intLitExpr),
DataTypes.BOOLEAN()),
- new GreaterThan(0, longLit)),
+ PredicateBuilder.greaterThan(0, longLit)),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.GREATER_THAN_OR_EQUAL,
Arrays.asList(longRefExpr, intLitExpr),
DataTypes.BOOLEAN()),
- new GreaterOrEqual(0, longLit)),
+ PredicateBuilder.greaterOrEqual(0, longLit)),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.LESS_THAN,
Arrays.asList(longRefExpr, intLitExpr),
DataTypes.BOOLEAN()),
- new LessThan(0, longLit)),
+ PredicateBuilder.lessThan(0, longLit)),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.LESS_THAN_OR_EQUAL,
Arrays.asList(longRefExpr, intLitExpr),
DataTypes.BOOLEAN()),
- new LessOrEqual(0, longLit)),
+ PredicateBuilder.lessOrEqual(0, longLit)),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.AND,
@@ -129,7 +129,9 @@ public class PredicateConverterTest {
Arrays.asList(doubleRefExpr,
floatLitExpr),
DataTypes.BOOLEAN())),
DataTypes.BOOLEAN()),
- new And(new LessOrEqual(0, longLit), new Equal(1,
doubleLit))),
+ PredicateBuilder.and(
+ PredicateBuilder.lessOrEqual(0, longLit),
+ PredicateBuilder.equal(1, doubleLit))),
Arguments.of(
CallExpression.permanent(
BuiltInFunctionDefinitions.OR,
@@ -143,6 +145,8 @@ public class PredicateConverterTest {
Arrays.asList(doubleRefExpr,
floatLitExpr),
DataTypes.BOOLEAN())),
DataTypes.BOOLEAN()),
- new Or(new NotEqual(0, longLit), new Equal(1,
doubleLit))));
+ PredicateBuilder.or(
+ PredicateBuilder.notEqual(0, longLit),
+ PredicateBuilder.equal(1, doubleLit))));
}
}
diff --git
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/predicate/PredicateTest.java
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/predicate/PredicateTest.java
index 8eeee3cf..744127f8 100644
---
a/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/predicate/PredicateTest.java
+++
b/flink-table-store-core/src/test/java/org/apache/flink/table/store/file/predicate/PredicateTest.java
@@ -82,7 +82,9 @@ public class PredicateTest {
.isEqualTo(false);
assertThat(predicate.negate().orElse(null))
- .isEqualTo(new NotEqual(0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
+ .isEqualTo(
+ PredicateBuilder.notEqual(
+ 0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
}
@Test
@@ -103,7 +105,9 @@ public class PredicateTest {
.isEqualTo(false);
assertThat(predicate.negate().orElse(null))
- .isEqualTo(new Equal(0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
+ .isEqualTo(
+ PredicateBuilder.equal(
+ 0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
}
@Test
@@ -128,7 +132,9 @@ public class PredicateTest {
.isEqualTo(false);
assertThat(predicate.negate().orElse(null))
- .isEqualTo(new LessOrEqual(0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
+ .isEqualTo(
+ PredicateBuilder.lessOrEqual(
+ 0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
}
@Test
@@ -153,7 +159,9 @@ public class PredicateTest {
.isEqualTo(false);
assertThat(predicate.negate().orElse(null))
- .isEqualTo(new LessThan(0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
+ .isEqualTo(
+ PredicateBuilder.lessThan(
+ 0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
}
@Test
@@ -174,7 +182,9 @@ public class PredicateTest {
.isEqualTo(false);
assertThat(predicate.negate().orElse(null))
- .isEqualTo(new GreaterOrEqual(0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
+ .isEqualTo(
+ PredicateBuilder.greaterOrEqual(
+ 0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
}
@Test
@@ -198,7 +208,9 @@ public class PredicateTest {
.isEqualTo(false);
assertThat(predicate.negate().orElse(null))
- .isEqualTo(new GreaterThan(0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
+ .isEqualTo(
+ PredicateBuilder.greaterThan(
+ 0, new
Literal(DataTypes.INT().getLogicalType(), 5)));
}
@Test
@@ -216,7 +228,7 @@ public class PredicateTest {
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(6, 7,
0)})).isEqualTo(false);
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(5, 7,
1)})).isEqualTo(true);
- assertThat(predicate.negate().orElse(null)).isEqualTo(new
IsNotNull(0));
+
assertThat(predicate.negate().orElse(null)).isEqualTo(PredicateBuilder.isNotNull(0));
}
@Test
@@ -236,7 +248,7 @@ public class PredicateTest {
assertThat(predicate.test(3, new FieldStats[] {new FieldStats(null,
null, 3)}))
.isEqualTo(false);
- assertThat(predicate.negate().orElse(null)).isEqualTo(new IsNull(0));
+
assertThat(predicate.negate().orElse(null)).isEqualTo(PredicateBuilder.isNull(0));
}
@Test
@@ -283,9 +295,11 @@ public class PredicateTest {
assertThat(predicate.negate().orElse(null))
.isEqualTo(
- new Or(
- new NotEqual(0, new
Literal(DataTypes.INT().getLogicalType(), 3)),
- new NotEqual(1, new
Literal(DataTypes.INT().getLogicalType(), 5))));
+ PredicateBuilder.or(
+ PredicateBuilder.notEqual(
+ 0, new
Literal(DataTypes.INT().getLogicalType(), 3)),
+ PredicateBuilder.notEqual(
+ 1, new
Literal(DataTypes.INT().getLogicalType(), 5))));
}
@Test
@@ -332,9 +346,11 @@ public class PredicateTest {
assertThat(predicate.negate().orElse(null))
.isEqualTo(
- new And(
- new NotEqual(0, new
Literal(DataTypes.INT().getLogicalType(), 3)),
- new NotEqual(1, new
Literal(DataTypes.INT().getLogicalType(), 5))));
+ PredicateBuilder.and(
+ PredicateBuilder.notEqual(
+ 0, new
Literal(DataTypes.INT().getLogicalType(), 3)),
+ PredicateBuilder.notEqual(
+ 1, new
Literal(DataTypes.INT().getLogicalType(), 5))));
}
@MethodSource("provideLikeExpressions")
diff --git
a/flink-table-store-hive/src/main/java/org/apache/flink/table/store/SearchArgumentToPredicateConverter.java
b/flink-table-store-hive/src/main/java/org/apache/flink/table/store/SearchArgumentToPredicateConverter.java
index b8058969..997447f1 100644
---
a/flink-table-store-hive/src/main/java/org/apache/flink/table/store/SearchArgumentToPredicateConverter.java
+++
b/flink-table-store-hive/src/main/java/org/apache/flink/table/store/SearchArgumentToPredicateConverter.java
@@ -21,10 +21,6 @@ package org.apache.flink.table.store;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.table.store.file.predicate.Equal;
-import org.apache.flink.table.store.file.predicate.IsNull;
-import org.apache.flink.table.store.file.predicate.LessOrEqual;
-import org.apache.flink.table.store.file.predicate.LessThan;
import org.apache.flink.table.store.file.predicate.Literal;
import org.apache.flink.table.store.file.predicate.Predicate;
import org.apache.flink.table.store.file.predicate.PredicateBuilder;
@@ -112,11 +108,11 @@ public class SearchArgumentToPredicateConverter {
LogicalType columnType = columnTypes.get(idx);
switch (leaf.getOperator()) {
case EQUALS:
- return new Equal(idx, toLiteral(columnType,
leaf.getLiteral()));
+ return PredicateBuilder.equal(idx, toLiteral(columnType,
leaf.getLiteral()));
case LESS_THAN:
- return new LessThan(idx, toLiteral(columnType,
leaf.getLiteral()));
+ return PredicateBuilder.lessThan(idx, toLiteral(columnType,
leaf.getLiteral()));
case LESS_THAN_EQUALS:
- return new LessOrEqual(idx, toLiteral(columnType,
leaf.getLiteral()));
+ return PredicateBuilder.lessOrEqual(idx, toLiteral(columnType,
leaf.getLiteral()));
case IN:
return PredicateBuilder.in(
idx,
@@ -130,7 +126,7 @@ public class SearchArgumentToPredicateConverter {
toLiteral(columnType, literalList.get(0)),
toLiteral(columnType, literalList.get(1)));
case IS_NULL:
- return new IsNull(idx);
+ return PredicateBuilder.isNull(idx);
default:
throw new UnsupportedOperationException(
"Unsupported operator " + leaf.getOperator());
diff --git
a/flink-table-store-hive/src/test/java/org/apache/flink/table/store/SearchArgumentToPredicateConverterTest.java
b/flink-table-store-hive/src/test/java/org/apache/flink/table/store/SearchArgumentToPredicateConverterTest.java
index 32213603..54583ae5 100644
---
a/flink-table-store-hive/src/test/java/org/apache/flink/table/store/SearchArgumentToPredicateConverterTest.java
+++
b/flink-table-store-hive/src/test/java/org/apache/flink/table/store/SearchArgumentToPredicateConverterTest.java
@@ -22,18 +22,9 @@ import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.DecimalData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;
-import org.apache.flink.table.store.file.predicate.And;
-import org.apache.flink.table.store.file.predicate.Equal;
-import org.apache.flink.table.store.file.predicate.GreaterOrEqual;
-import org.apache.flink.table.store.file.predicate.GreaterThan;
-import org.apache.flink.table.store.file.predicate.IsNotNull;
-import org.apache.flink.table.store.file.predicate.IsNull;
-import org.apache.flink.table.store.file.predicate.LessOrEqual;
-import org.apache.flink.table.store.file.predicate.LessThan;
import org.apache.flink.table.store.file.predicate.Literal;
-import org.apache.flink.table.store.file.predicate.NotEqual;
-import org.apache.flink.table.store.file.predicate.Or;
import org.apache.flink.table.store.file.predicate.Predicate;
+import org.apache.flink.table.store.file.predicate.PredicateBuilder;
import org.apache.flink.table.types.logical.LogicalType;
import org.apache.hadoop.hive.common.type.HiveDecimal;
@@ -103,7 +94,7 @@ public class SearchArgumentToPredicateConverterTest {
new SearchArgumentToPredicateConverter(
sarg, Collections.singletonList("a"),
Collections.singletonList(flinkType));
- Predicate expected = new Equal(0, new Literal(flinkType,
flinkLiteral));
+ Predicate expected = PredicateBuilder.equal(0, new Literal(flinkType,
flinkLiteral));
Predicate actual = converter.convert().orElse(null);
assertThat(actual).isEqualTo(expected);
}
@@ -120,7 +111,7 @@ public class SearchArgumentToPredicateConverterTest {
public void testEqual() {
SearchArgument.Builder builder = SearchArgumentFactory.newBuilder();
SearchArgument sarg = builder.equals("f_bigint",
PredicateLeaf.Type.LONG, 100L).build();
- Predicate expected = new Equal(1, new Literal(BIGINT_TYPE, 100L));
+ Predicate expected = PredicateBuilder.equal(1, new
Literal(BIGINT_TYPE, 100L));
assertExpected(sarg, expected);
}
@@ -129,7 +120,7 @@ public class SearchArgumentToPredicateConverterTest {
SearchArgument.Builder builder = SearchArgumentFactory.newBuilder();
SearchArgument sarg =
builder.startNot().equals("f_bigint", PredicateLeaf.Type.LONG,
100L).end().build();
- Predicate expected = new NotEqual(1, new Literal(BIGINT_TYPE, 100L));
+ Predicate expected = PredicateBuilder.notEqual(1, new
Literal(BIGINT_TYPE, 100L));
assertExpected(sarg, expected);
}
@@ -137,7 +128,7 @@ public class SearchArgumentToPredicateConverterTest {
public void testLessThan() {
SearchArgument.Builder builder = SearchArgumentFactory.newBuilder();
SearchArgument sarg = builder.lessThan("f_bigint",
PredicateLeaf.Type.LONG, 100L).build();
- Predicate expected = new LessThan(1, new Literal(BIGINT_TYPE, 100L));
+ Predicate expected = PredicateBuilder.lessThan(1, new
Literal(BIGINT_TYPE, 100L));
assertExpected(sarg, expected);
}
@@ -149,7 +140,7 @@ public class SearchArgumentToPredicateConverterTest {
.lessThan("f_bigint", PredicateLeaf.Type.LONG, 100L)
.end()
.build();
- Predicate expected = new GreaterOrEqual(1, new Literal(BIGINT_TYPE,
100L));
+ Predicate expected = PredicateBuilder.greaterOrEqual(1, new
Literal(BIGINT_TYPE, 100L));
assertExpected(sarg, expected);
}
@@ -158,7 +149,7 @@ public class SearchArgumentToPredicateConverterTest {
SearchArgument.Builder builder = SearchArgumentFactory.newBuilder();
SearchArgument sarg =
builder.lessThanEquals("f_bigint", PredicateLeaf.Type.LONG,
100L).build();
- Predicate expected = new LessOrEqual(1, new Literal(BIGINT_TYPE,
100L));
+ Predicate expected = PredicateBuilder.lessOrEqual(1, new
Literal(BIGINT_TYPE, 100L));
assertExpected(sarg, expected);
}
@@ -170,7 +161,7 @@ public class SearchArgumentToPredicateConverterTest {
.lessThanEquals("f_bigint", PredicateLeaf.Type.LONG,
100L)
.end()
.build();
- Predicate expected = new GreaterThan(1, new Literal(BIGINT_TYPE,
100L));
+ Predicate expected = PredicateBuilder.greaterThan(1, new
Literal(BIGINT_TYPE, 100L));
assertExpected(sarg, expected);
}
@@ -180,11 +171,10 @@ public class SearchArgumentToPredicateConverterTest {
SearchArgument sarg =
builder.in("f_bigint", PredicateLeaf.Type.LONG, 100L, 200L,
300L).build();
Predicate expected =
- new Or(
- new Or(
- new Equal(1, new Literal(BIGINT_TYPE, 100L)),
- new Equal(1, new Literal(BIGINT_TYPE, 200L))),
- new Equal(1, new Literal(BIGINT_TYPE, 300L)));
+ PredicateBuilder.or(
+ PredicateBuilder.equal(1, new Literal(BIGINT_TYPE,
100L)),
+ PredicateBuilder.equal(1, new Literal(BIGINT_TYPE,
200L)),
+ PredicateBuilder.equal(1, new Literal(BIGINT_TYPE,
300L)));
assertExpected(sarg, expected);
}
@@ -197,11 +187,10 @@ public class SearchArgumentToPredicateConverterTest {
.end()
.build();
Predicate expected =
- new And(
- new And(
- new NotEqual(1, new Literal(BIGINT_TYPE,
100L)),
- new NotEqual(1, new Literal(BIGINT_TYPE,
200L))),
- new NotEqual(1, new Literal(BIGINT_TYPE, 300L)));
+ PredicateBuilder.and(
+ PredicateBuilder.notEqual(1, new Literal(BIGINT_TYPE,
100L)),
+ PredicateBuilder.notEqual(1, new Literal(BIGINT_TYPE,
200L)),
+ PredicateBuilder.notEqual(1, new Literal(BIGINT_TYPE,
300L)));
assertExpected(sarg, expected);
}
@@ -211,9 +200,9 @@ public class SearchArgumentToPredicateConverterTest {
SearchArgument sarg =
builder.between("f_bigint", PredicateLeaf.Type.LONG, 100L,
200L).build();
Predicate expected =
- new And(
- new GreaterOrEqual(1, new Literal(BIGINT_TYPE, 100L)),
- new LessOrEqual(1, new Literal(BIGINT_TYPE, 200L)));
+ PredicateBuilder.and(
+ PredicateBuilder.greaterOrEqual(1, new
Literal(BIGINT_TYPE, 100L)),
+ PredicateBuilder.lessOrEqual(1, new
Literal(BIGINT_TYPE, 200L)));
assertExpected(sarg, expected);
}
@@ -226,9 +215,9 @@ public class SearchArgumentToPredicateConverterTest {
.end()
.build();
Predicate expected =
- new Or(
- new LessThan(1, new Literal(BIGINT_TYPE, 100L)),
- new GreaterThan(1, new Literal(BIGINT_TYPE, 200L)));
+ PredicateBuilder.or(
+ PredicateBuilder.lessThan(1, new Literal(BIGINT_TYPE,
100L)),
+ PredicateBuilder.greaterThan(1, new
Literal(BIGINT_TYPE, 200L)));
assertExpected(sarg, expected);
}
@@ -236,7 +225,7 @@ public class SearchArgumentToPredicateConverterTest {
public void testIsNull() {
SearchArgument.Builder builder = SearchArgumentFactory.newBuilder();
SearchArgument sarg = builder.isNull("f_bigint",
PredicateLeaf.Type.LONG).build();
- Predicate expected = new IsNull(1);
+ Predicate expected = PredicateBuilder.isNull(1);
assertExpected(sarg, expected);
}
@@ -245,7 +234,7 @@ public class SearchArgumentToPredicateConverterTest {
SearchArgument.Builder builder = SearchArgumentFactory.newBuilder();
SearchArgument sarg =
builder.startNot().isNull("f_bigint",
PredicateLeaf.Type.LONG).end().build();
- Predicate expected = new IsNotNull(1);
+ Predicate expected = PredicateBuilder.isNotNull(1);
assertExpected(sarg, expected);
}
@@ -260,11 +249,10 @@ public class SearchArgumentToPredicateConverterTest {
.end()
.build();
Predicate expected =
- new Or(
- new Or(
- new Equal(1, new Literal(BIGINT_TYPE, 100L)),
- new Equal(1, new Literal(BIGINT_TYPE, 200L))),
- new Equal(1, new Literal(BIGINT_TYPE, 300L)));
+ PredicateBuilder.or(
+ PredicateBuilder.equal(1, new Literal(BIGINT_TYPE,
100L)),
+ PredicateBuilder.equal(1, new Literal(BIGINT_TYPE,
200L)),
+ PredicateBuilder.equal(1, new Literal(BIGINT_TYPE,
300L)));
assertExpected(sarg, expected);
}
@@ -281,11 +269,10 @@ public class SearchArgumentToPredicateConverterTest {
.end()
.build();
Predicate expected =
- new And(
- new And(
- new NotEqual(1, new Literal(BIGINT_TYPE,
100L)),
- new NotEqual(1, new Literal(BIGINT_TYPE,
200L))),
- new NotEqual(1, new Literal(BIGINT_TYPE, 300L)));
+ PredicateBuilder.and(
+ PredicateBuilder.notEqual(1, new Literal(BIGINT_TYPE,
100L)),
+ PredicateBuilder.notEqual(1, new Literal(BIGINT_TYPE,
200L)),
+ PredicateBuilder.notEqual(1, new Literal(BIGINT_TYPE,
300L)));
assertExpected(sarg, expected);
}
@@ -306,11 +293,10 @@ public class SearchArgumentToPredicateConverterTest {
.end()
.build();
Predicate expected =
- new And(
- new And(
- new NotEqual(1, new Literal(BIGINT_TYPE,
100L)),
- new NotEqual(1, new Literal(BIGINT_TYPE,
200L))),
- new NotEqual(1, new Literal(BIGINT_TYPE, 300L)));
+ PredicateBuilder.and(
+ PredicateBuilder.notEqual(1, new Literal(BIGINT_TYPE,
100L)),
+ PredicateBuilder.notEqual(1, new Literal(BIGINT_TYPE,
200L)),
+ PredicateBuilder.notEqual(1, new Literal(BIGINT_TYPE,
300L)));
assertExpected(sarg, expected);
}
@@ -333,11 +319,10 @@ public class SearchArgumentToPredicateConverterTest {
.end()
.build();
Predicate expected =
- new Or(
- new Or(
- new Equal(1, new Literal(BIGINT_TYPE, 100L)),
- new Equal(1, new Literal(BIGINT_TYPE, 200L))),
- new Equal(1, new Literal(BIGINT_TYPE, 300L)));
+ PredicateBuilder.or(
+ PredicateBuilder.equal(1, new Literal(BIGINT_TYPE,
100L)),
+ PredicateBuilder.equal(1, new Literal(BIGINT_TYPE,
200L)),
+ PredicateBuilder.equal(1, new Literal(BIGINT_TYPE,
300L)));
assertExpected(sarg, expected);
}
