github-actions[bot] commented on code in PR #66085:
URL: https://github.com/apache/doris/pull/66085#discussion_r3655092228
##########
be/src/exec/operator/operator.cpp:
##########
@@ -325,77 +325,35 @@ Status OperatorXBase::do_projections(RuntimeState* state,
Block* origin_block,
}
Block input_block = *origin_block;
- size_t bytes_usage = 0;
- ColumnsWithTypeAndName new_columns;
+ std::vector<int> result_column_ids;
Review Comment:
[P1] Keep projection allocations in the next-round reservation estimate
This rewrite removes every update to `local_state->_estimate_memory_usage`.
`OperatorX::get_reserve_mem_size()` reads that value, and `PipelineTask`
consumes/resets it before the next `get_block_after_projects()` to decide
whether to pause/revoke/spill. Moving a result into `output_block` after
expression evaluation does not remove the allocation: a large string/complex
intermediate or final projection will now reserve only the minimum and can
allocate through the workload-group pressure gate. Please keep an observed or
sound pre-execution estimate for newly allocated projection data (without
counting genuinely shared pass-through storage) and cover a repeated large
variable-width projection with reservation enabled.
##########
be/src/exec/operator/operator.cpp:
##########
@@ -325,77 +325,35 @@ Status OperatorXBase::do_projections(RuntimeState* state,
Block* origin_block,
}
Block input_block = *origin_block;
- size_t bytes_usage = 0;
- ColumnsWithTypeAndName new_columns;
+ std::vector<int> result_column_ids;
for (const auto& projections : local_state->_intermediate_projections) {
- if (projections.empty()) {
- return Status::InternalError("meet empty intermediate projection,
node id: {}",
- node_id());
- }
- new_columns.resize(projections.size());
+ result_column_ids.resize(projections.size());
for (int i = 0; i < projections.size(); i++) {
- RETURN_IF_ERROR(projections[i]->execute(&input_block,
new_columns[i]));
- if (new_columns[i].column->size() != rows) {
- return Status::InternalError(
- "intermediate projection result column size {} not
equal input rows {}, "
- "expr: {}",
- new_columns[i].column->size(), rows,
- projections[i]->root()->debug_string());
- }
- }
- Block tmp_block {new_columns};
- bytes_usage += tmp_block.allocated_bytes();
- input_block.swap(tmp_block);
- }
-
- if (input_block.rows() != rows) {
- return Status::InternalError(
- "after intermediate projections input block rows {} not equal
origin rows {}, "
- "input_block: {}",
- input_block.rows(), rows, input_block.dump_structure());
- }
- auto insert_column_datas = [&](auto& to, ColumnPtr& from, size_t rows) {
- if (is_column_nullable(*to) && !is_column_nullable(*from)) {
- if (_keep_origin || !from->is_exclusive()) {
- auto& null_column = reinterpret_cast<ColumnNullable&>(*to);
- null_column.get_nested_column().insert_range_from(*from, 0,
rows);
- null_column.get_null_map_column().get_data().resize_fill(rows,
0);
- bytes_usage += null_column.allocated_bytes();
- } else {
- to = make_nullable(from, false)->assert_mutable();
- }
- } else {
- if (_keep_origin || !from->is_exclusive()) {
- to->insert_range_from(*from, 0, rows);
- bytes_usage += from->allocated_bytes();
- } else {
- to = from->assert_mutable();
- }
+ RETURN_IF_ERROR(projections[i]->execute(&input_block,
&result_column_ids[i]));
}
- };
+ input_block.shuffle_columns(result_column_ids);
+ }
+ DCHECK_EQ(rows, input_block.rows());
auto scoped_mutable_block =
VectorizedUtils::build_scoped_mutable_mem_reuse_block(
output_block, *_output_row_descriptor);
auto& mutable_block = scoped_mutable_block.mutable_block();
auto& mutable_columns = mutable_block.mutable_columns();
- if (rows != 0) {
- DCHECK_EQ(mutable_columns.size(), local_state->_projections.size()) <<
debug_string();
- for (int i = 0; i < mutable_columns.size(); ++i) {
- ColumnPtr column_ptr;
-
RETURN_IF_ERROR(local_state->_projections[i]->execute(&input_block,
column_ptr));
- if (column_ptr->size() != rows) {
- return Status::InternalError(
- "projection result column size {} not equal input rows
{}, expr: {}",
- column_ptr->size(), rows,
- local_state->_projections[i]->root()->debug_string());
- }
- column_ptr = column_ptr->convert_to_full_column_if_const();
- bytes_usage += column_ptr->allocated_bytes();
- insert_column_datas(mutable_columns[i], column_ptr, rows);
+ DCHECK_EQ(mutable_columns.size(), local_state->_projections.size()) <<
debug_string();
+
+ for (int i = 0; i < mutable_columns.size(); ++i) {
Review Comment:
[P2] Avoid cloning pass-through slots while discarding the reusable buffer
For a final slot ref, `origin_block`, the shallow `input_block`, and
`column_ptr` all own the same column when `IColumn::mutate()` runs, so
`shallow_mutate()` takes its deep-clone path. Meanwhile `mutable_columns[i]` is
the retained-capacity output column cleared by `PipelineTask`, and this
assignment destroys it. A slot-only projection therefore performs a fresh
allocation plus full-column copy on every block. Please publish the immutable
result without requiring mutable ownership, or release the source owners before
a real transfer, and add a reused-block slot projection benchmark/test.
##########
be/src/exec/operator/operator.cpp:
##########
@@ -325,77 +325,35 @@ Status OperatorXBase::do_projections(RuntimeState* state,
Block* origin_block,
}
Block input_block = *origin_block;
- size_t bytes_usage = 0;
- ColumnsWithTypeAndName new_columns;
+ std::vector<int> result_column_ids;
for (const auto& projections : local_state->_intermediate_projections) {
- if (projections.empty()) {
- return Status::InternalError("meet empty intermediate projection,
node id: {}",
- node_id());
- }
- new_columns.resize(projections.size());
+ result_column_ids.resize(projections.size());
for (int i = 0; i < projections.size(); i++) {
- RETURN_IF_ERROR(projections[i]->execute(&input_block,
new_columns[i]));
- if (new_columns[i].column->size() != rows) {
- return Status::InternalError(
- "intermediate projection result column size {} not
equal input rows {}, "
- "expr: {}",
- new_columns[i].column->size(), rows,
- projections[i]->root()->debug_string());
- }
- }
- Block tmp_block {new_columns};
- bytes_usage += tmp_block.allocated_bytes();
- input_block.swap(tmp_block);
- }
-
- if (input_block.rows() != rows) {
- return Status::InternalError(
- "after intermediate projections input block rows {} not equal
origin rows {}, "
- "input_block: {}",
- input_block.rows(), rows, input_block.dump_structure());
- }
- auto insert_column_datas = [&](auto& to, ColumnPtr& from, size_t rows) {
- if (is_column_nullable(*to) && !is_column_nullable(*from)) {
- if (_keep_origin || !from->is_exclusive()) {
- auto& null_column = reinterpret_cast<ColumnNullable&>(*to);
- null_column.get_nested_column().insert_range_from(*from, 0,
rows);
- null_column.get_null_map_column().get_data().resize_fill(rows,
0);
- bytes_usage += null_column.allocated_bytes();
- } else {
- to = make_nullable(from, false)->assert_mutable();
- }
- } else {
- if (_keep_origin || !from->is_exclusive()) {
- to->insert_range_from(*from, 0, rows);
- bytes_usage += from->allocated_bytes();
- } else {
- to = from->assert_mutable();
- }
+ RETURN_IF_ERROR(projections[i]->execute(&input_block,
&result_column_ids[i]));
}
- };
+ input_block.shuffle_columns(result_column_ids);
+ }
+ DCHECK_EQ(rows, input_block.rows());
auto scoped_mutable_block =
VectorizedUtils::build_scoped_mutable_mem_reuse_block(
output_block, *_output_row_descriptor);
auto& mutable_block = scoped_mutable_block.mutable_block();
auto& mutable_columns = mutable_block.mutable_columns();
- if (rows != 0) {
- DCHECK_EQ(mutable_columns.size(), local_state->_projections.size()) <<
debug_string();
- for (int i = 0; i < mutable_columns.size(); ++i) {
- ColumnPtr column_ptr;
-
RETURN_IF_ERROR(local_state->_projections[i]->execute(&input_block,
column_ptr));
- if (column_ptr->size() != rows) {
- return Status::InternalError(
- "projection result column size {} not equal input rows
{}, expr: {}",
- column_ptr->size(), rows,
- local_state->_projections[i]->root()->debug_string());
- }
- column_ptr = column_ptr->convert_to_full_column_if_const();
- bytes_usage += column_ptr->allocated_bytes();
- insert_column_datas(mutable_columns[i], column_ptr, rows);
+ DCHECK_EQ(mutable_columns.size(), local_state->_projections.size()) <<
debug_string();
+
+ for (int i = 0; i < mutable_columns.size(); ++i) {
+ ColumnPtr column_ptr;
+ RETURN_IF_ERROR(local_state->_projections[i]->execute(&input_block,
column_ptr));
+ column_ptr = column_ptr->convert_to_full_column_if_const();
+ if (mutable_columns[i]->is_nullable() != column_ptr->is_nullable()) {
+ throw Exception(ErrorCode::INTERNAL_ERROR, "Nullable mismatch");
}
- DCHECK(mutable_block.rows() == rows);
+ mutable_columns[i] = IColumn::mutate(std::move(column_ptr));
Review Comment:
[P1] Validate the result against the destination type before replacing its
pointer
This block keeps the `DataType` created from `_output_row_descriptor`, but
`VSlotRef::execute_type(block)` intentionally validates a returned slot column
against the runtime input type, which can differ from the expression's
declared/output type. With equal nullability (for example declared/output `INT`
but runtime `BIGINT`), the check above passes and this installs `ColumnInt64`
under `DataTypeInt32`. The old typed `insert_range_from` failed loudly through
release-enabled `assert_cast`; please validate against the destination type or
materialize into the typed destination before replacing it.
##########
be/src/exec/operator/operator.cpp:
##########
@@ -325,77 +325,35 @@ Status OperatorXBase::do_projections(RuntimeState* state,
Block* origin_block,
}
Block input_block = *origin_block;
- size_t bytes_usage = 0;
- ColumnsWithTypeAndName new_columns;
+ std::vector<int> result_column_ids;
for (const auto& projections : local_state->_intermediate_projections) {
- if (projections.empty()) {
- return Status::InternalError("meet empty intermediate projection,
node id: {}",
- node_id());
- }
- new_columns.resize(projections.size());
+ result_column_ids.resize(projections.size());
for (int i = 0; i < projections.size(); i++) {
- RETURN_IF_ERROR(projections[i]->execute(&input_block,
new_columns[i]));
- if (new_columns[i].column->size() != rows) {
- return Status::InternalError(
- "intermediate projection result column size {} not
equal input rows {}, "
- "expr: {}",
- new_columns[i].column->size(), rows,
- projections[i]->root()->debug_string());
- }
- }
- Block tmp_block {new_columns};
- bytes_usage += tmp_block.allocated_bytes();
- input_block.swap(tmp_block);
- }
-
- if (input_block.rows() != rows) {
- return Status::InternalError(
- "after intermediate projections input block rows {} not equal
origin rows {}, "
- "input_block: {}",
- input_block.rows(), rows, input_block.dump_structure());
- }
- auto insert_column_datas = [&](auto& to, ColumnPtr& from, size_t rows) {
- if (is_column_nullable(*to) && !is_column_nullable(*from)) {
- if (_keep_origin || !from->is_exclusive()) {
- auto& null_column = reinterpret_cast<ColumnNullable&>(*to);
- null_column.get_nested_column().insert_range_from(*from, 0,
rows);
- null_column.get_null_map_column().get_data().resize_fill(rows,
0);
- bytes_usage += null_column.allocated_bytes();
- } else {
- to = make_nullable(from, false)->assert_mutable();
- }
- } else {
- if (_keep_origin || !from->is_exclusive()) {
- to->insert_range_from(*from, 0, rows);
- bytes_usage += from->allocated_bytes();
- } else {
- to = from->assert_mutable();
- }
+ RETURN_IF_ERROR(projections[i]->execute(&input_block,
&result_column_ids[i]));
}
- };
+ input_block.shuffle_columns(result_column_ids);
+ }
+ DCHECK_EQ(rows, input_block.rows());
Review Comment:
[P2] Keep a release-mode rejection for an empty intermediate stage
`init()` still accepts an empty element in `intermediate_projections_list`;
`resize(0)` followed by `shuffle_columns({})` creates a zero-column block whose
`rows()` is zero. These replacement checks disappear in release builds, so
final expressions execute with count zero and a nonempty input can return
`Status::OK()` with no rows, whereas the deleted check returned
`InternalError`. Please restore the empty-stage/runtime row-count invariant
(the per-expression size checks can rely on `execute_column()`) and cover the
release behavior with a malformed-plan/unit test.
##########
be/src/exec/operator/operator.cpp:
##########
@@ -325,77 +325,35 @@ Status OperatorXBase::do_projections(RuntimeState* state,
Block* origin_block,
}
Block input_block = *origin_block;
- size_t bytes_usage = 0;
- ColumnsWithTypeAndName new_columns;
+ std::vector<int> result_column_ids;
for (const auto& projections : local_state->_intermediate_projections) {
- if (projections.empty()) {
- return Status::InternalError("meet empty intermediate projection,
node id: {}",
- node_id());
- }
- new_columns.resize(projections.size());
+ result_column_ids.resize(projections.size());
for (int i = 0; i < projections.size(); i++) {
- RETURN_IF_ERROR(projections[i]->execute(&input_block,
new_columns[i]));
- if (new_columns[i].column->size() != rows) {
- return Status::InternalError(
- "intermediate projection result column size {} not
equal input rows {}, "
- "expr: {}",
- new_columns[i].column->size(), rows,
- projections[i]->root()->debug_string());
- }
- }
- Block tmp_block {new_columns};
- bytes_usage += tmp_block.allocated_bytes();
- input_block.swap(tmp_block);
- }
-
- if (input_block.rows() != rows) {
- return Status::InternalError(
- "after intermediate projections input block rows {} not equal
origin rows {}, "
- "input_block: {}",
- input_block.rows(), rows, input_block.dump_structure());
- }
- auto insert_column_datas = [&](auto& to, ColumnPtr& from, size_t rows) {
- if (is_column_nullable(*to) && !is_column_nullable(*from)) {
- if (_keep_origin || !from->is_exclusive()) {
- auto& null_column = reinterpret_cast<ColumnNullable&>(*to);
- null_column.get_nested_column().insert_range_from(*from, 0,
rows);
- null_column.get_null_map_column().get_data().resize_fill(rows,
0);
- bytes_usage += null_column.allocated_bytes();
- } else {
- to = make_nullable(from, false)->assert_mutable();
- }
- } else {
- if (_keep_origin || !from->is_exclusive()) {
- to->insert_range_from(*from, 0, rows);
- bytes_usage += from->allocated_bytes();
- } else {
- to = from->assert_mutable();
- }
+ RETURN_IF_ERROR(projections[i]->execute(&input_block,
&result_column_ids[i]));
}
- };
+ input_block.shuffle_columns(result_column_ids);
+ }
+ DCHECK_EQ(rows, input_block.rows());
auto scoped_mutable_block =
VectorizedUtils::build_scoped_mutable_mem_reuse_block(
output_block, *_output_row_descriptor);
auto& mutable_block = scoped_mutable_block.mutable_block();
auto& mutable_columns = mutable_block.mutable_columns();
- if (rows != 0) {
- DCHECK_EQ(mutable_columns.size(), local_state->_projections.size()) <<
debug_string();
- for (int i = 0; i < mutable_columns.size(); ++i) {
- ColumnPtr column_ptr;
-
RETURN_IF_ERROR(local_state->_projections[i]->execute(&input_block,
column_ptr));
- if (column_ptr->size() != rows) {
- return Status::InternalError(
- "projection result column size {} not equal input rows
{}, expr: {}",
- column_ptr->size(), rows,
- local_state->_projections[i]->root()->debug_string());
- }
- column_ptr = column_ptr->convert_to_full_column_if_const();
- bytes_usage += column_ptr->allocated_bytes();
- insert_column_datas(mutable_columns[i], column_ptr, rows);
+ DCHECK_EQ(mutable_columns.size(), local_state->_projections.size()) <<
debug_string();
+
+ for (int i = 0; i < mutable_columns.size(); ++i) {
+ ColumnPtr column_ptr;
+ RETURN_IF_ERROR(local_state->_projections[i]->execute(&input_block,
column_ptr));
+ column_ptr = column_ptr->convert_to_full_column_if_const();
+ if (mutable_columns[i]->is_nullable() != column_ptr->is_nullable()) {
Review Comment:
[P1] Preserve the supported nullable widening here
`VExpr::check_expr_output_type()` explicitly accepts `T -> Nullable(T)`, and
`VExpr::execute_column()` also allows a declared nullable expression to return
a physically non-nullable column when the evaluated rows contain no nulls. Such
a plan passes prepare/expression execution but now throws here; the deleted
branch attached an all-zero null map. Please preserve this one-way mismatch
with `make_nullable(..., false)` (or equivalent) before transferring ownership,
while still rejecting `Nullable(T) -> T`, and add a focused positive-row test.
--
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]
