Repository: alibaba/GraphScope (language: python)

Issue:
Miss pk identify in some cases in GOpt
**Describe the bug**

Currently, in the GOpt-based compilation, the following query (on LDBC graph with "Person"'s "id" as pk) missed pk optimization in the plan:
```
g.V().has("PERSON", "id", 19791209300143)
```
The plan looks like:
```
  "plan": [{
    "opr": {
      "scan": {
        "params": {
          "tables": [{
            "id": 1
          }],
          "predicate": {
            "operators": [{
              "var": {
                "property": {
                  "key": {
                    "name": "id"
                  }
                },
                "nodeType": {
                  "dataType": "INT64"
                }
              },
              "nodeType": {
                "dataType": "INT64"
              }
            }, {
              "logical": "EQ",
              "nodeType": {
                "dataType": "BOOLEAN"
              }
            }, {
              "const": {
                "i64": "19791209300143"
              },
              "nodeType": {
                "dataType": "INT64"
              }
            }]
          },
          "sampleRatio": 1.0
        }
      }
    },
...
}
```

While the following query can correctly identify pk:
```
g.V().hasLabel("PERSON").has("id", 19791209300143)
```
The plan looks like:
```
  "plan": [{
    "opr": {
      "scan": {
        "params": {
          "tables": [{
            "id": 1
          }],
          "sampleRatio": 1.0
        },
        "idxPredicate": {
          "orPredicates": [{
            "predicates": [{
              "key": {
                "key": {
                  "name": "id"
                }
              },
              "const": {
                "i64": "19791209300143"
              }
            }]
          }]
        }
      }
    },
...
}
```



Discussion / hints:

Fix patch:
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rex/ClassifiedFilter.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rex/ClassifiedFilter.java
new file mode 100644
index 000000000000..ea6dde0057fe
--- /dev/null
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rex/ClassifiedFilter.java
@@ -0,0 +1,58 @@
+/*
+ *
+ *  * Copyright 2020 Alibaba Group Holding Limited.
+ *  *
+ *  * Licensed 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.graphscope.common.ir.rex;
+
+import org.apache.calcite.rex.RexNode;
+
+import java.util.Collections;
+import java.util.List;
+
+public class ClassifiedFilter {
+    private final List<RexNode> labelFilters;
+    private final List<Comparable> labelValues;
+    private final List<RexNode> uniqueKeyFilters;
+    private final List<RexNode> extraFilters;
+
+    public ClassifiedFilter(
+            List<RexNode> labelFilters,
+            List<Comparable> labelValues,
+            List<RexNode> uniqueKeyFilters,
+            List<RexNode> extraFilters) {
+        this.labelFilters = labelFilters;
+        this.labelValues = labelValues;
+        this.uniqueKeyFilters = uniqueKeyFilters;
+        this.extraFilters = extraFilters;
+    }
+
+    public List<RexNode> getLabelFilters() {
+        return Collections.unmodifiableList(labelFilters);
+    }
+
+    public List<Comparable> getLabelValues() {
+        return Collections.unmodifiableList(labelValues);
+    }
+
+    public List<RexNode> getUniqueKeyFilters() {
+        return Collections.unmodifiableList(uniqueKeyFilters);
+    }
+
+    public List<RexNode> getExtraFilters() {
+        return Collections.unmodifiableList(extraFilters);
+    }
+}
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rex/RexFilterClassifier.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rex/RexFilterClassifier.java
new file mode 100644
index 000000000000..4c268ff1e490
--- /dev/null
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rex/RexFilterClassifier.java
@@ -0,0 +1,430 @@
+/*
+ *
+ *  * Copyright 2020 Alibaba Group Holding Limited.
+ *  *
+ *  * Licensed 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.graphscope.common.ir.rex;
+
+import com.alibaba.graphscope.common.ir.rel.graph.AbstractBindableTableScan;
+import com.alibaba.graphscope.common.ir.rel.graph.GraphLogicalSource;
+import com.alibaba.graphscope.common.ir.rel.type.TableConfig;
+import com.alibaba.graphscope.common.ir.tools.GraphBuilder;
+import com.alibaba.graphscope.common.ir.tools.Utils;
+import com.alibaba.graphscope.common.ir.type.GraphLabelType;
+import com.alibaba.graphscope.common.ir.type.GraphNameOrId;
+import com.alibaba.graphscope.common.ir.type.GraphProperty;
+import com.alibaba.graphscope.common.ir.type.GraphSchemaType;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.*;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.Sarg;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.javatuples.Pair;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/**
+ * We use this class to decouple a complex condition that contains nested {@code OR} and {@code AND} operators into different filtering conditions.
+ * There are mainly three types of filtering:
+ * 1) filtering based on labels;
+ * 2) based on the scan's unique key;
+ * 3) querying other properties.
+ *
+ * For example: for the condition _.~label = 'person' AND _.~id = 1 AND _.name = 'marko',
+ * _.~label = 'person', _.~id = 1, and _.name = 'marko' would be decoupled into three different {@code RexNode} structures.
+ */
+public class RexFilterClassifier extends RexVisitorImpl<RexFilterClassifier.Filter> {
+    private final GraphBuilder builder;
+    private final @Nullable AbstractBindableTableScan tableScan;
+
+    public RexFilterClassifier(
+            GraphBuilder builder, @Nullable AbstractBindableTableScan tableScan) {
+        super(true);
+        this.builder = builder;
+        this.tableScan = tableScan;
+    }
+
+    public ClassifiedFilter classify(RexNode condition) {
+        Filter filter = condition.accept(this);
+        List<RexNode> labelFilters = Lists.newArrayList();
+        List<RexNode> uniqueKeyFilters = Lists.newArrayList();
+        filter.getSchemaFilters()
+                .forEach(
+                        k -> {
+                            switch (k.getSchemaType()) {
+                                case LABEL:
+                                    labelFilters.add(k.getFilter());
+                                    break;
+                                case UNIQUE_KEY:
+                                default:
+                                    uniqueKeyFilters.add(k.getFilter());
+                            }
+                        });
+        List<RexNode> extraFilters = Lists.newArrayList();
+        if (filter.getOtherFilter() != null) {
+            extraFilters.add(filter.getOtherFilter());
+        }
+        List<Comparable> labelValues = Lists.newArrayList();
+        labelFilters.forEach(k -> labelValues.addAll(getLabelValues(k)));
+        return new ClassifiedFilter(labelFilters, labelValues, uniqueKeyFilters, extraFilters);
+    }
+
+    private List<Comparable> getLabelValues(RexNode labelFilter) {
+        return labelFilter.accept(new LabelValueCollector());
+    }
+
+    @Override
+    public Filter visitCall(RexCall call) {
+        SqlOperator operator = call.getOperator();
+        List<RexNode> operands = call.getOperands();
+        switch (operator.getKind()) {
+            case AND:
+                return conjunctions(call, visitList(operands));
+            case OR:
+                return disjunctions(call);
+            case EQUALS:
+            case SEARCH:
+                RexVariableAliasCollector<Integer> aliasCollector =
+                        new RexVariableAliasCollector<>(
+                                true, (RexGraphVariable var) -> var.getAliasId());
+                if (isLabelEqualFilter(call)) {
+                    Integer tagId = call.accept(aliasCollector).get(0);
+                    return new Filter(
+                            ImmutableList.of(
+                                    new Filter.SchemaFilter(tagId, call, Filter.SchemaType.LABEL)),
+                            null);
+                } else if (tableScan != null && isUniqueKeyEqualFilter(call)) {
+                    Integer tagId = call.accept(aliasCollector).get(0);
+                    return new Filter(
+                            ImmutableList.of(
+                                    new Filter.SchemaFilter(
+                                            tagId, call, Filter.SchemaType.UNIQUE_KEY)),
+                            null);
+                }
+            default:
+                return new Filter(ImmutableList.of(), call);
+        }
+    }
+
+    private Filter conjunctions(RexNode original, List<Filter> filters) {
+        Map<Pair<Integer, Filter.SchemaType>, RexNode> schemaFilterMap = Maps.newLinkedHashMap();
+        List<RexNode> otherFilters = Lists.newArrayList();
+        filters.forEach(
+                k -> {
+                    k.getSchemaFilters()
+                            .forEach(
+                                    v -> {
+                                        Pair<Integer, Filter.SchemaType> key =
+                                                Pair.with(v.getTagId(), v.getSchemaType());
+                                        RexNode filtering = v.getFilter();
+                                        RexNode existing = schemaFilterMap.get(key);
+                                        if (existing != null) {
+                                            filtering = builder.and(existing, v.getFilter());
+                                        }
+                                        if (!filtering.equals(existing)) {
+                                            schemaFilterMap.put(key, filtering);
+                                        }
+                                    });
+                    if (k.getOtherFilter() != null) {
+                        otherFilters.add(k.getOtherFilter());
+                    }
+                });
+        List<Filter.SchemaFilter> andSchemaFilters = Lists.newArrayList();
+        schemaFilterMap.forEach(
+                (k, v) -> {
+                    andSchemaFilters.add(new Filter.SchemaFilter(k.getValue0(), v, k.getValue1()));
+                });
+        RexNode otherFilter =
+                otherFilters.isEmpty()
+                        ? null
+                        : RexUtil.composeConjunction(builder.getRexBuilder(), otherFilters, false);
+        return new Filter(andSchemaFilters, otherFilter);
+    }
+
+    private Filter disjunctions(RexCall original) {
+        SqlOperator operator = original.getOperator();
+        switch (operator.getKind()) {
+            case OR:
+                List<RexNode> operands = original.getOperands();
+                Filter.SchemaFilter schemaFilter = null;
+                for (RexNode operand : operands) {
+                    if (operand.getKind() != SqlKind.EQUALS
+                            && operand.getKind() != SqlKind.SEARCH) {
+                        return new Filter(ImmutableList.of(), original);
+                    }
+                    List<Filter.SchemaFilter> curFilters = operand.accept(this).getSchemaFilters();
+                    if (curFilters.size() != 1) {
+                        return new Filter(ImmutableList.of(), original);
+                    }
+                    Filter.SchemaFilter cur = curFilters.get(0);
+                    if (schemaFilter == null) {
+                        schemaFilter = cur;
+                    } else {
+                        if (schemaFilter.getTagId() == cur.getTagId()
+                                && schemaFilter.getSchemaType() == cur.getSchemaType()) {
+                            schemaFilter =
+                                    new Filter.SchemaFilter(
+                                            schemaFilter.getTagId(),
+                                            builder.or(schemaFilter.getFilter(), cur.getFilter()),
+                                            schemaFilter.getSchemaType());
+                        } else {
+                            return new Filter(ImmutableList.of(), original);
+                        }
+                    }
+                }
+            default:
+                return new Filter(ImmutableList.of(), original);
+        }
+    }
+
+    // check the condition if it is the pattern of label equal filter, i.e. ~label = 'person' or
+    // ~label within ['person', 'software']
+    // if it is then return the literal containing label values, otherwise null
+    private boolean isLabelEqualFilter(RexCall rexCall) {
+        return isLabelEqualFilter0(rexCall) != null;
+    }
+
+    // check the condition if it is the pattern of label equal filter, i.e. ~label = 'person' or
+    // ~label within ['person', 'software']
+    // if it is then return the literal containing label values, otherwise null
+    private static @Nullable RexLiteral isLabelEqualFilter0(RexNode condition) {
+        if (condition instanceof RexCall) {
+            RexCall rexCall = (RexCall) condition;
+            SqlOperator operator = rexCall.getOperator();
+            switch (operator.getKind()) {
+                case EQUALS:
+                case SEARCH:
+                    RexNode left = rexCall.getOperands().get(0);
+                    RexNode right = rexCall.getOperands().get(1);
+                    if (left.getType() instanceof GraphLabelType && right instanceof RexLiteral) {
+                        Comparable value = ((RexLiteral) right).getValue();
+                        // if Sarg is a continuous range then the filter is not the 'equal', i.e.
+                        // ~label SEARCH [[1, 10]] which means ~label >= 1 and ~label <= 10
+                        if (value instanceof Sarg && !((Sarg) value).isPoints()) {
+                            return null;
+                        }
+                        return (RexLiteral) right;
+                    } else if (right.getType() instanceof GraphLabelType
+                            && left instanceof RexLiteral) {
+                        Comparable value = ((RexLiteral) left).getValue();
+                        if (value instanceof Sarg && !((Sarg) value).isPoints()) {
+                            return null;
+                        }
+                        return (RexLiteral) left;
+                    }
+                default:
+                    return null;
+            }
+        } else {
+            return null;
+        }
+    }
+
+    private boolean isUniqueKeyEqualFilter(RexNode condition) {
+        if (!(tableScan instanceof GraphLogicalSource)) return false;
+        if (condition instanceof RexCall) {
+            RexCall rexCall = (RexCall) condition;
+            SqlOperator operator = rexCall.getOperator();
+            switch (operator.getKind()) {
+                case EQUALS:
+                case SEARCH:
+                    RexNode left = rexCall.getOperands().get(0);
+                    RexNode right = rexCall.getOperands().get(1);
+                    if (isUniqueKey(left, tableScan) && isLiteralOrDynamicParams(right)) {
+                        if (right instanceof RexLiteral) {
+                            Comparable value = ((RexLiteral) right).getValue();
+                            // if Sarg is a continuous range then the filter is not the 'equal',
+                            // i.e. ~id SEARCH [[1, 10]] which means ~id >= 1 and ~id <= 10
+                            if (value instanceof Sarg && !((Sarg) value).isPoints()) {
+                                return false;
+                            }
+                        }
+                        return true;
+                    } else if (isUniqueKey(right, tableScan) && isLiteralOrDynamicParams(left)) {
+                        if (left instanceof RexLiteral) {
+                            Comparable value = ((RexLiteral) left).getValue();
+                            if (value instanceof Sarg && !((Sarg) value).isPoints()) {
+                                return false;
+                            }
+                        }
+                        return true;
+                    }
+                default:
+                    return false;
+            }
+        } else {
+            return false;
+        }
+    }
+
+    private boolean isUniqueKey(RexNode rexNode, RelNode tableScan) {
+        if (rexNode instanceof RexGraphVariable) {
+            return isUniqueKey((RexGraphVariable) rexNode, tableScan);
+        }
+        return false;
+    }
+
+    private boolean isUniqueKey(RexGraphVariable var, RelNode tableScan) {
+        if (var.getProperty() == null) return false;
+        switch (var.getProperty().getOpt()) {
+            case ID:
+                return true;
+            case KEY:
+                GraphSchemaType schemaType =
+                        (GraphSchemaType) tableScan.getRowType().getFieldList().get(0).getType();
+                ImmutableBitSet propertyIds = getPropertyIds(var.getProperty(), schemaType);
+                TableConfig tableConfig = ((AbstractBindableTableScan) tableScan).getTableConfig();
+                if (!propertyIds.isEmpty()
+                        && tableConfig.getTables().stream().allMatch(k -> k.isKey(propertyIds))) {
+                    return true;
+                }
+            case LABEL:
+            case ALL:
+            case LEN:
+            default:
+                return false;
+        }
+    }
+
+    private ImmutableBitSet getPropertyIds(GraphProperty property, GraphSchemaType schemaType) {
+        if (property.getOpt() != GraphProperty.Opt.KEY) return ImmutableBitSet.of();
+        GraphNameOrId key = property.getKey();
+        if (key.getOpt() == GraphNameOrId.Opt.ID) {
+            return ImmutableBitSet.of(key.getId());
+        }
+        for (int i = 0; i < schemaType.getFieldList().size(); ++i) {
+            RelDataTypeField field = schemaType.getFieldList().get(i);
+            if (field.getName().equals(key.getName())) {
+                return ImmutableBitSet.of(i);
+            }
+        }
+        return ImmutableBitSet.of();
+    }
+
+    private boolean isLiteralOrDynamicParams(RexNode node) {
+        return node instanceof RexLiteral || node instanceof RexDynamicParam;
+    }
+
+    private static class LabelValueCollector extends RexVisitorImpl<List<Comparable>> {
+        public LabelValueCollector() {
+            super(true);
+        }
+
+        @Override
+        public List<Comparable> visitCall(RexCall call) {
+            SqlOperator operator = call.getOperator();
+            switch (operator.getKind()) {
+                case AND:
+                    List<Comparable> andLabels = Lists.newArrayList();
+                    call.getOperands()
+                            .forEach(
+                                    k -> {
+                                        List<Comparable> cur = k.accept(this);
+                                        if (andLabels.isEmpty()) {
+                                            andLabels.addAll(cur);
+                                        } else {
+                                            andLabels.retainAll(cur);
+                                        }
+                                        if (andLabels.isEmpty()) {
+                                            throw new IllegalArgumentException(
+                                                    "cannot find common labels between values="
+                                                            + andLabels
+                                                            + " and values="
+                                                            + cur);
+                                        }
+                                    });
+                case OR:
+                    List<Comparable> orLabels = Lists.newArrayList();
+                    call.getOperands()
+                            .forEach(
+                                    k -> {
+                                        orLabels.addAll(k.accept(this));
+                                    });
+                    return orLabels.stream().distinct().collect(Collectors.toList());
+                case EQUALS:
+                case SEARCH:
+                    RexLiteral labelLiteral = isLabelEqualFilter0(call);
+                    if (labelLiteral != null) {
+                        return Utils.getValuesAsList(labelLiteral.getValueAs(Comparable.class));
+                    }
+                default:
+                    return ImmutableList.of();
+            }
+        }
+    }
+
+    // Here we further differentiate the filter conditions for different tags to prevent the
+    // grouping of label conditions from various tags.
+    // For example, _.~label == 'person' OR _.~label = 'software' can be organized into _.~label IN
+    // ['person', 'software'],
+    // but a.~label = 'person' OR b.~label = 'software' cannot be.
+    public static class Filter {
+        private final List<SchemaFilter> schemaFilters;
+        private final @Nullable RexNode otherFilter;
+
+        public Filter(List<SchemaFilter> schemaFilters, RexNode otherFilter) {
+            this.schemaFilters = schemaFilters;
+            this.otherFilter = otherFilter;
+        }
+
+        public List<SchemaFilter> getSchemaFilters() {
+            return Collections.unmodifiableList(schemaFilters);
+        }
+
+        public @Nullable RexNode getOtherFilter() {
+            return otherFilter;
+        }
+
+        public static class SchemaFilter {
+            private final Integer tagId;
+            private final RexNode filter;
+            private final SchemaType schemaType;
+
+            public SchemaFilter(Integer tagId, RexNode filtering, SchemaType schemaType) {
+                this.tagId = tagId;
+                this.filter = filtering;
+                this.schemaType = schemaType;
+            }
+
+            public Integer getTagId() {
+                return tagId;
+            }
+
+            public RexNode getFilter() {
+                return filter;
+            }
+
+            public SchemaType getSchemaType() {
+                return schemaType;
+            }
+        }
+
+        public enum SchemaType {
+            LABEL,
+            UNIQUE_KEY
+        }
+    }
+}
diff --git a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphBuilder.java b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphBuilder.java
index 8c83cb7adf47..56b68bacd9ce 100644
--- a/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphBuilder.java
+++ b/interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphBuilder.java
@@ -59,10 +59,8 @@
 import org.apache.calcite.sql.type.IntervalSqlType;
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.calcite.tools.RelBuilder;
-import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.calcite.util.Litmus;
 import org.apache.calcite.util.Pair;
-import org.apache.calcite.util.Sarg;
 import org.apache.commons.lang3.ObjectUtils;
 import org.checkerframework.checker.nullness.qual.Nullable;
 
@@ -932,10 +930,11 @@ public GraphBuilder filter(Iterable<? extends RexNode> conditions) {
 
     private AbstractBindableTableScan fuseFilters(
             AbstractBindableTableScan tableScan, RexNode condition, GraphBuilder builder) {
-        List<Comparable> labelValues = Lists.newArrayList();
-        List<RexNode> uniqueKeyFilters = Lists.newArrayList();
-        List<RexNode> extraFilters = Lists.newArrayList();
-        classifyFilters(tableScan, condition, labelValues, uniqueKeyFilters, extraFilters);
+        RexFilterClassifier classifier = new RexFilterClassifier(builder, tableScan);
+        ClassifiedFilter filterResult = classifier.classify(condition);
+        List<Comparable> labelValues = filterResult.getLabelValues();
+        List<RexNode> uniqueKeyFilters = Lists.newArrayList(filterResult.getUniqueKeyFilters());
+        List<RexNode> extraFilters = Lists.newArrayList(filterResult.getExtraFilters());
         if (!labelValues.isEmpty()) {
             GraphLabelType labelType =
                     ((GraphSchemaType) tableScan.getRowType().getFieldList().get(0).getType())
@@ -983,6 +982,10 @@ private AbstractBindableTableScan fuseFilters(
                             originalUniqueKeyFilters.accept(propertyChecker);
                             builder.filter(originalUniqueKeyFilters);
                         }
+                        if (!uniqueKeyFilters.isEmpty()) {
+                            builder.filter(uniqueKeyFilters);
+                            uniqueKeyFilters.clear();
+                        }
                     }
                     ImmutableList originalFilters = tableScan.getFilters();
                     if (ObjectUtils.isNotEmpty(originalFilters)) {
@@ -991,6 +994,8 @@ private AbstractBindableTableScan fuseFilters(
                     }
                     if (!extraFilters.isEmpty()) {
                         extraFilters.forEach(k -> k.accept(propertyChecker));
+                        builder.filter(extraFilters);
+                        extraFilters.clear();
                     }
                     tableScan = (AbstractBindableTableScan) builder.build();
                 }
@@ -998,11 +1003,11 @@ private AbstractBindableTableScan fuseFilters(
         }
         if (tableScan instanceof GraphLogicalSource && !uniqueKeyFilters.isEmpty()) {
             GraphLogicalSource source = (GraphLogicalSource) tableScan;
-            Preconditions.checkArgument(
-                    source.getUniqueKeyFilters() == null,
-                    "can not add unique key filters if original is not empty");
-            source.setUniqueKeyFilters(
-                    RexUtil.composeDisjunction(this.getRexBuilder(), uniqueKeyFilters));
+            if (source.getUniqueKeyFilters() != null || uniqueKeyFilters.size() > 1) {
+                extraFilters.addAll(uniqueKeyFilters);
+            } else {
+                source.setUniqueKeyFilters(uniqueKeyFilters.get(0));
+            }
         }
         if (!extraFilters.isEmpty()) {
             ImmutableList originalFilters = tableScan.getFilters();
@@ -1031,14 +1036,10 @@ private AbstractLogicalMatch fuseFilters(
             RexNode condition,
             List<RexNode> extraFilters,
             GraphBuilder builder) {
-        List<RexNode> labelFilters = Lists.newArrayList();
-        for (RexNode conjunction : RelOptUtil.conjunctions(condition)) {
-            if (isLabelEqualFilter(conjunction) != null) {
-                labelFilters.add(conjunction);
-            } else {
-                extraFilters.add(conjunction);
-            }
-        }
+        RexFilterClassifier classifier = new RexFilterClassifier(builder, null);
+        ClassifiedFilter filter = classifier.classify(condition);
+        List<RexNode> labelFilters = filter.getLabelFilters();
+        extraFilters.addAll(filter.getExtraFilters());
         for (RexNode labelFilter : labelFilters) {
             PushFilterVisitor visitor = new PushFilterVisitor(builder, labelFilter);
             match = (AbstractLogicalMatch) match.accept(visitor);
@@ -1049,161 +1050,6 @@ private AbstractLogicalMatch fuseFilters(
         return match;
     }
 
-    private void classifyFilters(
-            AbstractBindableTableScan tableScan,
-            RexNode condition,
-            List<Comparable> labelValues,
-            List<RexNode> uniqueKeyFilters, // unique key filters int the list are composed by 'OR'
-            List<RexNode> filters) {
-        List<RexNode> conjunctions = RelOptUtil.conjunctions(condition);
-        List<RexNode> filtersToRemove = Lists.newArrayList();
-        for (RexNode conjunction : conjunctions) {
-            RexLiteral labelLiteral = isLabelEqualFilter(conjunction);
-            if (labelLiteral != null) {
-                filtersToRemove.add(conjunction);
-                labelValues.addAll(
-                        com.alibaba.graphscope.common.ir.tools.Utils.getValuesAsList(
-                                labelLiteral.getValueAs(Comparable.class)));
-                break;
-            }
-        }
-        if (tableScan instanceof GraphLogicalSource
-                && ((GraphLogicalSource) tableScan).getUniqueKeyFilters() == null) {
-            // try to extract unique key filters from the original condition
-            List<RexNode> disjunctions = RelOptUtil.disjunctions(condition);
-            for (RexNode disjunction : disjunctions) {
-                if (isUniqueKeyEqualFilter(disjunction, tableScan)) {
-                    filtersToRemove.add(disjunction);
-                    uniqueKeyFilters.add(disjunction);
-                }
-            }
-        }
-        if (!filtersToRemove.isEmpty()) {
-            conjunctions.removeAll(filtersToRemove);
-        }
-        filters.addAll(conjunctions);
-    }
-
-    // check the condition if it is the pattern of label equal filter, i.e. ~label = 'person' or
-    // ~label within ['person', 'software']
-    // if it is then return the literal containing label values, otherwise null
-    private @Nullable RexLiteral isLabelEqualFilter(RexNode condition) {
-        if (condition instanceof RexCall) {
-            RexCall rexCall = (RexCall) condition;
-            SqlOperator operator = rexCall.getOperator();
-            switch (operator.getKind()) {
-                case EQUALS:
-                case SEARCH:
-                    RexNode left = rexCall.getOperands().get(0);
-                    RexNode right = rexCall.getOperands().get(1);
-                    if (left.getType() instanceof GraphLabelType && right instanceof RexLiteral) {
-                        Comparable value = ((RexLiteral) right).getValue();
-                        // if Sarg is a continuous range then the filter is not the 'equal', i.e.
-                        // ~label SEARCH [[1, 10]] which means ~label >= 1 and ~label <= 10
-                        if (value instanceof Sarg && !((Sarg) value).isPoints()) {
-                            return null;
-                        }
-                        return (RexLiteral) right;
-                    } else if (right.getType() instanceof GraphLabelType
-                            && left instanceof RexLiteral) {
-                        Comparable value = ((RexLiteral) left).getValue();
-                        if (value instanceof Sarg && !((Sarg) value).isPoints()) {
-                            return null;
-                        }
-                        return (RexLiteral) left;
-                    }
-                default:
-                    return null;
-            }
-        } else {
-            return null;
-        }
-    }
-
-    // check the condition if it is the pattern of unique key equal filter, i.e. ~id = 1 or ~id
-    // within [1, 2]
-    private boolean isUniqueKeyEqualFilter(RexNode condition, RelNode tableScan) {
-        if (condition instanceof RexCall) {
-            RexCall rexCall = (RexCall) condition;
-            SqlOperator operator = rexCall.getOperator();
-            switch (operator.getKind()) {
-                case EQUALS:
-                case SEARCH:
-                    RexNode left = rexCall.getOperands().get(0);
-                    RexNode right = rexCall.getOperands().get(1);
-                    if (isUniqueKey(left, tableScan) && isLiteralOrDynamicParams(right)) {
-                        if (right instanceof RexLiteral) {
-                            Comparable value = ((RexLiteral) right).getValue();
-                            // if Sarg is a continuous range then the filter is not the 'equal',
-                            // i.e. ~id SEARCH [[1, 10]] which means ~id >= 1 and ~id <= 10
-                            if (value instanceof Sarg && !((Sarg) value).isPoints()) {
-                                return false;
-                            }
-                        }
-                        return true;
-                    } else if (isUniqueKey(right, tableScan) && isLiteralOrDynamicParams(left)) {
-                        if (left instanceof RexLiteral) {
-                            Comparable value = ((RexLiteral) left).getValue();
-                            if (value instanceof Sarg && !((Sarg) value).isPoints()) {
-                                return false;
-                            }
-                        }
-                        return true;
-                    }
-                default:
-                    return false;
-            }
-        } else {
-            return false;
-        }
-    }
-
-    private boolean isUniqueKey(RexNode rexNode, RelNode tableScan) {
-        if (rexNode instanceof RexGraphVariable) {
-            return isUniqueKey((RexGraphVariable) rexNode, tableScan);
-        }
-        return false;
-    }
-
-    private boolean isUniqueKey(RexGraphVariable var, RelNode tableScan) {
-        if (var.getProperty() == null) return false;
-        switch (var.getProperty().getOpt()) {
-            case ID:
-                return true;
-            case KEY:
-                GraphSchemaType schemaType =
-                        (GraphSchemaType) tableScan.getRowType().getFieldList().get(0).getType();
-                ImmutableBitSet propertyIds = getPropertyIds(var.getProperty(), schemaType);
-                if (!propertyIds.isEmpty() && tableScan.getTable().isKey(propertyIds)) {
-                    return true;
-                }
-            case LABEL:
-            case ALL:
-            case LEN:
-            default:
-                return false;
-        }
-    }
-
-    private ImmutableBitSet getPropertyIds(GraphProperty property, GraphSchemaType schemaType) {
-        if (property.getOpt() != GraphProperty.Opt.KEY) return ImmutableBitSet.of();
-        GraphNameOrId key = property.getKey();
-        if (key.getOpt() == GraphNameOrId.Opt.ID) {
-            return ImmutableBitSet.of(key.getId());
-        }
-        for (int i = 0; i < schemaType.getFieldList().size(); ++i) {
-            RelDataTypeField field = schemaType.getFieldList().get(i);
-            if (field.getName().equals(key.getName())) {
-                return ImmutableBitSet.of(i);
-            }
-        }
-        return ImmutableBitSet.of();
-    }
-
-    private boolean isLiteralOrDynamicParams(RexNode node) {
-        return node instanceof RexLiteral || node instanceof RexDynamicParam;
-    }
-
     // return the top node if its type is Filter, otherwise null
     private Filter topFilter() {
         if (this.size() > 0 && this.peek() instanceof Filter) {


Test patch:
diff --git a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/gremlin/antlr4x/GraphBuilderTest.java b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/gremlin/antlr4x/GraphBuilderTest.java
index 61eee357fdee..3adfb8cea4a4 100644
--- a/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/gremlin/antlr4x/GraphBuilderTest.java
+++ b/interactive_engine/compiler/src/test/java/com/alibaba/graphscope/gremlin/antlr4x/GraphBuilderTest.java
@@ -1595,4 +1595,26 @@ public void g_V_match_as_a_out_as_b_count_test() {
                     + "], matchOpt=[INNER])",
                 node.explain().trim());
     }
+
+    // id is the primary key of label 'person', should be fused as 'uniqueKeyFilters'
+    @Test
+    public void g_V_has_label_person_id_test() {
+        RelNode node = eval("g.V().has(\"person\", \"id\", 1)");
+        Assert.assertEquals(
+                "GraphLogicalProject($f0=[_], isAppend=[false])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[person]}],"
+                        + " alias=[_], opt=[VERTEX], uniqueKeyFilters=[=(_.id, 1)])",
+                node.explain().trim());
+    }
+
+    // id is not the primary key of label 'software', should be fused as 'fusedFilter'
+    @Test
+    public void g_V_has_label_software_id_test() {
+        RelNode node = eval("g.V().has(\"software\", \"id\", 1)");
+        Assert.assertEquals(
+                "GraphLogicalProject($f0=[_], isAppend=[false])\n"
+                        + "  GraphLogicalSource(tableConfig=[{isAll=false, tables=[software]}],"
+                        + " alias=[_], fusedFilter=[[=(_.id, 1)]], opt=[VERTEX])",
+                node.explain().trim());
+    }
 }