[fix][dingo-exec] SortOperator spillToDisk

Author: guojn1

dingo-calcite/src/main/java/io/dingodb/calcite/visitor/function/DingoSortVisitFun.java

@@ -17,10 +17,12 @@
 package io.dingodb.calcite.visitor.function;
 
 import io.dingodb.calcite.rel.dingo.DingoSort;
+import io.dingodb.calcite.type.converter.DefinitionMapper;
 import io.dingodb.calcite.visitor.DingoJobVisitor;
 import io.dingodb.common.ExecutionContext;
 import io.dingodb.common.Location;
 import io.dingodb.common.table.HybridSearchTable;
+import io.dingodb.common.type.DingoType;
 import io.dingodb.exec.base.IdGenerator;
 import io.dingodb.exec.base.Job;
 import io.dingodb.exec.dag.Vertex;
@@ -62,12 +64,13 @@ static class OperatorSupplier implements Supplier<Vertex> {
 
         @Override
         public Vertex get() {
+            DingoType schema = DefinitionMapper.mapToDingoType(rel.getInput().getRowType());
             SortParam param = new SortParam(
                 toSortCollation(rel.getCollation().getFieldCollations()),
                 rel.fetch == null ? -1 : RexLiteral.intValue(rel.fetch),
                 rel.offset == null ? 0 : RexLiteral.intValue(rel.offset),
                 rel.getHints().stream().anyMatch( e -> e.hintName.equalsIgnoreCase(HybridSearchTable.HINT_NAME)),
-                executionContext);
+                executionContext, schema, 0);
             return new Vertex(SORT, param);
         }
     }

dingo-common/src/main/java/io/dingodb/common/config/DingoConfiguration.java

@@ -54,6 +54,7 @@ public class DingoConfiguration {
     private SecurityConfiguration security = new SecurityConfiguration();
     private VariableConfiguration variable = new VariableConfiguration();
     private CommonConfiguration common = new CommonConfiguration();
+    private SpillConfiguration spill = new SpillConfiguration();
 
     public static synchronized void parse(final String configPath) throws Exception {
         if (configPath != null) {
@@ -63,6 +64,7 @@ public static synchronized void parse(final String configPath) throws Exception
         INSTANCE.security = INSTANCE.getConfig("security", SecurityConfiguration.class);
         INSTANCE.variable = INSTANCE.getConfig("variable", VariableConfiguration.class);
         INSTANCE.common = INSTANCE.getConfig("common", CommonConfiguration.class);
+        INSTANCE.spill = INSTANCE.getConfig("spill", SpillConfiguration.class);
     }
 
     private static void copyConfig(Map<String, Object> from, Map<String, Object> to) {
@@ -117,6 +119,15 @@ public static Integer lowerCaseTableNames() {
         return Optional.mapOrGet(INSTANCE.variable, VariableConfiguration::getLowerCaseTableNames, () -> 2);
     }
 
+    public static String spillDir() {
+        return Optional.mapOrGet(INSTANCE.spill, SpillConfiguration::getDir,
+            () -> System.getProperty("java.io.tmpdir") + java.io.File.separator + "dingo-spill");
+    }
+
+    public static int spillThreshold() {
+        return Optional.mapOrGet(INSTANCE.spill, SpillConfiguration::getThreshold, () -> 100000);
+    }
+
     public static CommonId serverId() {
         return INSTANCE.serverId;
     }

dingo-common/src/main/java/io/dingodb/common/config/SpillConfiguration.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright 2021 DataCanvas
+ *
+ * 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 io.dingodb.common.config;
+
+import com.fasterxml.jackson.annotation.JsonAutoDetect;
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+
+import java.io.File;
+
+@Getter
+@Setter
+@ToString
+@JsonAutoDetect(fieldVisibility = JsonAutoDetect.Visibility.ANY)
+public class SpillConfiguration {
+    private String dir = System.getProperty("java.io.tmpdir") + File.separator + "dingo-spill";
+
+    private Integer threshold = 100000;
+}

dingo-exec/src/main/java/io/dingodb/exec/codec/AvroTupleCodec.java

@@ -107,4 +107,22 @@ record = decodeBytes(is, record, reader);
         }
         return tuples;
     }
+
+    public @Nullable Object[] decodeOne(@NonNull BinaryDecoder decoder) throws IOException {
+        GenericRecord record;
+        try {
+            record = reader.read(null, decoder);
+        } catch (EOFException e) {
+            return null;
+        }
+        if (record == null) {
+            return null;
+        }
+        int size = schema.getFields().size();
+        Object[] tuple = new Object[size];
+        for (int i = 0; i < size; ++i) {
+            tuple[i] = record.get(i);
+        }
+        return (Object[]) type.convertFrom(tuple, AvroDataConverter.INSTANCE);
+    }
 }

dingo-exec/src/main/java/io/dingodb/exec/dag/Edge.java

@@ -75,7 +75,6 @@ public boolean transformToNext(Context context, Object[] tuple) {
                     try {
                         blocked.get();
                         //checkExecutorFinishedRevoking((MemoryRevoker) operator, blocked, next.getData());
-                        LogUtils.info(log, "memory revoke locked continue..");
                         break;
                     } catch (InterruptedException | ExecutionException e) {
                         throw new RuntimeException(e);
@@ -116,7 +115,7 @@ public ListenableFuture<?> handleMemoryRevoke(AbstractParams param, Operator ope
             ListenableFuture<?> future = memoryRevoker.startMemoryRevoke(param);
             if (future != null) {
                 RevokerParams revokerParams = (RevokerParams) param;
-                LogUtils.info(log, "async memory revoke, poolName:{}, pre pin:{}, param cnt:{}",
+                LogUtils.debug(log, "async memory revoke, poolName:{}, pre pin:{}, param cnt:{}",
                     memoryAllocatorCtx.getName(), previous.getPin(), revokerParams.getCacheSize());
             }
             return future;

dingo-exec/src/main/java/io/dingodb/exec/memory/OperatorMemoryAllocatorCtx.java

@@ -141,7 +141,7 @@ public synchronized void resetMemoryRevokingRequested() {
             return;
         }
         memoryRevokingRequestedFuture = SettableFuture.create();
-        LogUtils.info(log, "resetMemoryRevokingRequested");
+        LogUtils.debug(log, "resetMemoryRevokingRequested");
     }
 
     @Override
@@ -231,7 +231,7 @@ public synchronized long requestMemoryRevokingOrReturnRevokingBytes(long revokeF
         checkState(revocable, "requestMemoryRevoking for unRevocable operator");
         boolean alreadyRequested = isMemoryRevokingRequested();
         if (!alreadyRequested && revocableAllocated.get() > 0) {
-            LogUtils.info(log, "request memory revoking,name:{}, revocableAllocated:{}, revokeFlag:{}",
+            LogUtils.debug(log, "request memory revoking,name:{}, revocableAllocated:{}, revokeFlag:{}",
                 this.getName(), revocableAllocated.get(), revokeFlag);
             memoryRevokingRequestedFuture.set(null);
             return revocableAllocated.get();

dingo-exec/src/main/java/io/dingodb/exec/memory/RocksdbSpiller.java

@@ -0,0 +1,36 @@
+/*
+ * Copyright 2021 DataCanvas
+ *
+ * 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 io.dingodb.exec.memory;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+public class RocksdbSpiller implements Spiller {
+
+    @Override
+    public void spill(ConcurrentHashMap hashMap) {
+        // manager rocksdb instance
+        // loop map
+        // encode key:  jobId + opId + unique_id + TupleKey + inc
+        // encode val:  TupleWithJoinFlag
+        // write to rocksdb
+    }
+
+    @Override
+    public void close() {
+
+    }
+}

dingo-exec/src/main/java/io/dingodb/exec/memory/Spiller.java

@@ -0,0 +1,25 @@
+/*
+ * Copyright 2021 DataCanvas
+ *
+ * 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 io.dingodb.exec.memory;
+
+import java.util.concurrent.ConcurrentHashMap;
+
+public interface Spiller {
+    void spill(ConcurrentHashMap hashMap);
+
+    void close();
+}