feat: Optimizing Sorting Performance via Radix Sort Algorithm

geaflow/geaflow-common/src/main/java/org/apache/geaflow/common/type/Types.java

@@ -94,7 +94,7 @@ public static <T> IType<T> getType(Class<T> type) {
         return TYPE_IMMUTABLE_MAP.get(type);
     }
 
-    public static IType<?> of(String typeName) {
+    public static IType<?> of(String typeName, int precision) {
         if (typeName == null) {
             throw new IllegalArgumentException("typeName is null");
         }
@@ -116,7 +116,7 @@ public static IType<?> of(String typeName) {
             case TYPE_NAME_DECIMAL:
                 return DECIMAL;
             case TYPE_NAME_BINARY_STRING:
-                return BINARY_STRING;
+                return new BinaryStringType(precision);
             case TYPE_NAME_TIMESTAMP:
                 return TIMESTAMP;
             case TYPE_NAME_DATE:

geaflow/geaflow-common/src/main/java/org/apache/geaflow/common/type/primitive/BinaryStringType.java

@@ -27,6 +27,37 @@ public class BinaryStringType implements IType<BinaryString> {
 
     public static final BinaryStringType INSTANCE = new BinaryStringType();
 
+    private int precision;
+
+    public BinaryStringType() {
+
+    }
+
+    public BinaryStringType(int precision) {
+        this.precision = precision;
+    }
+
+    public int getPrecision() {
+        return precision;
+    }
+
+    @Override
+    public boolean equals(Object obj) {
+        if (this == obj) {
+            return true;
+        }
+
+        if (obj == null) {
+            return false;
+        }
+
+        if (getClass() != obj.getClass()) {
+            return false;
+        }
+
+        return true;
+    }
+
     @Override
     public String getName() {
         return Types.TYPE_NAME_BINARY_STRING;

geaflow/geaflow-dsl/geaflow-dsl-catalog/src/main/java/org/apache/geaflow/dsl/catalog/console/CatalogUtil.java

@@ -98,7 +98,7 @@ public static VertexTable convertToVertexTable(String instanceName, VertexModel
                 idFieldName = fieldModel.getName();
             }
             String typeName = convertTypeName(fieldModel.getType().name());
-            IType<?> fieldType = Types.of(typeName);
+            IType<?> fieldType = Types.of(typeName, -1);
             TableField field = new TableField(fieldModel.getName(), fieldType, false);
             fields.add(field);
         }
@@ -154,7 +154,7 @@ public static EdgeTable convertToEdgeTable(String instanceName, EdgeModel model)
                 default:
             }
             String typeName = convertTypeName(fieldModel.getType().name());
-            IType<?> fieldType = Types.of(typeName);
+            IType<?> fieldType = Types.of(typeName, -1);
             TableField field = new TableField(fieldModel.getName(), fieldType, false);
             fields.add(field);
         }
@@ -237,7 +237,7 @@ private static List<TableField> convertToTableField(List<FieldModel> fieldModels
         List<TableField> fields = new ArrayList<>(fieldModels.size());
         for (FieldModel fieldModel : fieldModels) {
             String typeName = convertTypeName(fieldModel.getType().name());
-            IType<?> fieldType = Types.of(typeName);
+            IType<?> fieldType = Types.of(typeName, -1);
             TableField field = new TableField(fieldModel.getName(), fieldType, false);
             fields.add(field);
         }

geaflow/geaflow-dsl/geaflow-dsl-parser/src/main/java/org/apache/geaflow/dsl/util/SqlTypeUtil.java

@@ -29,6 +29,7 @@
 import org.apache.calcite.sql.type.SqlTypeName;
 import org.apache.geaflow.common.type.IType;
 import org.apache.geaflow.common.type.Types;
+import org.apache.geaflow.common.type.primitive.BinaryStringType;
 import org.apache.geaflow.dsl.calcite.EdgeRecordType;
 import org.apache.geaflow.dsl.calcite.GraphRecordType;
 import org.apache.geaflow.dsl.calcite.PathRecordType;
@@ -47,7 +48,7 @@ public final class SqlTypeUtil {
     public static IType<?> convertType(SqlDataTypeSpec typeSpec) {
         String typeName = typeSpec.getTypeName().getSimple().toUpperCase();
         typeName = convertTypeName(typeName);
-        return Types.of(typeName);
+        return Types.of(typeName, typeSpec.getPrecision());
     }
 
     public static IType<?> convertType(RelDataType type) {
@@ -90,7 +91,7 @@ private static List<TableField> toTableFields(List<RelDataTypeField> fields) {
 
     public static IType<?> ofTypeName(SqlTypeName sqlTypeName) {
         String typeName = convertTypeName(sqlTypeName.getName());
-        return Types.of(typeName);
+        return Types.of(typeName, sqlTypeName.getPrecision());
     }
 
     public static RelDataType convertToRelType(IType<?> type, boolean isNullable,
@@ -129,7 +130,9 @@ public static RelDataType convertToRelType(IType<?> type, boolean isNullable,
             default:
                 if (type.isPrimitive()) {
                     String sqlTypeName = convertToSqlTypeName(type);
-                    SqlTypeName typeName = SqlTypeName.valueOf(sqlTypeName);
+                    SqlTypeName typeName = Types.getType(type.getTypeClass()) == Types.BINARY_STRING
+                        ? SqlTypeName.get(sqlTypeName, ((BinaryStringType) type).getPrecision())
+                        : SqlTypeName.get(sqlTypeName);
                     return typeFactory.createTypeWithNullability(typeFactory.createSqlType(typeName), isNullable);
                 } else {
                     throw new GeaFlowDSLException("Not support type: " + type);

geaflow/geaflow-dsl/geaflow-dsl-runtime/pom.xml

@@ -116,6 +116,17 @@
             <artifactId>testng</artifactId>
         </dependency>
 
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.openjdk.jmh</groupId>
+            <artifactId>jmh-generator-annprocess</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <!-- include kafka server for tests  -->
             <groupId>org.apache.kafka</groupId>

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/table/OrderByHeapSort.java

@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.geaflow.dsl.runtime.function.table;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.PriorityQueue;
+import org.apache.geaflow.dsl.common.data.Row;
+import org.apache.geaflow.dsl.common.function.FunctionContext;
+import org.apache.geaflow.dsl.runtime.function.table.order.SortInfo;
+import org.apache.geaflow.dsl.runtime.function.table.order.TopNRowComparator;
+
+public class OrderByHeapSort implements OrderByFunction {
+
+    private final SortInfo sortInfo;
+
+    private PriorityQueue<Row> topNQueue;
+
+    private TopNRowComparator<Row> topNRowComparator;
+
+    public OrderByHeapSort(SortInfo sortInfo) {
+        this.sortInfo = sortInfo;
+    }
+
+    @Override
+    public void open(FunctionContext context) {
+        this.topNRowComparator = new TopNRowComparator<>(sortInfo);
+        this.topNQueue = new PriorityQueue<>(
+            sortInfo.fetch, topNRowComparator.getNegativeComparator());
+    }
+
+    @Override
+    public void process(Row row) {
+        if (topNQueue.size() == sortInfo.fetch) {
+            if (sortInfo.orderByFields.isEmpty()) {
+                return;
+            }
+            Row top = topNQueue.peek();
+            if (topNQueue.comparator().compare(top, row) < 0) {
+                topNQueue.remove();
+                topNQueue.add(row);
+            }
+        } else {
+            topNQueue.add(row);
+        }
+    }
+
+    @Override
+    public Iterable<Row> finish() {
+        List<Row> results = new ArrayList<>();
+        while (!topNQueue.isEmpty()) {
+            results.add(topNQueue.remove());
+        }
+        Collections.reverse(results);
+        topNQueue.clear();
+        return results;
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/table/OrderByRadixSort.java

@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.geaflow.dsl.runtime.function.table;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.geaflow.dsl.common.data.Row;
+import org.apache.geaflow.dsl.common.function.FunctionContext;
+import org.apache.geaflow.dsl.runtime.function.table.order.MultiFieldRadixSort;
+import org.apache.geaflow.dsl.runtime.function.table.order.SortInfo;
+
+public class OrderByRadixSort implements OrderByFunction {
+
+    private final SortInfo sortInfo;
+
+    private List<Row> allRows;
+
+    public OrderByRadixSort(SortInfo sortInfo) {
+        this.sortInfo = sortInfo;
+    }
+
+    @Override
+    public void open(FunctionContext context) {
+        this.allRows = new ArrayList<>();
+    }
+
+    @Override
+    public void process(Row row) {
+        if (sortInfo.fetch == 0) {
+            return;
+        }
+        allRows.add(row);
+    }
+
+    @Override
+    public Iterable<Row> finish() {
+        List<Row> sortedRows = new ArrayList<>(allRows);
+        MultiFieldRadixSort.multiFieldRadixSort(sortedRows, sortInfo);
+        allRows.clear();
+        return sortedRows;
+    }
+}

geaflow/geaflow-dsl/geaflow-dsl-runtime/src/main/java/org/apache/geaflow/dsl/runtime/function/table/OrderByTimSort.java

@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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 org.apache.geaflow.dsl.runtime.function.table;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.geaflow.dsl.common.data.Row;
+import org.apache.geaflow.dsl.common.function.FunctionContext;
+import org.apache.geaflow.dsl.runtime.function.table.order.SortInfo;
+import org.apache.geaflow.dsl.runtime.function.table.order.TopNRowComparator;
+
+public class OrderByTimSort implements OrderByFunction {
+
+    private final SortInfo sortInfo;
+
+    private List<Row> allRows;
+
+    private TopNRowComparator<Row> topNRowComparator;
+
+    public OrderByTimSort(SortInfo sortInfo) {
+        this.sortInfo = sortInfo;
+    }
+
+    @Override
+    public void open(FunctionContext context) {
+        this.topNRowComparator = new TopNRowComparator<>(sortInfo);
+        this.allRows = new ArrayList<>();
+    }
+
+    @Override
+    public void process(Row row) {
+        if (sortInfo.fetch == 0) {
+            return;
+        }
+        allRows.add(row);
+    }
+
+    @Override
+    public Iterable<Row> finish() {
+        List<Row> sortedRows = new ArrayList<>(allRows);
+        sortedRows.sort(topNRowComparator);
+        allRows.clear();
+        return sortedRows;
+    }
+}