This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new f707d0874f1 branch-4.1: [fix](be) Replay schema-change WAL with
duplicate column IDs #70160 (#70299)
f707d0874f1 is described below
commit f707d0874f1e98b34d77d8c8ba71d912db249415
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Nov 10 11:40:07 2026 +0800
branch-4.1: [fix](be) Replay schema-change WAL with duplicate column IDs
#70160 (#70299)
Cherry-picked from #70160
Co-authored-by: Xin Liao <[email protected]>
---
be/src/format_v2/wal/wal_reader.cpp | 7 +-
be/test/format_v2/wal/wal_reader_test.cpp | 70 ++++++++++++++++-
.../test_group_commit_schema_change_wal.out | 16 ++++
.../test_group_commit_schema_change_wal.groovy | 91 ++++++++++++++++++++++
4 files changed, 177 insertions(+), 7 deletions(-)
diff --git a/be/src/format_v2/wal/wal_reader.cpp
b/be/src/format_v2/wal/wal_reader.cpp
index 5f57a422ff8..d319313b37c 100644
--- a/be/src/format_v2/wal/wal_reader.cpp
+++ b/be/src/format_v2/wal/wal_reader.cpp
@@ -21,7 +21,6 @@
#include <absl/strings/str_split.h>
#include <ranges>
-#include <unordered_set>
#include <utility>
#include "agent/be_exec_version_manager.h"
@@ -89,15 +88,13 @@ Status parse_wal_column_ids(const std::string& encoded,
std::vector<int32_t>* co
return Status::Corruption("WAL header contains no column ids");
}
- std::unordered_set<int32_t> seen;
+ // Schema change appends shadow columns with the original column ids.
Preserve all physical
+ // positions: the field-id mapper selects the first (base) column, like
the legacy WAL reader.
for (const absl::string_view token : absl::StrSplit(encoded, ',')) {
int32_t column_id = 0;
if (token.empty() || !absl::SimpleAtoi(token, &column_id)) {
return Status::Corruption("invalid WAL column id '{}'",
std::string(token));
}
- if (!seen.emplace(column_id).second) {
- return Status::Corruption("duplicate WAL column id {}", column_id);
- }
column_ids->push_back(column_id);
}
return Status::OK();
diff --git a/be/test/format_v2/wal/wal_reader_test.cpp
b/be/test/format_v2/wal/wal_reader_test.cpp
index 65e37bd5f89..0d2024e8116 100644
--- a/be/test/format_v2/wal/wal_reader_test.cpp
+++ b/be/test/format_v2/wal/wal_reader_test.cpp
@@ -34,6 +34,8 @@
#include "core/data_type/data_type_number.h"
#include "core/data_type/data_type_string.h"
#include "core/data_type/data_type_struct.h"
+#include "format_v2/column_mapper.h"
+#include "format_v2/table_reader.h"
#include "io/fs/local_file_system.h"
#include "load/group_commit/wal/wal_file_reader.h"
#include "load/group_commit/wal/wal_writer.h"
@@ -188,12 +190,76 @@ TEST(WalReaderV2Test, ParseColumnIdsPreservesHeaderOrder)
{
EXPECT_EQ(column_ids, (std::vector<int32_t> {17, 4, 99}));
}
-TEST(WalReaderV2Test, ParseColumnIdsRejectsMalformedOrAmbiguousHeaders) {
+TEST(WalReaderV2Test, ParseColumnIdsRejectsMalformedHeaders) {
std::vector<int32_t> column_ids;
EXPECT_FALSE(parse_wal_column_ids("", &column_ids).ok());
EXPECT_FALSE(parse_wal_column_ids("17,,99", &column_ids).ok());
EXPECT_FALSE(parse_wal_column_ids("17,nope,99", &column_ids).ok());
- EXPECT_FALSE(parse_wal_column_ids("17,4,17", &column_ids).ok());
+}
+
+TEST(WalReaderV2Test,
SchemaChangeDuplicateIdsKeepBaseColumnAndPhysicalPositions) {
+ const auto wal_path = temporary_wal_path();
+ const auto int_type = make_nullable(std::make_shared<DataTypeInt32>());
+ const auto bigint_type = make_nullable(std::make_shared<DataTypeInt64>());
+ Block source;
+ auto base = int_type->create_column();
+ base->insert(Field::create_field<TYPE_INT>(20));
+ source.insert({std::move(base), int_type, "v"});
+ auto shadow = bigint_type->create_column();
+ shadow->insert(Field::create_field<TYPE_BIGINT>(200));
+ source.insert({std::move(shadow), bigint_type, "__doris_shadow_v"});
+ auto trailing = int_type->create_column();
+ trailing->insert(Field::create_field<TYPE_INT>(30));
+ source.insert({std::move(trailing), int_type, "other"});
+ auto pblock = serialize_block(source);
+
+ WalWriter writer(wal_path);
+ ASSERT_TRUE(writer.init(io::global_local_filesystem()).ok());
+ ASSERT_TRUE(writer.append_header("15,15,17").ok());
+ ASSERT_TRUE(writer.append_blocks({&pblock}).ok());
+ ASSERT_TRUE(writer.finalize().ok());
+
+ std::shared_ptr<io::FileSystemProperties> properties;
+ std::unique_ptr<io::FileDescription> description;
+ WalReader reader(properties, description, nullptr, nullptr, {});
+ reader._wal_reader = std::make_shared<WalFileReader>(wal_path);
+ ASSERT_TRUE(reader._wal_reader->init().ok());
+ std::string encoded_ids;
+ ASSERT_TRUE(reader._wal_reader->read_header(reader._version,
encoded_ids).ok());
+ ASSERT_TRUE(parse_wal_column_ids(encoded_ids, &reader._column_ids).ok());
+ EXPECT_EQ(reader._column_ids, (std::vector<int32_t> {15, 15, 17}));
+
+ std::vector<ColumnDefinition> schema;
+ ASSERT_TRUE(reader.get_schema(&schema).ok());
+ ASSERT_EQ(schema.size(), 3);
+ // The ALTER has finished: the table expects BIGINT, but replay must read
the base INT.
+ auto changed = schema[0];
+ changed.type = bigint_type;
+ const std::vector<ColumnDefinition> projected {schema[2], changed};
+ auto mapper = reader.create_column_mapper({});
+ ASSERT_TRUE(mapper->create_mapping(projected, {}, schema).ok());
+ ASSERT_EQ(mapper->mappings().size(), 2);
+ EXPECT_EQ(mapper->mappings()[0].file_local_id, 2);
+ EXPECT_EQ(mapper->mappings()[1].file_local_id, 0);
+ EXPECT_TRUE(mapper->mappings()[1].file_type->equals(*int_type));
+ EXPECT_TRUE(mapper->mappings()[1].table_type->equals(*bigint_type));
+
+ auto request = std::make_shared<FileScanRequest>();
+ ASSERT_TRUE(mapper->create_scan_request({}, projected,
request.get()).ok());
+ const auto other_position = request->local_positions.at(LocalColumnId(2));
+ const auto base_position = request->local_positions.at(LocalColumnId(0));
+ ASSERT_TRUE(reader.open(std::move(request)).ok());
+ Block output({{int_type->create_column(), int_type, "first"},
+ {int_type->create_column(), int_type, "second"}});
+ size_t rows = 0;
+ bool eof = false;
+ ASSERT_TRUE(reader.get_block(&output, &rows, &eof).ok());
+ EXPECT_EQ(rows, 1);
+
EXPECT_EQ((*output.get_by_position(other_position.value()).column)[0].get<TYPE_INT>(),
30);
+
EXPECT_EQ((*output.get_by_position(base_position.value()).column)[0].get<TYPE_INT>(),
20);
+ ASSERT_TRUE(reader.close().ok());
+ std::filesystem::remove_all(
+
std::filesystem::path(wal_path).parent_path().parent_path().parent_path());
}
TEST(WalReaderV2Test,
WriterBackedReaderPreservesNestedSchemaAndMovesFirstPayload) {
diff --git
a/regression-test/data/insert_p0/group_commit/test_group_commit_schema_change_wal.out
b/regression-test/data/insert_p0/group_commit/test_group_commit_schema_change_wal.out
new file mode 100644
index 00000000000..67646d10923
--- /dev/null
+++
b/regression-test/data/insert_p0/group_commit/test_group_commit_schema_change_wal.out
@@ -0,0 +1,16 @@
+-- This file is automatically generated. You should know what you did if you
want to edit this
+-- !running --
+1 10 100
+2 20 200
+3 \N 300
+
+-- !finished --
+1 10 100
+2 20 200
+3 \N 300
+
+-- !cancelled --
+1 10 100
+2 20 200
+3 \N 300
+
diff --git
a/regression-test/suites/insert_p0/group_commit/test_group_commit_schema_change_wal.groovy
b/regression-test/suites/insert_p0/group_commit/test_group_commit_schema_change_wal.groovy
new file mode 100644
index 00000000000..f40d050c9ce
--- /dev/null
+++
b/regression-test/suites/insert_p0/group_commit/test_group_commit_schema_change_wal.groovy
@@ -0,0 +1,91 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you 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.
+
+import org.awaitility.Awaitility
+import static java.util.concurrent.TimeUnit.SECONDS
+
+suite("test_group_commit_schema_change_wal", "nonConcurrent") {
+ def alterPoint = "FE.SchemaChangeJobV2.runRunning.block"
+ def loadPoint = "LoadBlockQueue._finish_group_commit_load.load_error"
+ def replayPoint = "WalTable.replay_wals.stop"
+ def convertPoint = isCloudMode() ?
"CloudSchemaChangeJob::process_alter_tablet.block"
+ :
"SchemaChangeJob::_do_process_alter_tablet.block"
+
+ def awaitAlter = { expected ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(1, SECONDS).until
{
+ def jobs = sql_return_maparray """SHOW ALTER TABLE COLUMN
+ WHERE TableName='test_group_commit_schema_change_wal'
+ ORDER BY CreateTime DESC LIMIT 1"""
+ return !jobs.isEmpty() && jobs[0].State == expected
+ }
+ }
+ def replayInState = { state ->
+ sql "DROP TABLE IF EXISTS test_group_commit_schema_change_wal"
+ sql """CREATE TABLE test_group_commit_schema_change_wal (
+ id INT NOT NULL, v INT NULL, other INT NULL
+ ) DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES("replication_num"="1", "light_schema_change"="true",
+ "group_commit_interval_ms"="1000")"""
+ sql "INSERT INTO test_group_commit_schema_change_wal VALUES (1,10,100)"
+ try {
+ if (state == "CANCELLED") {
+ // Hold the BE task, leaving the FE job mutex available for
CANCEL ALTER.
+ GetDebugPoint().enableDebugPointForAllBEs(convertPoint)
+ } else {
+ GetDebugPoint().enableDebugPointForAllFEs(alterPoint)
+ }
+ sql "ALTER TABLE test_group_commit_schema_change_wal MODIFY COLUMN
v BIGINT NULL"
+ awaitAlter("RUNNING")
+ GetDebugPoint().enableDebugPointForAllBEs(loadPoint)
+ GetDebugPoint().enableDebugPointForAllBEs(replayPoint)
+ // The WAL contains both the base INT and the shadow BIGINT with
the same unique id.
+ sql "SET group_commit = async_mode"
+ sql "INSERT INTO test_group_commit_schema_change_wal VALUES
(2,20,200),(3,NULL,300)"
+ sql "SET group_commit = off_mode"
+ if (state == "FINISHED") {
+ // Let the normal WAL-drain timeout expire before replaying
with the new schema.
+ GetDebugPoint().disableDebugPointForAllFEs(alterPoint)
+ awaitAlter("FINISHED")
+ } else if (state == "CANCELLED") {
+ sql "CANCEL ALTER TABLE COLUMN FROM
test_group_commit_schema_change_wal"
+ awaitAlter("CANCELLED")
+ }
+ GetDebugPoint().disableDebugPointForAllBEs(replayPoint)
+ Awaitility.await().atMost(360, SECONDS).pollInterval(1,
SECONDS).until {
+ return (sql "SELECT count(*) FROM
test_group_commit_schema_change_wal")[0][0] == 3
+ }
+ GetDebugPoint().disableDebugPointForAllFEs(alterPoint)
+ if (state == "RUNNING") {
+ awaitAlter("FINISHED")
+ }
+ } finally {
+ sql "SET group_commit = off_mode"
+ GetDebugPoint().disableDebugPointForAllBEs(convertPoint)
+ GetDebugPoint().disableDebugPointForAllBEs(loadPoint)
+ GetDebugPoint().disableDebugPointForAllBEs(replayPoint)
+ GetDebugPoint().disableDebugPointForAllFEs(alterPoint)
+ }
+ }
+
+ sql "SET group_commit = off_mode"
+ replayInState("RUNNING")
+ order_qt_running "SELECT * FROM test_group_commit_schema_change_wal"
+ replayInState("FINISHED")
+ order_qt_finished "SELECT * FROM test_group_commit_schema_change_wal"
+ replayInState("CANCELLED")
+ order_qt_cancelled "SELECT * FROM test_group_commit_schema_change_wal"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
