Skip to content

[opt](exec) Align operator projection with scanner (without copy when do project) #66085

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
Mryange wants to merge 2 commits into
base: master
Choose a base branch
from
+30 −82
Open

[opt](exec) Align operator projection with scanner (without copy when do project) #66085

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
1962421
upd
Mryange Jul 27, 2026
71bf601
upd
Mryange Jul 27, 2026
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
86 changes: 21 additions & 65 deletions be/src/exec/operator/operator.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -325,77 +325,33 @@ 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;

@github-actions github-actions Bot Jul 27, 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] 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.

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);
}

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(rows, input_block.rows());

@github-actions github-actions Bot Jul 27, 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] 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.

if (!output_block->mem_reuse()) {
Block empty_output(VectorizedUtils::create_empty_block(*_output_row_descriptor));
output_block->swap(empty_output);
}
DCHECK_EQ(output_block->columns(), local_state->_projections.size()) << debug_string();

for (int i = 0; i < output_block->columns(); ++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 (output_block->get_by_position(i).column->is_nullable() != column_ptr->is_nullable()) {
throw Exception(ErrorCode::INTERNAL_ERROR, "Nullable mismatch");
}
DCHECK(mutable_block.rows() == rows);
output_block->replace_by_position(i, std::move(column_ptr));
}
local_state->_estimate_memory_usage += bytes_usage;

DCHECK_EQ(output_block->rows(), rows);

return Status::OK();
}
Expand Down
4 changes: 0 additions & 4 deletions be/src/exec/operator/operator.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1014,10 +1014,6 @@ class OperatorXBase : public OperatorBase {
std::string _op_name;
int _parallel_tasks = 0;

//_keep_origin is used to avoid copying during projection,
// currently set to false only in the nestloop join.
bool _keep_origin = true;

// _blockable is true if the operator contains expressions that may block execution
bool _blockable = false;
};
Expand Down
22 changes: 9 additions & 13 deletions be/src/exec/scan/scanner.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -222,27 +222,23 @@ Status Scanner::_do_projections(Block* origin_block, Block* output_block) {
}

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();

DCHECK_EQ(mutable_columns.size(), _projections.size());
if (!output_block->mem_reuse()) {
Block empty_output(VectorizedUtils::create_empty_block(*_output_row_descriptor));
output_block->swap(empty_output);
}
DCHECK_EQ(output_block->columns(), _projections.size());

for (int i = 0; i < mutable_columns.size(); ++i) {
for (int i = 0; i < output_block->columns(); ++i) {
ColumnPtr column_ptr;
RETURN_IF_ERROR(_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()) {
if (output_block->get_by_position(i).column->is_nullable() != column_ptr->is_nullable()) {
throw Exception(ErrorCode::INTERNAL_ERROR, "Nullable mismatch");
}
mutable_columns[i] = IColumn::mutate(std::move(column_ptr));
output_block->replace_by_position(i, std::move(column_ptr));
}

scoped_mutable_block.restore();

// origin columns was moved into output_block, so we need to set origin_block to empty columns
// Projection results may share columns with origin_block, so reset it before the next scan.
auto empty_columns = origin_block->clone_empty_columns();
origin_block->set_columns(std::move(empty_columns));
DCHECK_EQ(output_block->rows(), rows);
Expand Down
Loading