*: backport FTS/TiCI to release-8.5

.gitmodules

@@ -155,3 +155,6 @@
 [submodule "contrib/highfive"]
 	path = contrib/highfive
 	url = https://github.com/BlueBrain/HighFive
+[submodule "contrib/tici"]
+	path = contrib/tici
+	url = git@github.com:pingcap-inc/tici.git

CMakeLists.txt

@@ -415,6 +415,6 @@ include (cmake/print_include_directories.cmake)
 # Another way to do so is to use JOB_POOL with single concurrency. However, it is Ninja specific and it seems to have bugs
 # with a large number of threads.
 include (cmake/sequential.cmake)
-build_sequentially (process_metrics symbolization tiflash_proxy)
+build_sequentially (process_metrics rustbuild symbolization tiflash_proxy)
 
 print_flags ()

contrib/CMakeLists.txt

@@ -207,3 +207,5 @@ if (ENABLE_TESTS AND NOT CMAKE_BUILD_TYPE_UC STREQUAL "DEBUG")
     add_subdirectory(hdf5-cmake)
     add_subdirectory(highfive-cmake)
 endif ()
+
+add_subdirectory(tici-search-lib)

contrib/tici-search-lib/CMakeLists.txt

@@ -0,0 +1,39 @@
+set(TICI_PROJECT_DIR "${CMAKE_SOURCE_DIR}/contrib/tici")
+set(TICI_LIB_NAME "tici_search_lib")
+set(TICI_LIB "${CMAKE_CURRENT_BINARY_DIR}/release/${CMAKE_STATIC_LIBRARY_PREFIX}${TICI_LIB_NAME}${CMAKE_STATIC_LIBRARY_SUFFIX}")
+
+file(GLOB LIB_SOURCE_FILES "${TICI_PROJECT_DIR}/src/*")
+file(MAKE_DIRECTORY ${CMAKE_CURRENT_BINARY_DIR}/cxxbridge)
+
+add_custom_command(
+    OUTPUT ${TICI_LIB}
+    COMMAND cargo build -p tici-search-lib --release --target-dir ${CMAKE_CURRENT_BINARY_DIR} --manifest-path ${TICI_PROJECT_DIR}/Cargo.toml
+    WORKING_DIRECTORY ${TICI_PROJECT_DIR}
+    DEPENDS ${LIB_SOURCE_FILES}
+    COMMENT "Build Rust lib"
+)
+
+add_custom_target(rustbuild ALL DEPENDS ${TICI_LIB})
+add_library(tici_search_lib_static STATIC IMPORTED GLOBAL)
+set_target_properties(tici_search_lib_static PROPERTIES
+    IMPORTED_LOCATION ${TICI_LIB}
+)
+
+add_dependencies(tici_search_lib_static rustbuild)
+target_include_directories(tici_search_lib_static  INTERFACE
+    ${CMAKE_CURRENT_BINARY_DIR}/cxxbridge)
+
+add_library(tici_search_lib SHARED "${TiFlash_SOURCE_DIR}/libs/libclara-cmake/dummy.cpp")
+target_compile_options(tici_search_lib PRIVATE -pthread)
+target_link_options(tici_search_lib PRIVATE -pthread)
+target_link_libraries(tici_search_lib PRIVATE "$<LINK_LIBRARY:WHOLE_ARCHIVE,tici_search_lib_static>")
+if(APPLE)
+  target_link_libraries(tici_search_lib PRIVATE
+      "-framework Security"
+      "-framework CoreFoundation"
+      "-framework IOKit"
+      )
+endif()
+
+target_include_directories(tici_search_lib  INTERFACE
+    ${CMAKE_CURRENT_BINARY_DIR}/cxxbridge)

dbms/CMakeLists.txt

@@ -224,6 +224,7 @@ target_link_libraries (dbms
     absl::synchronization
     tiflash_contrib::aws_s3
     tiflash_vector_search
+    tici_search_lib
 
     etcdpb
     tiflash_parsers
@@ -373,6 +374,10 @@ if (ENABLE_TESTS)
                 DESTINATION ".")
     endif ()
 
+    install (TARGETS tici_search_lib
+            COMPONENT tiflash-gtest
+            DESTINATION ".")
+
     target_compile_options(gtests_dbms PRIVATE -Wno-unknown-pragmas -Wno-deprecated-copy)
     add_check(gtests_dbms)
 

dbms/src/Common/TiFlashMetrics.h

@@ -98,7 +98,8 @@ static_assert(RAFT_REGION_BIG_WRITE_THRES * 4 < RAFT_REGION_BIG_WRITE_MAX, "Inva
       F(type_partition_ts, {"type", "partition_table_scan"}),                                                                       \
       F(type_window, {"type", "window"}),                                                                                           \
       F(type_window_sort, {"type", "window_sort"}),                                                                                 \
-      F(type_expand, {"type", "expand"}))                                                                                           \
+      F(type_expand, {"type", "expand"}),                                                                                           \
+      F(type_tici, {"type", "tici_scan"}))                                                                                          \
     M(tiflash_memory_exceed_quota_count, "Total number of cases where memory exceeds quota", Counter)                               \
     M(tiflash_coprocessor_request_duration_seconds,                                                                                 \
       "Bucketed histogram of request duration",                                                                                     \
@@ -188,6 +189,20 @@ static_assert(RAFT_REGION_BIG_WRITE_THRES * 4 < RAFT_REGION_BIG_WRITE_MAX, "Inva
       Histogram,                                                                                                                    \
       F(type_sync_schema_apply_duration, {{"type", "sync_schema_duration"}}, ExpBuckets{0.001, 2, 20}),                             \
       F(type_sync_table_schema_apply_duration, {{"type", "sync_table_schema_duration"}}, ExpBuckets{0.001, 2, 20}))                 \
+    M(tiflash_gc_safepoint_request_count,                                                                                           \
+      "GC safepoint request events",                                                                                                \
+      Counter,                                                                                                                      \
+      F(type_get_gc_state, {{"type", "get_gc_state"}}),                                                                             \
+      F(type_zero_gc_safe_point, {{"type", "zero_gc_safe_point"}}),                                                                 \
+      F(type_rewind, {{"type", "rewind"}}),                                                                                         \
+      F(type_pd_response_error, {{"type", "pd_response_error"}}),                                                                   \
+      F(type_request_exception, {{"type", "request_exception"}}),                                                                   \
+      F(type_backoff_error, {{"type", "backoff_error"}}))                                                                           \
+    M(tiflash_gc_safepoint_backoff_count,                                                                                           \
+      "Bucketed histogram of GC safepoint request backoff count per call",                                                          \
+      Histogram,                                                                                                                    \
+      F(type_success, {{"type", "success"}}, ExpBuckets{1, 2, 8}),                                                                  \
+      F(type_failure, {{"type", "failure"}}, ExpBuckets{1, 2, 8}))                                                                  \
     M(tiflash_raft_read_index_count, "Total number of raft read index", Counter)                                                    \
     M(tiflash_stale_read_count, "Total number of stale read", Counter)                                                              \
     M(tiflash_raft_read_index_duration_seconds,                                                                                     \

dbms/src/Debug/dbgFuncSchema.cpp

@@ -183,7 +183,11 @@ void dbgFuncGcSchemas(Context & context, const ASTs & args, DBGInvoker::Printer
     Timestamp gc_safe_point = 0;
     bool ignore_remain_regions = false;
     if (args.empty())
-        gc_safe_point = PDClientHelper::getGCSafePointWithRetry(context.getTMTContext().getPDClient(), NullspaceID);
+        gc_safe_point = PDClientHelper::getGCSafePointWithRetry(
+            context.getTMTContext().getPDClient(),
+            NullspaceID,
+            30,
+            context.getSettingsRef().safe_point_get_max_backoff_ms);
     if (!args.empty())
         gc_safe_point = safeGet<Timestamp>(typeid_cast<const ASTLiteral &>(*args[0]).value);
     if (args.size() >= 2)

dbms/src/Debug/dbgKVStore/dbgFuncMockRaftSnapshot.cpp

@@ -547,8 +547,8 @@ RegionPtrWithBlock::CachePtr GenRegionPreDecodeBlockData(const RegionPtr & regio
             gc_safe_point = PDClientHelper::getGCSafePointWithRetry(
                 pd_client,
                 keyspace_id,
-                false,
-                context.getSettingsRef().safe_point_update_interval_seconds);
+                context.getSettingsRef().safe_point_update_interval_seconds,
+                context.getSettingsRef().safe_point_get_max_backoff_ms);
         }
         /**
          * In 5.0.1, feature `compaction filter` is enabled by default. Under such feature tikv will do gc in write & default cf individually.

dbms/src/Debug/dbgQueryExecutor.cpp

@@ -270,7 +270,10 @@ BlockInputStreamPtr executeMPPQuery(Context & context, const DAGProperties & pro
         mpp::DispatchTaskResponse resp;
         auto status = rpc.call(&client_context, req, &resp);
         if (!status.ok())
-            throw Exception("Meet grpc error while dispatch mpp task: " + rpc.errMsg(status));
+        {
+            std::string extra_msg = "addr: " + Debug::LOCAL_HOST;
+            throw Exception("Meet grpc error while dispatch mpp task: " + rpc.errMsg(status, extra_msg));
+        }
         if (resp.has_error())
             throw Exception("Meet error while dispatch mpp task: " + resp.error().msg());
     }
@@ -399,8 +402,17 @@ tipb::SelectResponse executeDAGRequest(
         region_id,
         RegionInfo(region_id, region_version, region_conf_version, std::move(key_ranges), nullptr));
 
-    DAGContext
-        dag_context(dag_request, std::move(tables_regions_info), NullspaceID, "", DAGRequestKind::Cop, "", 0, "", log);
+    DAGContext dag_context(
+        dag_request,
+        std::move(tables_regions_info),
+        QueryShardInfos(),
+        NullspaceID,
+        "",
+        DAGRequestKind::Cop,
+        "",
+        0,
+        "",
+        log);
     context.setDAGContext(&dag_context);
 
     DAGDriver<DAGRequestKind::Cop> driver(context, start_ts, DEFAULT_UNSPECIFIED_SCHEMA_VERSION, &dag_response, true);
@@ -434,8 +446,17 @@ bool runAndCompareDagReq(
         region_id,
         RegionInfo(region_id, region->version(), region->confVer(), std::move(key_ranges), nullptr));
 
-    DAGContext
-        dag_context(dag_request, std::move(tables_regions_info), NullspaceID, "", DAGRequestKind::Cop, "", 0, "", log);
+    DAGContext dag_context(
+        dag_request,
+        std::move(tables_regions_info),
+        QueryShardInfos(),
+        NullspaceID,
+        "",
+        DAGRequestKind::Cop,
+        "",
+        0,
+        "",
+        log);
     context.setDAGContext(&dag_context);
     DAGDriver<DAGRequestKind::Cop>
         driver(context, properties.start_ts, DEFAULT_UNSPECIFIED_SCHEMA_VERSION, &dag_response, true);

dbms/src/Flash/BatchCoprocessorHandler.cpp

@@ -52,8 +52,9 @@ grpc::Status BatchCoprocessorHandler::execute()
 
     try
     {
+        bool is_tici = cop_request->table_shard_infos_size() != 0;
         RUNTIME_CHECK_MSG(
-            !cop_context.db_context.getSharedContextDisagg()->isDisaggregatedComputeMode(),
+            !cop_context.db_context.getSharedContextDisagg()->isDisaggregatedComputeMode() || is_tici,
             "cannot run cop or batchCop request on tiflash_compute node");
 
         switch (cop_request->tp())
@@ -79,6 +80,7 @@ grpc::Status BatchCoprocessorHandler::execute()
             DAGContext dag_context(
                 dag_request,
                 std::move(tables_regions_info),
+                QueryShardInfos::create(cop_request->table_shard_infos()),
                 RequestUtils::deriveKeyspaceID(cop_request->context()),
                 cop_context.db_context.getClientInfo().current_address.toString(),
                 DAGRequestKind::BatchCop,

dbms/src/Flash/Coprocessor/DAGContext.cpp

@@ -49,6 +49,7 @@ bool strictSqlMode(UInt64 sql_mode)
 DAGContext::DAGContext(
     tipb::DAGRequest & dag_request_,
     TablesRegionsInfo && tables_regions_info_,
+    QueryShardInfos && query_shard_infos_,
     KeyspaceID keyspace_id_,
     const String & tidb_host_,
     DAGRequestKind kind_,
@@ -65,6 +66,7 @@ DAGContext::DAGContext(
     , kind(kind_)
     , is_root_mpp_task(false)
     , tables_regions_info(std::move(tables_regions_info_))
+    , query_shard_infos(std::move(query_shard_infos_))
     , log(std::move(log_))
     , operator_spill_contexts(std::make_shared<TaskOperatorSpillContexts>())
     , flags(dag_request->flags())

dbms/src/Flash/Coprocessor/DAGContext.h

@@ -31,6 +31,7 @@
 #include <Flash/Coprocessor/DAGRequest.h>
 #include <Flash/Coprocessor/FineGrainedShuffle.h>
 #include <Flash/Coprocessor/RuntimeFilterMgr.h>
+#include <Flash/Coprocessor/ShardInfo.h>
 #include <Flash/Coprocessor/TablesRegionsInfo.h>
 #include <Flash/Executor/toRU.h>
 #include <Flash/Mpp/MPPTaskId.h>
@@ -155,6 +156,7 @@ class DAGContext
     DAGContext(
         tipb::DAGRequest & dag_request_,
         TablesRegionsInfo && tables_regions_info_,
+        QueryShardInfos && query_shard_infos_,
         KeyspaceID keyspace_id_,
         const String & tidb_host_,
         DAGRequestKind cop_kind_,
@@ -380,6 +382,8 @@ class DAGContext
     TablesRegionsInfo tables_regions_info;
     // part of regions_for_local_read + regions_for_remote_read, only used for batch-cop
     RegionInfoList retry_regions;
+    QueryShardInfos query_shard_infos;
+    ShardInfoList retry_shards;
 
     LoggerPtr log;
 

dbms/src/Flash/Coprocessor/DAGDriver.cpp

@@ -221,6 +221,18 @@ try
             batch_cop_writer->Write(response);
         }
 
+        if (!dag_context.retry_shards.empty())
+        {
+            coprocessor::BatchResponse response;
+            for (const auto & shard : dag_context.retry_shards)
+            {
+                auto * retry_shard = response.add_retry_shards();
+                retry_shard->set_shard_id(shard.shard_id);
+                retry_shard->set_shard_epoch(shard.shard_epoch);
+            }
+            batch_cop_writer->Write(response);
+        }
+
         auto streaming_writer = std::make_shared<BatchCopStreamWriter>(batch_cop_writer);
         TiDB::TiDBCollators collators;
         auto response_writer = std::make_unique<StreamingDAGResponseWriter<BatchCopStreamWriterPtr>>(

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.cpp

@@ -1227,6 +1227,32 @@ bool DAGExpressionAnalyzer::appendExtraCastsAfterTS(
     return true;
 }
 
+bool DAGExpressionAnalyzer::appendExtraCastsAfterTiCI(
+    ExpressionActionsChain & chain,
+    const std::vector<UInt8> & may_need_add_cast_column,
+    const TiCIScan & tici_scan)
+{
+    auto & step = initAndGetLastStep(chain);
+    auto & actions = step.actions;
+
+    auto [has_cast, casted_columns]
+        = buildExtraCastsAfterTS(actions, may_need_add_cast_column, tici_scan.getReturnColumns());
+
+    if (!has_cast)
+        return false;
+
+    // Add a projection to replace the original columns with the casted columns.
+    NamesWithAliases project_cols;
+    for (size_t i = 0; i < may_need_add_cast_column.size(); ++i)
+        project_cols.emplace_back(casted_columns[i], source_columns[i].name);
+    actions->add(ExpressionAction::project(project_cols));
+
+    for (auto & col : source_columns)
+        step.required_output.push_back(col.name);
+
+    return true;
+}
+
 String DAGExpressionAnalyzer::appendDurationCast(
     const String & fsp_expr,
     const String & dur_expr,

dbms/src/Flash/Coprocessor/DAGExpressionAnalyzer.h

@@ -22,6 +22,7 @@
 #include <Flash/Coprocessor/DAGSet.h>
 #include <Flash/Coprocessor/DAGUtils.h>
 #include <Flash/Coprocessor/RuntimeFilterMgr.h>
+#include <Flash/Coprocessor/TiCIScan.h>
 #include <Flash/Coprocessor/TiDBTableScan.h>
 #include <Interpreters/AggregateDescription.h>
 #include <Interpreters/ExpressionActions.h>
@@ -113,6 +114,11 @@ class DAGExpressionAnalyzer : private boost::noncopyable
         const std::vector<UInt8> & may_need_add_cast_column,
         const TiDBTableScan & table_scan);
 
+    bool appendExtraCastsAfterTiCI(
+        ExpressionActionsChain & chain,
+        const std::vector<UInt8> & may_need_add_cast_column,
+        const TiCIScan & tici_scan);
+
     /// return true if some actions is needed
     bool appendJoinKeyAndJoinFilters(
         ExpressionActionsChain & chain,

dbms/src/Flash/Coprocessor/GenSchemaAndColumn.cpp

@@ -91,6 +91,19 @@ NamesAndTypes genNamesAndTypes(const TiDBTableScan & table_scan, const StringRef
     return genNamesAndTypes(table_scan.getColumns(), column_prefix);
 }
 
+NamesAndTypes genNamesAndTypesForTiCI(const TiDB::ColumnInfos & column_infos, const StringRef & column_prefix)
+{
+    NamesAndTypes names_and_types;
+    names_and_types.reserve(column_infos.size());
+    for (const auto & column_info : column_infos)
+    {
+        names_and_types.emplace_back(
+            fmt::format("{}_{}", column_prefix, column_info.id),
+            getDataTypeByColumnInfoForComputingLayer(column_info));
+    }
+    return names_and_types;
+}
+
 std::tuple<DM::ColumnDefinesPtr, int> genColumnDefinesForDisaggregatedRead(const TiDBTableScan & table_scan)
 {
     auto column_defines = std::make_shared<DM::ColumnDefines>();

dbms/src/Flash/Coprocessor/GenSchemaAndColumn.h

@@ -30,6 +30,7 @@ String genNameForExchangeReceiver(Int32 col_index);
 
 NamesAndTypes genNamesAndTypes(const TiDBTableScan & table_scan, const StringRef & column_prefix);
 NamesAndTypes genNamesAndTypes(const TiDB::ColumnInfos & column_infos, const StringRef & column_prefix);
+NamesAndTypes genNamesAndTypesForTiCI(const TiDB::ColumnInfos & column_infos, const StringRef & column_prefix);
 ColumnsWithTypeAndName getColumnWithTypeAndName(const NamesAndTypes & names_and_types);
 NamesAndTypes toNamesAndTypes(const DAGSchema & dag_schema);