This is an automated cherry-pick of pingcap#10530

Signed-off-by: ti-chi-bot <ti-community-prow-bot@tidb.io>

Author: gengliqi

dbms/src/Flash/Coprocessor/GenSchemaAndColumn.cpp

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#include <DataStreams/GeneratedColumnPlaceholderBlockInputStream.h>
 #include <Flash/Coprocessor/GenSchemaAndColumn.h>
 #include <Storages/DeltaMerge/DeltaMergeDefines.h>
 #include <Storages/MutableSupport.h>
@@ -91,14 +92,23 @@ NamesAndTypes genNamesAndTypes(const TiDBTableScan & table_scan, const StringRef
     return genNamesAndTypes(table_scan.getColumns(), column_prefix);
 }
 
-std::tuple<DM::ColumnDefinesPtr, int> genColumnDefinesForDisaggregatedRead(const TiDBTableScan & table_scan)
+std::tuple<DM::ColumnDefinesPtr, int, std::vector<std::tuple<UInt64, String, DataTypePtr>>> genColumnDefinesForDisaggregatedRead(
+    const TiDBTableScan & table_scan)
 {
     auto column_defines = std::make_shared<DM::ColumnDefines>();
     int extra_table_id_index = InvalidColumnID;
     column_defines->reserve(table_scan.getColumnSize());
+    std::vector<std::tuple<UInt64, String, DataTypePtr>> generated_column_infos;
     for (Int32 i = 0; i < table_scan.getColumnSize(); ++i)
     {
         const auto & column_info = table_scan.getColumns()[i];
+        if (column_info.hasGeneratedColumnFlag())
+        {
+            const auto & data_type = getDataTypeByColumnInfoForComputingLayer(column_info);
+            const auto & col_name = GeneratedColumnPlaceholderBlockInputStream::getColumnName(i);
+            generated_column_infos.push_back(std::make_tuple(i, col_name, data_type));
+            continue;
+        }
         // Now the upper level seems treat disagg read as an ExchangeReceiver output, so
         // use this as output column prefix.
         // Even if the id is pk_column or extra_table_id, we still output it as
@@ -114,10 +124,13 @@ std::tuple<DM::ColumnDefinesPtr, int> genColumnDefinesForDisaggregatedRead(const
             break;
         case ExtraTableIDColumnID:
         {
+<<<<<<< HEAD
             column_defines->emplace_back(DM::ColumnDefine{
                 ExtraTableIDColumnID,
                 output_name, // MutableSupport::extra_table_id_column_name
                 MutableSupport::extra_table_id_column_type});
+=======
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
             extra_table_id_index = i;
             break;
         }
@@ -130,7 +143,7 @@ std::tuple<DM::ColumnDefinesPtr, int> genColumnDefinesForDisaggregatedRead(const
             break;
         }
     }
-    return {std::move(column_defines), extra_table_id_index};
+    return {std::move(column_defines), extra_table_id_index, std::move(generated_column_infos)};
 }
 
 ColumnsWithTypeAndName getColumnWithTypeAndName(const NamesAndTypes & names_and_types)

dbms/src/Flash/Coprocessor/GenSchemaAndColumn.h

@@ -33,7 +33,8 @@ NamesAndTypes genNamesAndTypes(const TiDB::ColumnInfos & column_infos, const Str
 ColumnsWithTypeAndName getColumnWithTypeAndName(const NamesAndTypes & names_and_types);
 NamesAndTypes toNamesAndTypes(const DAGSchema & dag_schema);
 
-// The column defines and `extra table id index`
-std::tuple<DM::ColumnDefinesPtr, int> genColumnDefinesForDisaggregatedRead(const TiDBTableScan & table_scan);
+// The column defines, `extra table id index` and `generated columns info` for disaggregated read.
+std::tuple<DM::ColumnDefinesPtr, int, std::vector<std::tuple<UInt64, String, DataTypePtr>>> genColumnDefinesForDisaggregatedRead(
+    const TiDBTableScan & table_scan);
 
 } // namespace DB

dbms/src/Storages/StorageDisaggregated.h

@@ -105,8 +105,15 @@ class StorageDisaggregated : public IStorage
     std::shared_ptr<disaggregated::EstablishDisaggTaskRequest> buildEstablishDisaggTaskReq(
         const Context & db_context,
         const pingcap::coprocessor::BatchCopTask & batch_cop_task);
+<<<<<<< HEAD
     DM::RSOperatorPtr buildRSOperator(const Context & db_context, const DM::ColumnDefinesPtr & columns_to_read);
     std::variant<DM::Remote::RNWorkersPtr, DM::SegmentReadTaskPoolPtr> packSegmentReadTasks(
+=======
+    std::tuple<DM::RSOperatorPtr, DM::ColumnRangePtr> buildRSOperatorAndColumnRange(
+        const Context & db_context,
+        const DM::ColumnDefinesPtr & columns_to_read);
+    std::tuple<std::variant<DM::Remote::RNWorkersPtr, DM::SegmentReadTaskPoolPtr>, DM::ColumnDefinesPtr> packSegmentReadTasks(
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
         const Context & db_context,
         DM::SegmentReadTasks && read_tasks,
         const DM::ColumnDefinesPtr & column_defines,
@@ -154,5 +161,12 @@ class StorageDisaggregated : public IStorage
     const FilterConditions & filter_conditions;
 
     std::unique_ptr<DAGExpressionAnalyzer> analyzer;
+<<<<<<< HEAD
+=======
+    static constexpr auto ZONE_LABEL_KEY = "zone";
+    std::optional<String> zone_label;
+    // For generated column, just need a placeholder, and TiDB will fill this column.
+    std::vector<std::tuple<UInt64, String, DataTypePtr>> generated_column_infos;
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
 };
 } // namespace DB

dbms/src/Storages/StorageDisaggregatedRemote.cpp

@@ -103,7 +103,18 @@ BlockInputStreams StorageDisaggregated::readThroughS3(const Context & db_context
 {
     // Build InputStream according to the remote segment read tasks
     DAGPipeline pipeline;
+<<<<<<< HEAD
     buildRemoteSegmentInputStreams(db_context, buildReadTaskWithBackoff(db_context), num_streams, pipeline);
+=======
+    buildRemoteSegmentInputStreams(
+        db_context,
+        buildReadTaskWithBackoff(db_context, scan_context),
+        num_streams,
+        pipeline,
+        scan_context);
+    // handle generated column if necessary.
+    executeGeneratedColumnPlaceholder(generated_column_infos, log, pipeline);
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
 
     NamesAndTypes source_columns;
     source_columns.reserve(table_scan.getColumnSize());
@@ -131,8 +142,16 @@ void StorageDisaggregated::readThroughS3(
         exec_context,
         group_builder,
         db_context,
+<<<<<<< HEAD
         buildReadTaskWithBackoff(db_context),
         num_streams);
+=======
+        buildReadTaskWithBackoff(db_context, scan_context),
+        num_streams,
+        scan_context);
+    // handle generated column if necessary.
+    executeGeneratedColumnPlaceholder(exec_context, group_builder, generated_column_infos, log);
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
 
     NamesAndTypes source_columns;
     auto header = group_builder.getCurrentHeader();
@@ -496,12 +515,23 @@ DM::RSOperatorPtr StorageDisaggregated::buildRSOperator(
     return DM::RSOperator::build(dag_query, table_scan.getColumns(), *columns_to_read, enable_rs_filter, log);
 }
 
+<<<<<<< HEAD
 std::variant<DM::Remote::RNWorkersPtr, DM::SegmentReadTaskPoolPtr> StorageDisaggregated::packSegmentReadTasks(
     const Context & db_context,
     DM::SegmentReadTasks && read_tasks,
     const DM::ColumnDefinesPtr & column_defines,
     size_t num_streams,
     int extra_table_id_index)
+=======
+std::tuple<std::variant<DM::Remote::RNWorkersPtr, DM::SegmentReadTaskPoolPtr>, DM::ColumnDefinesPtr> StorageDisaggregated::
+    packSegmentReadTasks(
+        const Context & db_context,
+        DM::SegmentReadTasks && read_tasks,
+        const DM::ColumnDefinesPtr & column_defines,
+        const DM::ScanContextPtr & scan_context,
+        size_t num_streams,
+        int extra_table_id_index)
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
 {
     const auto & executor_id = table_scan.getTableScanExecutorID();
 
@@ -520,20 +550,29 @@ std::variant<DM::Remote::RNWorkersPtr, DM::SegmentReadTaskPoolPtr> StorageDisagg
         push_down_filter);
     const UInt64 start_ts = sender_target_mpp_task_id.gather_id.query_id.start_ts;
     const auto enable_read_thread = db_context.getSettingsRef().dt_enable_read_thread;
+<<<<<<< HEAD
+=======
+    const auto & final_columns_defines = push_down_executor && push_down_executor->extra_cast
+        ? push_down_executor->columns_after_cast
+        : column_defines;
+    RUNTIME_CHECK(num_streams > 0, num_streams);
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
     LOG_INFO(
         log,
         "packSegmentReadTasks: enable_read_thread={} read_mode={} is_fast_scan={} keep_order={} task_count={} "
-        "num_streams={} column_defines={}",
+        "num_streams={} column_defines={} final_columns_defines={}",
         enable_read_thread,
         magic_enum::enum_name(read_mode),
         table_scan.isFastScan(),
         table_scan.keepOrder(),
         read_tasks.size(),
         num_streams,
-        *column_defines);
+        *column_defines,
+        *final_columns_defines);
 
     if (enable_read_thread)
     {
+<<<<<<< HEAD
         return std::make_shared<DM::SegmentReadTaskPool>(
             extra_table_id_index,
             *column_defines,
@@ -561,6 +600,44 @@ std::variant<DM::Remote::RNWorkersPtr, DM::SegmentReadTaskPoolPtr> StorageDisagg
                 .read_mode = read_mode,
             },
             num_streams);
+=======
+        // Under disagg arch, now we use blocking IO to read data from cloud storage. So it require more active
+        // segments to fully utilize the read threads.
+        const size_t read_thread_num_active_seg = 10 * num_streams;
+        return {
+            std::make_shared<DM::SegmentReadTaskPool>(
+                extra_table_id_index,
+                *final_columns_defines,
+                push_down_executor,
+                start_ts,
+                db_context.getSettingsRef().max_block_size,
+                read_mode,
+                std::move(read_tasks),
+                /*after_segment_read*/ [](const DM::DMContextPtr &, const DM::SegmentPtr &) {},
+                log->identifier(),
+                /*enable_read_thread*/ true,
+                num_streams,
+                read_thread_num_active_seg,
+                context.getDAGContext()->getKeyspaceID(),
+                context.getDAGContext()->getResourceGroupName()),
+            final_columns_defines};
+    }
+    else
+    {
+        return {
+            DM::Remote::RNWorkers::create(
+                db_context,
+                std::move(read_tasks),
+                {
+                    .log = log->getChild(executor_id),
+                    .columns_to_read = final_columns_defines,
+                    .start_ts = start_ts,
+                    .push_down_executor = push_down_executor,
+                    .read_mode = read_mode,
+                },
+                num_streams),
+            final_columns_defines};
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
     }
 }
 
@@ -598,15 +675,29 @@ void StorageDisaggregated::buildRemoteSegmentInputStreams(
     DAGPipeline & pipeline)
 {
     // Build the input streams to read blocks from remote segments
+<<<<<<< HEAD
     auto [column_defines, extra_table_id_index] = genColumnDefinesForDisaggregatedRead(table_scan);
     auto packed_read_tasks
         = packSegmentReadTasks(db_context, std::move(read_tasks), column_defines, num_streams, extra_table_id_index);
     RUNTIME_CHECK(num_streams > 0, num_streams);
+=======
+    DM::ColumnDefinesPtr column_defines;
+    int extra_table_id_index;
+    std::tie(column_defines, extra_table_id_index, generated_column_infos)
+        = genColumnDefinesForDisaggregatedRead(table_scan);
+    auto [packed_read_tasks, final_column_defines] = packSegmentReadTasks(
+        db_context,
+        std::move(read_tasks),
+        column_defines,
+        scan_context,
+        num_streams,
+        extra_table_id_index);
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
     pipeline.streams.reserve(num_streams);
 
     InputStreamBuilder builder{
         .tracing_id = log->identifier(),
-        .columns_to_read = column_defines,
+        .columns_to_read = final_column_defines,
         .extra_table_id_index = extra_table_id_index,
     };
     for (size_t stream_idx = 0; stream_idx < num_streams; ++stream_idx)
@@ -661,14 +752,28 @@ void StorageDisaggregated::buildRemoteSegmentSourceOps(
     size_t num_streams)
 {
     // Build the input streams to read blocks from remote segments
+<<<<<<< HEAD
     auto [column_defines, extra_table_id_index] = genColumnDefinesForDisaggregatedRead(table_scan);
     auto packed_read_tasks
         = packSegmentReadTasks(db_context, std::move(read_tasks), column_defines, num_streams, extra_table_id_index);
+=======
+    DM::ColumnDefinesPtr column_defines;
+    int extra_table_id_index;
+    std::tie(column_defines, extra_table_id_index, generated_column_infos)
+        = genColumnDefinesForDisaggregatedRead(table_scan);
+    auto [packed_read_tasks, final_column_defines] = packSegmentReadTasks(
+        db_context,
+        std::move(read_tasks),
+        column_defines,
+        scan_context,
+        num_streams,
+        extra_table_id_index);
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530))
 
     RUNTIME_CHECK(num_streams > 0, num_streams);
     SrouceOpBuilder builder{
         .tracing_id = log->identifier(),
-        .column_defines = column_defines,
+        .column_defines = final_column_defines,
         .extra_table_id_index = extra_table_id_index,
         .exec_context = exec_context,
     };

dbms/src/TableFunctions/CMakeLists.txt

@@ -18,5 +18,16 @@ add_headers_and_sources(tiflash_table_functions .)
 list(REMOVE_ITEM tiflash_table_functions_sources ITableFunction.cpp TableFunctionFactory.cpp)
 list(REMOVE_ITEM tiflash_table_functions_headers ITableFunction.h TableFunctionFactory.h)
 
+<<<<<<< HEAD:dbms/src/TableFunctions/CMakeLists.txt
 add_library(tiflash_table_functions ${tiflash_table_functions_sources})
 target_link_libraries(tiflash_table_functions tiflash_storages_system dbms ${Poco_Foundation_LIBRARY})
+=======
+tikv-worker-url = "tikv-worker0:19000"
+
+enable-telemetry = false
+temp-dir = "/data/tmp"
+[performance]
+tcp-keep-alive = true
+[security]
+enable-sem = false
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530)):tests/docker/next-gen-config/tidb.toml

release-linux-llvm/dockerfiles/misc/install_cmake.sh

@@ -13,6 +13,16 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+<<<<<<< HEAD:release-linux-llvm/dockerfiles/misc/install_cmake.sh
+=======
+[storage]
+# No space is reserved at all for testing
+reserve-space = "0"
+# Enable keyspace and ttl for next-gen
+api-version = 2
+enable-ttl = true
+low-space-threshold = 0
+>>>>>>> a5e14033f8 (Fix three schema mismatch bugs under disaggregated arch  (#10530)):tests/docker/next-gen-config/tikv.toml
 
 # Install cmake for CI/CD.
 # Require: wget

tests/docker/next-gen-config/tikv-worker.toml

@@ -0,0 +1,28 @@
+# Copyright 2025 PingCAP, Inc.
+#
+# 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.
+
+# object storage for next-gen
+[dfs]
+prefix = "tikv"
+s3-endpoint = "http://minio0:9000"
+s3-key-id = "minioadmin"
+s3-secret-key = "minioadmin"
+s3-bucket = "tiflash-test"
+s3-region = "local"
+
+[schema-manager]
+dir = "/data/schemas"
+enabled = true
+keyspace-refresh-interval = "10s"
+schema-refresh-threshold = 1