Add UT for list struct based executors

dbms/src/Flash/tests/gtest_interpreter.cpp

@@ -631,5 +631,50 @@ CreatingSets
 }
 CATCH
 
+TEST_F(InterpreterExecuteTest, ListBase)
+try
+{
+    {
+        auto request = context
+                           .scan("test_db", "test_table")
+                           .filter(eq(col("s1"), col("s2")))
+                           .aggregation(Max(col("s1")), col("s2"))
+                           .limit(10)
+                           .buildToListStruct(context);
+        String expected = R"(
+Limit, limit = 10
+ Expression: <final projection>
+  Aggregating
+   Concat
+    Expression: <before aggregation>
+     Filter: <execute where>
+      MockTableScan)";
+        ASSERT_BLOCKINPUTSTREAM_EQAUL(expected, request, 1);
+    }
+
+    {
+        auto request = context
+                           .scan("test_db", "test_table")
+                           .filter(eq(col("s1"), col("s2")))
+                           .aggregation(Max(col("s1")), col("s2"))
+                           .topN("s2", false, 10)
+                           .buildToListStruct(context);
+        String expected = R"(
+Union: <for test>
+ SharedQuery x 20: <restore concurrency>
+  Expression: <final projection>
+   MergeSorting, limit = 10
+    Union: <for partial order>
+     PartialSorting x 20: limit = 10
+      SharedQuery: <restore concurrency>
+       ParallelAggregating, max_threads: 20, final: true
+        Expression x 20: <before aggregation>
+         Filter: <execute where>
+          MockTableScan)";
+        ASSERT_BLOCKINPUTSTREAM_EQAUL(expected, request, 20);
+    }
+}
+CATCH
+
 } // namespace tests
 } // namespace DB

dbms/src/Flash/tests/gtest_planner_interpreter.cpp

@@ -984,5 +984,57 @@ CreatingSets
 }
 CATCH
 
+TEST_F(PlannerInterpreterExecuteTest, ListBase)
+try
+{
+    {
+        auto request = context
+                           .scan("test_db", "test_table")
+                           .filter(eq(col("s1"), col("s2")))
+                           .aggregation(Max(col("s1")), col("s2"))
+                           .filter(eq(col("s2"), lit(Field("1", 1))))
+                           .limit(10)
+                           .buildToListStruct(context);
+        String expected = R"(
+Expression: <final projection>
+ Limit, limit = 10
+  Filter
+   Expression: <expr after aggregation>
+    Aggregating
+     Concat
+      Expression: <before aggregation>
+       Filter
+        MockTableScan)";
+        ASSERT_BLOCKINPUTSTREAM_EQAUL(expected, request, 1);
+    }
+
+    {
+        auto request = context
+                           .scan("test_db", "test_table")
+                           .filter(eq(col("s1"), col("s2")))
+                           .aggregation(Max(col("s1")), col("s2"))
+                           .filter(eq(col("s2"), lit(Field("1", 1))))
+                           .topN("s2", false, 10)
+                           .buildToListStruct(context);
+        String expected = R"(
+Union: <for test>
+ Expression x 20: <final projection>
+  SharedQuery: <restore concurrency>
+   MergeSorting, limit = 10
+    Union: <for partial order>
+     PartialSorting x 20: limit = 10
+      Expression: <before TopN>
+       Filter
+        Expression: <expr after aggregation>
+         SharedQuery: <restore concurrency>
+          ParallelAggregating, max_threads: 20, final: true
+           Expression x 20: <before aggregation>
+            Filter
+             MockTableScan)";
+        ASSERT_BLOCKINPUTSTREAM_EQAUL(expected, request, 20);
+    }
+}
+CATCH
+
 } // namespace tests
 } // namespace DB

dbms/src/TestUtils/executorSerializer.cpp

@@ -57,29 +57,6 @@ void toString(const Columns & columns, FmtBuffer & buf)
     }
     buf.fmtAppend("<{}, {}>", bound, getColumnTypeName(columns.at(bound)));
 }
-} // namespace
-
-String ExecutorSerializer::serialize(const tipb::DAGRequest * dag_request)
-{
-    assert((dag_request->executors_size() > 0) != dag_request->has_root_executor());
-    if (dag_request->has_root_executor())
-    {
-        serialize(dag_request->root_executor(), 0);
-        return buf.toString();
-    }
-    else
-    {
-        FmtBuffer buffer;
-        String prefix;
-        traverseExecutors(dag_request, [this, &prefix](const tipb::Executor & executor) {
-            assert(executor.has_executor_id());
-            buf.fmtAppend("{}{}\n", prefix, executor.executor_id());
-            prefix.append(" ");
-            return true;
-        });
-        return buffer.toString();
-    }
-}
 
 void serializeTableScan(const String & executor_id, const tipb::TableScan & ts, FmtBuffer & buf)
 {
@@ -263,8 +240,55 @@ void serializeSort(const String & executor_id, const tipb::Sort & sort [[maybe_u
         ", ");
     buf.append("}\n");
 }
+} // namespace
+
+String ExecutorSerializer::serialize(const tipb::DAGRequest * dag_request)
+{
+    assert((dag_request->executors_size() > 0) != dag_request->has_root_executor());
+    if (dag_request->has_root_executor())
+    {
+        serializeTreeStruct(dag_request->root_executor(), 0);
+    }
+    else
+    {
+        serializeListStruct(dag_request);
+    }
+    return buf.toString();
+}
+
+void ExecutorSerializer::serializeListStruct(const tipb::DAGRequest * dag_request)
+{
+    String prefix;
+    traverseExecutors(dag_request, [this, &prefix](const tipb::Executor & executor) {
+        buf.append(prefix);
+        switch (executor.tp())
+        {
+        case tipb::ExecType::TypeTableScan:
+            serializeTableScan("TableScan", executor.tbl_scan(), buf);
+            break;
+        case tipb::ExecType::TypeSelection:
+            serializeSelection("Selection", executor.selection(), buf);
+            break;
+        case tipb::ExecType::TypeAggregation:
+        // stream agg is not supported, treated as normal agg
+        case tipb::ExecType::TypeStreamAgg:
+            serializeAggregation("Aggregation", executor.aggregation(), buf);
+            break;
+        case tipb::ExecType::TypeTopN:
+            serializeTopN("TopN", executor.topn(), buf);
+            break;
+        case tipb::ExecType::TypeLimit:
+            serializeLimit("Limit", executor.limit(), buf);
+            break;
+        default:
+            throw TiFlashException("Should not reach here", Errors::Coprocessor::Internal);
+        }
+        prefix.append(" ");
+        return true;
+    });
+}
 
-void ExecutorSerializer::serialize(const tipb::Executor & root_executor, size_t level)
+void ExecutorSerializer::serializeTreeStruct(const tipb::Executor & root_executor, size_t level)
 {
     auto append_str = [&level, this](const tipb::Executor & executor) {
         assert(executor.has_executor_id());
@@ -324,7 +348,7 @@ void ExecutorSerializer::serialize(const tipb::Executor & root_executor, size_t
         if (executor.has_join())
         {
             for (const auto & child : executor.join().children())
-                serialize(child, level);
+                serializeTreeStruct(child, level);
             return false;
         }
         return true;

dbms/src/TestUtils/executorSerializer.h

@@ -28,7 +28,8 @@ class ExecutorSerializer
     String serialize(const tipb::DAGRequest * dag_request);
 
 private:
-    void serialize(const tipb::Executor & root_executor, size_t level);
+    void serializeListStruct(const tipb::DAGRequest * dag_request);
+    void serializeTreeStruct(const tipb::Executor & root_executor, size_t level);
     void addPrefix(size_t level) { buf.append(String(level, ' ')); }
 
 private:

dbms/src/TestUtils/mockExecutor.cpp

@@ -15,6 +15,7 @@
 #include <Debug/MockComputeServerManager.h>
 #include <Debug/astToExecutor.h>
 #include <Debug/dbgFuncCoprocessor.h>
+#include <Flash/Statistics/traverseExecutors.h>
 #include <Interpreters/Context.h>
 #include <Parsers/ASTExpressionList.h>
 #include <Parsers/ASTFunction.h>
@@ -96,6 +97,20 @@ std::shared_ptr<tipb::DAGRequest> DAGRequestBuilder::build(MockDAGRequestContext
     return dag_request_ptr;
 }
 
+std::shared_ptr<tipb::DAGRequest> DAGRequestBuilder::buildToListStruct(MockDAGRequestContext & mock_context)
+{
+    auto dag_request_ptr = build(mock_context);
+    auto & mutable_executors = *dag_request_ptr->mutable_executors();
+    traverseExecutorsReverse(dag_request_ptr.get(), [&](const tipb::Executor & executor) -> bool {
+        auto * mutable_executor = mutable_executors.Add();
+        (*mutable_executor) = executor;
+        mutable_executor->clear_executor_id();
+        return true;
+    });
+    dag_request_ptr->release_root_executor();
+    return dag_request_ptr;
+}
+
 // Currently Sort and Window Executors don't support columnPrune.
 // TODO: support columnPrume for Sort and Window.
 void columnPrune(ExecutorPtr executor)

dbms/src/TestUtils/mockExecutor.h

@@ -71,6 +71,7 @@ class DAGRequestBuilder
     }
 
     std::shared_ptr<tipb::DAGRequest> build(MockDAGRequestContext & mock_context);
+    std::shared_ptr<tipb::DAGRequest> buildToListStruct(MockDAGRequestContext & mock_context);
     QueryTasks buildMPPTasks(MockDAGRequestContext & mock_context);
     QueryTasks buildMPPTasks(MockDAGRequestContext & mock_context, const DAGProperties & properties);
 

dbms/src/TestUtils/tests/gtest_mock_executors.cpp

@@ -340,5 +340,44 @@ try
     }
 }
 CATCH
+
+TEST_F(MockDAGRequestTest, ListBase)
+try
+{
+    {
+        auto request = context
+                           .scan("test_db", "test_table")
+                           .filter(eq(col("s1"), col("s2")))
+                           .aggregation(Max(col("s1")), col("s2"))
+                           .filter(eq(col("s2"), lit(Field("1", 1))))
+                           .limit(10)
+                           .buildToListStruct(context);
+        String expected = R"(
+Limit | 10
+ Selection | equals(<1, String>, <-5692549928996306944, String>)}
+  Aggregation | group_by: {<1, String>}, agg_func: {max(<0, String>)}
+   Selection | equals(<0, String>, <1, String>)}
+    TableScan | {<0, String>, <1, String>})";
+        ASSERT_DAGREQUEST_EQAUL(expected, request);
+    }
+
+    {
+        auto request = context
+                           .scan("test_db", "test_table")
+                           .filter(eq(col("s1"), col("s2")))
+                           .aggregation(Max(col("s1")), col("s2"))
+                           .filter(eq(col("s2"), lit(Field("1", 1))))
+                           .topN("s2", false, 10)
+                           .buildToListStruct(context);
+        String expected = R"(
+TopN | order_by: {(<1, String>, desc: false)}, limit: 10
+ Selection | equals(<1, String>, <-5692549928996306944, String>)}
+  Aggregation | group_by: {<1, String>}, agg_func: {max(<0, String>)}
+   Selection | equals(<0, String>, <1, String>)}
+    TableScan | {<0, String>, <1, String>})";
+        ASSERT_DAGREQUEST_EQAUL(expected, request);
+    }
+}
+CATCH
 } // namespace tests
 } // namespace DB