fix(interactive): Support Path Expand Until Condition in Gremlin-Calcite

interactive_engine/compiler/src/main/antlr4/GremlinGS.g4

@@ -189,6 +189,7 @@ traversalMethod_bothE
 traversalMethod_with
     : 'with' LPAREN StringLiteral COMMA oC_Literal RPAREN
     | 'with' LPAREN evaluationTimeoutKey COMMA evaluationTimeoutValue RPAREN
+    | 'with' LPAREN StringLiteral COMMA (ANON_TRAVERSAL_ROOT DOT)? traversalMethod_expr RPAREN // to support path until condition in gremlin-calcite, i.e. with('UNTIL', expr(_.age > 20))
     ;
 
 evaluationTimeoutKey

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/GraphFieldTrimmer.java

@@ -450,6 +450,7 @@ public TrimResult trimFields(GraphLogicalPathExpand pathExpand, UsedFields field
                         pathExpand.getFetch(),
                         pathExpand.getResultOpt(),
                         pathExpand.getPathOpt(),
+                        pathExpand.getUntilCondition(),
                         pathExpand.getAliasName(),
                         pathExpand.getStartAlias());
         return result(newPathExpand, mapping, pathExpand);

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/GraphIOProcessor.java

@@ -213,6 +213,10 @@ public void visit(RelNode node, int ordinal, @Nullable RelNode parent) {
                                         parent instanceof GraphLogicalGetV,
                                         "there should be a getV operator after path expand since"
                                                 + " edge in patten should have two endpoints");
+                                Preconditions.checkArgument(
+                                        ((GraphLogicalPathExpand) node).getUntilCondition() == null,
+                                        "cannot apply optimization if path expand has until"
+                                                + " conditions");
                                 PatternVertex vertex = visitAndAddVertex((GraphLogicalGetV) parent);
                                 visitAndAddPxdEdge(
                                         (GraphLogicalPathExpand) node, lastVisited, vertex);
@@ -809,7 +813,8 @@ private RelNode createExpandGetV(
                                 edge.getElementDetails().getRange().getOffset(),
                                 edge.getElementDetails().getRange().getFetch());
                 GraphLogicalPathExpand pxd =
-                        (GraphLogicalPathExpand) builder.pathExpand(pxdBuilder.build()).build();
+                        (GraphLogicalPathExpand)
+                                builder.pathExpand(pxdBuilder.buildConfig()).build();
                 GraphLogicalExpand expand = (GraphLogicalExpand) pxd.getExpand();
                 GraphSchemaType edgeType =
                         (GraphSchemaType) expand.getRowType().getFieldList().get(0).getType();
@@ -858,6 +863,7 @@ private GraphLogicalPathExpand createPathExpandWithOptional(
                         pxd.getFetch(),
                         pxd.getResultOpt(),
                         pxd.getPathOpt(),
+                        pxd.getUntilCondition(),
                         pxd.getAliasName(),
                         pxd.getStartAlias(),
                         optional);
@@ -872,6 +878,7 @@ private GraphLogicalPathExpand createPathExpandWithOptional(
                         pxd.getFetch(),
                         pxd.getResultOpt(),
                         pxd.getPathOpt(),
+                        pxd.getUntilCondition(),
                         pxd.getAliasName(),
                         pxd.getStartAlias(),
                         optional);

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/planner/rules/ExpandGetVFusionRule.java

@@ -308,6 +308,7 @@ public void onMatch(RelOptRuleCall call) {
                             pathExpand.getFetch(),
                             pathExpand.getResultOpt(),
                             pathExpand.getPathOpt(),
+                            pathExpand.getUntilCondition(),
                             pathExpand.getAliasName(),
                             pathExpand.getStartAlias(),
                             pathExpand.isOptional());

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rel/graph/GraphLogicalPathExpand.java

@@ -63,6 +63,8 @@ public class GraphLogicalPathExpand extends SingleRel {
 
     private final boolean optional;
 
+    private final @Nullable RexNode untilCondition;
+
     protected GraphLogicalPathExpand(
             GraphOptCluster cluster,
             @Nullable List<RelHint> hints,
@@ -73,6 +75,7 @@ protected GraphLogicalPathExpand(
             @Nullable RexNode fetch,
             GraphOpt.PathExpandResult resultOpt,
             GraphOpt.PathExpandPath pathOpt,
+            @Nullable RexNode untilCondition,
             @Nullable String aliasName,
             AliasNameWithId startAlias,
             boolean optional) {
@@ -97,6 +100,7 @@ protected GraphLogicalPathExpand(
                 (this.aliasId == AliasInference.DEFAULT_ID)
                         ? resultOpt
                         : GraphOpt.PathExpandResult.ALL_V_E;
+        this.untilCondition = untilCondition;
     }
 
     protected GraphLogicalPathExpand(
@@ -108,6 +112,7 @@ protected GraphLogicalPathExpand(
             @Nullable RexNode fetch,
             GraphOpt.PathExpandResult resultOpt,
             GraphOpt.PathExpandPath pathOpt,
+            @Nullable RexNode untilCondition,
             @Nullable String aliasName,
             AliasNameWithId startAlias,
             boolean optional) {
@@ -125,6 +130,7 @@ protected GraphLogicalPathExpand(
         this.aliasId = cluster.getIdGenerator().generate(this.aliasName);
         this.startAlias = Objects.requireNonNull(startAlias);
         this.optional = optional;
+        this.untilCondition = untilCondition;
     }
 
     public static GraphLogicalPathExpand create(
@@ -137,6 +143,7 @@ public static GraphLogicalPathExpand create(
             @Nullable RexNode fetch,
             GraphOpt.PathExpandResult resultOpt,
             GraphOpt.PathExpandPath pathOpt,
+            @Nullable RexNode untilCondition,
             String aliasName,
             AliasNameWithId startAlias,
             boolean optional) {
@@ -150,6 +157,7 @@ public static GraphLogicalPathExpand create(
                 fetch,
                 resultOpt,
                 pathOpt,
+                untilCondition,
                 aliasName,
                 startAlias,
                 optional);
@@ -165,6 +173,7 @@ public static GraphLogicalPathExpand create(
             @Nullable RexNode fetch,
             GraphOpt.PathExpandResult resultOpt,
             GraphOpt.PathExpandPath pathOpt,
+            @Nullable RexNode untilCondition,
             String aliasName,
             AliasNameWithId startAlias) {
         return create(
@@ -177,6 +186,7 @@ public static GraphLogicalPathExpand create(
                 fetch,
                 resultOpt,
                 pathOpt,
+                untilCondition,
                 aliasName,
                 startAlias,
                 false);
@@ -191,6 +201,7 @@ public static GraphLogicalPathExpand create(
             @Nullable RexNode fetch,
             GraphOpt.PathExpandResult resultOpt,
             GraphOpt.PathExpandPath pathOpt,
+            @Nullable RexNode untilCondition,
             String aliasName,
             AliasNameWithId startAlias) {
         return create(
@@ -202,6 +213,7 @@ public static GraphLogicalPathExpand create(
                 fetch,
                 resultOpt,
                 pathOpt,
+                untilCondition,
                 aliasName,
                 startAlias,
                 false);
@@ -216,6 +228,7 @@ public static GraphLogicalPathExpand create(
             @Nullable RexNode fetch,
             GraphOpt.PathExpandResult resultOpt,
             GraphOpt.PathExpandPath pathOpt,
+            @Nullable RexNode untilCondition,
             String aliasName,
             AliasNameWithId startAlias,
             boolean optional) {
@@ -231,6 +244,7 @@ public static GraphLogicalPathExpand create(
                 fetch,
                 resultOpt,
                 pathOpt,
+                untilCondition,
                 aliasName,
                 startAlias,
                 optional);
@@ -246,6 +260,7 @@ public RelWriter explainTerms(RelWriter pw) {
                 .itemIf("fetch", fetch, fetch != null)
                 .item("path_opt", getPathOpt())
                 .item("result_opt", getResultOpt())
+                .itemIf("until_condition", untilCondition, untilCondition != null)
                 .item("alias", AliasInference.SIMPLE_NAME(getAliasName()))
                 .itemIf(
                         "aliasId",
@@ -306,6 +321,10 @@ public boolean isOptional() {
         return optional;
     }
 
+    public @Nullable RexNode getUntilCondition() {
+        return untilCondition;
+    }
+
     @Override
     protected RelDataType deriveRowType() {
         return new RelRecordType(
@@ -352,6 +371,7 @@ public GraphLogicalPathExpand copy(RelTraitSet traitSet, List<RelNode> inputs) {
                     getFetch(),
                     getResultOpt(),
                     getPathOpt(),
+                    getUntilCondition(),
                     getAliasName(),
                     getStartAlias(),
                     isOptional());
@@ -366,6 +386,7 @@ public GraphLogicalPathExpand copy(RelTraitSet traitSet, List<RelNode> inputs) {
                     getFetch(),
                     getResultOpt(),
                     getPathOpt(),
+                    getUntilCondition(),
                     getAliasName(),
                     getStartAlias(),
                     isOptional());

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/runtime/proto/GraphRelToProtoConverter.java

@@ -260,6 +260,12 @@
         pathExpandBuilder.setResultOpt(Utils.protoPathResultOpt(pxd.getResultOpt()));
         GraphAlgebra.Range range = buildRange(pxd.getOffset(), pxd.getFetch());
         pathExpandBuilder.setHopRange(range);
+        if (pxd.getUntilCondition() != null) {
+            OuterExpression.Expression untilCondition =
+                    pxd.getUntilCondition()
+                            .accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder));
+            pathExpandBuilder.setCondition(untilCondition);
+        }
         if (pxd.getAliasId() != AliasInference.DEFAULT_ID) {
             pathExpandBuilder.setAlias(Utils.asAliasId(pxd.getAliasId()));
         }
@@ -456,162 +462,162 @@
         return applyBuilder;
     }
 
     @Override
     public RelNode visit(GraphLogicalAggregate aggregate) {
         visitChildren(aggregate);
         List<RelDataTypeField> fields = aggregate.getRowType().getFieldList();
         List<GraphAggCall> groupCalls = aggregate.getAggCalls();
         GraphGroupKeys keys = aggregate.getGroupKey();
         if (groupCalls.isEmpty()) { // transform to project + dedup by keys
             Preconditions.checkArgument(
                     keys.groupKeyCount() > 0,
                     "group keys should not be empty while group calls is empty");
             GraphAlgebraPhysical.Project.Builder projectBuilder =
                     GraphAlgebraPhysical.Project.newBuilder();
             for (int i = 0; i < keys.groupKeyCount(); ++i) {
                 RexNode var = keys.getVariables().get(i);
                 Preconditions.checkArgument(
                         var instanceof RexGraphVariable,
                         "each group key should be type %s, but is %s",
                         RexGraphVariable.class,
                         var.getClass());
                 OuterExpression.Expression expr =
                         var.accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder));
                 int aliasId;
                 if (i >= fields.size()
                         || (aliasId = fields.get(i).getIndex()) == AliasInference.DEFAULT_ID) {
                     throw new IllegalArgumentException(
                             "each group key should have an alias if need dedup");
                 }
                 GraphAlgebraPhysical.Project.ExprAlias.Builder projectExprAliasBuilder =
                         GraphAlgebraPhysical.Project.ExprAlias.newBuilder();
                 projectExprAliasBuilder.setExpr(expr);
                 if (aliasId != AliasInference.DEFAULT_ID) {
                     projectExprAliasBuilder.setAlias(Utils.asAliasId(aliasId));
                 }
                 projectBuilder.addMappings(projectExprAliasBuilder.build());
             }
             GraphAlgebra.Dedup.Builder dedupBuilder = GraphAlgebra.Dedup.newBuilder();
             for (int i = 0; i < keys.groupKeyCount(); ++i) {
                 RelDataTypeField field = fields.get(i);
                 RexVariable rexVar =
                         RexGraphVariable.of(
                                 field.getIndex(),
                                 AliasInference.DEFAULT_COLUMN_ID,
                                 field.getName(),
                                 field.getType());
                 OuterExpression.Variable exprVar =
                         rexVar.accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
                                 .getOperators(0)
                                 .getVar();
                 dedupBuilder.addKeys(exprVar);
             }
             GraphAlgebraPhysical.PhysicalOpr.Builder projectOprBuilder =
                     GraphAlgebraPhysical.PhysicalOpr.newBuilder();
             projectOprBuilder.setOpr(
                     GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
                             .setProject(projectBuilder));
             GraphAlgebraPhysical.PhysicalOpr.Builder dedupOprBuilder =
                     GraphAlgebraPhysical.PhysicalOpr.newBuilder();
             dedupOprBuilder.setOpr(
                     GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setDedup(dedupBuilder));
             if (isPartitioned) {
                 Map<Integer, Set<GraphNameOrId>> tagColumns =
                         Utils.extractTagColumnsFromRexNodes(keys.getVariables());
                 if (preCacheEdgeProps) {
                     Utils.removeEdgeProperties(
                             com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(
                                     aggregate.getInput()),
                             tagColumns);
                 }
                 lazyPropertyFetching(tagColumns);
             }
             physicalBuilder.addPlan(projectOprBuilder.build());
             physicalBuilder.addPlan(dedupOprBuilder.build());
         } else {
             GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
                     GraphAlgebraPhysical.PhysicalOpr.newBuilder();
             GraphAlgebraPhysical.GroupBy.Builder groupByBuilder =
                     GraphAlgebraPhysical.GroupBy.newBuilder();
             for (int i = 0; i < keys.groupKeyCount(); ++i) {
                 RexNode var = keys.getVariables().get(i);
                 Preconditions.checkArgument(
                         var instanceof RexGraphVariable,
                         "each group key should be type %s, but is %s",
                         RexGraphVariable.class,
                         var.getClass());
                 OuterExpression.Variable exprVar =
                         var.accept(new RexToProtoConverter(true, isColumnId, this.rexBuilder))
                                 .getOperators(0)
                                 .getVar();
                 int aliasId = fields.get(i).getIndex();
                 GraphAlgebraPhysical.GroupBy.KeyAlias.Builder keyAliasBuilder =
                         GraphAlgebraPhysical.GroupBy.KeyAlias.newBuilder();
                 keyAliasBuilder.setKey(exprVar);
                 if (aliasId != AliasInference.DEFAULT_ID) {
                     keyAliasBuilder.setAlias(Utils.asAliasId(aliasId));
                 }
                 groupByBuilder.addMappings(keyAliasBuilder);
             }
             for (int i = 0; i < groupCalls.size(); ++i) {
                 List<RexNode> operands = groupCalls.get(i).getOperands();
                 if (operands.isEmpty()) {
                     throw new IllegalArgumentException(
                             "operands in aggregate call should not be empty");
                 }
 
                 GraphAlgebraPhysical.GroupBy.AggFunc.Builder aggFnAliasBuilder =
                         GraphAlgebraPhysical.GroupBy.AggFunc.newBuilder();
                 for (RexNode operand : operands) {
                     Preconditions.checkArgument(
                             operand instanceof RexGraphVariable,
                             "each expression in aggregate call should be type %s, but is %s",
                             RexGraphVariable.class,
                             operand.getClass());
                     OuterExpression.Variable var =
                             operand.accept(
                                             new RexToProtoConverter(
                                                     true, isColumnId, this.rexBuilder))
                                     .getOperators(0)
                                     .getVar();
                     aggFnAliasBuilder.addVars(var);
                 }
                 GraphAlgebraPhysical.GroupBy.AggFunc.Aggregate aggOpt =
                         Utils.protoAggOpt(groupCalls.get(i));
                 aggFnAliasBuilder.setAggregate(aggOpt);
                 int aliasId = fields.get(i + keys.groupKeyCount()).getIndex();
                 if (aliasId != AliasInference.DEFAULT_ID) {
                     aggFnAliasBuilder.setAlias(Utils.asAliasId(aliasId));
                 }
                 groupByBuilder.addFunctions(aggFnAliasBuilder);
             }
             oprBuilder.setOpr(
                     GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
                             .setGroupBy(groupByBuilder));
             oprBuilder.addAllMetaData(
                     Utils.physicalProtoRowType(aggregate.getRowType(), isColumnId));
             if (isPartitioned) {
                 List<RexNode> keysAndAggs = Lists.newArrayList();
                 keysAndAggs.addAll(keys.getVariables());
                 keysAndAggs.addAll(
                         groupCalls.stream()
                                 .flatMap(k -> k.getOperands().stream())
                                 .collect(Collectors.toList()));
                 Map<Integer, Set<GraphNameOrId>> tagColumns =
                         Utils.extractTagColumnsFromRexNodes(keysAndAggs);
                 if (preCacheEdgeProps) {
                     Utils.removeEdgeProperties(
                             com.alibaba.graphscope.common.ir.tools.Utils.getOutputType(
                                     aggregate.getInput()),
                             tagColumns);
                 }
                 lazyPropertyFetching(tagColumns);
             }
             physicalBuilder.addPlan(oprBuilder.build());
         }
         return aggregate;
     }
 
     @Override
     public RelNode visit(GraphLogicalDedupBy dedupBy) {
         visitChildren(dedupBy);

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/tools/GraphBuilder.java

@@ -217,6 +217,7 @@
                         fetchNode,
                         pxdConfig.getResultOpt(),
                         pxdConfig.getPathOpt(),
+                        pxdConfig.getUntilCondition(),
                         pxdConfig.getAlias(),
                         getAliasNameWithId(
                                 pxdConfig.getStartAlias(),
@@ -786,251 +787,251 @@
         }
     }
 
     private boolean isCurrentSupported(SqlOperator operator) {
         SqlKind sqlKind = operator.getKind();
         return sqlKind.belongsTo(SqlKind.BINARY_ARITHMETIC)
                 || sqlKind.belongsTo(SqlKind.COMPARISON)
                 || sqlKind == SqlKind.AND
                 || sqlKind == SqlKind.OR
                 || sqlKind == SqlKind.DESCENDING
                 || (sqlKind == SqlKind.OTHER_FUNCTION
                         && (operator.getName().equals("POWER")
                                 || operator.getName().equals("<<")
                                 || operator.getName().equals(">>")))
                 || (sqlKind == SqlKind.MINUS_PREFIX)
                 || (sqlKind == SqlKind.CASE)
                 || (sqlKind == SqlKind.PROCEDURE_CALL)
                 || (sqlKind == SqlKind.NOT)
                 || sqlKind == SqlKind.ARRAY_VALUE_CONSTRUCTOR
                 || sqlKind == SqlKind.MAP_VALUE_CONSTRUCTOR
                 || sqlKind == SqlKind.IS_NULL
                 || sqlKind == SqlKind.IS_NOT_NULL
                 || sqlKind == SqlKind.EXTRACT
                 || sqlKind == SqlKind.SEARCH
                 || sqlKind == SqlKind.POSIX_REGEX_CASE_SENSITIVE
                 || sqlKind == SqlKind.AS
                 || sqlKind == SqlKind.BIT_AND
                 || sqlKind == SqlKind.BIT_OR
                 || sqlKind == SqlKind.BIT_XOR
                 || (sqlKind == SqlKind.OTHER
                         && (operator.getName().equals("IN")
                                 || operator.getName().equals("DATETIME_MINUS")))
                 || sqlKind == SqlKind.ARRAY_CONCAT;
     }
 
     @Override
     public GraphBuilder filter(RexNode... conditions) {
         return filter(ImmutableList.copyOf(conditions));
     }
 
     @Override
     public GraphBuilder filter(Iterable<? extends RexNode> conditions) {
         conditions = flatExprs((Iterable<RexNode>) conditions);
 
         RexVisitor propertyChecker = new RexPropertyChecker(true, this);
         for (RexNode condition : conditions) {
             RelDataType type = condition.getType();
             // make sure all conditions have the Boolean return type
             if (!(type instanceof BasicSqlType) || type.getSqlTypeName() != SqlTypeName.BOOLEAN) {
                 throw new IllegalArgumentException(
                         "filter condition "
                                 + condition
                                 + " should return Boolean value, but is "
                                 + type);
             }
             // check property existence for specific label
             condition.accept(propertyChecker);
         }
 
         // specific implementation for gremlin, project will change the 'head' before the current
         // filter, which need to be recovered later
         RelDataTypeField recoverHead = null;
 
         RexSubQueryPreComputer preComputer = new RexSubQueryPreComputer(this);
         List<RexNode> newConditions = Lists.newArrayList();
         for (RexNode condition : conditions) {
             // if the condition has subquery as its operand, i.e. where(out().out().count() > 2),
             // subquery(out().out().count()) needs to be precomputed in advance, the condition also
             // needs to be adapted
             newConditions.add(preComputer.precompute(condition));
         }
         // project subquery in advance
         if (!preComputer.getSubQueryNodes().isEmpty()) {
             RelNode input = requireNonNull(peek(), "frame stack is empty");
             if (input.getRowType().getFieldList().size() == 1) {
                 RelDataTypeField field = input.getRowType().getFieldList().get(0);
                 // give a non-default alias to the head, so that it can be recovered later
                 if (field.getName() == AliasInference.DEFAULT_NAME) {
                     Set<String> uniqueAliases = AliasInference.getUniqueAliasList(input, true);
                     uniqueAliases.addAll(preComputer.getSubQueryAliases());
                     String nonDefault = AliasInference.inferAliasWithPrefix("$f", uniqueAliases);
                     // set the non default alias to the input
                     as(nonDefault);
                     recoverHead =
                             new RelDataTypeFieldImpl(
                                     nonDefault, generateAliasId(nonDefault), field.getType());
                 } else {
                     recoverHead = field;
                 }
             }
             project(preComputer.getSubQueryNodes(), preComputer.getSubQueryAliases(), true);
             conditions =
                     newConditions.stream()
                             .map(k -> k.accept(new RexTmpVariableConverter(true, this)))
                             .collect(Collectors.toList());
         }
         super.filter(ImmutableSet.of(), conditions);
         // fuse filter with the previous table scan if meets the conditions
         Filter filter = topFilter();
         if (filter != null) {
             GraphBuilder builder =
                     GraphBuilder.create(
                             this.configs, (GraphOptCluster) getCluster(), getRelOptSchema());
             RexNode condition = filter.getCondition();
             RelNode input = !filter.getInputs().isEmpty() ? filter.getInput(0) : null;
             if (input instanceof AbstractBindableTableScan) {
                 AbstractBindableTableScan tableScan = (AbstractBindableTableScan) input;
                 List<Integer> aliasIds =
                         condition.accept(
                                 new RexVariableAliasCollector<>(
                                         true, RexGraphVariable::getAliasId));
                 // fuze all conditions into table scan
                 if (!aliasIds.isEmpty()
                         && ImmutableList.of(AliasInference.DEFAULT_ID, tableScan.getAliasId())
                                 .containsAll(aliasIds)) {
                     condition =
                             condition.accept(
                                     new RexVariableAliasConverter(
                                             true,
                                             this,
                                             AliasInference.SIMPLE_NAME(AliasInference.DEFAULT_NAME),
                                             AliasInference.DEFAULT_ID));
                     // add condition into table scan
                     // pop the filter from the inner stack
                     replaceTop(fuseFilters(tableScan, condition, builder));
                 }
             } else if (input instanceof AbstractLogicalMatch) {
                 List<RexNode> extraFilters = Lists.newArrayList();
                 AbstractLogicalMatch match =
                         fuseFilters((AbstractLogicalMatch) input, condition, extraFilters, builder);
                 if (!match.equals(input)) {
                     if (extraFilters.isEmpty()) {
                         replaceTop(match);
                     } else {
                         replaceTop(builder.push(match).filter(extraFilters).build());
                     }
                 }
             }
         }
         if (recoverHead != null) {
             project(ImmutableList.of(variable(recoverHead.getName())), ImmutableList.of(), true);
         }
         return this;
     }
 
     private AbstractBindableTableScan fuseFilters(
             AbstractBindableTableScan tableScan, RexNode condition, GraphBuilder builder) {
         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())
                             .getLabelType();
             List<String> labelsToKeep =
                     labelType.getLabelsEntry().stream()
                             .filter(k -> labelValues.contains(k.getLabel()))
                             .map(k -> k.getLabel())
                             .collect(Collectors.toList());
             Preconditions.checkArgument(
                     !labelsToKeep.isEmpty(),
                     "cannot find common labels between values= " + labelValues + " and label=",
                     labelType);
             if (labelsToKeep.size() < labelType.getLabelsEntry().size()) {
                 LabelConfig newLabelConfig = new LabelConfig(false);
                 labelsToKeep.forEach(k -> newLabelConfig.addLabel(k));
                 if (tableScan instanceof GraphLogicalSource) {
                     builder.source(
                             new SourceConfig(
                                     ((GraphLogicalSource) tableScan).getOpt(),
                                     newLabelConfig,
                                     tableScan.getAliasName()));
                 } else if (tableScan instanceof GraphLogicalExpand) {
                     builder.push(tableScan.getInput(0))
                             .expand(
                                     new ExpandConfig(
                                             ((GraphLogicalExpand) tableScan).getOpt(),
                                             newLabelConfig,
                                             tableScan.getAliasName()));
                 } else if (tableScan instanceof GraphLogicalGetV) {
                     builder.push(tableScan.getInput(0))
                             .getV(
                                     new GetVConfig(
                                             ((GraphLogicalGetV) tableScan).getOpt(),
                                             newLabelConfig,
                                             tableScan.getAliasName()));
                 }
                 if (builder.size() > 0) {
                     // check if the property still exist after updating the label type
                     RexVisitor propertyChecker = new RexPropertyChecker(true, builder);
                     if (tableScan instanceof GraphLogicalSource) {
                         RexNode originalUniqueKeyFilters =
                                 ((GraphLogicalSource) tableScan).getUniqueKeyFilters();
                         if (originalUniqueKeyFilters != null) {
                             originalUniqueKeyFilters.accept(propertyChecker);
                             builder.filter(originalUniqueKeyFilters);
                         }
                         if (!uniqueKeyFilters.isEmpty()) {
                             builder.filter(uniqueKeyFilters);
                             uniqueKeyFilters.clear();
                         }
                     }
                     ImmutableList originalFilters = tableScan.getFilters();
                     if (ObjectUtils.isNotEmpty(originalFilters)) {
                         originalFilters.forEach(k -> ((RexNode) k).accept(propertyChecker));
                         builder.filter(originalFilters);
                     }
                     if (!extraFilters.isEmpty()) {
                         extraFilters.forEach(k -> k.accept(propertyChecker));
                         builder.filter(extraFilters);
                         extraFilters.clear();
                     }
                     tableScan = (AbstractBindableTableScan) builder.build();
                 }
             }
         }
         if (tableScan instanceof GraphLogicalSource && !uniqueKeyFilters.isEmpty()) {
             GraphLogicalSource source = (GraphLogicalSource) tableScan;
             if (source.getUniqueKeyFilters() != null || uniqueKeyFilters.size() > 1) {
                 extraFilters.addAll(uniqueKeyFilters);
             } else {
                 source.setUniqueKeyFilters(uniqueKeyFilters.get(0));
             }
         }
         if (!extraFilters.isEmpty()) {
             ImmutableList originalFilters = tableScan.getFilters();
             if (ObjectUtils.isNotEmpty(originalFilters)) {
                 for (int i = 0; i < originalFilters.size(); ++i) {
                     extraFilters.add(i, (RexNode) originalFilters.get(i));
                 }
             }
             tableScan.setFilters(
                     ImmutableList.of(
                             RexUtil.composeConjunction(this.getRexBuilder(), extraFilters)));
         }
         return tableScan;
     }
 
     /**
      * fuse label filters into the {@code match} if possible
      * @param match
      * @param condition
      * @param extraFilters
      * @param builder
      * @return
      */
     private AbstractLogicalMatch fuseFilters(
             AbstractLogicalMatch match,
             RexNode condition,
@@ -1418,122 +1419,122 @@
      * @param nodes build limit() if empty
      * @return
      */
     @Override
     public GraphBuilder sortLimit(
             @Nullable RexNode offsetNode,
             @Nullable RexNode fetchNode,
             Iterable<? extends RexNode> nodes) {
         if (offsetNode != null && !(offsetNode instanceof RexLiteral)) {
             throw new IllegalArgumentException("OFFSET node must be RexLiteral");
         }
         if (offsetNode != null && !(offsetNode instanceof RexLiteral)) {
             throw new IllegalArgumentException("FETCH node must be RexLiteral");
         }
 
         nodes = flatExprs((Iterable<RexNode>) nodes);
 
         RelNode input = requireNonNull(peek(), "frame stack is empty");
 
         List<RelDataTypeField> originalFields = input.getRowType().getFieldList();
 
         Registrar registrar = new Registrar(this, input, true);
         List<RexNode> registerNodes = registrar.registerExpressions(ImmutableList.copyOf(nodes));
 
         // expressions need to be projected in advance
         if (!registrar.getExtraNodes().isEmpty()) {
             if (input.getRowType().getFieldList().size() == 1) {
                 RelDataTypeField field = input.getRowType().getFieldList().get(0);
                 // give a non-default alias to the head, so that the tail project can preserve the
                 // head field
                 if (field.getName() == AliasInference.DEFAULT_NAME) {
                     Set<String> uniqueAliases = AliasInference.getUniqueAliasList(input, true);
                     uniqueAliases.addAll(registrar.getExtraAliases());
                     String nonDefault = AliasInference.inferAliasWithPrefix("$f", uniqueAliases);
                     // set the non default alias to the input
                     as(nonDefault);
                     originalFields =
                             Lists.newArrayList(
                                     new RelDataTypeFieldImpl(
                                             nonDefault,
                                             generateAliasId(nonDefault),
                                             field.getType()));
                 }
             }
             project(registrar.getExtraNodes(), registrar.getExtraAliases(), registrar.isAppend());
             RexTmpVariableConverter converter = new RexTmpVariableConverter(true, this);
             registerNodes =
                     registerNodes.stream()
                             .map(k -> k.accept(converter))
                             .collect(Collectors.toList());
             input = requireNonNull(peek(), "frame stack is empty");
         }
 
         List<RelFieldCollation> fieldCollations = fieldCollations(registerNodes);
         Config config = Utils.getFieldValue(RelBuilder.class, this, "config");
 
         // limit 0 -> return empty value
         if ((fetchNode != null && RexLiteral.intValue(fetchNode) == 0) && config.simplifyLimit()) {
             return (GraphBuilder) empty();
         }
 
         // output all results without any order -> skip
         if (offsetNode == null && fetchNode == null && fieldCollations.isEmpty()) {
             return this; // sort is trivial
         }
         // sortLimit is actually limit if collations are empty
         if (fieldCollations.isEmpty()) {
             // fuse limit with the previous sort operator
             // order + limit -> topK
             if (input instanceof Sort) {
                 Sort sort2 = (Sort) input;
                 // output all results without any limitations
                 if (sort2.offset == null && sort2.fetch == null) {
                     RelNode sort =
                             GraphLogicalSort.create(
                                     sort2.getInput(), sort2.collation, offsetNode, fetchNode);
                     replaceTop(sort);
                     return this;
                 }
             }
             // order + project + limit -> topK + project
             if (input instanceof Project) {
                 Project project = (Project) input;
                 if (project.getInput() instanceof Sort) {
                     Sort sort2 = (Sort) project.getInput();
                     if (sort2.offset == null && sort2.fetch == null) {
                         RelNode sort =
                                 GraphLogicalSort.create(
                                         sort2.getInput(), sort2.collation, offsetNode, fetchNode);
                         replaceTop(
                                 GraphLogicalProject.create(
                                         (GraphOptCluster) project.getCluster(),
                                         project.getHints(),
                                         sort,
                                         project.getProjects(),
                                         project.getRowType(),
                                         ((GraphLogicalProject) project).isAppend()));
                         return this;
                     }
                 }
             }
         }
         RelNode sort =
                 GraphLogicalSort.create(
                         input, GraphRelCollations.of(fieldCollations), offsetNode, fetchNode);
         replaceTop(sort);
         // to remove the extra columns we have added
         if (!registrar.getExtraAliases().isEmpty()) {
             List<RexNode> originalExprs = new ArrayList<>();
             List<String> originalAliases = new ArrayList<>();
             for (RelDataTypeField field : originalFields) {
                 originalExprs.add(variable(field.getName()));
                 originalAliases.add(field.getName());
             }
             project(originalExprs, originalAliases, false);
         }
         return this;
     }
 
     public GraphBuilder dedupBy(Iterable<? extends RexNode> nodes) {
         RelNode input = requireNonNull(peek(), "frame stack is empty");
 
@@ -1743,98 +1744,98 @@
      * @param alias
      * @return
      */
     @Override
     public GraphBuilder as(String alias) {
         RelNode top = requireNonNull(peek(), "frame stack is empty");
         // skip intermediate operations which make no changes to the row type, i.e.
         // filter/limit/dedup...
         while (!top.getInputs().isEmpty() && top.getInput(0).getRowType() == top.getRowType()) {
             top = top.getInput(0);
         }
         if (top instanceof AbstractBindableTableScan
                 || top instanceof GraphLogicalPathExpand
                 || top instanceof GraphLogicalProject
                 || top instanceof GraphLogicalAggregate) {
             RelDataType rowType = top.getRowType();
             // we can assign the alias only if the top node has only one field, otherwise we skip
             // the
             // operation
             if (rowType.getFieldList().size() != 1) {
                 return this;
             }
             build();
             if (!top.getInputs().isEmpty()) {
                 push(top.getInput(0));
             }
             if (top instanceof GraphLogicalSource) {
                 GraphLogicalSource source = (GraphLogicalSource) top;
                 source(
                         new SourceConfig(
                                 source.getOpt(), getLabelConfig(source.getTableConfig()), alias));
                 if (source.getUniqueKeyFilters() != null) {
                     filter(source.getUniqueKeyFilters());
                 }
                 if (ObjectUtils.isNotEmpty(source.getFilters())) {
                     filter(source.getFilters());
                 }
             } else if (top instanceof GraphLogicalExpand) {
                 GraphLogicalExpand expand = (GraphLogicalExpand) top;
                 expand(
                         new ExpandConfig(
                                 expand.getOpt(), getLabelConfig(expand.getTableConfig()), alias));
                 if (ObjectUtils.isNotEmpty(expand.getFilters())) {
                     filter(expand.getFilters());
                 }
             } else if (top instanceof GraphLogicalGetV) {
                 GraphLogicalGetV getV = (GraphLogicalGetV) top;
                 getV(new GetVConfig(getV.getOpt(), getLabelConfig(getV.getTableConfig()), alias));
                 if (ObjectUtils.isNotEmpty(getV.getFilters())) {
                     filter(getV.getFilters());
                 }
             } else if (top instanceof GraphLogicalPathExpand) {
                 GraphLogicalPathExpand pxdExpand = (GraphLogicalPathExpand) top;
                 GraphLogicalExpand expand = (GraphLogicalExpand) pxdExpand.getExpand();
                 GraphLogicalGetV getV = (GraphLogicalGetV) pxdExpand.getGetV();
                 PathExpandConfig.Builder pxdBuilder = PathExpandConfig.newBuilder(this);
                 RexNode offset = pxdExpand.getOffset(), fetch = pxdExpand.getFetch();
                 pxdBuilder
                         .expand(
                                 new ExpandConfig(
                                         expand.getOpt(),
                                         getLabelConfig(expand.getTableConfig()),
                                         expand.getAliasName()))
                         .getV(
                                 new GetVConfig(
                                         getV.getOpt(),
                                         getLabelConfig(getV.getTableConfig()),
                                         getV.getAliasName()))
                         .pathOpt(pxdExpand.getPathOpt())
                         .resultOpt(pxdExpand.getResultOpt())
                         .range(
                                 offset == null
                                         ? 0
                                         : ((RexLiteral) offset).getValueAs(Integer.class),
                                 fetch == null ? -1 : ((RexLiteral) fetch).getValueAs(Integer.class))
                         .startAlias(pxdExpand.getStartAlias().getAliasName())
                         .alias(alias);
-                pathExpand(pxdBuilder.build());
+                pathExpand(pxdBuilder.buildConfig());
             } else if (top instanceof GraphLogicalProject) {
                 GraphLogicalProject project = (GraphLogicalProject) top;
                 project(project.getProjects(), Lists.newArrayList(alias), project.isAppend());
             } else if (top instanceof GraphLogicalAggregate) {
                 GraphLogicalAggregate aggregate = (GraphLogicalAggregate) top;
                 // if group key is empty, we can assign the alias to the single aggregated value in
                 // group
                 if (aggregate.getGroupKey().groupKeyCount() == 0
                         && aggregate.getAggCalls().size() == 1) {
                     GraphAggCall aggCall = aggregate.getAggCalls().get(0);
                     aggregate(aggregate.getGroupKey(), ImmutableList.of(aggCall.as(alias)));
                 }
             }
         }
         return this;
     }
 
     private LabelConfig getLabelConfig(TableConfig tableConfig) {
         List<String> labels =
                 tableConfig.getTables().stream()