github-actions[bot] commented on code in PR #66230:
URL: https://github.com/apache/doris/pull/66230#discussion_r3674229055
##########
be/src/load/group_commit/wal/wal_reader.cpp:
##########
@@ -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={}), "
+ "source_position={}, error={}",
+ _wal_path, slot_desc->col_name(),
slot_desc->col_unique_id(), pos,
Review Comment:
[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.
##########
be/src/load/group_commit/wal/wal_reader.cpp:
##########
@@ -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={}), "
+ "source_position={}, error={}",
+ _wal_path, slot_desc->col_name(),
slot_desc->col_unique_id(), pos,
+ 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);
+ auto replay_status = replay_column.check_type_and_column_match();
Review Comment:
[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.
##########
be/src/load/group_commit/wal/wal_reader.cpp:
##########
@@ -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={}), "
Review Comment:
[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.
##########
be/src/load/group_commit/wal/wal_reader.cpp:
##########
@@ -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={}), "
+ "source_position={}, error={}",
+ _wal_path, slot_desc->col_name(),
slot_desc->col_unique_id(), pos,
+ 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);
+ auto replay_status = replay_column.check_type_and_column_match();
Review Comment:
[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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
