add microbenchmark for exchange and window function (#5137)

close #4276, close #5138

dbms/CMakeLists.txt

@@ -316,6 +316,7 @@ if (ENABLE_TESTS)
         ${TiFlash_SOURCE_DIR}/dbms/src/AggregateFunctions/AggregateFunctionSum.cpp
         )
     target_include_directories(bench_dbms BEFORE PRIVATE ${SPARCEHASH_INCLUDE_DIR} ${benchmark_SOURCE_DIR}/include)
+    target_compile_definitions(bench_dbms PUBLIC DBMS_PUBLIC_GTEST)
 
     target_link_libraries(bench_dbms gtest dbms test_util_bench_main benchmark clickhouse_functions)
     if (ENABLE_TIFLASH_DTWORKLOAD)

dbms/src/Debug/astToExecutor.h

@@ -350,4 +350,4 @@ ExecutorPtr compileWindow(ExecutorPtr input, size_t & executor_index, ASTPtr fun
 ExecutorPtr compileSort(ExecutorPtr input, size_t & executor_index, ASTPtr order_by_expr_list, bool is_partial_sort);
 
 void literalFieldToTiPBExpr(const ColumnInfo & ci, const Field & field, tipb::Expr * expr, Int32 collator_id);
-} // namespace DB
+} // namespace DB

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.h

@@ -153,7 +153,9 @@ class DAGExpressionAnalyzer : private boost::noncopyable
         const tipb::Window & window,
         size_t window_columns_start_index);
 
+#ifndef DBMS_PUBLIC_GTEST
 private:
+#endif
     NamesAndTypes buildOrderColumns(
         const ExpressionActionsPtr & actions,
         const ::google::protobuf::RepeatedPtrField<tipb::ByItem> & order_by);

dbms/src/Flash/Coprocessor/DAGQueryBlockInterpreter.h

@@ -54,7 +54,9 @@ class DAGQueryBlockInterpreter
 
     BlockInputStreams execute();
 
+#ifndef DBMS_PUBLIC_GTEST
 private:
+#endif
     void executeImpl(DAGPipeline & pipeline);
     void handleMockTableScan(const TiDBTableScan & table_scan, DAGPipeline & pipeline);
     void handleTableScan(const TiDBTableScan & table_scan, DAGPipeline & pipeline);

dbms/src/Flash/tests/CMakeLists.txt

@@ -13,14 +13,3 @@
 # limitations under the License.
 
 include_directories (${CMAKE_CURRENT_BINARY_DIR})
-
-add_executable (exchange_perftest
-    exchange_perftest.cpp
-    ${TiFlash_SOURCE_DIR}/dbms/src/Server/StorageConfigParser.cpp
-    ${TiFlash_SOURCE_DIR}/dbms/src/Functions/FunctionsConversion.cpp)
-target_link_libraries (exchange_perftest
-    gtest_main
-    dbms
-    clickhouse_functions
-    clickhouse_aggregate_functions
-    tiflash-dttool-lib)

dbms/src/Flash/tests/WindowTestUtil.h

@@ -0,0 +1,81 @@
+// Copyright 2022 PingCAP, Ltd.
+//
+// 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.
+
+#pragma once
+
+#include <Flash/Coprocessor/DAGQueryBlockInterpreter.h>
+
+namespace DB
+{
+namespace tests
+{
+
+inline std::shared_ptr<DB::DAGQueryBlockInterpreter> mockInterpreter(Context & context, const std::vector<DB::NameAndTypePair> & source_columns, int concurrency)
+{
+    std::vector<BlockInputStreams> mock_input_streams_vec = {};
+    DAGQueryBlock mock_query_block(0, static_cast<const google::protobuf::RepeatedPtrField<tipb::Executor>>(nullptr));
+    std::vector<SubqueriesForSets> mock_subqueries_for_sets = {};
+    std::shared_ptr<DAGQueryBlockInterpreter> mock_interpreter = std::make_shared<DAGQueryBlockInterpreter>(context,
+                                                                                                            mock_input_streams_vec,
+                                                                                                            mock_query_block,
+                                                                                                            concurrency);
+    mock_interpreter->analyzer = std::make_unique<DAGExpressionAnalyzer>(std::move(source_columns), context);
+    return mock_interpreter;
+}
+
+inline void mockExecuteProject(std::shared_ptr<DAGQueryBlockInterpreter> & mock_interpreter, DAGPipeline & pipeline, NamesWithAliases & final_project)
+{
+    mock_interpreter->executeProject(pipeline, final_project);
+}
+
+inline void mockExecuteWindowOrder(std::shared_ptr<DAGQueryBlockInterpreter> & mock_interpreter, DAGPipeline & pipeline, const tipb::Sort & sort)
+{
+    mock_interpreter->handleWindowOrder(pipeline, sort);
+    mock_interpreter->input_streams_vec[0] = pipeline.streams;
+    NamesWithAliases final_project;
+    for (const auto & column : (*mock_interpreter->analyzer).source_columns)
+    {
+        final_project.push_back({column.name, ""});
+    }
+    mockExecuteProject(mock_interpreter, pipeline, final_project);
+}
+
+inline void mockExecuteWindowOrder(std::shared_ptr<DAGQueryBlockInterpreter> & mock_interpreter, DAGPipeline & pipeline, const String & sort_json)
+{
+    tipb::Sort sort;
+    ::google::protobuf::util::JsonStringToMessage(sort_json, &sort);
+    mockExecuteWindowOrder(mock_interpreter, pipeline, sort);
+}
+
+inline void mockExecuteWindow(std::shared_ptr<DAGQueryBlockInterpreter> & mock_interpreter, DAGPipeline & pipeline, const tipb::Window & window)
+{
+    mock_interpreter->handleWindow(pipeline, window);
+    mock_interpreter->input_streams_vec[0] = pipeline.streams;
+    NamesWithAliases final_project;
+    for (const auto & column : (*mock_interpreter->analyzer).source_columns)
+    {
+        final_project.push_back({column.name, ""});
+    }
+    mockExecuteProject(mock_interpreter, pipeline, final_project);
+}
+
+inline void mockExecuteWindow(std::shared_ptr<DAGQueryBlockInterpreter> & mock_interpreter, DAGPipeline & pipeline, std::string window_json_str)
+{
+    tipb::Window window;
+    google::protobuf::util::JsonStringToMessage(window_json_str, &window);
+    mockExecuteWindow(mock_interpreter, pipeline, window);
+}
+
+} // namespace tests
+} // namespace DB