This is an automated email from the ASF dual-hosted git repository.
dataroaring pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new c73391b75d0 [Fix](autoinc) Hanlde the processing of auto_increment
column on exchange node rather than on TabletWriter when using
`TABLET_SINK_SHUFFLE_PARTITIONED` (#36836)
c73391b75d0 is described below
commit c73391b75d03aca3646f7cf6dcdd091668f89c13
Author: bobhan1 <bh2444151...@outlook.com>
AuthorDate: Sun Jun 30 13:07:27 2024 +0800
[Fix](autoinc) Hanlde the processing of auto_increment column on exchange
node rather than on TabletWriter when using `TABLET_SINK_SHUFFLE_PARTITIONED`
(#36836)
## Proposed changes
Issue Number: close #36638
https://github.com/apache/doris/pull/30914 add partition tablet sink
shuffle and the processing of auto_increment column should be handled on
exchange node raher than TabletWriter when using partition tablet sink
shuffle.
branch-2.1-pick: https://github.com/apache/doris/pull/37029
---
be/src/pipeline/exec/exchange_sink_operator.cpp | 5 +-
be/src/vec/sink/vtablet_block_convertor.cpp | 5 +-
be/src/vec/sink/writer/vtablet_writer.cpp | 2 +
be/src/vec/sink/writer/vtablet_writer_v2.cpp | 2 +
.../unique/test_unique_table_auto_inc.out | 62 ++++++++++++++++++++++
.../unique/test_unique_table_auto_inc.groovy | 38 +++++++++++++
6 files changed, 110 insertions(+), 4 deletions(-)
diff --git a/be/src/pipeline/exec/exchange_sink_operator.cpp
b/be/src/pipeline/exec/exchange_sink_operator.cpp
index 0ccded0b825..198bc555024 100644
--- a/be/src/pipeline/exec/exchange_sink_operator.cpp
+++ b/be/src/pipeline/exec/exchange_sink_operator.cpp
@@ -193,9 +193,12 @@ Status ExchangeSinkLocalState::open(RuntimeState* state) {
std::make_unique<vectorized::OlapTabletFinder>(_vpartition.get(),
find_tablet_mode);
_tablet_sink_tuple_desc =
_state->desc_tbl().get_tuple_descriptor(p._tablet_sink_tuple_id);
_tablet_sink_row_desc = p._pool->add(new
RowDescriptor(_tablet_sink_tuple_desc, false));
- //_block_convertor no need init_autoinc_info here
+ // if _part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED,
we handle the processing of auto_increment column
+ // on exchange node rather than on TabletWriter
_block_convertor =
std::make_unique<vectorized::OlapTableBlockConvertor>(_tablet_sink_tuple_desc);
+ _block_convertor->init_autoinc_info(_schema->db_id(),
_schema->table_id(),
+ _state->batch_size());
_location = p._pool->add(new
OlapTableLocationParam(p._tablet_sink_location));
_row_distribution.init(
{.state = _state,
diff --git a/be/src/vec/sink/vtablet_block_convertor.cpp
b/be/src/vec/sink/vtablet_block_convertor.cpp
index 7f7f4c76008..feb6633511e 100644
--- a/be/src/vec/sink/vtablet_block_convertor.cpp
+++ b/be/src/vec/sink/vtablet_block_convertor.cpp
@@ -505,8 +505,7 @@ Status
OlapTableBlockConvertor::_fill_auto_inc_cols(vectorized::Block* block, si
vectorized::ColumnInt64::Container& dst_values = dst_column->get_data();
vectorized::ColumnPtr src_column_ptr = block->get_by_position(idx).column;
- if (const vectorized::ColumnConst* const_column =
- check_and_get_column<vectorized::ColumnConst>(src_column_ptr))
{
+ if (const auto* const_column =
check_and_get_column<vectorized::ColumnConst>(src_column_ptr)) {
// for insert stmt like "insert into tbl1 select null,col1,col2,...
from tbl2" or
// "insert into tbl1 select 1,col1,col2,... from tbl2", the type of
literal's column
// will be `ColumnConst`
@@ -529,7 +528,7 @@ Status
OlapTableBlockConvertor::_fill_auto_inc_cols(vectorized::Block* block, si
int64_t value = const_column->get_int(0);
dst_values.resize_fill(rows, value);
}
- } else if (const vectorized::ColumnNullable* src_nullable_column =
+ } else if (const auto* src_nullable_column =
check_and_get_column<vectorized::ColumnNullable>(src_column_ptr)) {
auto src_nested_column_ptr =
src_nullable_column->get_nested_column_ptr();
const auto& null_map_data = src_nullable_column->get_null_map_data();
diff --git a/be/src/vec/sink/writer/vtablet_writer.cpp
b/be/src/vec/sink/writer/vtablet_writer.cpp
index 5d36ca37805..6d388a7d958 100644
--- a/be/src/vec/sink/writer/vtablet_writer.cpp
+++ b/be/src/vec/sink/writer/vtablet_writer.cpp
@@ -1197,6 +1197,8 @@ Status VTabletWriter::_init(RuntimeState* state,
RuntimeProfile* profile) {
}
_block_convertor =
std::make_unique<OlapTableBlockConvertor>(_output_tuple_desc);
+ // if partition_type is TABLET_SINK_SHUFFLE_PARTITIONED, we handle the
processing of auto_increment column
+ // on exchange node rather than on TabletWriter
_block_convertor->init_autoinc_info(
_schema->db_id(), _schema->table_id(), _state->batch_size(),
_schema->is_partial_update() &&
!_schema->auto_increment_coulumn().empty(),
diff --git a/be/src/vec/sink/writer/vtablet_writer_v2.cpp
b/be/src/vec/sink/writer/vtablet_writer_v2.cpp
index 9bd154ce212..fbefd7a6d83 100644
--- a/be/src/vec/sink/writer/vtablet_writer_v2.cpp
+++ b/be/src/vec/sink/writer/vtablet_writer_v2.cpp
@@ -214,6 +214,8 @@ Status VTabletWriterV2::_init(RuntimeState* state,
RuntimeProfile* profile) {
}
_block_convertor =
std::make_unique<OlapTableBlockConvertor>(_output_tuple_desc);
+ // if partition_type is TABLET_SINK_SHUFFLE_PARTITIONED, we handle the
processing of auto_increment column
+ // on exchange node rather than on TabletWriter
_block_convertor->init_autoinc_info(
_schema->db_id(), _schema->table_id(), _state->batch_size(),
_schema->is_partial_update() &&
!_schema->auto_increment_coulumn().empty(),
diff --git
a/regression-test/data/data_model_p0/unique/test_unique_table_auto_inc.out
b/regression-test/data/data_model_p0/unique/test_unique_table_auto_inc.out
index 800e97b1daf..c6910870c09 100644
--- a/regression-test/data/data_model_p0/unique/test_unique_table_auto_inc.out
+++ b/regression-test/data/data_model_p0/unique/test_unique_table_auto_inc.out
@@ -217,3 +217,65 @@ NNereids 9998
3 test1 test2
4 test3 test4
+-- !sql --
+1 A
+2 B
+3 C
+
+-- !sql --
+6 6
+
+-- !sql --
+A
+B
+C
+a
+b
+c
+
+-- !sql --
+12 12
+
+-- !sql --
+A
+B
+C
+Jack
+Jack
+Jack
+Jack
+Jack
+Jack
+a
+b
+c
+
+-- !ql --
+\N Alice
+\N Mick
+\N Steve
+100 John
+300 Bob
+
+-- !sql --
+17 17
+
+-- !sql --
+A
+Alice
+B
+Bob
+C
+Jack
+Jack
+Jack
+Jack
+Jack
+Jack
+John
+Mick
+Steve
+a
+b
+c
+
diff --git
a/regression-test/suites/data_model_p0/unique/test_unique_table_auto_inc.groovy
b/regression-test/suites/data_model_p0/unique/test_unique_table_auto_inc.groovy
index ee6c45b92a4..7a532aba3f4 100644
---
a/regression-test/suites/data_model_p0/unique/test_unique_table_auto_inc.groovy
+++
b/regression-test/suites/data_model_p0/unique/test_unique_table_auto_inc.groovy
@@ -425,5 +425,43 @@ suite("test_unique_table_auto_inc") {
sql """insert into ${table12} select r_regionkey, "test3", "test4" from
${table12} where r_regionkey=4;"""
qt_sql "select * from ${table12} order by r_regionkey;"
sql "drop table if exists ${table12};"
+
+
+ def table13 = "test_unique_tab_auto_inc_col_insert_select3"
+ sql "drop table if exists ${table13}"
+ sql """ CREATE TABLE ${table13} (
+ mykey BIGINT NOT NULL AUTO_INCREMENT,
+ name VARCHAR(64)
+ ) UNIQUE KEY(mykey)
+ DISTRIBUTED BY HASH(mykey)
+ BUCKETS AUTO PROPERTIES("replication_num" = "1"); """
+ def table14 = "test_unique_tab_auto_inc_col_insert_select4"
+ sql "drop table if exists ${table14}"
+ sql """ CREATE TABLE ${table14} (
+ mykey BIGINT NULL,
+ name VARCHAR(64)
+ ) DUPLICATE KEY(mykey)
+ DISTRIBUTED BY HASH(mykey)
+ BUCKETS AUTO PROPERTIES("replication_num" = "1"); """
+
+ sql """insert into ${table13}(name) values("A"), ("B"), ("C")"""
+ qt_sql "select * from ${table13} order by mykey;"
+ sql """insert into ${table13}(name) select lower(name) from ${table13};"""
+ qt_sql """ select count(mykey), count(distinct mykey) from ${table13}"""
+ qt_sql "select name from ${table13} order by name;"
+ sql """ insert into ${table13}(name) select "Jack" as name from
${table13};"""
+ qt_sql """ select count(mykey), count(distinct mykey) from ${table13}"""
+ qt_sql "select name from ${table13} order by name;"
+
+ sql """ insert into ${table14} values
+ (100,"John"), (null, "Mick"), (300, "Bob"), (null, "Alice"), (null,
"Steve");"""
+ qt_ql "select * from ${table14} order by mykey, name;"
+ sql """ insert into ${table13} select mykey, name from ${table14};"""
+ qt_sql """ select count(mykey), count(distinct mykey) from ${table13}"""
+ qt_sql "select name from ${table13} order by name;"
+
+ sql "drop table if exists ${table13};"
+ sql "drop table if exists ${table14};"
+
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@doris.apache.org
For additional commands, e-mail: commits-h...@doris.apache.org
