fix(interactive): Fix Result Parsing Mismatch Errors of Map Structu…

…re (#4006)

For the `map` structure, there is inconsistency between the type
maintenance on the compiler side and the implementation on the runtime
side. Specifically, the compiler maintains the types of entries in the
map according to the order specified by the user, while the runtime uses
a `TreeMap` to maintain entries based on the internal order of keys.
This inconsistency has caused parsing issues for `map` results by the
compiler.

To address this issue, this PR primarily focuses on the following two
aspects:
1. specifying the columns to be output and their order concretely in the
`sink` operator to ensure columns are output in the order specified by
the user;
2. maintaining specific mappings from keys to types in the `map` type,
see
[here](https://github.com/alibaba/GraphScope/pull/4006/files#diff-49e26198c19ddbe7e665b8d5a66ceced8b8d02916e07def597fd4a7cdfffeffa).
During result parsing, this ensures that the type corresponding to the
returned key is found in this map, thereby ensuring consistency.

Co-authored-by: Longbin Lai <[email protected]>

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/rex/operator/SqlMapValueConstructor.java

@@ -16,10 +16,14 @@
 
 package com.alibaba.graphscope.common.ir.rex.operator;
 
+import com.alibaba.graphscope.common.ir.rex.RexCallBinding;
+import com.alibaba.graphscope.common.ir.type.ArbitraryMapType;
 import com.alibaba.graphscope.common.ir.type.GraphTypeFactoryImpl;
+import com.google.common.collect.Maps;
 
 import org.apache.calcite.rel.type.RelDataType;
 import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.SqlCallBinding;
 import org.apache.calcite.sql.SqlKind;
 import org.apache.calcite.sql.SqlOperatorBinding;
@@ -31,6 +35,7 @@
 import org.checkerframework.checker.nullness.qual.Nullable;
 
 import java.util.List;
+import java.util.Map;
 
 /**
  * This operator is used to fold columns into a map, i.e. {name: a.name, age: a.age} in cypher queries.
@@ -56,8 +61,16 @@ public RelDataType inferReturnType(SqlOperatorBinding opBinding) {
                 && valueType.getSqlTypeName() != SqlTypeName.ANY) {
             return SqlTypeUtil.createMapType(opBinding.getTypeFactory(), keyType, valueType, false);
         } else {
+            Map<RexNode, ArbitraryMapType.KeyValueType> keyValueTypeMap = Maps.newHashMap();
+            List<RexNode> operands = ((RexCallBinding) opBinding).getRexOperands();
+            for (int i = 0; i < operands.size(); i += 2) {
+                keyValueTypeMap.put(
+                        operands.get(i),
+                        new ArbitraryMapType.KeyValueType(
+                                keyTypes.get(i / 2), valueTypes.get(i / 2)));
+            }
             return ((GraphTypeFactoryImpl) typeFactory)
-                    .createArbitraryMapType(keyTypes, valueTypes, false);
+                    .createArbitraryMapType(keyValueTypeMap, false);
         }
     }
 

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

@@ -24,9 +24,11 @@
 import com.alibaba.graphscope.common.ir.rel.GraphShuttle;
 import com.alibaba.graphscope.common.ir.runtime.PhysicalBuilder;
 import com.alibaba.graphscope.common.ir.runtime.PhysicalPlan;
+import com.alibaba.graphscope.common.ir.tools.AliasInference;
 import com.alibaba.graphscope.common.ir.tools.LogicalPlan;
 import com.alibaba.graphscope.gaia.proto.GraphAlgebra;
 import com.alibaba.graphscope.gaia.proto.GraphAlgebraPhysical;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
@@ -57,20 +59,27 @@ public class GraphRelProtoPhysicalBuilder extends PhysicalBuilder {
     // `g.V().out().union(out(), out())`,
     // `g.V().out()` is a common sub-plan, the pair of <union, g.V().out()> is recorded in this map
     private final IdentityHashMap<RelNode, List<CommonTableScan>> relToCommons;
+    private final boolean skipSinkColumns;
 
     public GraphRelProtoPhysicalBuilder(
             Configs graphConfig, IrMeta irMeta, LogicalPlan logicalPlan) {
+        this(graphConfig, irMeta, logicalPlan, false);
+    }
+
+    @VisibleForTesting
+    public GraphRelProtoPhysicalBuilder(
+            Configs graphConfig, IrMeta irMeta, LogicalPlan logicalPlan, boolean skipSinkColumns) {
         super(logicalPlan);
         this.physicalBuilder = GraphAlgebraPhysical.PhysicalPlan.newBuilder();
         this.relToCommons = createRelToCommons(logicalPlan);
-
         this.relShuttle =
                 new GraphRelToProtoConverter(
                         irMeta.getSchema().isColumnId(),
                         graphConfig,
                         this.physicalBuilder,
                         this.relToCommons,
                         createExtraParams(irMeta));
+        this.skipSinkColumns = skipSinkColumns;
     }
 
     @Override
@@ -79,7 +88,11 @@ public PhysicalPlan build() {
         try {
             RelNode regularQuery = this.logicalPlan.getRegularQuery();
             regularQuery.accept(this.relShuttle);
-            appendDefaultSink();
+            physicalBuilder.addPlan(
+                    GraphAlgebraPhysical.PhysicalOpr.newBuilder()
+                            .setOpr(
+                                    GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder()
+                                            .setSink(getSinkByColumns(regularQuery))));
             plan = getPlanAsJson(physicalBuilder.build());
             int planId = Objects.hash(logicalPlan);
             physicalBuilder.setPlanId(planId);
@@ -92,17 +105,23 @@ public PhysicalPlan build() {
         }
     }
 
-    private void appendDefaultSink() {
-        GraphAlgebraPhysical.PhysicalOpr.Builder oprBuilder =
-                GraphAlgebraPhysical.PhysicalOpr.newBuilder();
+    private GraphAlgebraPhysical.Sink getSinkByColumns(RelNode regularQuery) {
         GraphAlgebraPhysical.Sink.Builder sinkBuilder = GraphAlgebraPhysical.Sink.newBuilder();
-        GraphAlgebra.Sink.SinkTarget.Builder sinkTargetBuilder =
-                GraphAlgebra.Sink.SinkTarget.newBuilder();
-        sinkTargetBuilder.setSinkDefault(GraphAlgebra.SinkDefault.newBuilder().build());
-        sinkBuilder.setSinkTarget(sinkTargetBuilder);
-        oprBuilder.setOpr(
-                GraphAlgebraPhysical.PhysicalOpr.Operator.newBuilder().setSink(sinkBuilder));
-        physicalBuilder.addPlan(oprBuilder);
+        sinkBuilder.setSinkTarget(
+                GraphAlgebra.Sink.SinkTarget.newBuilder()
+                        .setSinkDefault(GraphAlgebra.SinkDefault.newBuilder().build()));
+        regularQuery
+                .getRowType()
+                .getFieldList()
+                .forEach(
+                        k -> {
+                            if (!skipSinkColumns && k.getIndex() != AliasInference.DEFAULT_ID) {
+                                sinkBuilder.addTags(
+                                        GraphAlgebraPhysical.Sink.OptTag.newBuilder()
+                                                .setTag(Utils.asAliasId(k.getIndex())));
+                            }
+                        });
+        return sinkBuilder.build();
     }
 
     private String getPlanAsJson(GraphAlgebraPhysical.PhysicalPlan physicalPlan) {

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

@@ -17,39 +17,39 @@
 package com.alibaba.graphscope.common.ir.type;
 
 import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexNode;
 import org.apache.calcite.sql.type.AbstractSqlType;
 import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.util.Pair;
 
 import java.util.Collections;
-import java.util.List;
-import java.util.Objects;
+import java.util.Map;
 
 /**
  * introduce a new map type to allow different keys or value types in a single map,
  * to support {@code MapLiteral} in cypher, i.e. [name: a.name, a: a, age: b.age]
  */
 public class ArbitraryMapType extends AbstractSqlType {
-    private final List<RelDataType> keyTypes;
-    private final List<RelDataType> valueTypes;
+    private final Map<RexNode, KeyValueType> keyValueTypeMap;
 
-    protected ArbitraryMapType(
-            List<RelDataType> keyTypes, List<RelDataType> valueTypes, boolean isNullable) {
+    protected ArbitraryMapType(Map<RexNode, KeyValueType> keyValueTypeMap, boolean isNullable) {
         super(SqlTypeName.MAP, isNullable, null);
-        this.keyTypes = Objects.requireNonNull(keyTypes);
-        this.valueTypes = Objects.requireNonNull(valueTypes);
+        this.keyValueTypeMap = keyValueTypeMap;
         this.computeDigest();
     }
 
     @Override
     protected void generateTypeString(StringBuilder sb, boolean withDetail) {
-        sb.append("(" + keyTypes.toString() + ", " + valueTypes.toString() + ") MAP");
+        sb.append("(" + keyValueTypeMap + ") MAP");
     }
 
-    public List<RelDataType> getKeyTypes() {
-        return Collections.unmodifiableList(this.keyTypes);
+    public Map<RexNode, KeyValueType> getKeyValueTypeMap() {
+        return Collections.unmodifiableMap(this.keyValueTypeMap);
     }
 
-    public List<RelDataType> getValueTypes() {
-        return Collections.unmodifiableList(this.valueTypes);
+    public static class KeyValueType extends Pair<RelDataType, RelDataType> {
+        public KeyValueType(RelDataType left, RelDataType right) {
+            super(left, right);
+        }
     }
 }