[common] Introduce Predicate interface and basic predicate expressions (#515)

Author: Alibaba-HZY

fluss-common/src/main/java/com/alibaba/fluss/predicate/And.java

@@ -0,0 +1,67 @@
+/*
+ * 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 com.alibaba.fluss.predicate;
+
+import com.alibaba.fluss.row.InternalRow;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** A {@link CompoundPredicate.Function} to eval and. */
+public class And extends CompoundPredicate.Function {
+
+    private static final long serialVersionUID = -2977938814804928712L;
+
+    public static final And INSTANCE = new And();
+
+    private And() {}
+
+    @Override
+    public boolean test(InternalRow row, List<Predicate> children) {
+        for (Predicate child : children) {
+            if (!child.test(row)) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Override
+    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));
+    }
+
+    @Override
+    public <T> T visit(FunctionVisitor<T> visitor, List<T> children) {
+        return visitor.visitAnd(children);
+    }
+}

fluss-common/src/main/java/com/alibaba/fluss/predicate/CompareUtils.java

@@ -0,0 +1,60 @@
+/*
+ * 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 com.alibaba.fluss.predicate;
+
+import com.alibaba.fluss.row.BinaryString;
+import com.alibaba.fluss.types.DataType;
+
+import static java.lang.Math.min;
+
+/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** Utils for comparator. */
+public class CompareUtils {
+    private CompareUtils() {}
+
+    public static int compareLiteral(DataType type, Object v1, Object v2) {
+        if (v1 instanceof Comparable) {
+            // because BinaryString can not serialize so v1 or v2 may be BinaryString convert to
+            // String for compare
+            if (v1 instanceof BinaryString) {
+                v1 = ((BinaryString) v1).toString();
+            }
+            if (v2 instanceof BinaryString) {
+                v2 = ((BinaryString) v2).toString();
+            }
+            return ((Comparable<Object>) v1).compareTo(v2);
+        } else if (v1 instanceof byte[]) {
+            return compare((byte[]) v1, (byte[]) v2);
+        } else {
+            throw new RuntimeException("Unsupported type: " + type);
+        }
+    }
+
+    private static int compare(byte[] first, byte[] second) {
+        for (int x = 0; x < min(first.length, second.length); x++) {
+            int cmp = first[x] - second[x];
+            if (cmp != 0) {
+                return cmp;
+            }
+        }
+        return first.length - second.length;
+    }
+}

fluss-common/src/main/java/com/alibaba/fluss/predicate/CompoundPredicate.java

@@ -0,0 +1,114 @@
+/*
+ * 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 com.alibaba.fluss.predicate;
+
+import com.alibaba.fluss.row.InternalRow;
+
+import java.io.Serializable;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * Non-leaf node in a {@link Predicate} tree. Its evaluation result depends on the results of its
+ * children.
+ */
+public class CompoundPredicate implements Predicate {
+
+    private final Function function;
+    private final List<Predicate> children;
+
+    public CompoundPredicate(Function function, List<Predicate> children) {
+        this.function = function;
+        this.children = children;
+    }
+
+    public Function function() {
+        return function;
+    }
+
+    public List<Predicate> children() {
+        return children;
+    }
+
+    @Override
+    public boolean test(InternalRow row) {
+        return function.test(row, children);
+    }
+
+    @Override
+    public Optional<Predicate> negate() {
+        return function.negate(children);
+    }
+
+    @Override
+    public <T> T visit(PredicateVisitor<T> visitor) {
+        return visitor.visit(this);
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (!(o instanceof CompoundPredicate)) {
+            return false;
+        }
+        CompoundPredicate that = (CompoundPredicate) o;
+        return Objects.equals(function, that.function) && Objects.equals(children, that.children);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(function, children);
+    }
+
+    @Override
+    public String toString() {
+        return function + "(" + children + ")";
+    }
+
+    /** Evaluate the predicate result based on multiple {@link Predicate}s. */
+    public abstract static class Function implements Serializable {
+
+        public abstract boolean test(InternalRow row, List<Predicate> children);
+
+        public abstract Optional<Predicate> negate(List<Predicate> children);
+
+        public abstract <T> T visit(FunctionVisitor<T> visitor, List<T> children);
+
+        @Override
+        public int hashCode() {
+            return this.getClass().getName().hashCode();
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) {
+                return true;
+            }
+            return o != null && getClass() == o.getClass();
+        }
+
+        @Override
+        public String toString() {
+            return getClass().getSimpleName();
+        }
+    }
+}

fluss-common/src/main/java/com/alibaba/fluss/predicate/Contains.java

@@ -0,0 +1,62 @@
+/*
+ * 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 com.alibaba.fluss.predicate;
+
+import com.alibaba.fluss.types.DataType;
+
+import java.util.List;
+import java.util.Optional;
+
+/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/** A {@link NullFalseLeafBinaryFunction} to evaluate {@code filter like '%abc%'}. */
+public class Contains extends NullFalseLeafBinaryFunction {
+
+    public static final Contains INSTANCE = new Contains();
+
+    private Contains() {}
+
+    @Override
+    public boolean test(DataType type, Object field, Object patternLiteral) {
+        String fieldString = field.toString();
+        return fieldString.contains((String) patternLiteral);
+    }
+
+    @Override
+    public boolean test(
+            DataType type,
+            long rowCount,
+            Object min,
+            Object max,
+            Long nullCount,
+            Object patternLiteral) {
+        return true;
+    }
+
+    @Override
+    public Optional<LeafFunction> negate() {
+        return Optional.empty();
+    }
+
+    @Override
+    public <T> T visit(FunctionVisitor<T> visitor, FieldRef fieldRef, List<Object> literals) {
+        return visitor.visitContains(fieldRef, literals.get(0));
+    }
+}

fluss-common/src/main/java/com/alibaba/fluss/predicate/EndsWith.java

@@ -0,0 +1,65 @@
+/*
+ * 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 com.alibaba.fluss.predicate;
+
+import com.alibaba.fluss.types.DataType;
+
+import java.util.List;
+import java.util.Optional;
+
+/* This file is based on source code of Apache Paimon Project (https://paimon.apache.org/), licensed by the Apache
+ * Software Foundation (ASF) under the Apache License, Version 2.0. See the NOTICE file distributed with this work for
+ * additional information regarding copyright ownership. */
+
+/**
+ * A {@link NullFalseLeafBinaryFunction} to evaluate {@code filter like '%abc' or filter like
+ * '_abc'}.
+ */
+public class EndsWith extends NullFalseLeafBinaryFunction {
+
+    public static final EndsWith INSTANCE = new EndsWith();
+
+    private EndsWith() {}
+
+    @Override
+    public boolean test(DataType type, Object field, Object patternLiteral) {
+        String fieldString = field.toString();
+        return fieldString.endsWith((String) patternLiteral);
+    }
+
+    @Override
+    public boolean test(
+            DataType type,
+            long rowCount,
+            Object min,
+            Object max,
+            Long nullCount,
+            Object patternLiteral) {
+        return true;
+    }
+
+    @Override
+    public Optional<LeafFunction> negate() {
+        return Optional.empty();
+    }
+
+    @Override
+    public <T> T visit(FunctionVisitor<T> visitor, FieldRef fieldRef, List<Object> literals) {
+        return visitor.visitEndsWith(fieldRef, literals.get(0));
+    }
+}