fix(interactive): consider path direction in path concat (#3920)

<!--
Thanks for your contribution! please review
https://github.com/alibaba/GraphScope/blob/main/CONTRIBUTING.md before
opening an issue.
-->

## What do these changes do?

<!-- Please give a short brief about these changes. -->

As titled. 
Now we specify the concat position in PathConcat. e.g., Assuming we want
to concat left path L with right path R, then we would specify the
concat position as follows:
1. concat L=[1,2,3] and R=[3,4,5] with L.end and R.start,
2. concat L=[1,2,3] and R=[5,4,3] with L.end and R.end,
3. concat L=[3,2,1] and R=[3,4,5] with L.start and R.start,
4. concat L=[3,2,1] and R=[5,4,3] with L.start and R.end.

The concatenated result, actually, can be either [1,2,3,4,5] or
[5,4,3,2,1]. We choose to keep the left path in the left part in the
result, i.e., the output concatenated path would be in order of
[1,2,3,4,5].

## Related issue number

<!-- Are there any issues opened that will be resolved by merging this
change? -->

Fixes #3765

---------

Co-authored-by: shirly121 <[email protected]>

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

@@ -35,6 +35,7 @@
 import com.alibaba.graphscope.common.ir.rex.RexGraphVariable;
 import com.alibaba.graphscope.common.ir.tools.AliasInference;
 import com.alibaba.graphscope.common.ir.tools.GraphBuilder;
+import com.alibaba.graphscope.common.ir.tools.GraphStdOperatorTable;
 import com.alibaba.graphscope.common.ir.tools.Utils;
 import com.alibaba.graphscope.common.ir.tools.config.*;
 import com.alibaba.graphscope.common.ir.type.GraphLabelType;
@@ -59,7 +60,6 @@
 import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.rex.RexUtil;
 import org.apache.calcite.rex.RexVariable;
-import org.apache.calcite.sql.fun.SqlLibraryOperators;
 import org.apache.calcite.util.ImmutableBitSet;
 import org.apache.commons.lang3.ObjectUtils;
 import org.checkerframework.checker.nullness.qual.Nullable;
@@ -574,10 +574,12 @@ public RelNode visit(GraphJoinDecomposition decomposition) {
                             decomposition.getBuildPattern(),
                             buildOrderMap,
                             new ParentPattern(decomposition.getParentPatten(), 1));
+            RelNode joinLeft = decomposition.getLeft();
+            RelNode joinRight = decomposition.getRight();
             this.details = probeDetails;
-            RelNode newLeft = visitChild(decomposition, 0, decomposition.getLeft()).getInput(0);
+            RelNode newLeft = visitChild(decomposition, 0, joinLeft).getInput(0);
             this.details = buildDetails;
-            RelNode newRight = visitChild(decomposition, 1, decomposition.getRight()).getInput(1);
+            RelNode newRight = visitChild(decomposition, 1, joinRight).getInput(1);
             RexNode joinCondition =
                     createJoinFilter(
                             jointVertices,
@@ -651,9 +653,19 @@ public RelNode visit(GraphJoinDecomposition decomposition) {
                                 String buildAlias = buildValue.getAlias();
                                 concatExprs.add(
                                         builder.call(
-                                                SqlLibraryOperators.ARRAY_CONCAT,
+                                                GraphStdOperatorTable.PATH_CONCAT,
                                                 builder.variable(probeAlias),
-                                                builder.variable(buildAlias)));
+                                                builder.getRexBuilder()
+                                                        .makeFlag(
+                                                                getConcatDirection(
+                                                                        probeJointVertex,
+                                                                        joinLeft)),
+                                                builder.variable(buildAlias),
+                                                builder.getRexBuilder()
+                                                        .makeFlag(
+                                                                getConcatDirection(
+                                                                        buildJointVertex,
+                                                                        joinRight))));
                                 concatAliases.add(probeValue.getParentAlias());
                             }
                         }
@@ -667,6 +679,12 @@ public RelNode visit(GraphJoinDecomposition decomposition) {
             return builder.build();
         }
 
+        private GraphOpt.GetV getConcatDirection(PatternVertex concatVertex, RelNode splitPattern) {
+            ConcatDirectionVisitor visitor = new ConcatDirectionVisitor(concatVertex);
+            visitor.go(splitPattern);
+            return visitor.direction;
+        }
+
         private RexNode createJoinFilter(
                 List<GraphJoinDecomposition.JoinVertexPair> jointVertices,
                 Map<DataKey, DataValue> vertexDetails,
@@ -1163,5 +1181,52 @@ private EdgeDataKey createEdgeKey(ExtendEdge edge, GlogueExtendIntersectEdge glo
             }
             return new EdgeDataKey(srcOrderId, targetOrderId, edge.getDirection());
         }
+
+        // given a concat vertex, help to determine its direction in the split path expand
+        private class ConcatDirectionVisitor extends RelVisitor {
+            private GraphOpt.GetV direction;
+            private final PatternVertex concatVertex;
+
+            public ConcatDirectionVisitor(PatternVertex concatVertex) {
+                this.concatVertex = concatVertex;
+                this.direction = null;
+            }
+
+            @Override
+            public void visit(RelNode node, int ordinal, @Nullable RelNode parent) {
+                if (direction != null) return;
+                if (node instanceof GraphExtendIntersect) {
+                    GlogueExtendIntersectEdge intersect =
+                            ((GraphExtendIntersect) node).getGlogueEdge();
+                    ExtendStep extendStep = intersect.getExtendStep();
+                    PatternVertex dstVertex =
+                            intersect
+                                    .getDstPattern()
+                                    .getVertexByOrder(extendStep.getTargetVertexOrder());
+                    if (dstVertex.equals(concatVertex)) {
+                        direction = GraphOpt.GetV.END;
+                        return;
+                    }
+                    for (ExtendEdge edge : extendStep.getExtendEdges()) {
+                        PatternVertex srcVertex =
+                                intersect
+                                        .getSrcPattern()
+                                        .getVertexByOrder(edge.getSrcVertexOrder());
+                        if (srcVertex.equals(concatVertex)) {
+                            direction = GraphOpt.GetV.START;
+                            return;
+                        }
+                    }
+                } else if (node instanceof GraphPattern) {
+                    PatternVertex vertex =
+                            ((GraphPattern) node).getPattern().getVertexSet().iterator().next();
+                    if (vertex.equals(concatVertex)) {
+                        direction = GraphOpt.GetV.START;
+                        return;
+                    }
+                }
+                node.childrenAccept(this);
+            }
+        }
     }
 }

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

@@ -19,6 +19,7 @@
 import com.alibaba.graphscope.common.ir.rex.RexGraphVariable;
 import com.alibaba.graphscope.common.ir.tools.AliasInference;
 import com.alibaba.graphscope.common.ir.tools.GraphStdOperatorTable;
+import com.alibaba.graphscope.common.ir.tools.config.GraphOpt;
 import com.alibaba.graphscope.gaia.proto.Common;
 import com.alibaba.graphscope.gaia.proto.DataType;
 import com.alibaba.graphscope.gaia.proto.OuterExpression;
@@ -58,8 +59,9 @@ public OuterExpression.Expression visitCall(RexCall call) {
             return visitArrayValueConstructor(call);
         } else if (operator.getKind() == SqlKind.MAP_VALUE_CONSTRUCTOR) {
             return visitMapValueConstructor(call);
-        } else if (operator.getKind() == SqlKind.ARRAY_CONCAT) {
-            return visitArrayConcat(call);
+        } else if (operator.getKind() == SqlKind.OTHER
+                && operator.getName().equals("PATH_CONCAT")) {
+            return visitPathConcat(call);
         } else if (operator.getKind() == SqlKind.EXTRACT) {
             return visitExtract(call);
         } else if (operator.getKind() == SqlKind.OTHER
@@ -72,6 +74,48 @@ public OuterExpression.Expression visitCall(RexCall call) {
         }
     }
 
+    private OuterExpression.Expression visitPathConcat(RexCall call) {
+        List<RexNode> operands = call.getOperands();
+        return OuterExpression.Expression.newBuilder()
+                .addOperators(
+                        OuterExpression.ExprOpr.newBuilder()
+                                .setPathConcat(
+                                        OuterExpression.PathConcat.newBuilder()
+                                                .setLeft(convertPathInfo(operands))
+                                                .setRight(
+                                                        convertPathInfo(
+                                                                operands.subList(
+                                                                        2, operands.size())))))
+                .build();
+    }
+
+    private OuterExpression.PathConcat.ConcatPathInfo convertPathInfo(List<RexNode> operands) {
+        Preconditions.checkArgument(
+                operands.size() >= 2
+                        && operands.get(0) instanceof RexGraphVariable
+                        && operands.get(1) instanceof RexLiteral);
+        OuterExpression.Variable variable = operands.get(0).accept(this).getOperators(0).getVar();
+        GraphOpt.GetV direction = ((RexLiteral) operands.get(1)).getValueAs(GraphOpt.GetV.class);
+        return OuterExpression.PathConcat.ConcatPathInfo.newBuilder()
+                .setPathTag(variable)
+                .setEndpoint(convertPathDirection(direction))
+                .build();
+    }
+
+    private OuterExpression.PathConcat.Endpoint convertPathDirection(GraphOpt.GetV direction) {
+        switch (direction) {
+            case START:
+                return OuterExpression.PathConcat.Endpoint.START;
+            case END:
+                return OuterExpression.PathConcat.Endpoint.END;
+            default:
+                throw new IllegalArgumentException(
+                        "invalid path concat direction ["
+                                + direction.name()
+                                + "], cannot convert to any physical expression");
+        }
+    }
+
     private OuterExpression.Expression visitDateMinus(RexCall call) {
         OuterExpression.Expression.Builder exprBuilder = OuterExpression.Expression.newBuilder();
         RexLiteral interval = (RexLiteral) call.getOperands().get(2);
@@ -124,25 +168,6 @@ private OuterExpression.Expression visitCase(RexCall call) {
                 .build();
     }
 
-    private OuterExpression.Expression visitArrayConcat(RexCall call) {
-        OuterExpression.Concat.Builder concatBuilder = OuterExpression.Concat.newBuilder();
-        call.getOperands()
-                .forEach(
-                        operand -> {
-                            Preconditions.checkArgument(
-                                    operand instanceof RexGraphVariable,
-                                    "parameters of 'CONCAT' should be"
-                                            + " 'variable' in ir core structure");
-                            concatBuilder.addVars(operand.accept(this).getOperators(0).getVar());
-                        });
-        return OuterExpression.Expression.newBuilder()
-                .addOperators(
-                        OuterExpression.ExprOpr.newBuilder()
-                                .setConcat(concatBuilder)
-                                .setNodeType(Utils.protoIrDataType(call.getType(), isColumnId)))
-                .build();
-    }
-
     private OuterExpression.Expression visitArrayValueConstructor(RexCall call) {
         OuterExpression.VariableKeys.Builder varsBuilder =
                 OuterExpression.VariableKeys.newBuilder();

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

@@ -547,6 +547,45 @@ public RexGraphVariable variable(@Nullable String alias, String property) {
                     columnField.left,
                     varName,
                     getTypeFactory().createSqlType(SqlTypeName.ANY));
+        } else if (property.equals(GraphProperty.START_V_KEY)) {
+            if (!(aliasField.getType() instanceof GraphPathType)) {
+                throw new ClassCastException(
+                        "cannot get property='start_v' from type class ["
+                                + aliasField.getType().getClass()
+                                + "], should be ["
+                                + GraphPathType.class
+                                + "]");
+            } else {
+                Preconditions.checkArgument(size() > 0, "frame stack is empty");
+                RelNode peek = peek();
+                Preconditions.checkArgument(
+                        peek != null && !peek.getInputs().isEmpty(),
+                        "path expand should have start vertex");
+                RelNode input = peek.getInput(0);
+                return RexGraphVariable.of(
+                        aliasField.getIndex(),
+                        new GraphProperty(GraphProperty.Opt.START_V),
+                        columnField.left,
+                        varName,
+                        input.getRowType().getFieldList().get(0).getType());
+            }
+        } else if (property.equals(GraphProperty.END_V_KEY)) {
+            if (!(aliasField.getType() instanceof GraphPathType)) {
+                throw new ClassCastException(
+                        "cannot get property='end_v' from type class ["
+                                + aliasField.getType().getClass()
+                                + "], should be ["
+                                + GraphPathType.class
+                                + "]");
+            } else {
+                GraphPathType pathType = (GraphPathType) aliasField.getType();
+                return RexGraphVariable.of(
+                        aliasField.getIndex(),
+                        new GraphProperty(GraphProperty.Opt.END_V),
+                        columnField.left,
+                        varName,
+                        pathType.getComponentType().getGetVType());
+            }
         }
         GraphSchemaType graphType = (GraphSchemaType) aliasField.getType();
         List<String> properties = new ArrayList<>();
@@ -815,7 +854,8 @@ private boolean isCurrentSupported(SqlOperator operator) {
                 || sqlKind == SqlKind.BIT_XOR
                 || (sqlKind == SqlKind.OTHER
                         && (operator.getName().equals("IN")
-                                || operator.getName().equals("DATETIME_MINUS")))
+                                || operator.getName().equals("DATETIME_MINUS")
+                                || operator.getName().equals("PATH_CONCAT")))
                 || sqlKind == SqlKind.ARRAY_CONCAT;
     }
 

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

@@ -20,6 +20,8 @@
 import com.alibaba.graphscope.common.ir.rex.operator.CaseOperator;
 import com.alibaba.graphscope.common.ir.rex.operator.SqlArrayValueConstructor;
 import com.alibaba.graphscope.common.ir.rex.operator.SqlMapValueConstructor;
+import com.alibaba.graphscope.common.ir.type.GraphTypeFamily;
+import com.google.common.collect.ImmutableList;
 
 import org.apache.calcite.sql.*;
 import org.apache.calcite.sql.fun.ExtSqlPosixRegexOperator;
@@ -296,4 +298,27 @@ public static final SqlFunction USER_DEFINED_PROCEDURE(StoredProcedureMeta meta)
                     ReturnTypes.BOOLEAN_NULLABLE,
                     GraphInferTypes.IN_OPERANDS_TYPE,
                     OperandTypes.ANY);
+
+    public static final SqlOperator PATH_CONCAT =
+            new SqlFunction(
+                    "PATH_CONCAT",
+                    SqlKind.OTHER,
+                    ReturnTypes.ARG0,
+                    null,
+                    GraphOperandTypes.operandMetadata(
+                            ImmutableList.of(
+                                    GraphTypeFamily.PATH,
+                                    SqlTypeFamily.IGNORE,
+                                    GraphTypeFamily.PATH,
+                                    SqlTypeFamily.IGNORE),
+                            typeFactory -> ImmutableList.of(),
+                            i ->
+                                    ImmutableList.of(
+                                                    "LeftPath",
+                                                    "LeftDirection",
+                                                    "RightPath",
+                                                    "RightDirection")
+                                            .get(i),
+                            i -> false),
+                    SqlFunctionCategory.SYSTEM);
 }

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

@@ -16,10 +16,8 @@
 
 package com.alibaba.graphscope.common.ir.tools.config;
 
-import org.apache.calcite.rel.type.RelDataTypeFamily;
-
 public abstract class GraphOpt {
-    public enum Source implements RelDataTypeFamily {
+    public enum Source {
         VERTEX,
         EDGE
     }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/type/GraphPathType.java

@@ -17,6 +17,7 @@
 package com.alibaba.graphscope.common.ir.type;
 
 import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeFamily;
 import org.apache.calcite.sql.type.AbstractSqlType;
 import org.apache.calcite.sql.type.ArraySqlType;
 import org.apache.calcite.sql.type.SqlTypeName;
@@ -78,4 +79,9 @@ public String getFullTypeString() {
             return sb.toString();
         }
     }
+
+    @Override
+    public RelDataTypeFamily getFamily() {
+        return GraphTypeFamily.PATH;
+    }
 }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/type/GraphProperty.java

@@ -25,6 +25,8 @@ public class GraphProperty {
     public static final String ALL_KEY = "~all";
     public static final String ID_KEY = "~id";
     public static final String LABEL_KEY = "~label";
+    public static final String START_V_KEY = "~start_v";
+    public static final String END_V_KEY = "~end_v";
 
     private final Opt opt;
     private final @Nullable GraphNameOrId key;
@@ -65,6 +67,8 @@ public enum Opt {
         LABEL,
         LEN,
         ALL,
-        KEY
+        KEY,
+        START_V,
+        END_V
     }
 }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/type/GraphSchemaType.java

@@ -191,7 +191,7 @@ public boolean isStruct() {
 
     @Override
     public RelDataTypeFamily getFamily() {
-        return scanOpt;
+        return scanOpt == GraphOpt.Source.VERTEX ? GraphTypeFamily.VERTEX : GraphTypeFamily.EDGE;
     }
 
     public List<GraphSchemaType> getSchemaTypeAsList() {

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/type/GraphTypeFamily.java

@@ -0,0 +1,27 @@
+/*
+ *
+ *  * 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.type;
+
+import org.apache.calcite.rel.type.RelDataTypeFamily;
+
+public enum GraphTypeFamily implements RelDataTypeFamily {
+    PATH,
+    VERTEX,
+    EDGE,
+}