fix(interactive): Add Flex Type Benchmark

.github/workflows/interactive.yml

@@ -156,6 +156,9 @@ jobs:
         cp ${GITHUB_WORKSPACE}/flex/interactive/examples/new_graph_algo/import.yaml ${INTERACTIVE_WORKSPACE}/data/new_graph_algo/import.yaml
         mkdir -p ${INTERACTIVE_WORKSPACE}/data/modern_graph
         cp ${GITHUB_WORKSPACE}/flex/interactive/examples/modern_graph/graph.yaml ${INTERACTIVE_WORKSPACE}/data/modern_graph/graph.yaml
+        mkdir -p ${INTERACTIVE_WORKSPACE}/data/type_test
+        cp ${GITHUB_WORKSPACE}/interactive_engine/compiler/src/test/resources/flex_bench/modern.yaml ${INTERACTIVE_WORKSPACE}/data/type_test/graph.yaml
+        cp ${GITHUB_WORKSPACE}/interactive_engine/compiler/src/test/resources/flex_bench/import.yaml ${INTERACTIVE_WORKSPACE}/data/type_test/import.yaml
 
         # load graph
         cd ${GITHUB_WORKSPACE}/flex/build
@@ -169,6 +172,15 @@ jobs:
         GLOG_v=10 ./bin/bulk_loader -g ${INTERACTIVE_WORKSPACE}/data/new_graph_algo/graph.yaml -l ${INTERACTIVE_WORKSPACE}/data/new_graph_algo/import.yaml -d ${INTERACTIVE_WORKSPACE}/data/new_graph_algo/indices/
         export FLEX_DATA_DIR=../interactive/examples/modern_graph
         GLOG_v=10 ./bin/bulk_loader -g ${INTERACTIVE_WORKSPACE}/data/modern_graph/graph.yaml -l ../interactive/examples/modern_graph/bulk_load.yaml -d ${INTERACTIVE_WORKSPACE}/data/modern_graph/indices/
+        export FLEX_DATA_DIR=${GITHUB_WORKSPACE}/interactive_engine/compiler/src/test/resources/flex_bench/data
+        GLOG_v=10 ./bin/bulk_loader -g ${INTERACTIVE_WORKSPACE}/data/type_test/graph.yaml -l ${INTERACTIVE_WORKSPACE}/data/type_test/import.yaml -d ${INTERACTIVE_WORKSPACE}/data/type_test/indices/
+
+    - name: Interactive Type Test
+      env:
+        INTERACTIVE_WORKSPACE: /tmp/interactive_workspace
+      run: |
+        cd ${GITHUB_WORKSPACE}/flex/tests/hqps
+        bash hqps_type_test.sh ${INTERACTIVE_WORKSPACE} ./interactive_config_test.yaml
 
     - name: Test HQPS admin http service
       env: 

flex/tests/hqps/hqps_type_test.sh

@@ -0,0 +1,94 @@
+#!/bin/bash
+# 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.
+set -e
+SCRIPT_DIR=$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" &>/dev/null && pwd)
+FLEX_HOME=${SCRIPT_DIR}/../../
+BULK_LOADER=${FLEX_HOME}/build/bin/bulk_loader
+SERVER_BIN=${FLEX_HOME}/build/bin/interactive_server
+GIE_HOME=${FLEX_HOME}/../interactive_engine/
+
+# 
+if [ $# -ne 2 ]; then
+  echo "Receives: $# args, need 2 args"
+  echo "Usage: $0 <INTERACTIVE_WORKSPACE> <ENGINE_CONFIG>"
+  exit 1
+fi
+
+INTERACTIVE_WORKSPACE=$1
+ENGINE_CONFIG_PATH=$2
+
+
+RED='\033[0;31m'
+GREEN='\033[0;32m'
+NC='\033[0m' # No Color
+err() {
+  echo -e "${RED}[$(date +'%Y-%m-%d %H:%M:%S')] -ERROR- $* ${NC}" >&2
+}
+
+info() {
+  echo -e "${GREEN}[$(date +'%Y-%m-%d %H:%M:%S')] -INFO- $* ${NC}"
+}
+
+
+kill_service(){
+    info "Kill Service first"
+    ps -ef | grep "com.alibaba.graphscope.GraphServer" | awk '{print $2}' | xargs kill -9 || true
+    ps -ef | grep "interactive_server" |  awk '{print $2}' | xargs kill -9  || true
+    sleep 3
+    # check if service is killed
+    info "Kill Service success"
+}
+
+# kill service when exit
+trap kill_service EXIT
+
+
+# start engine service and load ldbc graph
+start_engine_service(){
+    # suppose graph has been loaded, check ${GRAPH_CSR_DATA_DIR} exists
+
+    #check SERVER_BIN exists
+    if [ ! -f ${SERVER_BIN} ]; then
+        err "SERVER_BIN not found"
+        exit 1
+    fi
+    cmd="${SERVER_BIN} -c ${ENGINE_CONFIG_PATH} --start-compiler true "
+    cmd="${cmd} -w ${INTERACTIVE_WORKSPACE} --enable-admin-service true > /tmp/engine.log 2>&1 &"
+
+    info "Start engine service with command: ${cmd}"
+    ${cmd} &
+    sleep 5
+    #check interactive_server is running, if not, exit
+    ps -ef | grep "interactive_server" | grep -v grep
+
+    info "Start engine service success"
+}
+
+
+
+run_type_test() {
+    echo "run type test"
+    pushd ${GIE_HOME}/compiler
+    cmd="mvn test -Dtest=com.alibaba.graphscope.cypher.integration.flex.bench.FlexTypeTest"
+    info "Run type test with command: ${cmd}"
+    ${cmd}
+    info "Run type test success"
+    popd
+}
+
+kill_service
+start_engine_service
+run_type_test
+kill_service

flex/tests/hqps/interactive_config_test.yaml

@@ -12,13 +12,17 @@ compute_engine:
     type: file # file/sqlite/etcd
   wal_uri: file://{GRAPH_DATA_DIR}/wal # Could be file://{GRAPH_DATA_DIR}/wal or other supported storage class. GRAPH_DATA_DIR is the placeholder for the graph data directory.
 compiler:
+  physical:
+    opt:
+      config: proto
   planner:
     is_on: true
-    opt: RBO
+    opt: CBO
     rules:
       - FilterIntoJoinRule
       - FilterMatchRule
-      - NotMatchToAntiJoinRule
+      - ExtendIntersectRule
+      - ExpandGetVFusionRule
   meta:
     reader:
       schema:

interactive_engine/compiler/Makefile

@@ -25,6 +25,7 @@ physical:=
 procedure:=
 extra:=
 config.path:=conf/ir.compiler.properties
+disable.expr.simplify:=
 
 build:
 	cd $(CUR_DIR)/.. && \
@@ -63,6 +64,7 @@ run:
 	  -Dgraph.planner.rules=${graph.planner.rules} \
 	  -Dgraph.planner.opt=${graph.planner.opt} \
 	  -Dgraph.statistics=${graph.statistics} \
+	  -Ddisable.expr.simplify={disable.expr.simplify} \
 	  com.alibaba.graphscope.GraphServer ${config.path}
 
 # make physical_plan config.path='<path to the config file>'

interactive_engine/compiler/pom.xml

@@ -362,6 +362,7 @@
                         <exclude>**/IrPatternTest.java</exclude>
                         <exclude>**/MovieTest.java</exclude>
                         <exclude>**/GraphAlgoTest.java</exclude>
+                        <exclude>**/FlexTypeTest.java</exclude>
                     </excludes>
                 </configuration>
             </plugin>

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/config/FrontendConfig.java

@@ -66,4 +66,7 @@ public class FrontendConfig {
 
     public static final Config<Long> METRICS_TOOL_INTERVAL_MS =
             Config.longConfig("metrics.tool.interval.ms", 5 * 60 * 1000L);
+
+    public static final Config<Boolean> DISABLE_EXPR_SIMPLIFY =
+            Config.boolConfig("disable.expr.simplify", false);
 }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/meta/fetcher/StaticIrMetaFetcher.java

@@ -52,7 +52,7 @@ public StaticIrMetaFetcher(IrMetaReader dataReader, List<IrMetaTracker> tracker)
         this.metaStats =
                 new IrMetaStats(
                         meta.getSnapshotId(), meta.getSchema(), meta.getStoredProcedures(), stats);
-        if (tracker != null && stats != null) {
+        if (tracker != null) {
             tracker.forEach(t -> t.onStatsChanged(this.metaStats));
         }
     }

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/meta/glogue/calcite/handler/GraphCollationHandler.java

@@ -1,19 +1,17 @@
 /*
+ * Copyright 2020 Alibaba Group Holding Limited.
  *
- *  * 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.
+ * 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.meta.glogue.calcite.handler;

interactive_engine/compiler/src/main/java/com/alibaba/graphscope/common/ir/meta/schema/IrDataTypeConvertor.java

@@ -36,6 +36,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.math.BigDecimal;
 import java.util.Map;
 import java.util.Objects;
 
@@ -45,6 +46,15 @@
 public interface IrDataTypeConvertor<T> {
     Logger logger = LoggerFactory.getLogger(IrDataTypeConvertor.class);
 
+    // support unsigned type as decimal type with fixed precision and scale
+    int UINT32_PRECISION = 10;
+    int UINT32_SCALE = 0;
+    int UINT64_PRECISION = 20;
+    int UINT64_SCALE = 0;
+
+    BigDecimal UINT32_MAX = new BigDecimal("4294967295");
+    BigDecimal UINT64_MAX = new BigDecimal("18446744073709551615");
+
     RelDataType convert(T dataFrom);
 
     T convert(RelDataType dataFrom);
@@ -58,402 +68,416 @@
             this.throwsOnFail = throwsOnFail;
         }
 
         @Override
         public RelDataType convert(DataType from) {
             requireNonNull(typeFactory, "typeFactory should not be null");
             switch (from) {
                 case BOOL:
                     return typeFactory.createSqlType(SqlTypeName.BOOLEAN);
                 case CHAR:
                     // single character
                     return typeFactory.createSqlType(SqlTypeName.CHAR, 1);
                 case STRING:
                     // string with unlimited length
                     return typeFactory.createSqlType(
                             SqlTypeName.VARCHAR, RelDataType.PRECISION_NOT_SPECIFIED);
                 case SHORT:
                     // 2-bytes signed integer
                     return typeFactory.createSqlType(SqlTypeName.SMALLINT);
                 case INT:
                     // 4-bytes signed integer
                     return typeFactory.createSqlType(SqlTypeName.INTEGER);
                 case LONG:
                     // 8-bytes signed integer
                     return typeFactory.createSqlType(SqlTypeName.BIGINT);
                 case FLOAT:
                     // single precision floating point, 4 bytes
                     return typeFactory.createSqlType(SqlTypeName.FLOAT);
                 case DOUBLE:
                     // double precision floating point, 8 bytes
                     return typeFactory.createSqlType(SqlTypeName.DOUBLE);
                 case DATE:
                     // int32 days since 1970-01-01
                     return typeFactory.createSqlType(SqlTypeName.DATE);
                 case TIME32:
                     // int32 milliseconds past midnight
                     return typeFactory.createSqlType(SqlTypeName.TIME);
                 case TIMESTAMP:
                     // int64 milliseconds since 1970-01-01 00:00:00.000000
                     return typeFactory.createSqlType(SqlTypeName.TIMESTAMP);
                 case INT_LIST:
                     // array of 4-bytes signed integer, unlimited size
                     return typeFactory.createArrayType(
                             convert(DataType.INT), RelDataType.PRECISION_NOT_SPECIFIED);
                 case LONG_LIST:
                     // array of 8-bytes signed integer, unlimited size
                     return typeFactory.createArrayType(
                             convert(DataType.LONG), RelDataType.PRECISION_NOT_SPECIFIED);
                 case FLOAT_LIST:
                     // array of single precision floating point, unlimited size
                     return typeFactory.createArrayType(
                             convert(DataType.FLOAT), RelDataType.PRECISION_NOT_SPECIFIED);
                 case DOUBLE_LIST:
                     // array of double precision floating point, unlimited size
                     return typeFactory.createArrayType(
                             convert(DataType.DOUBLE), RelDataType.PRECISION_NOT_SPECIFIED);
                 case STRING_LIST:
                     // array of string, unlimited size
                     return typeFactory.createArrayType(
                             convert(DataType.STRING), RelDataType.PRECISION_NOT_SPECIFIED);
                 case UNKNOWN:
                     return typeFactory.createSqlType(SqlTypeName.UNKNOWN);
                 case BYTES:
                 case BYTES_LIST:
                 default:
                     if (throwsOnFail) {
                         throw new UnsupportedOperationException(
                                 "convert GrootDataType ["
                                         + from.name()
                                         + "] to RelDataType is unsupported yet");
                     }
                     return typeFactory.createSqlType(SqlTypeName.ANY);
             }
         }
 
         @Override
         public DataType convert(RelDataType dataFrom) {
             SqlTypeName typeName = dataFrom.getSqlTypeName();
             switch (typeName) {
                 case BOOLEAN:
                     return DataType.BOOL;
                 case CHAR:
                     if (dataFrom.getPrecision() == 1) {
                         return DataType.CHAR;
                     }
                     break;
                 case VARCHAR:
                     if (dataFrom.getPrecision() == RelDataType.PRECISION_NOT_SPECIFIED) {
                         return DataType.STRING;
                     }
                     break;
                 case SMALLINT:
                     return DataType.SHORT;
                 case INTEGER:
                     return DataType.INT;
                 case BIGINT:
                     return DataType.LONG;
                 case FLOAT:
                     return DataType.FLOAT;
                 case DOUBLE:
                     return DataType.DOUBLE;
                 case DATE:
                     return DataType.DATE;
                 case TIME:
                     return DataType.TIME32;
                 case TIMESTAMP:
                     return DataType.TIMESTAMP;
                 case MULTISET:
                 case ARRAY:
                     RelDataType componentType = dataFrom.getComponentType();
                     // check the array or set is an unlimited size list of primitive type
                     if (componentType != null
                             && dataFrom.getPrecision() == RelDataType.PRECISION_NOT_SPECIFIED) {
                         switch (componentType.getSqlTypeName()) {
                             case INTEGER:
                                 return DataType.INT_LIST;
                             case BIGINT:
                                 return DataType.LONG_LIST;
                             case FLOAT:
                                 return DataType.FLOAT_LIST;
                             case DOUBLE:
                                 return DataType.DOUBLE_LIST;
                             case VARCHAR:
                                 if (componentType.getPrecision()
                                         == RelDataType.PRECISION_NOT_SPECIFIED) {
                                     return DataType.STRING_LIST;
                                 }
                         }
                     }
                     break;
                 case UNKNOWN:
                     return DataType.UNKNOWN;
                 default:
             }
             if (throwsOnFail) {
                 throw new UnsupportedOperationException(
                         "convert RelDataType ["
                                 + dataFrom
                                 + "] to GrootDataType is unsupported yet");
             }
             return DataType.UNKNOWN;
         }
     }
 
     class Flex implements IrDataTypeConvertor<GSDataTypeDesc> {
         private final RelDataTypeFactory typeFactory;
         private final boolean throwsOnFail;
 
         public Flex(RelDataTypeFactory typeFactory, boolean throwsOnFail) {
             this.typeFactory = typeFactory;
             this.throwsOnFail = throwsOnFail;
         }
 
         @Override
         public RelDataType convert(GSDataTypeDesc from) {
             Objects.requireNonNull(typeFactory, "typeFactory should not be null");
             Map<String, Object> typeMap = from.getYamlDesc();
             Object value;
             if ((value = typeMap.get("primitive_type")) != null) {
                 switch (value.toString()) {
                     case "DT_NULL":
                         return typeFactory.createSqlType(SqlTypeName.NULL);
                     case "DT_ANY":
                         // any type
                         return typeFactory.createSqlType(SqlTypeName.ANY);
+                    case "DT_UNSIGNED_INT32":
+                        return typeFactory.createSqlType(
+                                SqlTypeName.DECIMAL, UINT32_PRECISION, UINT32_SCALE);
                     case "DT_SIGNED_INT32":
                         // 4-bytes signed integer
                         return typeFactory.createSqlType(SqlTypeName.INTEGER);
+                    case "DT_UNSIGNED_INT64":
+                        return typeFactory.createSqlType(
+                                SqlTypeName.DECIMAL, UINT64_PRECISION, UINT64_SCALE);
                     case "DT_SIGNED_INT64":
                         // 8-bytes signed integer
                         return typeFactory.createSqlType(SqlTypeName.BIGINT);
                     case "DT_BOOL":
                         // boolean type
                         return typeFactory.createSqlType(SqlTypeName.BOOLEAN);
                     case "DT_FLOAT":
                         // single precision floating point, 4 bytes
                         return typeFactory.createSqlType(SqlTypeName.FLOAT);
                     case "DT_DOUBLE":
                         // double precision floating point, 8 bytes
                         return typeFactory.createSqlType(SqlTypeName.DOUBLE);
                 }
             } else if ((value = typeMap.get("string")) != null) {
                 Map<String, Object> strType = (Map<String, Object>) value;
                 if (strType.containsKey("long_text")) {
                     // string with unlimited length
                     return typeFactory.createSqlType(
                             SqlTypeName.VARCHAR, RelDataType.PRECISION_NOT_SPECIFIED);
                 } else if (strType.containsKey("char")) {
                     Object charValue = strType.get("char");
                     Integer fixedLen = getIntValue(charValue, "fixed_length");
                     if (fixedLen == null) {
                         fixedLen =
                                 typeFactory.getTypeSystem().getDefaultPrecision(SqlTypeName.CHAR);
                         logger.warn(
                                 "can not convert {} to a valid fixed length, use default"
                                         + " length {} instead",
                                 charValue,
                                 fixedLen);
                     }
                     // string with fixed length
                     return typeFactory.createSqlType(SqlTypeName.CHAR, fixedLen);
                 } else if (strType.containsKey("var_char")) {
                     Object varCharValue = strType.get("var_char");
                     Integer maxLen = getIntValue(varCharValue, "max_length");
                     if (maxLen == null) {
                         maxLen =
                                 typeFactory
                                         .getTypeSystem()
                                         .getDefaultPrecision(SqlTypeName.VARCHAR);
                         logger.warn(
                                 "can not convert {} to a valid max length, use default"
                                         + " length {} instead",
                                 varCharValue,
                                 maxLen);
                     }
                     // string with variable length, bound by max length
                     return typeFactory.createSqlType(SqlTypeName.VARCHAR, maxLen);
                 }
             } else if ((value = typeMap.get("temporal")) != null) {
                 Map<String, Object> temporalType = (Map<String, Object>) value;
                 if (temporalType.containsKey("date32")) {
                     // int32 days since 1970-01-01
                     return typeFactory.createSqlType(SqlTypeName.DATE);
                 } else if (temporalType.containsKey("time32")) {
                     // int32 milliseconds past midnight
                     return typeFactory.createSqlType(SqlTypeName.TIME);
                 } else if (temporalType.containsKey("timestamp")) {
                     // int64 milliseconds since 1970-01-01 00:00:00.000000
                     return typeFactory.createSqlType(SqlTypeName.TIMESTAMP);
                 }
             } else if ((value = typeMap.get("array")) != null) {
                 Map<String, Object> arrayType = (Map<String, Object>) value;
                 Map<String, Object> componentType =
                         (Map<String, Object>) arrayType.get("component_type");
                 Preconditions.checkArgument(
                         componentType != null, "field 'component_type' is required in array type");
                 // array of component type, unlimited size
                 return typeFactory.createArrayType(
                         convert(new GSDataTypeDesc(componentType)),
                         RelDataType.PRECISION_NOT_SPECIFIED);
             } else if ((value = typeMap.get("map")) != null) {
                 Map<String, Object> mapType = (Map<String, Object>) value;
                 Map<String, Object> keyType = (Map<String, Object>) mapType.get("key_type");
                 Preconditions.checkArgument(
                         keyType != null, "field 'key_type' is required in map type");
                 Map<String, Object> valueType = (Map<String, Object>) mapType.get("value_type");
                 Preconditions.checkArgument(
                         valueType != null, "field 'value_type' is required in map type");
                 // map of key type to value type
                 return typeFactory.createMapType(
                         convert(new GSDataTypeDesc(keyType)),
                         convert(new GSDataTypeDesc(valueType)));
             } else if ((value = typeMap.get("decimal")) != null) {
                 Integer precision = getIntValue(value, "precision");
                 if (precision == null) {
                     precision =
                             typeFactory.getTypeSystem().getDefaultPrecision(SqlTypeName.DECIMAL);
                     logger.warn(
                             "can not convert {} to a valid precision, use default"
                                     + " precision {} instead",
                             value,
                             precision);
                 }
                 Integer scale = getIntValue(value, "scale");
                 if (scale == null) {
                     scale = typeFactory.getTypeSystem().getMaxScale(SqlTypeName.DECIMAL);
                     logger.warn(
                             "can not convert {} to a valid scale, use max" + " scale {} instead",
                             value,
                             scale);
                 }
                 // decimal type with precision and scale
                 return typeFactory.createSqlType(SqlTypeName.DECIMAL, precision, scale);
             }
             if (throwsOnFail) {
                 throw new UnsupportedOperationException(
                         "convert GSDataTypeDesc [" + from + "] to RelDataType is unsupported yet");
             }
             return typeFactory.createSqlType(SqlTypeName.ANY);
         }
 
         @Override
         public GSDataTypeDesc convert(RelDataType from) {
             if (from instanceof IntervalSqlType) {
                 return new GSDataTypeDesc(ImmutableMap.of("primitive_type", "DT_SIGNED_INT64"));
             } else if (from instanceof GraphLabelType) {
                 return new GSDataTypeDesc(ImmutableMap.of("primitive_type", "DT_SIGNED_INT32"));
             }
             SqlTypeName typeName = from.getSqlTypeName();
             Map<String, Object> yamlDesc;
             switch (typeName) {
                 case ANY:
                     yamlDesc = ImmutableMap.of("primitive_type", "DT_ANY");
                     break;
                 case NULL:
                     yamlDesc = ImmutableMap.of("primitive_type", "DT_NULL");
                     break;
                 case INTEGER:
                     yamlDesc = ImmutableMap.of("primitive_type", "DT_SIGNED_INT32");
                     break;
                 case BIGINT:
                     yamlDesc = ImmutableMap.of("primitive_type", "DT_SIGNED_INT64");
                     break;
                 case BOOLEAN:
                     yamlDesc = ImmutableMap.of("primitive_type", "DT_BOOL");
                     break;
                 case FLOAT:
                     yamlDesc = ImmutableMap.of("primitive_type", "DT_FLOAT");
                     break;
                 case DOUBLE:
                     yamlDesc = ImmutableMap.of("primitive_type", "DT_DOUBLE");
                     break;
                 case CHAR:
                     Map charMap = Maps.newHashMap();
                     charMap.put("char", ImmutableMap.of("fixed_length", from.getPrecision()));
                     yamlDesc = ImmutableMap.of("string", charMap);
                     break;
                 case VARCHAR:
                     if (from.getPrecision() == RelDataType.PRECISION_NOT_SPECIFIED) {
                         Map longTextMap = Maps.newHashMap();
                         longTextMap.put("long_text", ImmutableMap.of());
                         yamlDesc = ImmutableMap.of("string", longTextMap);
                     } else {
                         Map varCharMap = Maps.newHashMap();
                         varCharMap.put(
                                 "var_char", ImmutableMap.of("max_length", from.getPrecision()));
                         yamlDesc = ImmutableMap.of("string", varCharMap);
                     }
                     break;
                 case DATE:
                     Map dateMap = Maps.newHashMap();
                     dateMap.put("date32", ImmutableMap.of());
                     yamlDesc = ImmutableMap.of("temporal", dateMap);
                     break;
                 case TIME:
                     Map timeMap = Maps.newHashMap();
                     timeMap.put("time32", ImmutableMap.of());
                     yamlDesc = ImmutableMap.of("temporal", timeMap);
                     break;
                 case TIMESTAMP:
                     Map timestampMap = Maps.newHashMap();
                     timestampMap.put("timestamp", ImmutableMap.of());
                     yamlDesc = ImmutableMap.of("temporal", timestampMap);
                     break;
                 case ARRAY:
                 case MULTISET:
                     Map<String, Object> componentType;
                     if (from instanceof ArbitraryArrayType) {
                         componentType = ImmutableMap.of("primitive_type", "DT_ANY");
                     } else {
                         componentType = convert(from.getComponentType()).getYamlDesc();
                     }
                     yamlDesc =
                             ImmutableMap.of(
                                     "array",
                                     ImmutableMap.of(
                                             "component_type",
                                             componentType,
                                             "max_length",
                                             from.getPrecision()
                                                             == RelDataType.PRECISION_NOT_SPECIFIED
                                                     ? Integer.MAX_VALUE
                                                     : from.getPrecision()));
                     break;
                 case MAP:
                     Map<String, Object> keyType;
                     Map<String, Object> valueType;
                     if (from instanceof ArbitraryMapType) {
                         keyType = ImmutableMap.of("primitive_type", "DT_ANY");
                         valueType = ImmutableMap.of("primitive_type", "DT_ANY");
                     } else {
                         keyType = convert(from.getKeyType()).getYamlDesc();
                         valueType = convert(from.getValueType()).getYamlDesc();
                     }
                     yamlDesc =
                             ImmutableMap.of(
                                     "map",
                                     ImmutableMap.of("key_type", keyType, "value_type", valueType));
                     break;
                 case DECIMAL:
-                    yamlDesc =
-                            ImmutableMap.of(
-                                    "decimal",
-                                    ImmutableMap.of(
-                                            "precision", from.getPrecision(),
-                                            "scale", from.getScale()));
+                    if (from.getPrecision() == UINT32_PRECISION
+                            && from.getScale() == UINT32_SCALE) {
+                        yamlDesc = ImmutableMap.of("primitive_type", "DT_UNSIGNED_INT32");
+                    } else if (from.getPrecision() == UINT64_PRECISION
+                            && from.getScale() == UINT64_SCALE) {
+                        yamlDesc = ImmutableMap.of("primitive_type", "DT_UNSIGNED_INT64");
+                    } else {
+                        yamlDesc =
+                                ImmutableMap.of(
+                                        "decimal",
+                                        ImmutableMap.of(
+                                                "precision", from.getPrecision(),
+                                                "scale", from.getScale()));
+                    }
                     break;
                 default:
                     if (throwsOnFail) {
                         throw new UnsupportedOperationException(
                                 "convert RelDataType ["
                                         + from
                                         + "] to GSDataTypeDesc is unsupported yet");
                     }
                     yamlDesc = ImmutableMap.of("primitive_type", "DT_ANY");
             }
             return new GSDataTypeDesc(yamlDesc);
         }
 
         private @Nullable Integer getIntValue(Object valueMap, String key) {
             if (valueMap instanceof Map) {
                 Object value = ((Map) valueMap).get(key);

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

@@ -46,6 +46,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.math.BigDecimal;
 import java.util.*;
 import java.util.stream.Collectors;
 
@@ -55,75 +56,94 @@
 public abstract class Utils {
     private static final Logger logger = LoggerFactory.getLogger(Utils.class);
 
     public static final Common.Value protoValue(RexLiteral literal) {
         if (literal.getTypeName() == SqlTypeName.SARG) {
             Sarg sarg = literal.getValueAs(Sarg.class);
             if (!sarg.isPoints()) {
                 throw new UnsupportedOperationException(
                         "can not convert continuous ranges to ir core array, sarg=" + sarg);
             }
             List<Comparable> values =
                     com.alibaba.graphscope.common.ir.tools.Utils.getValuesAsList(sarg);
             switch (literal.getType().getSqlTypeName()) {
                 case INTEGER:
                     Common.I32Array.Builder i32Array = Common.I32Array.newBuilder();
                     values.forEach(value -> i32Array.addItem(((Number) value).intValue()));
                     return Common.Value.newBuilder().setI32Array(i32Array).build();
                 case BIGINT:
                     Common.I64Array.Builder i64Array = Common.I64Array.newBuilder();
                     values.forEach(value -> i64Array.addItem(((Number) value).longValue()));
                     return Common.Value.newBuilder().setI64Array(i64Array).build();
                 case CHAR:
                     Common.StringArray.Builder stringArray = Common.StringArray.newBuilder();
                     values.forEach(
                             value ->
                                     stringArray.addItem(
                                             (value instanceof NlsString)
                                                     ? ((NlsString) value).getValue()
                                                     : (String) value));
                     return Common.Value.newBuilder().setStrArray(stringArray).build();
                 case DECIMAL:
                 case FLOAT:
                 case DOUBLE:
                     Common.DoubleArray.Builder doubleArray = Common.DoubleArray.newBuilder();
                     values.forEach(value -> doubleArray.addItem(((Number) value).doubleValue()));
                     return Common.Value.newBuilder().setF64Array(doubleArray).build();
                 default:
                     throw new UnsupportedOperationException(
                             "can not convert sarg=" + sarg + " ir core array");
             }
         }
         switch (literal.getType().getSqlTypeName()) {
             case NULL:
                 return Common.Value.newBuilder().setNone(Common.None.newBuilder().build()).build();
             case BOOLEAN:
                 return Common.Value.newBuilder().setBoolean((Boolean) literal.getValue()).build();
             case INTEGER:
                 return Common.Value.newBuilder()
                         .setI32(((Number) literal.getValue()).intValue())
                         .build();
             case BIGINT:
                 return Common.Value.newBuilder()
                         .setI64(((Number) literal.getValue()).longValue())
                         .build();
             case CHAR:
                 String valueStr =
                         (literal.getValue() instanceof NlsString)
                                 ? ((NlsString) literal.getValue()).getValue()
                                 : (String) literal.getValue();
                 return Common.Value.newBuilder().setStr(valueStr).build();
             case DECIMAL:
+                if (literal.getType().getPrecision() == IrDataTypeConvertor.UINT32_PRECISION
+                        && literal.getType().getScale() == IrDataTypeConvertor.UINT32_SCALE) {
+                    BigDecimal uint32Value = (BigDecimal) literal.getValue();
+                    return Common.Value.newBuilder().setU32(uint32Value.intValue()).build();
+                }
+                if (literal.getType().getPrecision() == IrDataTypeConvertor.UINT64_PRECISION
+                        && literal.getType().getScale() == IrDataTypeConvertor.UINT64_SCALE) {
+                    BigDecimal uint32Value = (BigDecimal) literal.getValue();
+                    return Common.Value.newBuilder().setU64(uint32Value.longValue()).build();
+                }
+                throw new UnsupportedOperationException(
+                        "decimal type with precision="
+                                + literal.getType().getPrecision()
+                                + ", scale="
+                                + literal.getType().getScale()
+                                + " is unsupported yet");
             case FLOAT:
+                return Common.Value.newBuilder()
+                        .setF32(((Number) literal.getValue()).floatValue())
+                        .build();
             case DOUBLE:
                 return Common.Value.newBuilder()
                         .setF64(((Number) literal.getValue()).doubleValue())
                         .build();
             default:
                 throw new UnsupportedOperationException(
                         "literal type " + literal.getTypeName() + " is unsupported yet");
         }
     }
 
     public static final OuterExpression.Property protoProperty(GraphProperty property) {
         switch (property.getOpt()) {
             case ID:
@@ -160,118 +180,118 @@
         }
     }
 
     public static final OuterExpression.ExprOpr protoOperator(SqlOperator operator) {
         switch (operator.getKind()) {
             case PLUS:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setArith(OuterExpression.Arithmetic.ADD)
                         .build();
             case MINUS:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setArith(OuterExpression.Arithmetic.SUB)
                         .build();
             case TIMES:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setArith(OuterExpression.Arithmetic.MUL)
                         .build();
             case DIVIDE:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setArith(OuterExpression.Arithmetic.DIV)
                         .build();
             case MOD:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setArith(OuterExpression.Arithmetic.MOD)
                         .build();
             case OTHER_FUNCTION:
                 if (operator.getName().equals("POWER")) {
                     return OuterExpression.ExprOpr.newBuilder()
                             .setArith(OuterExpression.Arithmetic.EXP)
                             .build();
                 } else if (operator.getName().equals("<<")) {
                     return OuterExpression.ExprOpr.newBuilder()
                             .setArith(OuterExpression.Arithmetic.BITLSHIFT)
                             .build();
                 } else if (operator.getName().equals(">>")) {
                     return OuterExpression.ExprOpr.newBuilder()
                             .setArith(OuterExpression.Arithmetic.BITRSHIFT)
                             .build();
                 }
             case EQUALS:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.EQ)
                         .build();
             case NOT_EQUALS:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.NE)
                         .build();
             case GREATER_THAN:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.GT)
                         .build();
             case GREATER_THAN_OR_EQUAL:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.GE)
                         .build();
             case LESS_THAN:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.LT)
                         .build();
             case LESS_THAN_OR_EQUAL:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.LE)
                         .build();
             case AND:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.AND)
                         .build();
             case OR:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.OR)
                         .build();
             case NOT:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.NOT)
                         .build();
             case IS_NULL:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.ISNULL)
                         .build();
             case SEARCH:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.WITHIN)
                         .build();
             case POSIX_REGEX_CASE_SENSITIVE:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setLogical(OuterExpression.Logical.REGEX)
                         .build();
             case BIT_AND:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setArith(OuterExpression.Arithmetic.BITAND)
                         .build();
             case BIT_OR:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setArith(OuterExpression.Arithmetic.BITOR)
                         .build();
             case BIT_XOR:
                 return OuterExpression.ExprOpr.newBuilder()
                         .setArith(OuterExpression.Arithmetic.BITXOR)
                         .build();
             case OTHER:
                 if (operator.getName().equals("IN")) {
                     return OuterExpression.ExprOpr.newBuilder()
                             .setLogical(OuterExpression.Logical.WITHIN)
                             .build();
                 }
             default:
                 throw new UnsupportedOperationException(
                         "operator type="
                                 + operator.getKind()
                                 + ", name="
                                 + operator.getName()
                                 + " is unsupported yet");
         }
     }
 
     public static final com.alibaba.graphscope.proto.type.Common.DataType protoBasicDataType(
             RelDataType basicType) {
         try {

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

@@ -42,6 +42,7 @@
 import com.alibaba.graphscope.common.ir.type.*;
 import com.alibaba.graphscope.gremlin.Utils;
 import com.alibaba.graphscope.proto.frontend.Code;
+import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
 import com.google.common.collect.ImmutableList;
 import com.google.common.collect.ImmutableSet;
@@ -96,6 +97,9 @@
                 new GraphRexSimplify(
                         cluster.getRexBuilder(), RelOptPredicateList.EMPTY, RexUtil.EXECUTOR));
         this.configs = context.unwrapOrThrow(Configs.class);
+        if (FrontendConfig.DISABLE_EXPR_SIMPLIFY.get(this.configs)) {
+            disableSimplify();
+        }
     }
 
     /**
@@ -109,6 +113,12 @@
         return new GraphBuilder(context, cluster, relOptSchema);
     }
 
+    @VisibleForTesting
+    public void disableSimplify() {
+        Config config = Utils.getFieldValue(RelBuilder.class, this, "config");
+        config.withSimplify(false);
+    }
+
     public Context getContext() {
         return this.configs;
     }
@@ -498,148 +508,148 @@
      * @param property
      * @return
      */
     public RexGraphVariable variable(@Nullable String alias, String property) {
         alias = AliasInference.isDefaultAlias(alias) ? AliasInference.DEFAULT_NAME : alias;
         Objects.requireNonNull(property);
         String varName = AliasInference.SIMPLE_NAME(alias) + AliasInference.DELIMITER + property;
         List<ColumnField> columnFields = getAliasField(alias);
         if (columnFields.size() != 1) {
             throw new FrontendException(
                     Code.PROPERTY_NOT_FOUND,
                     "cannot get property="
                             + property
                             + " from alias="
                             + alias
                             + ", expected one column, but found "
                             + columnFields.size());
         }
         ColumnField columnField = columnFields.get(0);
         RelDataTypeField aliasField = columnField.right;
         if (property.equals(GraphProperty.LEN_KEY)) {
             if (!(aliasField.getType() instanceof GraphPathType)) {
                 throw new FrontendException(
                         Code.PROPERTY_NOT_FOUND,
                         "cannot get property='len' from type class ["
                                 + aliasField.getType().getClass()
                                 + "], should be ["
                                 + GraphPathType.class
                                 + "]");
             } else {
                 return RexGraphVariable.of(
                         aliasField.getIndex(),
                         new GraphProperty(GraphProperty.Opt.LEN),
                         columnField.left,
                         varName,
                         getTypeFactory().createSqlType(SqlTypeName.INTEGER));
             }
         }
         if (!(aliasField.getType() instanceof GraphSchemaType)) {
             throw new FrontendException(
                     Code.PROPERTY_NOT_FOUND,
                     "cannot get property=['id', 'label', 'all', 'key'] from type class ["
                             + aliasField.getType().getClass()
                             + "], should be ["
                             + GraphSchemaType.class
                             + "]");
         }
         if (property.equals(GraphProperty.LABEL_KEY)) {
             GraphSchemaType schemaType = (GraphSchemaType) aliasField.getType();
             return RexGraphVariable.of(
                     aliasField.getIndex(),
                     new GraphProperty(GraphProperty.Opt.LABEL),
                     columnField.left,
                     varName,
                     schemaType.getLabelType());
         } else if (property.equals(GraphProperty.ID_KEY)) {
             return RexGraphVariable.of(
                     aliasField.getIndex(),
                     new GraphProperty(GraphProperty.Opt.ID),
                     columnField.left,
                     varName,
                     getTypeFactory().createSqlType(SqlTypeName.BIGINT));
         } else if (property.equals(GraphProperty.ALL_KEY)) {
             return RexGraphVariable.of(
                     aliasField.getIndex(),
                     new GraphProperty(GraphProperty.Opt.ALL),
                     columnField.left,
                     varName,
                     getTypeFactory().createSqlType(SqlTypeName.ANY));
         } else if (property.equals(GraphProperty.START_V_KEY)) {
             if (!(aliasField.getType() instanceof GraphPathType)) {
                 throw new FrontendException(
                         Code.PROPERTY_NOT_FOUND,
                         "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 FrontendException(
                         Code.PROPERTY_NOT_FOUND,
                         "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<>();
         boolean isColumnId =
                 (relOptSchema instanceof GraphOptSchema)
                         ? ((GraphOptSchema) relOptSchema).getRootSchema().isColumnId()
                         : false;
         for (RelDataTypeField pField : graphType.getFieldList()) {
             if (pField.getName().equals(property)) {
                 return RexGraphVariable.of(
                         aliasField.getIndex(),
                         isColumnId
                                 ? new GraphProperty(new GraphNameOrId(pField.getIndex()))
                                 : new GraphProperty(new GraphNameOrId(pField.getName())),
                         columnField.left,
                         varName,
                         pField.getType());
             }
             properties.add(pField.getName());
         }
         throw new FrontendException(
                 Code.PROPERTY_NOT_FOUND,
                 "{property="
                         + property
                         + "} "
                         + "not found; expected properties are: "
                         + properties);
     }
 
     /**
      * get {@code RelDataTypeField} by the given alias, for {@code RexGraphVariable} inference
      *
      * @param alias
      * @return
      */
     private List<ColumnField> getAliasField(String alias) {
         Objects.requireNonNull(alias);
         if (alias.equals(AliasInference.STAR)) {
@@ -856,254 +866,254 @@
         }
     }
 
     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")
                                         || operator.getName().equals("PATH_CONCAT")
                                         || operator.getName().equals("PATH_FUNCTION"))
                         || operator.getName().startsWith(GraphFunctions.FUNCTION_PREFIX))
                 || 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,
@@ -1491,122 +1501,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");
 
@@ -1739,7 +1749,11 @@
             return rexBuilder.makeLiteral((Boolean) value);
         } else if (value instanceof BigDecimal) {
             return rexBuilder.makeExactLiteral((BigDecimal) value);
-        } else if (value instanceof Float || value instanceof Double) {
+        } else if (value instanceof Float) {
+            return rexBuilder.makeApproxLiteral(
+                    BigDecimal.valueOf(((Number) value).floatValue()),
+                    getTypeFactory().createSqlType(SqlTypeName.FLOAT));
+        } else if (value instanceof Double) {
             return rexBuilder.makeApproxLiteral(BigDecimal.valueOf(((Number) value).doubleValue()));
         } else if (value instanceof Long) { // convert long to BIGINT, i.e. 2l
             return rexBuilder.makeBigintLiteral(BigDecimal.valueOf(((Number) value).longValue()));
@@ -1830,104 +1844,104 @@
      * @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...
         RelNode parent = null;
         while (!top.getInputs().isEmpty() && top.getInput(0).getRowType() == top.getRowType()) {
             parent = top;
             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.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)));
                 }
             }
             if (parent != null && peek() != top) {
                 parent.replaceInput(0, build());
                 push(parent);
             }
         }
         return this;
     }
 
     private LabelConfig getLabelConfig(TableConfig tableConfig) {
         List<String> labels =
                 tableConfig.getTables().stream()