[BugFix] FSE-v2 missed setting the schema for shared-data sorted schema change (backport #72235)

be/src/storage/lake/delta_writer.cpp

@@ -92,7 +92,8 @@ class DeltaWriterImpl {
                              int64_t schema_id, const PartialUpdateMode& partial_update_mode,
                              const std::map<string, string>* column_to_expr_value, PUniqueId load_id,
                              RuntimeProfile* profile, BundleWritableFileContext* bundle_writable_file_context,
-                             GlobalDictByNameMaps* global_dicts, bool is_multi_statements_txn)
+                             GlobalDictByNameMaps* global_dicts, bool is_multi_statements_txn,
+                             std::shared_ptr<const TabletSchema> tablet_schema)
             : _tablet_manager(tablet_manager),
               _tablet_id(tablet_id),
               _txn_id(txn_id),
@@ -103,6 +104,7 @@ class DeltaWriterImpl {
               _mem_tracker(mem_tracker),
               _slots(slots),
               _max_buffer_size(max_buffer_size > 0 ? max_buffer_size : config::write_buffer_size),
+              _tablet_schema(std::move(tablet_schema)),
               _immutable_tablet_size(immutable_tablet_size),
               _merge_condition(std::move(merge_condition)),
               _miss_auto_increment_column(miss_auto_increment_column),
@@ -1123,11 +1125,15 @@ StatusOr<DeltaWriterBuilder::DeltaWriterPtr> DeltaWriterBuilder::build() {
     if (UNLIKELY(_schema_id == 0)) {
         return Status::InvalidArgument("schema_id not set");
     }
-    auto impl =
-            new DeltaWriterImpl(_tablet_mgr, _tablet_id, _txn_id, _partition_id, _slots, _merge_condition,
-                                _miss_auto_increment_column, _db_id, _table_id, _immutable_tablet_size, _mem_tracker,
-                                _max_buffer_size, _schema_id, _partial_update_mode, _column_to_expr_value, _load_id,
-                                _profile, _bundle_writable_file_context, _global_dicts, _is_multi_statements_txn);
+    if (UNLIKELY(_tablet_schema != nullptr && _tablet_schema->id() != _schema_id)) {
+        return Status::InvalidArgument(
+                fmt::format("tablet_schema id {} mismatches schema_id {}", _tablet_schema->id(), _schema_id));
+    }
+    auto impl = new DeltaWriterImpl(_tablet_mgr, _tablet_id, _txn_id, _partition_id, _slots, _merge_condition,
+                                    _miss_auto_increment_column, _db_id, _table_id, _immutable_tablet_size,
+                                    _mem_tracker, _max_buffer_size, _schema_id, _partial_update_mode,
+                                    _column_to_expr_value, _load_id, _profile, _bundle_writable_file_context,
+                                    _global_dicts, _is_multi_statements_txn, _tablet_schema);
     return std::make_unique<DeltaWriter>(impl);
 }
 

be/src/storage/lake/delta_writer.h

@@ -267,6 +267,15 @@ class DeltaWriterBuilder {
         return *this;
     }
 
+    // Pre-set the tablet schema. When provided, DeltaWriter skips the schema lookup that
+    // would otherwise consult TableSchemaService (cache + tablet metadata + FE RPC). Useful
+    // when the caller already holds the exact schema and wants to avoid the dependency on
+    // catalog ids and the schema cache, e.g. lake schema-change rewrites.
+    DeltaWriterBuilder& set_tablet_schema(std::shared_ptr<const TabletSchema> tablet_schema) {
+        _tablet_schema = std::move(tablet_schema);
+        return *this;
+    }
+
     DeltaWriterBuilder& set_partial_update_mode(const PartialUpdateMode& partial_update_mode) {
         _partial_update_mode = partial_update_mode;
         return *this;
@@ -325,6 +334,7 @@ class DeltaWriterBuilder {
     BundleWritableFileContext* _bundle_writable_file_context{nullptr};
     GlobalDictByNameMaps* _global_dicts = nullptr;
     bool _is_multi_statements_txn = false;
+    std::shared_ptr<const TabletSchema> _tablet_schema;
 };
 
 } // namespace starrocks::lake

be/src/storage/lake/schema_change.cpp

@@ -276,14 +276,19 @@ Status SortedSchemaChange::process(RowsetPtr rowset, RowsetMetadata* new_rowset_
     RETURN_IF_ERROR(reader->prepare());
     RETURN_IF_ERROR(reader->open(_read_params));
 
-    // create writer
+    // Pass the new tablet schema directly so DeltaWriter does not consult TableSchemaService.
+    // The schema-service lookup is keyed by (db_id, table_id, schema_id) and falls back to an
+    // FE RPC; without catalog ids on the alter task the RPC goes out with zeros and FE replies
+    // "Table not exist". The schema-change task already holds the exact schema, so we sidestep
+    // the lookup entirely.
     ASSIGN_OR_RETURN(auto writer, DeltaWriterBuilder()
                                           .set_tablet_manager(_tablet_manager)
                                           .set_tablet_id(_new_tablet.id())
                                           .set_txn_id(_txn_id)
                                           .set_max_buffer_size(_max_buffer_size)
                                           .set_mem_tracker(CurrentThread::mem_tracker())
-                                          .set_schema_id(_new_tablet_schema->id()) // TODO: pass tablet schema directly
+                                          .set_schema_id(_new_tablet_schema->id())
+                                          .set_tablet_schema(_new_tablet_schema)
                                           .build());
     RETURN_IF_ERROR(writer->open());
     DeferOp defer([&]() { writer->close(); });

be/test/storage/lake/schema_change_test.cpp

@@ -36,6 +36,8 @@
 #include "storage/metadata_util.h"
 #include "testutil/assert.h"
 #include "testutil/id_generator.h"
+#include "testutil/sync_point.h"
+#include "util/defer_op.h"
 
 namespace starrocks::lake {
 
@@ -2764,4 +2766,128 @@ TEST_F(SchemaChangeBaseTabletReadSchemaTest, test_generated_column) {
     ASSERT_EQ(3, row_count);
 }
 
+// Regression for the SortedSchemaChange path: DeltaWriter must not consult the schema
+// service (which falls back to an FE RPC) because the schema-change task already holds
+// the exact target schema. The fix passes the tablet schema directly into DeltaWriter,
+// so the schema lookup -- and its FE RPC fallback -- is bypassed entirely.
+TEST_F(SchemaChangeBaseTabletReadSchemaTest, test_sorted_schema_change_skips_schema_rpc) {
+    SyncPoint::GetInstance()->EnableProcessing();
+    DeferOp restore([] {
+        SyncPoint::GetInstance()->ClearAllCallBacks();
+        SyncPoint::GetInstance()->DisableProcessing();
+    });
+
+    std::atomic_int rpc_invocations = 0;
+    SyncPoint::GetInstance()->SetCallBack(
+            "TableSchemaService::_fetch_schema_via_rpc::test_hook",
+            [&rpc_invocations](void* arg) { rpc_invocations.fetch_add(1, std::memory_order_relaxed); });
+
+    auto base_metadata = create_base_tablet_metadata();
+    auto base_tablet_id = base_metadata->id();
+    CHECK_OK(_tablet_manager->put_tablet_metadata(*base_metadata));
+
+    // Build an AGG_KEYS new tablet so SchemaChangeUtils::parse_request flips sc_sorting=true.
+    auto new_metadata = std::make_shared<TabletMetadata>();
+    new_metadata->set_id(next_id());
+    new_metadata->set_version(1);
+    auto new_schema = new_metadata->mutable_schema();
+    new_schema->set_id(next_id());
+    new_schema->set_num_short_key_columns(1);
+    new_schema->set_keys_type(AGG_KEYS);
+    new_schema->set_num_rows_per_row_block(65535);
+    {
+        auto c0 = new_schema->add_column();
+        c0->set_unique_id(1);
+        c0->set_name("c0");
+        c0->set_type("INT");
+        c0->set_is_key(true);
+        c0->set_is_nullable(false);
+    }
+    {
+        auto c1 = new_schema->add_column();
+        c1->set_unique_id(2);
+        c1->set_name("c1");
+        c1->set_type("VARCHAR");
+        c1->set_length(20);
+        c1->set_is_key(true);
+        c1->set_is_nullable(false);
+    }
+    {
+        auto c2_sum = new_schema->add_column();
+        c2_sum->set_unique_id(9);
+        c2_sum->set_name("c2_sum");
+        c2_sum->set_type("BIGINT");
+        c2_sum->set_is_key(false);
+        c2_sum->set_is_nullable(true);
+        c2_sum->set_aggregation("SUM");
+    }
+    CHECK_OK(_tablet_manager->put_tablet_metadata(*new_metadata));
+
+    int64_t version = 1;
+    int64_t txn_id = next_id();
+    auto base_tablet_schema = TabletSchema::create(base_metadata->schema());
+    auto base_schema = std::make_shared<VSchema>(ChunkHelper::convert_schema(base_tablet_schema));
+
+    auto col_c0 = Int32Column::create();
+    auto col_c1 = BinaryColumn::create();
+    auto col_c2 = Int32Column::create();
+    auto col_c3 = Int32Column::create();
+    auto col_c4 = Int32Column::create();
+    col_c0->append_datum(Datum(1));
+    col_c1->append_datum(Datum(Slice("row0")));
+    col_c2->append_datum(Datum(100));
+    col_c3->append_datum(Datum(1000));
+    col_c4->append_datum(Datum(10));
+    VChunk chunk0({std::move(col_c0), std::move(col_c1), std::move(col_c2), std::move(col_c3), std::move(col_c4)},
+                  base_schema);
+    uint32_t indexes[1] = {0};
+
+    ASSIGN_OR_ABORT(auto delta_writer, DeltaWriterBuilder()
+                                               .set_tablet_manager(_tablet_manager.get())
+                                               .set_tablet_id(base_tablet_id)
+                                               .set_txn_id(txn_id)
+                                               .set_partition_id(_partition_id)
+                                               .set_mem_tracker(_mem_tracker.get())
+                                               .set_schema_id(base_tablet_schema->id())
+                                               .set_tablet_schema(base_tablet_schema)
+                                               .build());
+    ASSERT_OK(delta_writer->open());
+    ASSERT_OK(delta_writer->write(chunk0, indexes, sizeof(indexes) / sizeof(indexes[0])));
+    ASSERT_OK(delta_writer->finish_with_txnlog());
+    delta_writer->close();
+    ASSERT_OK(TEST_publish_single_version(_tablet_manager.get(), base_tablet_id, version + 1, txn_id).status());
+    version++;
+
+    // Drop the BE schema cache so the only way DeltaWriter could resolve the new schema is
+    // an FE RPC. With the fix, the schema is preset and the lookup is skipped, so the hook
+    // above must observe zero invocations.
+    _tablet_manager->update_metacache_limit(0);
+    _tablet_manager->prune_metacache();
+
+    auto new_tablet_id = new_metadata->id();
+    auto alter_txn_id = next_id();
+    {
+        auto t_read_schema = create_base_tablet_read_schema(next_id());
+
+        TAlterTabletReqV2 request;
+        request.base_tablet_id = base_tablet_id;
+        request.new_tablet_id = new_tablet_id;
+        request.alter_version = version;
+        request.txn_id = alter_txn_id;
+        request.__set_base_tablet_read_schema(t_read_schema);
+        request.__set_base_table_column_names({"c0", "c1", "c6", "c2", "c4"});
+        TAlterMaterializedViewParam mv_param;
+        mv_param.__set_column_name("c2_sum");
+        mv_param.__set_origin_column_name("c2");
+        request.__set_materialized_view_params({mv_param});
+
+        SchemaChangeHandler handler(_tablet_manager.get());
+        ASSERT_OK(handler.process_alter_tablet(request));
+    }
+    ASSERT_OK(publish_version_for_schema_change(new_tablet_id, version + 1, alter_txn_id));
+
+    EXPECT_EQ(0, rpc_invocations.load(std::memory_order_relaxed))
+            << "SortedSchemaChange must not call the FE schema RPC when DeltaWriter is given a preset schema";
+}
+
 } // namespace starrocks::lake

test/sql/test_fast_schema_evolution/R/test_fse_alter_with_metacache_off

@@ -0,0 +1,46 @@
+-- name: test_lake_alter_add_first_key_no_metacache @cloud
+update information_schema.be_configs set value = "0" where name = "lake_metadata_cache_limit";
+-- result:
+-- !result
+create database test_db_${uuid0};
+-- result:
+-- !result
+use test_db_${uuid0};
+-- result:
+-- !result
+create table t1 (k1 int, k2 int, v1 int) duplicate key(k1)
+  distributed by hash(k1) buckets 1 properties("replication_num" = "1");
+-- result:
+-- !result
+insert into t1 values (1, 10, 100), (2, 20, 200), (3, 30, 300);
+-- result:
+-- !result
+alter table t1 add column k0 int key default "0" first;
+-- result:
+-- !result
+function: wait_alter_table_finish()
+-- result:
+None
+-- !result
+select * from t1 order by k1;
+-- result:
+0	1	10	100
+0	2	20	200
+0	3	30	300
+-- !result
+insert into t1 values (-1, 4, 40, 400);
+-- result:
+-- !result
+select * from t1 order by k1;
+-- result:
+0	1	10	100
+0	2	20	200
+0	3	30	300
+-1	4	40	400
+-- !result
+drop database test_db_${uuid0};
+-- result:
+-- !result
+update information_schema.be_configs set value = "2147483648" where name = "lake_metadata_cache_limit";
+-- result:
+-- !result

test/sql/test_fast_schema_evolution/T/test_fse_alter_with_metacache_off

@@ -0,0 +1,25 @@
+-- name: test_lake_alter_add_first_key_no_metacache @cloud
+-- Regression for the SortedSchemaChange path on lake tables: when the BE schema cache is
+-- disabled, the schema-change writer must still find the new schema. The fix passes the
+-- tablet schema directly into DeltaWriter, bypassing TableSchemaService entirely. Before
+-- the fix the writer fell back to an FE schema RPC and the alter was cancelled with
+-- "table not found, db id: 0, table id: 0" because the alter task carried no catalog ids.
+update information_schema.be_configs set value = "0" where name = "lake_metadata_cache_limit";
+create database test_db_${uuid0};
+use test_db_${uuid0};
+
+create table t1 (k1 int, k2 int, v1 int) duplicate key(k1)
+  distributed by hash(k1) buckets 1 properties("replication_num" = "1");
+insert into t1 values (1, 10, 100), (2, 20, 200), (3, 30, 300);
+
+-- Trigger SortedSchemaChange by adding a key column at the head: this rewrites rowsets
+-- through DeltaWriter, the only schema-change path that built a writer without a schema.
+alter table t1 add column k0 int key default "0" first;
+function: wait_alter_table_finish()
+
+select * from t1 order by k1;
+insert into t1 values (-1, 4, 40, 400);
+select * from t1 order by k1;
+
+drop database test_db_${uuid0};
+update information_schema.be_configs set value = "2147483648" where name = "lake_metadata_cache_limit";