Add coprocessor support in TiFlash

.gitmodules

@@ -37,6 +37,10 @@
 [submodule "contrib/kvproto"]
 	path = contrib/kvproto
 	url = https://github.com/pingcap/kvproto.git
+[submodule "contrib/tipb"]
+	path = contrib/tipb
+	url = https://github.com/pingcap/tipb.git
+	branch = master
 [submodule "contrib/client-c"]
 	path = contrib/client-c
 	url = [email protected]:tikv/client-c.git

CMakeLists.txt

@@ -272,6 +272,7 @@ include (cmake/find_capnp.cmake)
 include (cmake/find_llvm.cmake)
 include (cmake/find_grpc.cmake)
 include (cmake/find_kvproto.cmake)
+include (cmake/find_tipb.cmake)
 
 
 include (cmake/find_contrib_lib.cmake)

cmake/find_tipb.cmake

@@ -0,0 +1,10 @@
+
+if (NOT EXISTS "${ClickHouse_SOURCE_DIR}/contrib/tipb/cpp/tipb/select.pb.h")
+	if (EXISTS "${ClickHouse_SOURCE_DIR}/contrib/tipb/proto/select.proto")
+		message (FATAL_ERROR "tipb cpp files in contrib/tipb is missing. Try go to contrib/tipb, and run ./generate_cpp.sh")
+	else()
+		message (FATAL_ERROR "tipb submodule in contrib/tipb is missing. Try run 'git submodule update --init --recursive', and go to contrib/tipb, and run ./generate_cpp.sh")
+	endif()
+endif ()
+
+message(STATUS "Using tipb: ${ClickHouse_SOURCE_DIR}/contrib/tipb/cpp")

contrib/CMakeLists.txt

@@ -1,5 +1,6 @@
 add_subdirectory (kvproto/cpp)
 add_subdirectory (client-c)
+add_subdirectory (tipb/cpp)
 
 if (NOT MSVC)
     set (CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -Wno-old-style-cast")

dbms/CMakeLists.txt

@@ -60,6 +60,7 @@ add_headers_and_sources(dbms src/Storages/Page)
 add_headers_and_sources(dbms src/Raft)
 add_headers_and_sources(dbms src/TiDB)
 add_headers_and_sources(dbms src/Client)
+add_headers_only(dbms src/Flash/Coprocessor)
 add_headers_only(dbms src/Server)
 
 list (APPEND clickhouse_common_io_sources ${CONFIG_BUILD})
@@ -149,8 +150,10 @@ target_link_libraries (dbms
     clickhouse_parsers
     clickhouse_common_config
     clickhouse_common_io
+    flash_service
     kvproto
     kv_client
+    tipb
     ${Protobuf_LIBRARIES}
     gRPC::grpc++_unsecure
     ${CURL_LIBRARIES}

dbms/src/CMakeLists.txt

@@ -13,4 +13,5 @@ add_subdirectory (AggregateFunctions)
 add_subdirectory (Server)
 add_subdirectory (Client)
 add_subdirectory (TableFunctions)
-add_subdirectory (Analyzers)
+add_subdirectory (Analyzers)
+add_subdirectory (Flash)

dbms/src/Common/ErrorCodes.cpp

@@ -395,6 +395,7 @@ namespace ErrorCodes
     extern const int SCHEMA_SYNC_ERROR = 10003;
     extern const int SCHEMA_VERSION_ERROR = 10004;
     extern const int DDL_ERROR = 10005;
+    extern const int COP_BAD_DAG_REQUEST = 10006;
 }
 
 }

dbms/src/Common/MyTime.cpp

@@ -473,4 +473,15 @@ void convertTimeZone(UInt64 from_time, UInt64 & to_time, const DateLUTImpl & tim
     to_time = to_my_time.toPackedUInt();
 }
 
+void convertTimeZoneByOffset(UInt64 from_time, UInt64 & to_time, Int64 offset, const DateLUTImpl & time_zone)
+{
+    MyDateTime from_my_time(from_time);
+    time_t epoch = time_zone.makeDateTime(
+        from_my_time.year, from_my_time.month, from_my_time.day, from_my_time.hour, from_my_time.minute, from_my_time.second);
+    epoch += offset;
+    MyDateTime to_my_time(time_zone.toYear(epoch), time_zone.toMonth(epoch), time_zone.toDayOfMonth(epoch),
+                          time_zone.toHour(epoch), time_zone.toMinute(epoch), time_zone.toSecond(epoch), from_my_time.micro_second);
+    to_time = to_my_time.toPackedUInt();
+}
+
 } // namespace DB

dbms/src/Common/MyTime.h

@@ -65,4 +65,6 @@ Field parseMyDateTime(const String & str);
 
 void convertTimeZone(UInt64 from_time, UInt64 & to_time, const DateLUTImpl & time_zone_from, const DateLUTImpl & time_zone_to);
 
+void convertTimeZoneByOffset(UInt64 from_time, UInt64 & to_time, Int64 offset, const DateLUTImpl & time_zone);
+
 } // namespace DB

dbms/src/Core/Defines.h

@@ -29,6 +29,8 @@
 #define DEFAULT_MAX_READ_TSO 0xFFFFFFFFFFFFFFFF
 #define DEFAULT_UNSPECIFIED_SCHEMA_VERSION -1
 
+#define DEFAULT_DAG_RECORDS_PER_CHUNK 64L
+
 /** Which blocks by default read the data (by number of rows).
   * Smaller values give better cache locality, less consumption of RAM, but more overhead to process the query.
   */

dbms/src/DataStreams/BlockStreamProfileInfo.h

@@ -29,6 +29,11 @@ struct BlockStreamProfileInfo
     size_t rows = 0;
     size_t blocks = 0;
     size_t bytes = 0;
+    // execution time is the total time spent on current stream and all its children streams
+    // note that it is different from total_stopwatch.elapsed(), which includes not only the
+    // time spent on current stream and all its children streams, but also the time of its
+    // parent streams
+    UInt64 execution_time = 0;
 
     using BlockStreamProfileInfos = std::vector<const BlockStreamProfileInfo *>;
 
@@ -45,6 +50,8 @@ struct BlockStreamProfileInfo
 
     void update(Block & block);
 
+    void updateExecutionTime(UInt64 time) { execution_time += time; }
+
     /// Binary serialization and deserialization of main fields.
     /// Writes only main fields i.e. fields that required by internal transmission protocol.
     void read(ReadBuffer & in);

dbms/src/DataStreams/IProfilingBlockInputStream.cpp

@@ -42,6 +42,8 @@ Block IProfilingBlockInputStream::read()
     if (isCancelledOrThrowIfKilled())
         return res;
 
+    auto start_time = info.total_stopwatch.elapsed();
+
     if (!checkTimeLimit())
         limit_exceeded_need_break = true;
 
@@ -83,6 +85,7 @@ Block IProfilingBlockInputStream::read()
     }
 #endif
 
+    info.updateExecutionTime(info.total_stopwatch.elapsed() - start_time);
     return res;
 }
 

dbms/src/DataTypes/DataTypeMyDateTime.cpp

@@ -20,7 +20,7 @@ DataTypeMyDateTime::DataTypeMyDateTime(int fraction_)
 {
     fraction = fraction_;
     if (fraction < 0 || fraction > 6)
-        throw Exception("fraction must >= 0 and < 6", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
+        throw Exception("fraction must >= 0 and <= 6", ErrorCodes::ILLEGAL_TYPE_OF_ARGUMENT);
 }
 
 void DataTypeMyDateTime::serializeText(const IColumn & column, size_t row_num, WriteBuffer & ostr) const

dbms/src/Debug/DBGInvoker.cpp

@@ -4,6 +4,7 @@
 #include <DataStreams/StringStreamBlockInputStream.h>
 #include <Debug/ClusterManage.h>
 #include <Debug/DBGInvoker.h>
+#include <Debug/dbgFuncCoprocessor.h>
 #include <Debug/dbgFuncMockTiDBData.h>
 #include <Debug/dbgFuncMockTiDBTable.h>
 #include <Debug/dbgFuncRegion.h>
@@ -30,48 +31,50 @@ void dbgFuncSleep(Context &, const ASTs & args, DBGInvoker::Printer output)
 
 DBGInvoker::DBGInvoker()
 {
-    regFunc("echo", dbgFuncEcho);
+    regSchemalessFunc("echo", dbgFuncEcho);
     // TODO: remove this, use sleep in bash script
-    regFunc("sleep", dbgFuncSleep);
-
-    regFunc("mock_tidb_table", MockTiDBTable::dbgFuncMockTiDBTable);
-    regFunc("mock_tidb_db", MockTiDBTable::dbgFuncMockTiDBDB);
-    regFunc("mock_tidb_partition", MockTiDBTable::dbgFuncMockTiDBPartition);
-    regFunc("drop_tidb_partition", MockTiDBTable::dbgFuncDropTiDBPartition);
-    regFunc("drop_tidb_table", MockTiDBTable::dbgFuncDropTiDBTable);
-    regFunc("drop_tidb_db", MockTiDBTable::dbgFuncDropTiDBDB);
-    regFunc("add_column_to_tidb_table", MockTiDBTable::dbgFuncAddColumnToTiDBTable);
-    regFunc("drop_column_from_tidb_table", MockTiDBTable::dbgFuncDropColumnFromTiDBTable);
-    regFunc("modify_column_in_tidb_table", MockTiDBTable::dbgFuncModifyColumnInTiDBTable);
-    regFunc("rename_column_in_tidb_table", MockTiDBTable::dbgFuncRenameColumnInTiDBTable);
-    regFunc("rename_tidb_table", MockTiDBTable::dbgFuncRenameTiDBTable);
-    regFunc("truncate_tidb_table", MockTiDBTable::dbgFuncTruncateTiDBTable);
-
-    regFunc("set_flush_threshold", dbgFuncSetFlushThreshold);
-
-    regFunc("raft_insert_row", dbgFuncRaftInsertRow);
-    regFunc("raft_insert_row_full", dbgFuncRaftInsertRowFull);
-    regFunc("raft_insert_rows", dbgFuncRaftInsertRows);
-    regFunc("raft_update_rows", dbgFuncRaftUpdateRows);
-    regFunc("raft_delete_rows", dbgFuncRaftDelRows);
-    regFunc("raft_delete_row", dbgFuncRaftDeleteRow);
-
-    regFunc("put_region", dbgFuncPutRegion);
-    regFunc("region_snapshot", dbgFuncRegionSnapshot);
-    regFunc("region_snapshot_data", dbgFuncRegionSnapshotWithData);
-
-    regFunc("try_flush", dbgFuncTryFlush);
-    regFunc("try_flush_region", dbgFuncTryFlushRegion);
-
-    regFunc("dump_all_region", dbgFuncDumpAllRegion);
-    regFunc("dump_all_mock_region", dbgFuncDumpAllMockRegion);
-
-    regFunc("enable_schema_sync_service", dbgFuncEnableSchemaSyncService);
-    regFunc("refresh_schemas", dbgFuncRefreshSchemas);
-    regFunc("reset_schemas", dbgFuncResetSchemas);
-
-    regFunc("dump_region_table", ClusterManage::dumpRegionTable);
-    regFunc("find_region_by_range", ClusterManage::findRegionByRange);
+    regSchemalessFunc("sleep", dbgFuncSleep);
+
+    regSchemalessFunc("mock_tidb_table", MockTiDBTable::dbgFuncMockTiDBTable);
+    regSchemalessFunc("mock_tidb_db", MockTiDBTable::dbgFuncMockTiDBDB);
+    regSchemalessFunc("mock_tidb_partition", MockTiDBTable::dbgFuncMockTiDBPartition);
+    regSchemalessFunc("drop_tidb_table", MockTiDBTable::dbgFuncDropTiDBTable);
+    regSchemalessFunc("drop_tidb_db", MockTiDBTable::dbgFuncDropTiDBDB);
+    regSchemalessFunc("drop_tidb_partition", MockTiDBTable::dbgFuncDropTiDBPartition);
+    regSchemalessFunc("add_column_to_tidb_table", MockTiDBTable::dbgFuncAddColumnToTiDBTable);
+    regSchemalessFunc("drop_column_from_tidb_table", MockTiDBTable::dbgFuncDropColumnFromTiDBTable);
+    regSchemalessFunc("modify_column_in_tidb_table", MockTiDBTable::dbgFuncModifyColumnInTiDBTable);
+    regSchemalessFunc("rename_column_in_tidb_table", MockTiDBTable::dbgFuncRenameColumnInTiDBTable);
+    regSchemalessFunc("rename_tidb_table", MockTiDBTable::dbgFuncRenameTiDBTable);
+    regSchemalessFunc("truncate_tidb_table", MockTiDBTable::dbgFuncTruncateTiDBTable);
+
+    regSchemalessFunc("set_flush_threshold", dbgFuncSetFlushThreshold);
+
+    regSchemalessFunc("raft_insert_row", dbgFuncRaftInsertRow);
+    regSchemalessFunc("raft_insert_row_full", dbgFuncRaftInsertRowFull);
+    regSchemalessFunc("raft_insert_rows", dbgFuncRaftInsertRows);
+    regSchemalessFunc("raft_update_rows", dbgFuncRaftUpdateRows);
+    regSchemalessFunc("raft_delete_rows", dbgFuncRaftDelRows);
+    regSchemalessFunc("raft_delete_row", dbgFuncRaftDeleteRow);
+
+    regSchemalessFunc("put_region", dbgFuncPutRegion);
+    regSchemalessFunc("region_snapshot", dbgFuncRegionSnapshot);
+    regSchemalessFunc("region_snapshot_data", dbgFuncRegionSnapshotWithData);
+
+    regSchemalessFunc("try_flush", dbgFuncTryFlush);
+    regSchemalessFunc("try_flush_region", dbgFuncTryFlushRegion);
+
+    regSchemalessFunc("dump_all_region", dbgFuncDumpAllRegion);
+    regSchemalessFunc("dump_all_mock_region", dbgFuncDumpAllMockRegion);
+    regSchemalessFunc("dump_region_table", ClusterManage::dumpRegionTable);
+    regSchemalessFunc("find_region_by_range", ClusterManage::findRegionByRange);
+
+    regSchemalessFunc("enable_schema_sync_service", dbgFuncEnableSchemaSyncService);
+    regSchemalessFunc("refresh_schemas", dbgFuncRefreshSchemas);
+    regSchemalessFunc("reset_schemas", dbgFuncResetSchemas);
+
+    regSchemafulFunc("dag", dbgFuncDAG);
+    regSchemafulFunc("mock_dag", dbgFuncMockDAG);
 }
 
 void replaceSubstr(std::string & str, const std::string & target, const std::string & replacement)
@@ -103,10 +106,25 @@ BlockInputStreamPtr DBGInvoker::invoke(Context & context, const std::string & or
         name = ori_name.substr(prefix_not_print_res.size(), ori_name.size() - prefix_not_print_res.size());
     }
 
-    auto it = funcs.find(name);
-    if (it == funcs.end())
-        throw Exception("DBG function not found", ErrorCodes::BAD_ARGUMENTS);
+    BlockInputStreamPtr res;
+    auto it_schemaless = schemaless_funcs.find(name);
+    if (it_schemaless != schemaless_funcs.end())
+        res = invokeSchemaless(context, name, it_schemaless->second, args);
+    else
+    {
+        auto it_schemaful = schemaful_funcs.find(name);
+        if (it_schemaful != schemaful_funcs.end())
+            res = invokeSchemaful(context, name, it_schemaful->second, args);
+        else
+            throw Exception("DBG function not found", ErrorCodes::BAD_ARGUMENTS);
+    }
+
+    return print_res ? res : std::shared_ptr<StringStreamBlockInputStream>();
+}
 
+BlockInputStreamPtr DBGInvoker::invokeSchemaless(
+    Context & context, const std::string & name, const SchemalessDBGFunc & func, const ASTs & args)
+{
     std::stringstream col_name;
     col_name << name << "(";
     for (size_t i = 0; i < args.size(); ++i)
@@ -119,9 +137,14 @@ BlockInputStreamPtr DBGInvoker::invoke(Context & context, const std::string & or
     std::shared_ptr<StringStreamBlockInputStream> res = std::make_shared<StringStreamBlockInputStream>(col_name.str());
     Printer printer = [&](const std::string & s) { res->append(s); };
 
-    (it->second)(context, args, printer);
+    func(context, args, printer);
 
-    return print_res ? res : std::shared_ptr<StringStreamBlockInputStream>();
+    return res;
+}
+
+BlockInputStreamPtr DBGInvoker::invokeSchemaful(Context & context, const std::string &, const SchemafulDBGFunc & func, const ASTs & args)
+{
+    return func(context, args);
 }
 
 } // namespace DB

dbms/src/Debug/DBGInvoker.h

@@ -25,16 +25,21 @@ class DBGInvoker
 {
 public:
     using Printer = std::function<void(const std::string &)>;
-    using DBGFunc = std::function<void(Context & context, const ASTs & args, Printer printer)>;
+    using SchemalessDBGFunc = std::function<void(Context & context, const ASTs & args, Printer printer)>;
+    using SchemafulDBGFunc = std::function<BlockInputStreamPtr(Context & context, const ASTs & args)>;
 
     DBGInvoker();
 
-    void regFunc(const std::string & name, DBGFunc func) { funcs[name] = func; }
+    void regSchemalessFunc(const std::string & name, SchemalessDBGFunc func) { schemaless_funcs[name] = func; }
+    void regSchemafulFunc(const std::string & name, SchemafulDBGFunc func) { schemaful_funcs[name] = func; }
 
     BlockInputStreamPtr invoke(Context & context, const std::string & ori_name, const ASTs & args);
+    BlockInputStreamPtr invokeSchemaless(Context & context, const std::string & name, const SchemalessDBGFunc & func, const ASTs & args);
+    BlockInputStreamPtr invokeSchemaful(Context & context, const std::string & name, const SchemafulDBGFunc & func, const ASTs & args);
 
 private:
-    std::unordered_map<std::string, DBGFunc> funcs;
+    std::unordered_map<std::string, SchemalessDBGFunc> schemaless_funcs;
+    std::unordered_map<std::string, SchemafulDBGFunc> schemaful_funcs;
 };
 
 } // namespace DB