Skip to content

[fix](be) Validate WAL replay column types #66230

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
mymeiyi wants to merge 1 commit into
base: master
Choose a base branch
from
+36 −7
Open

[fix](be) Validate WAL replay column types #66230

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
43 changes: 36 additions & 7 deletions be/src/load/group_commit/wal/wal_reader.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -99,13 +99,42 @@ Status WalReader::_do_get_next_block(Block* block, size_t* read_rows, bool* eof)
return Status::InternalError("read wal {} fail, pos {}, columns size {}", _wal_path,
pos, src_block.columns());
}
ColumnPtr column_ptr = src_block.get_by_position(pos).column;
if (!column_ptr && slot_desc->is_nullable()) {
column_ptr = make_nullable(column_ptr);
const auto& source_column = src_block.get_by_position(pos);
const auto& target_column = output_block_columns[index];
auto source_status = source_column.check_type_and_column_match();
if (!source_status.ok()) {
return Status::InternalError(
"Invalid WAL column while replaying WAL {}: slot={} (unique_id={}), "

@github-actions github-actions Bot Jul 29, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P2] Add executable coverage for the mismatch-and-retry path

This changes a kernel replay path to replace a fatal projection cast with compatibility checks and retryable statuses, but the PR adds no test for that behavior. The helper test only checks hand-built columns, the WAL writer/reader test round-trips matching types, and the only legacy scanner tests are disabled/count-only. Please add enabled serialized-WAL coverage that distinguishes a supported promotion (which should reach the existing CAST) from a genuinely incompatible mapping and AGG_STATE signature/layout (which should return a contextual error), and verify authenticated rollback, WAL retention/retry, and wait-mode exhaustion signaling after a fragment failure.

"source_position={}, error={}",
_wal_path, slot_desc->col_name(), slot_desc->col_unique_id(), pos,

@github-actions github-actions Bot Jul 29, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Roll back the replay transaction before requeueing this error

These new mismatch returns propagate through the fragment future, but WalTable::_handle_stream_load calls RETURN_IF_ERROR(ctx->load_status_future.get()) before its explicit rollback_txn. This native-insert path gets txn_id from the FE plan and leaves need_rollback false, so neither the callback nor StreamLoadContext's destructor rolls the failed transaction back. Normal mode only attempts an abort at the start of the next retry; with group_commit_wait_replay_wal_finish=true that abort is skipped entirely, and the same label can exhaust its retries before the WAL is moved to tmp and the waiter is notified without a failure value. Please propagate the future failure into both the returned status and ctx->status, restore internal token/auth-code authentication, and complete rollback before requeueing it.

source_status.to_string());
}
dst_block.insert(index, ColumnWithTypeAndName(std::move(column_ptr),
output_block_columns[index].type,
output_block_columns[index].name));

ColumnWithTypeAndName replay_column(source_column.column, target_column.type,
target_column.name);
Comment on lines +102 to +114
auto replay_status = replay_column.check_type_and_column_match();

@github-actions github-actions Bot Jul 29, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Preserve the serialized source type through compatible schema changes

This target-type validation rejects retained-ID schema changes before the load scanner can cast them. For example, a light ARRAY<INT> -> ARRAY<BIGINT> change can leave an older WAL whose ColumnArray contains Int32; relabeling it as the current ARRAY<BIGINT> fails here even though FileScanner::_cast_to_input_block has the element-wise conversion. Heavy scalar changes are normally fenced by the WAL drain, but its timeout or an unavailable BE can likewise leave an old INT WAL after the current type becomes BIGINT/STRING, and retry will never make it match. Please retain source_column.type and apply a schema-compatibility decision before the existing CAST for supported same-shape conversions; shape evolution such as appended STRUCT fields also needs explicit field/default mapping because the legacy struct CAST requires equal arity.

@github-actions github-actions Bot Jul 29, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[P1] Validate the complete AGG_STATE signature and storage layout

This compatibility check is too weak for AGG_STATE. If an old sum(INT) WAL survives the bounded schema-change drain after the column becomes avg(BIGINT), FE has accepted a changed AGG_STATE type, but BE DataTypeAggState::equals still compares only the dynamic type and its DataTypeFixedLengthObject checker ignores item_size. The source check, this replay check, and the later equality can therefore accept payloads with different sizes and meanings. The downstream CAST also reduces both states to DataTypeFixedLengthObject and selects the identity path, so sum bytes can be interpreted as an avg state. Please compare the full function/argument/nullability/version signature and validate the serialized layout (including fixed item size) before relabeling the column.

if (!replay_status.ok()) {
return Status::InternalError(
"WAL replay column type mismatch: wal={}, slot={} (unique_id={}), "
"source_position={}, source_name={}, source_type={}, source_column={}, "
"target_name={}, target_type={}, error={}",
_wal_path, slot_desc->col_name(), slot_desc->col_unique_id(), pos,
source_column.name, source_column.type->get_name(),
source_column.column->get_name(), target_column.name,
target_column.type == nullptr ? "null" : target_column.type->get_name(),
replay_status.to_string());
}
Comment on lines +116 to +126
if (!source_column.type->equals(*target_column.type)) {
return Status::InternalError(
"WAL replay logical type mismatch: wal={}, slot={} (unique_id={}), "
"source_position={}, source_name={}, source_type={}, target_name={}, "
"target_type={}",
_wal_path, slot_desc->col_name(), slot_desc->col_unique_id(), pos,
source_column.name, source_column.type->get_name(), target_column.name,
target_column.type->get_name());
}
Comment on lines +127 to +135
dst_block.insert(index, ColumnWithTypeAndName(source_column.column, target_column.type,
target_column.name));
index++;
}
block->swap(dst_block);
Expand Down Expand Up @@ -142,4 +171,4 @@ Status WalReader::_get_columns_impl(std::unordered_map<std::string, DataTypePtr>
return Status::OK();
}

} // namespace doris
} // namespace doris
Loading