Skip to content

[opt](column) avoid redundant column clones #65367

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 1 commit into
base: master
Choose a base branch
from
Open

[opt](column) avoid redundant column clones #65367

Show file tree
Hide file tree
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
2 changes: 1 addition & 1 deletion be/src/exec/operator/multi_cast_data_stream_source.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -110,7 +110,7 @@ Status MultiCastDataStreamerSourceOperatorX::get_block_impl(RuntimeState* state,
if (!local_state._output_expr_contexts.empty() && output_block->rows() > 0) {
SCOPED_TIMER(local_state._materialize_data_timer);
RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(
local_state._output_expr_contexts, *output_block, block, true));
local_state._output_expr_contexts, *output_block, block));
materialize_block_inplace(*block);
}
return Status::OK();
Expand Down
21 changes: 3 additions & 18 deletions be/src/exec/sink/writer/iceberg/viceberg_table_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -206,7 +206,7 @@ Status VIcebergTableWriter::write(RuntimeState* state, Block& block) {
}
Block output_block;
RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(_vec_output_expr_ctxs, block,
&output_block, false));
&output_block));
materialize_block_inplace(output_block);
return _write_prepared_block(output_block);
}
Expand Down Expand Up @@ -436,23 +436,8 @@ Status VIcebergTableWriter::_write_prepared_block(Block& output_block) {

Status VIcebergTableWriter::_filter_block(doris::Block& block, const IColumn::Filter* filter,
doris::Block* output_block) {
const ColumnsWithTypeAndName& columns_with_type_and_name =
block.get_columns_with_type_and_name();
ColumnsWithTypeAndName result_columns;
for (const auto& col : columns_with_type_and_name) {
result_columns.emplace_back(col.column->clone_resized(col.column->size()), col.type,
col.name);
}
*output_block = {std::move(result_columns)};

std::vector<uint32_t> columns_to_filter;
int column_to_keep = output_block->columns();
columns_to_filter.resize(column_to_keep);
for (uint32_t i = 0; i < column_to_keep; ++i) {
columns_to_filter[i] = i;
}

Block::filter_block_internal(output_block, columns_to_filter, *filter);
*output_block = block;
Block::filter_block_internal(output_block, *filter);
return Status::OK();
}

Expand Down
2 changes: 1 addition & 1 deletion be/src/exec/sink/writer/maxcompute/vmc_table_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ Status VMCTableWriter::write(RuntimeState* state, Block& block) {

Block output_block;
RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(_vec_output_expr_ctxs, block,
&output_block, false));
&output_block));
materialize_block_inplace(output_block);

_row_count += output_block.rows();
Expand Down
22 changes: 3 additions & 19 deletions be/src/exec/sink/writer/vhive_table_writer.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -92,7 +92,7 @@ Status VHiveTableWriter::write(RuntimeState* state, Block& block) {
}
Block output_block;
RETURN_IF_ERROR(VExprContext::get_output_block_after_execute_exprs(_vec_output_expr_ctxs, block,
&output_block, false));
&output_block));
materialize_block_inplace(output_block);

std::unordered_map<std::shared_ptr<VHivePartitionWriter>, IColumn::Filter> writer_positions;
Expand Down Expand Up @@ -220,24 +220,8 @@ Status VHiveTableWriter::write(RuntimeState* state, Block& block) {

Status VHiveTableWriter::_filter_block(doris::Block& block, const IColumn::Filter* filter,
doris::Block* output_block) {
const ColumnsWithTypeAndName& columns_with_type_and_name =
block.get_columns_with_type_and_name();
ColumnsWithTypeAndName result_columns;
for (int i = 0; i < columns_with_type_and_name.size(); ++i) {
const auto& col = columns_with_type_and_name[i];
result_columns.emplace_back(col.column->clone_resized(col.column->size()), col.type,
col.name);
}
*output_block = {std::move(result_columns)};

std::vector<uint32_t> columns_to_filter;
int column_to_keep = output_block->columns();
columns_to_filter.resize(column_to_keep);
for (uint32_t i = 0; i < column_to_keep; ++i) {
columns_to_filter[i] = i;
}

Block::filter_block_internal(output_block, columns_to_filter, *filter);
*output_block = block;
Block::filter_block_internal(output_block, *filter);
return Status::OK();
}

Expand Down
4 changes: 2 additions & 2 deletions be/src/exprs/function/if.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -420,8 +420,8 @@ class FunctionIf : public IFunction {
handled = false;

if (cond_is_null) {
block.replace_by_position(result,
arg_else.column->clone_resized(arg_cond.column->size()));
DCHECK_EQ(arg_else.column->size(), arg_cond.column->size());
block.replace_by_position(result, arg_else.column);
handled = true;
return Status::OK();
}
Expand Down
18 changes: 8 additions & 10 deletions be/src/exprs/lambda_function/varray_map_function.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -121,7 +121,7 @@ class ArrayMapFunction : public LambdaFunction {
auto outside_null_map = ColumnUInt8::create(
arguments[0].column->convert_to_full_column_if_const()->size(), 0);
// offset column
MutableColumnPtr array_column_offset;
ColumnPtr array_column_offset;
size_t nested_array_column_rows = 0;
ColumnPtr first_array_offsets = nullptr;
//2. get the result column from executed expr, and the needed is nested column of array
Expand Down Expand Up @@ -158,8 +158,7 @@ class ArrayMapFunction : public LambdaFunction {
if (i == 0) {
nested_array_column_rows = col_array.get_data_ptr()->size();
first_array_offsets = col_array.get_offsets_ptr();
const auto& off_data = col_array.get_offsets_column();
array_column_offset = off_data.clone_resized(col_array.get_offsets_column().size());
array_column_offset = first_array_offsets;
args_info.offsets_ptr = &col_array.get_offsets();
} else {
// select array_map((x,y)->x+y,c_array1,[0,1,2,3]) from array_test2;
Expand Down Expand Up @@ -200,8 +199,8 @@ class ArrayMapFunction : public LambdaFunction {
auto empty_nested_column = assert_cast<const DataTypeArray*>(nested_type.get())
->get_nested_type()
->create_column();
auto result_array_column = ColumnArray::create(std::move(empty_nested_column),
std::move(array_column_offset));
auto result_array_column =
ColumnArray::create(std::move(empty_nested_column), array_column_offset);

if (is_nullable) {
result_column = ColumnNullable::create(std::move(result_array_column),
Expand Down Expand Up @@ -300,7 +299,7 @@ class ArrayMapFunction : public LambdaFunction {
if (result_type->is_nullable()) {
if (res_type->is_nullable()) {
result_column = ColumnNullable::create(
ColumnArray::create(std::move(result_col), std::move(array_column_offset)),
ColumnArray::create(std::move(result_col), array_column_offset),
std::move(outside_null_map));
} else {
// deal with eg: select array_map(x -> x is null, [null, 1, 2]);
Expand All @@ -310,19 +309,18 @@ class ArrayMapFunction : public LambdaFunction {
result_column = ColumnNullable::create(
ColumnArray::create(ColumnNullable::create(std::move(result_col),
std::move(nested_null_map)),
std::move(array_column_offset)),
array_column_offset),
std::move(outside_null_map));
}
} else {
if (res_type->is_nullable()) {
result_column =
ColumnArray::create(std::move(result_col), std::move(array_column_offset));
result_column = ColumnArray::create(std::move(result_col), array_column_offset);
} else {
auto nested_null_map = ColumnUInt8::create(result_col->size(), 0);

result_column = ColumnArray::create(
ColumnNullable::create(std::move(result_col), std::move(nested_null_map)),
std::move(array_column_offset));
array_column_offset);
}
}
return Status::OK();
Expand Down
3 changes: 2 additions & 1 deletion be/src/exprs/vcondition_expr.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -369,7 +369,8 @@ Status VectorizedIfExpr::execute_for_null_condition(Block& block, const ColumnNu
handled = false;

if (cond_is_null) {
block.replace_by_position(result, arg_else.column->clone_resized(arg_cond.column->size()));
DCHECK_EQ(arg_else.column->size(), arg_cond.column->size());
block.replace_by_position(result, arg_else.column);
handled = true;
return Status::OK();
}
Expand Down
16 changes: 2 additions & 14 deletions be/src/exprs/vexpr_context.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -435,15 +435,8 @@ Status VExprContext::execute_conjuncts_and_filter_block(const VExprContextSPtrs&
return Status::OK();
}

// do_projection: for some query(e.g. in MultiCastDataStreamerSourceOperator::get_block()),
// output_vexpr_ctxs will output the same column more than once, and if the output_block
// is mem-reused later, it will trigger DCHECK_EQ(d.column->use_count(), 1) failure when
// doing Block::clear_column_data, set do_projection to true to copy the column data to
// avoid this problem.
Status VExprContext::get_output_block_after_execute_exprs(
const VExprContextSPtrs& output_vexpr_ctxs, const Block& input_block, Block* output_block,
bool do_projection) {
auto rows = input_block.rows();
const VExprContextSPtrs& output_vexpr_ctxs, const Block& input_block, Block* output_block) {
ColumnsWithTypeAndName result_columns;
_reset_memory_usage(output_vexpr_ctxs);

Expand All @@ -455,12 +448,7 @@ Status VExprContext::get_output_block_after_execute_exprs(
const auto& name = vexpr_ctx->expr_name();

vexpr_ctx->_memory_usage += result_column->allocated_bytes();
if (do_projection) {
result_columns.emplace_back(result_column->clone_resized(rows), type, name);

} else {
result_columns.emplace_back(result_column, type, name);
}
result_columns.emplace_back(result_column, type, name);
}
*output_block = {result_columns};
return Status::OK();
Expand Down
3 changes: 1 addition & 2 deletions be/src/exprs/vexpr_context.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -329,8 +329,7 @@ class VExprContext {
int column_to_keep, IColumn::Filter& filter);

[[nodiscard]] static Status get_output_block_after_execute_exprs(const VExprContextSPtrs&,
const Block&, Block*,
bool do_projection = false);
const Block&, Block*);

int get_last_result_column_id() const {
DCHECK(_last_result_column_id != -1);
Expand Down
Loading