This is an automated email from the ASF dual-hosted git repository.
liaoxin01 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 17e561b0492 [fix](be) Replay schema-change WAL with duplicate column
IDs (#70160)
17e561b0492 is described below
commit 17e561b04925ac2531c998bad7a3fa5207432a7c
Author: Xin Liao <[email protected]>
AuthorDate: Thu Nov 5 20:45:03 2026 +0800
[fix](be) Replay schema-change WAL with duplicate column IDs (#70160)
### What problem does this PR solve?
Heavy ALTER operations can write base and shadow columns with the same
ID into Group Commit WALs. V2 rejects these valid headers as corruption.
Accept duplicate IDs while preserving physical column order and the
existing first-column mapping.
Related PR: #66008
### Release note
Fix Group Commit WAL replay failures during heavy schema changes with
FileScannerV2.
---
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]
