You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by yi...@apache.org on 2024/02/29 20:26:02 UTC
(doris) branch branch-2.1 updated (0aa7108ee25 -> 0d77fc48474)
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a change to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
from 0aa7108ee25 [fix](merge-on-write) incorrect result caused by key range filter with pk (#31456)
new 819ab6fc002 [feature](sink) support paritition tablet sink shuffle (#30914)
new 82faa7469bd Support analyze rollup. (#31576)
new 1f825ee2d68 [improve](export) Support partition data consistency (#31290)
new ae926d0d8c1 [pipelineX](bug) Fix core dump if union has no children (#31611)
new 2047b9416f5 [Fix](nereids) Only rewrite the slots that appear both in trival-agg func and grouping sets (#31600)
new 336f82e4267 [fix](test)Fix mtmv switch is false in incremtn case (#31584)
new 07703633dd2 [pipelineX](conf) enable pipelineX by default (#29894)
new 05f525e922b [improvement](docs)add faq for flink-connector-doris (#31426)
new cb0ab6886bb Test the unique model by modify a value type from BOOLEAN to other type (#31589)
new 6b2af74a63f [doc](fix) add k8s operator doc for upgrade doris (#31586)
new 1398f112461 [improvement](group_commit) Empty wal should be deleted when replaying it (#31427)
new b613de3e6d8 [fix](regression) set rf type to 4 to forbid fuzzy rf_type in shape check #31567
new d8b9909675f [Fix](Status) Handle returned Status correctly #31434
new 0d77fc48474 [nereids] explode function stats configurable (#31332)
The 14 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/agent/task_worker_pool.cpp | 2 +-
be/src/exec/odbc_connector.cpp | 2 +-
be/src/exprs/runtime_filter.cpp | 14 +-
be/src/exprs/runtime_filter.h | 2 +-
be/src/exprs/runtime_filter_slots.h | 2 +-
be/src/http/action/restore_tablet_action.cpp | 3 +-
be/src/http/action/tablet_migration_action.cpp | 8 +-
be/src/index-tools/index_tool.cpp | 7 +-
be/src/olap/tablet_manager.cpp | 3 +-
be/src/olap/tablet_manager.h | 2 +-
be/src/olap/wal/wal_reader.cpp | 3 +
be/src/olap/wal/wal_table.cpp | 2 +-
be/src/pipeline/exec/analytic_source_operator.cpp | 7 +-
be/src/pipeline/exec/analytic_source_operator.h | 2 +-
be/src/pipeline/exec/result_file_sink_operator.cpp | 4 +-
be/src/pipeline/exec/result_sink_operator.cpp | 4 +-
be/src/pipeline/exec/union_source_operator.cpp | 3 +-
be/src/pipeline/pipeline_x/operator.cpp | 2 +-
.../pipeline_x/pipeline_x_fragment_context.cpp | 4 +-
be/src/runtime/result_buffer_mgr.cpp | 5 +-
be/src/runtime/result_buffer_mgr.h | 2 +-
be/src/vec/exec/vanalytic_eval_node.cpp | 15 +-
be/src/vec/exec/vanalytic_eval_node.h | 2 +-
be/src/vec/exec/vsort_node.cpp | 2 +-
be/src/vec/exec/vunion_node.cpp | 2 +-
be/src/vec/functions/function_rpc.cpp | 10 +-
be/src/vec/functions/function_rpc.h | 6 +-
be/src/vec/olap/vertical_merge_iterator.cpp | 4 +-
be/src/vec/olap/vgeneric_iterators.cpp | 4 +-
be/src/vec/sink/async_writer_sink.h | 2 +-
be/src/vec/sink/autoinc_buffer.cpp | 10 +-
be/src/vec/sink/autoinc_buffer.h | 2 +-
be/src/vec/sink/vdata_stream_sender.cpp | 104 +++++++-
be/src/vec/sink/vdata_stream_sender.h | 41 ++-
be/src/vec/sink/vresult_file_sink.cpp | 6 +-
be/src/vec/sink/vresult_sink.cpp | 6 +-
be/src/vec/sink/vrow_distribution.h | 3 +
be/src/vec/sink/writer/async_result_writer.cpp | 8 +-
be/src/vec/sink/writer/async_result_writer.h | 2 +-
be/src/vec/sink/writer/vfile_result_writer.cpp | 5 +-
be/src/vec/sink/writer/vfile_result_writer.h | 2 +-
docs/en/docs/ecosystem/flink-doris-connector.md | 4 +
.../install/k8s-deploy/doris-cluster-upgrade.md | 286 ++++++++++++++++++++
docs/sidebars.json | 1 +
docs/zh-CN/docs/ecosystem/flink-doris-connector.md | 4 +
.../install/k8s-deploy/doris-cluster-upgrade.md | 291 +++++++++++++++++++++
.../java/org/apache/doris/analysis/ExportStmt.java | 15 ++
.../java/org/apache/doris/catalog/OlapTable.java | 8 +-
.../main/java/org/apache/doris/load/ExportJob.java | 87 ++++--
.../main/java/org/apache/doris/load/ExportMgr.java | 1 +
.../org/apache/doris/load/ExportTaskExecutor.java | 5 +-
.../glue/translator/PhysicalPlanTranslator.java | 3 +
...y.java => DistributionSpecTabletIdShuffle.java} | 10 +-
.../nereids/properties/PhysicalProperties.java | 3 +
.../nereids/rules/analysis/NormalizeRepeat.java | 84 ++++--
.../doris/nereids/stats/StatsCalculator.java | 3 +-
.../trees/plans/commands/ExportCommand.java | 13 +
.../trees/plans/commands/InsertExecutor.java | 26 +-
.../plans/commands/InsertIntoTableCommand.java | 15 +-
.../plans/physical/PhysicalOlapTableSink.java | 21 +-
.../org/apache/doris/planner/DataPartition.java | 34 +--
.../org/apache/doris/planner/DataStreamSink.java | 46 ++++
.../org/apache/doris/planner/OlapTableSink.java | 22 +-
.../java/org/apache/doris/qe/SessionVariable.java | 44 +++-
.../java/org/apache/doris/qe/ShowExecutor.java | 7 +-
.../apache/doris/statistics/AnalysisManager.java | 10 +-
.../statistics/StatisticsAutoCollectorTest.java | 14 +
gensrc/thrift/DataSinks.thrift | 7 +-
..._basic.out => test_export_data_consistency.out} | 263 -------------------
...ot_both_appear_in_agg_fun_and_grouping_sets.out | 61 +++++
....groovy => test_export_data_consistency.groovy} | 131 +++++-----
.../external_table_p0/hive/test_hive_other.groovy | 2 +-
...both_appear_in_agg_fun_and_grouping_sets.groovy | 30 +++
.../cross_join_list_str_increment_create.groovy | 2 +-
.../cross_join_range_date_increment_create.groovy | 2 +-
...cross_join_range_number_increment_create.groovy | 2 +-
.../full_join_list_str_increment_create.groovy | 2 +-
.../full_join_range_date_increment_create.groovy | 2 +-
.../full_join_range_number_increment_create.groovy | 2 +-
.../inner_join_list_str_increment_create.groovy | 2 +-
.../inner_join_range_date_increment_create.groovy | 2 +-
...inner_join_range_number_increment_create.groovy | 2 +-
...left_anti_join_list_str_increment_create.groovy | 2 +-
...ft_anti_join_range_date_increment_create.groovy | 2 +-
..._anti_join_range_number_increment_create.groovy | 2 +-
.../left_join_list_str_increment_create.groovy | 2 +-
.../left_join_range_date_increment_create.groovy | 2 +-
.../left_join_range_number_increment_create.groovy | 2 +-
...left_semi_join_list_str_increment_create.groovy | 2 +-
...ft_semi_join_range_date_increment_create.groovy | 2 +-
..._semi_join_range_number_increment_create.groovy | 2 +-
...ight_anti_join_list_str_increment_create.groovy | 2 +-
...ht_anti_join_range_date_increment_create.groovy | 2 +-
..._anti_join_range_number_increment_create.groovy | 2 +-
.../right_join_list_str_increment_create.groovy | 2 +-
.../right_join_range_date_increment_create.groovy | 2 +-
...right_join_range_number_increment_create.groovy | 2 +-
...ight_semi_join_list_str_increment_create.groovy | 2 +-
...ht_semi_join_range_date_increment_create.groovy | 2 +-
..._semi_join_range_number_increment_create.groovy | 2 +-
.../self_conn_list_str_increment_create.groovy | 2 +-
.../self_conn_range_date_increment_create.groovy | 2 +-
.../self_conn_range_number_increment_create.groovy | 2 +-
.../suites/nereids_shape_check/load.groovy | 2 +-
.../test_unique_model_schema_value_change.groovy | 151 ++++++++++-
.../suites/statistics/test_analyze_mv.groovy | 64 ++++-
106 files changed, 1583 insertions(+), 564 deletions(-)
create mode 100644 docs/en/docs/install/k8s-deploy/doris-cluster-upgrade.md
create mode 100644 docs/zh-CN/docs/install/k8s-deploy/doris-cluster-upgrade.md
copy fe/fe-core/src/main/java/org/apache/doris/nereids/properties/{DistributionSpecAny.java => DistributionSpecTabletIdShuffle.java} (73%)
copy regression-test/data/export_p0/{test_export_basic.out => test_export_data_consistency.out} (60%)
copy regression-test/suites/export_p0/{test_export_orc.groovy => test_export_data_consistency.groovy} (60%)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 12/14: [fix](regression) set rf type to 4 to forbid fuzzy rf_type in shape check #31567
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit b613de3e6d880283e1ccd1bbe9fa0318f4cbd8bb
Author: minghong <en...@gmail.com>
AuthorDate: Thu Feb 29 23:36:08 2024 +0800
[fix](regression) set rf type to 4 to forbid fuzzy rf_type in shape check #31567
---
regression-test/suites/nereids_shape_check/load.groovy | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/regression-test/suites/nereids_shape_check/load.groovy b/regression-test/suites/nereids_shape_check/load.groovy
index 9ce7bd030d5..db6bcdebcb1 100644
--- a/regression-test/suites/nereids_shape_check/load.groovy
+++ b/regression-test/suites/nereids_shape_check/load.groovy
@@ -53,7 +53,7 @@ suite("broadcastJoin") {
sql "set enable_nereids_planner=true"
sql "set forbid_unknown_col_stats=true"
sql "set enable_fallback_to_original_planner=false"
-
+ sql "set runtime_filter_type=4"
sql """
alter table t1 modify column code set stats('row_count'='7846', 'ndv'='10000', 'num_nulls'='0', 'min_value'='999999', 'max_value'='999999', 'data_size'='5.7');
"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 09/14: Test the unique model by modify a value type from BOOLEAN to other type (#31589)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit cb0ab6886bbc454eecb6c2d5da6c3fe6739c9ae2
Author: kkop <45...@users.noreply.github.com>
AuthorDate: Thu Feb 29 23:32:36 2024 +0800
Test the unique model by modify a value type from BOOLEAN to other type (#31589)
---
.../test_unique_model_schema_value_change.groovy | 151 ++++++++++++++++++++-
1 file changed, 147 insertions(+), 4 deletions(-)
diff --git a/regression-test/suites/schema_change_p0/test_unique_model_schema_value_change.groovy b/regression-test/suites/schema_change_p0/test_unique_model_schema_value_change.groovy
index 12f69a3b8b2..a8ef1265ab0 100644
--- a/regression-test/suites/schema_change_p0/test_unique_model_schema_value_change.groovy
+++ b/regression-test/suites/schema_change_p0/test_unique_model_schema_value_change.groovy
@@ -129,7 +129,7 @@ suite("test_unique_model_schema_value_change","p0") {
- //Test the unique model by adding a value column with FLOAT
+ //TODO Test the unique model by adding a value column with FLOAT
//java.sql.SQLException: errCode = 2, detailMessage = Default value will loose precision: 166.68f
/* sql initTable
sql initTableData
@@ -143,7 +143,7 @@ suite("test_unique_model_schema_value_change","p0") {
- //Test the unique model by adding a value column with DOUBLE
+ //TODO Test the unique model by adding a value column with DOUBLE
//java.sql.SQLException: errCode = 2, detailMessage = Default value will loose precision: 166.689
/* sql initTable
sql initTableData
@@ -218,7 +218,7 @@ suite("test_unique_model_schema_value_change","p0") {
time 60
}, insertSql, true,"${tbName}")
- //Test the unique model by adding a value column with HLL
+ //TODO Test the unique model by adding a value column with HLL
//java.sql.SQLException: errCode = 2, detailMessage = Can not assign aggregation method on column in Unique data model table: comment
/* sql initTable
sql initTableData
@@ -229,7 +229,7 @@ suite("test_unique_model_schema_value_change","p0") {
time 60
}, insertSql, true,"${tbName}")*/
- //Test the unique model by adding a value column with bitmap
+ //TODO Test the unique model by adding a value column with bitmap
//java.sql.SQLException: errCode = 2, detailMessage = Can not assign aggregation method on column in Unique data model table: device_id
/*
sql initTable
@@ -268,4 +268,147 @@ suite("test_unique_model_schema_value_change","p0") {
}, insertSql, true,"${tbName}")
+ /**
+ * Test the unique model by modify a value type
+ */
+
+
+ initTable = " CREATE TABLE IF NOT EXISTS ${tbName}\n" +
+ " (\n" +
+ " `user_id` LARGEINT NOT NULL COMMENT \"用户id\",\n" +
+ " `username` VARCHAR(50) NOT NULL COMMENT \"用户昵称\",\n" +
+ " `is_teacher` BOOLEAN COMMENT \"是否是老师\",\n" +
+ " `city` VARCHAR(20) COMMENT \"用户所在城市\",\n" +
+ " `age` SMALLINT COMMENT \"用户年龄\",\n" +
+ " `sex` TINYINT COMMENT \"用户性别\",\n" +
+ " `phone` LARGEINT COMMENT \"用户电话\",\n" +
+ " `address` VARCHAR(500) COMMENT \"用户地址\",\n" +
+ " `register_time` DATETIME COMMENT \"用户注册时间\"\n" +
+ " )\n" +
+ " UNIQUE KEY(`user_id`, `username`)\n" +
+ " DISTRIBUTED BY HASH(`user_id`) BUCKETS 1\n" +
+ " PROPERTIES (\n" +
+ " \"replication_allocation\" = \"tag.location.default: 1\",\n" +
+ " \"enable_unique_key_merge_on_write\" = \"true\"\n" +
+ " );"
+
+ initTableData = "insert into ${tbName} values(123456789, 'Alice', 0, 'Beijing', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00')," +
+ " (234567890, 'Bob', 0, 'Shanghai', 30, 1, 13998765432, 'No. 456 Street, Shanghai', '2022-02-02 12:00:00')," +
+ " (345678901, 'Carol', 1, 'Guangzhou', 28, 0, 13724681357, 'No. 789 Street, Guangzhou', '2022-03-03 14:00:00')," +
+ " (456789012, 'Dave', 0, 'Shenzhen', 35, 1, 13680864279, 'No. 987 Street, Shenzhen', '2022-04-04 16:00:00')," +
+ " (567890123, 'Eve', 0, 'Chengdu', 27, 0, 13572468091, 'No. 654 Street, Chengdu', '2022-05-05 18:00:00')," +
+ " (678901234, 'Frank', 1, 'Hangzhou', 32, 1, 13467985213, 'No. 321 Street, Hangzhou', '2022-06-06 20:00:00')," +
+ " (789012345, 'Grace', 0, 'Xian', 29, 0, 13333333333, 'No. 222 Street, Xian', '2022-07-07 22:00:00');"
+
+ //Test the unique model by modify a value type from BOOLEAN to TINYINT
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = Can not change BOOLEAN to TINYINT
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher TINYINT DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', 1, 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
+
+ //Test the unique model by modify a value type from BOOLEAN to SMALLINT
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = Can not change BOOLEAN to SMALLINT
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher SMALLINT DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', 1, 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
+
+ //Test the unique model by modify a value type from BOOLEAN to INT
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = Can not change BOOLEAN to INT
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher INT DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', 1, 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
+
+
+ //Test the unique model by modify a value type from BOOLEAN to BIGINT
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = Can not change BOOLEAN to BIGINT
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher BIGINT DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', 1, 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
+
+
+ //Test the unique model by modify a value type from BOOLEAN to FLOAT
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = Can not change BOOLEAN to FLOAT
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher FLOAT DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', 1.0, 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
+
+ //Test the unique model by modify a value type from BOOLEAN to DOUBLE
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = Can not change BOOLEAN to DOUBLE
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher DOUBLE DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', 1.0, 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
+
+ //Test the unique model by modify a value type from BOOLEAN to DECIMAL
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = Can not change BOOLEAN to DECIMAL32
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher DECIMAL DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', 1.0, 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
+
+ //Test the unique model by modify a value type from BOOLEAN to CHAR
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = Can not change BOOLEAN to CHAR
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher CHAR DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', '1', 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
+
+ //Test the unique model by modify a value type from BOOLEAN to STRING
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = Can not change BOOLEAN to STRING
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher STRING DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', '1', 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
+
+ //Test the unique model by modify a value type from BOOLEAN to VARCHAR
+ //TODO java.sql.SQLException: errCode = 2, detailMessage = String Type should not be used in key column[is_teacher].
+/* sql initTable
+ sql initTableData
+ sql """ alter table ${tbName} MODIFY column is_teacher VARCHAR(32) DEFAULT "0" """
+ insertSql = "insert into ${tbName} values(123456689, 'Alice', '1', 'Yaan', 25, 0, 13812345678, 'No. 123 Street, Beijing', '2022-01-01 10:00:00'); "
+ waitForSchemaChangeDone({
+ sql getTableStatusSql
+ time 60
+ }, insertSql, true,"${tbName}")*/
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 13/14: [Fix](Status) Handle returned Status correctly #31434
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit d8b9909675f18b1013749d9e67488a4f60b12ef3
Author: Uniqueyou <13...@users.noreply.github.com>
AuthorDate: Thu Feb 29 23:47:33 2024 +0800
[Fix](Status) Handle returned Status correctly #31434
---
be/src/agent/task_worker_pool.cpp | 2 +-
be/src/exec/odbc_connector.cpp | 2 +-
be/src/exprs/runtime_filter.cpp | 14 ++++++++------
be/src/exprs/runtime_filter.h | 2 +-
be/src/exprs/runtime_filter_slots.h | 2 +-
be/src/http/action/restore_tablet_action.cpp | 3 +--
be/src/http/action/tablet_migration_action.cpp | 8 ++++----
be/src/index-tools/index_tool.cpp | 7 ++++++-
be/src/olap/tablet_manager.cpp | 3 +--
be/src/olap/tablet_manager.h | 2 +-
be/src/pipeline/exec/analytic_source_operator.cpp | 7 +++----
be/src/pipeline/exec/analytic_source_operator.h | 2 +-
be/src/pipeline/exec/result_file_sink_operator.cpp | 4 ++--
be/src/pipeline/exec/result_sink_operator.cpp | 4 ++--
be/src/pipeline/pipeline_x/operator.cpp | 2 +-
be/src/runtime/result_buffer_mgr.cpp | 5 ++---
be/src/runtime/result_buffer_mgr.h | 2 +-
be/src/vec/exec/vanalytic_eval_node.cpp | 15 +++++++--------
be/src/vec/exec/vanalytic_eval_node.h | 2 +-
be/src/vec/exec/vsort_node.cpp | 2 +-
be/src/vec/exec/vunion_node.cpp | 2 +-
be/src/vec/functions/function_rpc.cpp | 10 ++++++----
be/src/vec/functions/function_rpc.h | 6 +++---
be/src/vec/olap/vertical_merge_iterator.cpp | 4 ++--
be/src/vec/olap/vgeneric_iterators.cpp | 4 ++--
be/src/vec/sink/async_writer_sink.h | 2 +-
be/src/vec/sink/autoinc_buffer.cpp | 10 ++++++----
be/src/vec/sink/autoinc_buffer.h | 2 +-
be/src/vec/sink/vdata_stream_sender.cpp | 2 +-
be/src/vec/sink/vresult_file_sink.cpp | 6 +++---
be/src/vec/sink/vresult_sink.cpp | 6 +++---
be/src/vec/sink/writer/async_result_writer.cpp | 8 +++++---
be/src/vec/sink/writer/async_result_writer.h | 2 +-
be/src/vec/sink/writer/vfile_result_writer.cpp | 5 ++---
be/src/vec/sink/writer/vfile_result_writer.h | 2 +-
35 files changed, 84 insertions(+), 77 deletions(-)
diff --git a/be/src/agent/task_worker_pool.cpp b/be/src/agent/task_worker_pool.cpp
index 194a74b9a32..00c16cfa12d 100644
--- a/be/src/agent/task_worker_pool.cpp
+++ b/be/src/agent/task_worker_pool.cpp
@@ -927,7 +927,7 @@ void report_tablet_callback(StorageEngine& engine, const TMasterInfo& master_inf
request.__isset.tablets = true;
uint64_t report_version = s_report_version;
- static_cast<void>(engine.tablet_manager()->build_all_report_tablets_info(&request.tablets));
+ engine.tablet_manager()->build_all_report_tablets_info(&request.tablets);
if (report_version < s_report_version) {
// TODO llj This can only reduce the possibility for report error, but can't avoid it.
// If FE create a tablet in FE meta and send CREATE task to this BE, the tablet may not be included in this
diff --git a/be/src/exec/odbc_connector.cpp b/be/src/exec/odbc_connector.cpp
index e25d6ff62f2..0a0f240cec7 100644
--- a/be/src/exec/odbc_connector.cpp
+++ b/be/src/exec/odbc_connector.cpp
@@ -151,7 +151,7 @@ Status ODBCConnector::open(RuntimeState* state, bool read) {
LOG(INFO) << "connect success:" << _connect_string.substr(0, _connect_string.find("Pwd="));
_is_open = true;
- static_cast<void>(begin_trans());
+ RETURN_IF_ERROR(begin_trans());
return Status::OK();
}
diff --git a/be/src/exprs/runtime_filter.cpp b/be/src/exprs/runtime_filter.cpp
index 786876cc796..24c41613be4 100644
--- a/be/src/exprs/runtime_filter.cpp
+++ b/be/src/exprs/runtime_filter.cpp
@@ -331,7 +331,7 @@ public:
return Status::OK();
}
- void change_to_bloom_filter(bool need_init_bf = false) {
+ Status change_to_bloom_filter(bool need_init_bf = false) {
CHECK(_filter_type == RuntimeFilterType::IN_OR_BLOOM_FILTER)
<< "Can not change to bloom filter because of runtime filter type is "
<< IRuntimeFilter::to_string(_filter_type);
@@ -339,11 +339,12 @@ public:
BloomFilterFuncBase* bf = _context.bloom_filter_func.get();
if (need_init_bf) {
// BloomFilter may be not init
- static_cast<void>(bf->init_with_fixed_length());
+ RETURN_IF_ERROR(bf->init_with_fixed_length());
insert_to_bloom_filter(bf);
}
// release in filter
_context.hybrid_set.reset();
+ return Status::OK();
}
Status init_bloom_filter(const size_t build_bf_cardinality) {
@@ -508,12 +509,12 @@ public:
VLOG_DEBUG << " change runtime filter to bloom filter(id=" << _filter_id
<< ") because: in_num(" << _context.hybrid_set->size()
<< ") >= max_in_num(" << _max_in_num << ")";
- change_to_bloom_filter(true);
+ RETURN_IF_ERROR(change_to_bloom_filter(true));
}
} else {
VLOG_DEBUG << " change runtime filter to bloom filter(id=" << _filter_id
<< ") because: already exist a bloom filter";
- change_to_bloom_filter();
+ RETURN_IF_ERROR(change_to_bloom_filter());
RETURN_IF_ERROR(_context.bloom_filter_func->merge(
wrapper->_context.bloom_filter_func.get()));
}
@@ -1307,8 +1308,9 @@ Status IRuntimeFilter::create_wrapper(const UpdateRuntimeFilterParamsV2* param,
}
}
-void IRuntimeFilter::change_to_bloom_filter() {
- _wrapper->change_to_bloom_filter();
+Status IRuntimeFilter::change_to_bloom_filter() {
+ RETURN_IF_ERROR(_wrapper->change_to_bloom_filter());
+ return Status::OK();
}
Status IRuntimeFilter::init_bloom_filter(const size_t build_bf_cardinality) {
diff --git a/be/src/exprs/runtime_filter.h b/be/src/exprs/runtime_filter.h
index 74b2580a4e6..91456cccced 100644
--- a/be/src/exprs/runtime_filter.h
+++ b/be/src/exprs/runtime_filter.h
@@ -278,7 +278,7 @@ public:
static Status create_wrapper(const UpdateRuntimeFilterParamsV2* param,
RuntimePredicateWrapper** wrapper);
- void change_to_bloom_filter();
+ Status change_to_bloom_filter();
Status init_bloom_filter(const size_t build_bf_cardinality);
Status update_filter(const UpdateRuntimeFilterParams* param);
void update_filter(RuntimePredicateWrapper* filter_wrapper, int64_t merge_time,
diff --git a/be/src/exprs/runtime_filter_slots.h b/be/src/exprs/runtime_filter_slots.h
index c9c1a996064..1fadf81e809 100644
--- a/be/src/exprs/runtime_filter_slots.h
+++ b/be/src/exprs/runtime_filter_slots.h
@@ -105,7 +105,7 @@ public:
if (over_max_in_num &&
runtime_filter->type() == RuntimeFilterType::IN_OR_BLOOM_FILTER) {
- runtime_filter->change_to_bloom_filter();
+ RETURN_IF_ERROR(runtime_filter->change_to_bloom_filter());
}
if (runtime_filter->is_bloomfilter()) {
diff --git a/be/src/http/action/restore_tablet_action.cpp b/be/src/http/action/restore_tablet_action.cpp
index d4161340406..a2c2ca05c19 100644
--- a/be/src/http/action/restore_tablet_action.cpp
+++ b/be/src/http/action/restore_tablet_action.cpp
@@ -183,8 +183,7 @@ Status RestoreTabletAction::_restore(const std::string& key, int64_t tablet_id,
Status s = _create_hard_link_recursive(latest_tablet_path, restore_schema_hash_path);
if (!s.ok()) {
// do not check the status of delete_directory, return status of link operation
- static_cast<void>(
- io::global_local_filesystem()->delete_directory(restore_schema_hash_path));
+ RETURN_IF_ERROR(io::global_local_filesystem()->delete_directory(restore_schema_hash_path));
return s;
}
std::string restore_shard_path = store->get_absolute_shard_path(tablet_meta.shard_id());
diff --git a/be/src/http/action/tablet_migration_action.cpp b/be/src/http/action/tablet_migration_action.cpp
index df75403497a..241f63c61a0 100644
--- a/be/src/http/action/tablet_migration_action.cpp
+++ b/be/src/http/action/tablet_migration_action.cpp
@@ -40,10 +40,10 @@ const static std::string HEADER_JSON = "application/json";
void TabletMigrationAction::_init_migration_action() {
int32_t max_thread_num = config::max_tablet_migration_threads;
int32_t min_thread_num = config::min_tablet_migration_threads;
- static_cast<void>(ThreadPoolBuilder("MigrationTaskThreadPool")
- .set_min_threads(min_thread_num)
- .set_max_threads(max_thread_num)
- .build(&_migration_thread_pool));
+ THROW_IF_ERROR(ThreadPoolBuilder("MigrationTaskThreadPool")
+ .set_min_threads(min_thread_num)
+ .set_max_threads(max_thread_num)
+ .build(&_migration_thread_pool));
}
void TabletMigrationAction::handle(HttpRequest* req) {
diff --git a/be/src/index-tools/index_tool.cpp b/be/src/index-tools/index_tool.cpp
index ade72ae6809..91beb905ae2 100644
--- a/be/src/index-tools/index_tool.cpp
+++ b/be/src/index-tools/index_tool.cpp
@@ -265,7 +265,12 @@ int main(int argc, char** argv) {
std::vector<FileInfo> files;
bool exists = false;
std::filesystem::path root_dir(FLAGS_directory);
- static_cast<void>(fs->list(root_dir, true, &files, &exists));
+ doris::Status status = fs->list(root_dir, true, &files, &exists);
+ if (!status.ok) {
+ std::cerr << "can't search from directory's all files,err : " << status
+ << std::endl;
+ return -1;
+ }
if (!exists) {
std::cerr << FLAGS_directory << " is not exists" << std::endl;
return -1;
diff --git a/be/src/olap/tablet_manager.cpp b/be/src/olap/tablet_manager.cpp
index 52fdec5b14b..6751225d86e 100644
--- a/be/src/olap/tablet_manager.cpp
+++ b/be/src/olap/tablet_manager.cpp
@@ -994,7 +994,7 @@ Status TabletManager::report_tablet_info(TTabletInfo* tablet_info) {
return res;
}
-Status TabletManager::build_all_report_tablets_info(std::map<TTabletId, TTablet>* tablets_info) {
+void TabletManager::build_all_report_tablets_info(std::map<TTabletId, TTablet>* tablets_info) {
DCHECK(tablets_info != nullptr);
VLOG_NOTICE << "begin to build all report tablets info";
@@ -1033,7 +1033,6 @@ Status TabletManager::build_all_report_tablets_info(std::map<TTabletId, TTablet>
DorisMetrics::instance()->tablet_version_num_distribution->set_histogram(
tablet_version_num_hist);
LOG(INFO) << "success to build all report tablets info. tablet_count=" << tablets_info->size();
- return Status::OK();
}
Status TabletManager::start_trash_sweep() {
diff --git a/be/src/olap/tablet_manager.h b/be/src/olap/tablet_manager.h
index 04f79b4f0f0..178ba50d9eb 100644
--- a/be/src/olap/tablet_manager.h
+++ b/be/src/olap/tablet_manager.h
@@ -132,7 +132,7 @@ public:
// Status::Error<INVALID_ARGUMENT>(), if tables is null
Status report_tablet_info(TTabletInfo* tablet_info);
- Status build_all_report_tablets_info(std::map<TTabletId, TTablet>* tablets_info);
+ void build_all_report_tablets_info(std::map<TTabletId, TTablet>* tablets_info);
Status start_trash_sweep();
diff --git a/be/src/pipeline/exec/analytic_source_operator.cpp b/be/src/pipeline/exec/analytic_source_operator.cpp
index 2f0f827d088..cb98b2e6466 100644
--- a/be/src/pipeline/exec/analytic_source_operator.cpp
+++ b/be/src/pipeline/exec/analytic_source_operator.cpp
@@ -240,13 +240,12 @@ Status AnalyticLocalState::init(RuntimeState* state, LocalStateInfo& info) {
return Status::OK();
}
-Status AnalyticLocalState::_reset_agg_status() {
+void AnalyticLocalState::_reset_agg_status() {
for (size_t i = 0; i < _agg_functions_size; ++i) {
_agg_functions[i]->reset(
_fn_place_ptr +
_parent->cast<AnalyticSourceOperatorX>()._offsets_of_aggregate_states[i]);
}
- return Status::OK();
}
Status AnalyticLocalState::_create_agg_status() {
@@ -359,7 +358,7 @@ Status AnalyticLocalState::_get_next_for_rows(size_t current_block_rows) {
range_end = _shared_state->current_row_position +
1; //going on calculate,add up data, no need to reset state
} else {
- static_cast<void>(_reset_agg_status());
+ _reset_agg_status();
if (!_parent->cast<AnalyticSourceOperatorX>()
._window.__isset
.window_start) { //[preceding, offset] --unbound: [preceding, following]
@@ -437,7 +436,7 @@ bool AnalyticLocalState::init_next_partition(vectorized::BlockRowPos found_parti
_partition_by_start = _shared_state->partition_by_end;
_shared_state->partition_by_end = found_partition_end;
_shared_state->current_row_position = _partition_by_start.pos;
- static_cast<void>(_reset_agg_status());
+ _reset_agg_status();
return true;
}
return false;
diff --git a/be/src/pipeline/exec/analytic_source_operator.h b/be/src/pipeline/exec/analytic_source_operator.h
index b2ab5b24b3c..eeb790ebf94 100644
--- a/be/src/pipeline/exec/analytic_source_operator.h
+++ b/be/src/pipeline/exec/analytic_source_operator.h
@@ -88,7 +88,7 @@ private:
bool need_check_first = false);
bool _whether_need_next_partition(vectorized::BlockRowPos& found_partition_end);
- Status _reset_agg_status();
+ void _reset_agg_status();
Status _create_agg_status();
Status _destroy_agg_status();
diff --git a/be/src/pipeline/exec/result_file_sink_operator.cpp b/be/src/pipeline/exec/result_file_sink_operator.cpp
index 88646ba4db3..4ed414a0774 100644
--- a/be/src/pipeline/exec/result_file_sink_operator.cpp
+++ b/be/src/pipeline/exec/result_file_sink_operator.cpp
@@ -187,9 +187,9 @@ Status ResultFileSinkLocalState::close(RuntimeState* state, Status exec_status)
_sender->update_return_rows(_writer == nullptr ? 0 : _writer->get_written_rows());
static_cast<void>(_sender->close(final_status));
}
- static_cast<void>(state->exec_env()->result_mgr()->cancel_at_time(
+ state->exec_env()->result_mgr()->cancel_at_time(
time(nullptr) + config::result_buffer_cancelled_interval_time,
- state->fragment_instance_id()));
+ state->fragment_instance_id());
} else {
if (final_status.ok()) {
bool all_receiver_eof = true;
diff --git a/be/src/pipeline/exec/result_sink_operator.cpp b/be/src/pipeline/exec/result_sink_operator.cpp
index 11a208e1392..ddaf1cd4a2b 100644
--- a/be/src/pipeline/exec/result_sink_operator.cpp
+++ b/be/src/pipeline/exec/result_sink_operator.cpp
@@ -182,9 +182,9 @@ Status ResultSinkLocalState::close(RuntimeState* state, Status exec_status) {
}
static_cast<void>(_sender->close(final_status));
}
- static_cast<void>(state->exec_env()->result_mgr()->cancel_at_time(
+ state->exec_env()->result_mgr()->cancel_at_time(
time(nullptr) + config::result_buffer_cancelled_interval_time,
- state->fragment_instance_id()));
+ state->fragment_instance_id());
RETURN_IF_ERROR(Base::close(state, exec_status));
return final_status;
}
diff --git a/be/src/pipeline/pipeline_x/operator.cpp b/be/src/pipeline/pipeline_x/operator.cpp
index 30ff16dde80..a29516a29af 100644
--- a/be/src/pipeline/pipeline_x/operator.cpp
+++ b/be/src/pipeline/pipeline_x/operator.cpp
@@ -490,7 +490,7 @@ template <typename Writer, typename Parent>
requires(std::is_base_of_v<vectorized::AsyncResultWriter, Writer>)
Status AsyncWriterSink<Writer, Parent>::open(RuntimeState* state) {
RETURN_IF_ERROR(Base::open(state));
- _writer->start_writer(state, _profile);
+ RETURN_IF_ERROR(_writer->start_writer(state, _profile));
return Status::OK();
}
diff --git a/be/src/runtime/result_buffer_mgr.cpp b/be/src/runtime/result_buffer_mgr.cpp
index a2009c5ec3c..04831fcefb2 100644
--- a/be/src/runtime/result_buffer_mgr.cpp
+++ b/be/src/runtime/result_buffer_mgr.cpp
@@ -92,7 +92,7 @@ Status ResultBufferMgr::create_sender(const TUniqueId& query_id, int buffer_size
// details see issue https://github.com/apache/doris/issues/16203
// add extra 5s for avoid corner case
int64_t max_timeout = time(nullptr) + exec_timout + 5;
- static_cast<void>(cancel_at_time(max_timeout, query_id));
+ cancel_at_time(max_timeout, query_id);
}
*sender = control_block;
return Status::OK();
@@ -173,7 +173,7 @@ Status ResultBufferMgr::cancel(const TUniqueId& query_id) {
return Status::OK();
}
-Status ResultBufferMgr::cancel_at_time(time_t cancel_time, const TUniqueId& query_id) {
+void ResultBufferMgr::cancel_at_time(time_t cancel_time, const TUniqueId& query_id) {
std::lock_guard<std::mutex> l(_timeout_lock);
TimeoutMap::iterator iter = _timeout_map.find(cancel_time);
@@ -184,7 +184,6 @@ Status ResultBufferMgr::cancel_at_time(time_t cancel_time, const TUniqueId& quer
}
iter->second.push_back(query_id);
- return Status::OK();
}
void ResultBufferMgr::cancel_thread() {
diff --git a/be/src/runtime/result_buffer_mgr.h b/be/src/runtime/result_buffer_mgr.h
index 7995496cbf9..06d13104205 100644
--- a/be/src/runtime/result_buffer_mgr.h
+++ b/be/src/runtime/result_buffer_mgr.h
@@ -74,7 +74,7 @@ public:
Status cancel(const TUniqueId& fragment_id);
// cancel one query at a future time.
- Status cancel_at_time(time_t cancel_time, const TUniqueId& query_id);
+ void cancel_at_time(time_t cancel_time, const TUniqueId& query_id);
private:
using BufferMap = std::unordered_map<TUniqueId, std::shared_ptr<BufferControlBlock>>;
diff --git a/be/src/vec/exec/vanalytic_eval_node.cpp b/be/src/vec/exec/vanalytic_eval_node.cpp
index 4491d291e2d..fbd49aa145a 100644
--- a/be/src/vec/exec/vanalytic_eval_node.cpp
+++ b/be/src/vec/exec/vanalytic_eval_node.cpp
@@ -203,7 +203,7 @@ Status VAnalyticEvalNode::prepare(RuntimeState* state) {
}
_fn_place_ptr = _agg_arena_pool->aligned_alloc(_total_size_of_aggregate_states,
_align_aggregate_states);
- RETURN_IF_CATCH_EXCEPTION(static_cast<void>(_create_agg_status()));
+ RETURN_IF_ERROR(_create_agg_status());
_executor.insert_result =
std::bind<void>(&VAnalyticEvalNode::_insert_result_info, this, std::placeholders::_1);
_executor.execute =
@@ -211,7 +211,7 @@ Status VAnalyticEvalNode::prepare(RuntimeState* state) {
std::placeholders::_2, std::placeholders::_3, std::placeholders::_4);
for (const auto& ctx : _agg_expr_ctxs) {
- static_cast<void>(VExpr::prepare(ctx, state, child(0)->row_desc()));
+ RETURN_IF_ERROR(VExpr::prepare(ctx, state, child(0)->row_desc()));
}
if (!_partition_by_eq_expr_ctxs.empty() || !_order_by_eq_expr_ctxs.empty()) {
vector<TTupleId> tuple_ids;
@@ -274,9 +274,9 @@ Status VAnalyticEvalNode::pull(doris::RuntimeState* /*state*/, vectorized::Block
return Status::OK();
}
_next_partition = _init_next_partition(_found_partition_end);
- static_cast<void>(_init_result_columns());
+ RETURN_IF_ERROR(_init_result_columns());
size_t current_block_rows = _input_blocks[_output_block_index].rows();
- static_cast<void>(_executor.get_next(current_block_rows));
+ RETURN_IF_ERROR(_executor.get_next(current_block_rows));
if (_window_end_position == current_block_rows) {
break;
}
@@ -373,7 +373,7 @@ Status VAnalyticEvalNode::_get_next_for_rows(size_t current_block_rows) {
range_end = _current_row_position +
1; //going on calculate,add up data, no need to reset state
} else {
- static_cast<void>(_reset_agg_status());
+ _reset_agg_status();
if (!_window.__isset
.window_start) { //[preceding, offset] --unbound: [preceding, following]
range_start = _partition_by_start.pos;
@@ -610,7 +610,7 @@ bool VAnalyticEvalNode::_init_next_partition(BlockRowPos found_partition_end) {
_partition_by_start = _partition_by_end;
_partition_by_end = found_partition_end;
_current_row_position = _partition_by_start.pos;
- static_cast<void>(_reset_agg_status());
+ _reset_agg_status();
return true;
}
return false;
@@ -730,11 +730,10 @@ Status VAnalyticEvalNode::_init_result_columns() {
return Status::OK();
}
-Status VAnalyticEvalNode::_reset_agg_status() {
+void VAnalyticEvalNode::_reset_agg_status() {
for (size_t i = 0; i < _agg_functions_size; ++i) {
_agg_functions[i]->reset(_fn_place_ptr + _offsets_of_aggregate_states[i]);
}
- return Status::OK();
}
Status VAnalyticEvalNode::_create_agg_status() {
diff --git a/be/src/vec/exec/vanalytic_eval_node.h b/be/src/vec/exec/vanalytic_eval_node.h
index 8c26c1ad79e..45f7ce5b1e8 100644
--- a/be/src/vec/exec/vanalytic_eval_node.h
+++ b/be/src/vec/exec/vanalytic_eval_node.h
@@ -98,7 +98,7 @@ private:
void _execute_for_win_func(int64_t partition_start, int64_t partition_end, int64_t frame_start,
int64_t frame_end);
- Status _reset_agg_status();
+ void _reset_agg_status();
Status _init_result_columns();
Status _create_agg_status();
Status _destroy_agg_status();
diff --git a/be/src/vec/exec/vsort_node.cpp b/be/src/vec/exec/vsort_node.cpp
index 4206d9b6a2f..848256a0aee 100644
--- a/be/src/vec/exec/vsort_node.cpp
+++ b/be/src/vec/exec/vsort_node.cpp
@@ -188,7 +188,7 @@ Status VSortNode::open(RuntimeState* state) {
} while (!eos);
- static_cast<void>(child(0)->close(state));
+ RETURN_IF_ERROR(child(0)->close(state));
mem_tracker()->consume(_sorter->data_size());
COUNTER_UPDATE(_sort_blocks_memory_usage, _sorter->data_size());
diff --git a/be/src/vec/exec/vunion_node.cpp b/be/src/vec/exec/vunion_node.cpp
index cc7eabc3bd2..2405373a062 100644
--- a/be/src/vec/exec/vunion_node.cpp
+++ b/be/src/vec/exec/vunion_node.cpp
@@ -276,7 +276,7 @@ Status VUnionNode::get_next(RuntimeState* state, Block* block, bool* eos) {
// The previous child needs to be closed if passthrough was enabled for it. In the non
// passthrough case, the child was already closed in the previous call to get_next().
DCHECK(is_child_passthrough(_to_close_child_idx));
- static_cast<void>(child(_to_close_child_idx)->close(state));
+ RETURN_IF_ERROR(child(_to_close_child_idx)->close(state));
_to_close_child_idx = -1;
}
diff --git a/be/src/vec/functions/function_rpc.cpp b/be/src/vec/functions/function_rpc.cpp
index a900436ffc5..ba171ffbbc9 100644
--- a/be/src/vec/functions/function_rpc.cpp
+++ b/be/src/vec/functions/function_rpc.cpp
@@ -26,6 +26,7 @@
#include <memory>
#include <utility>
+#include "common/status.h"
#include "runtime/exec_env.h"
#include "util/brpc_client_cache.h"
#include "vec/columns/column.h"
@@ -46,7 +47,7 @@ Status RPCFnImpl::vec_call(FunctionContext* context, Block& block, const ColumnN
PFunctionCallRequest request;
PFunctionCallResponse response;
request.set_function_name(_function_name);
- _convert_block_to_proto(block, arguments, input_rows_count, &request);
+ RETURN_IF_ERROR(_convert_block_to_proto(block, arguments, input_rows_count, &request));
brpc::Controller cntl;
_client->fn_call(&cntl, &request, &response, nullptr);
if (cntl.Failed()) {
@@ -65,16 +66,17 @@ Status RPCFnImpl::vec_call(FunctionContext* context, Block& block, const ColumnN
return Status::OK();
}
-void RPCFnImpl::_convert_block_to_proto(Block& block, const ColumnNumbers& arguments,
- size_t input_rows_count, PFunctionCallRequest* request) {
+Status RPCFnImpl::_convert_block_to_proto(Block& block, const ColumnNumbers& arguments,
+ size_t input_rows_count, PFunctionCallRequest* request) {
size_t row_count = std::min(block.rows(), input_rows_count);
for (size_t col_idx : arguments) {
PValues* arg = request->add_args();
ColumnWithTypeAndName& column = block.get_by_position(col_idx);
arg->set_has_null(column.column->has_null(row_count));
auto col = column.column->convert_to_full_column_if_const();
- static_cast<void>(column.type->get_serde()->write_column_to_pb(*col, *arg, 0, row_count));
+ RETURN_IF_ERROR(column.type->get_serde()->write_column_to_pb(*col, *arg, 0, row_count));
}
+ return Status::OK();
}
void RPCFnImpl::_convert_to_block(Block& block, const PValues& result, size_t pos) {
diff --git a/be/src/vec/functions/function_rpc.h b/be/src/vec/functions/function_rpc.h
index fa23e5b97cb..ae71632f974 100644
--- a/be/src/vec/functions/function_rpc.h
+++ b/be/src/vec/functions/function_rpc.h
@@ -52,9 +52,9 @@ public:
bool available() { return _client != nullptr; }
private:
- void _convert_block_to_proto(vectorized::Block& block,
- const vectorized::ColumnNumbers& arguments,
- size_t input_rows_count, PFunctionCallRequest* request);
+ Status _convert_block_to_proto(vectorized::Block& block,
+ const vectorized::ColumnNumbers& arguments,
+ size_t input_rows_count, PFunctionCallRequest* request);
void _convert_to_block(vectorized::Block& block, const PValues& result, size_t pos);
std::shared_ptr<PFunctionService_Stub> _client;
diff --git a/be/src/vec/olap/vertical_merge_iterator.cpp b/be/src/vec/olap/vertical_merge_iterator.cpp
index 5fbd1ed2c2e..49916048b5c 100644
--- a/be/src/vec/olap/vertical_merge_iterator.cpp
+++ b/be/src/vec/olap/vertical_merge_iterator.cpp
@@ -363,7 +363,7 @@ Status VerticalMergeIteratorContext::_load_next_block() {
}
for (auto it = _block_list.begin(); it != _block_list.end(); it++) {
if (it->use_count() == 1) {
- static_cast<void>(block_reset(*it));
+ RETURN_IF_ERROR(block_reset(*it));
_block = *it;
_block_list.erase(it);
break;
@@ -371,7 +371,7 @@ Status VerticalMergeIteratorContext::_load_next_block() {
}
if (_block == nullptr) {
_block = std::make_shared<Block>();
- static_cast<void>(block_reset(_block));
+ RETURN_IF_ERROR(block_reset(_block));
}
Status st = _iter->next_batch(_block.get());
if (!st.ok()) {
diff --git a/be/src/vec/olap/vgeneric_iterators.cpp b/be/src/vec/olap/vgeneric_iterators.cpp
index 0e97f1d1720..2d45c884a78 100644
--- a/be/src/vec/olap/vgeneric_iterators.cpp
+++ b/be/src/vec/olap/vgeneric_iterators.cpp
@@ -292,7 +292,7 @@ Status VMergeIteratorContext::_load_next_block() {
}
for (auto it = _block_list.begin(); it != _block_list.end(); it++) {
if (it->use_count() == 1) {
- static_cast<void>(block_reset(*it));
+ RETURN_IF_ERROR(block_reset(*it));
_block = *it;
_block_list.erase(it);
break;
@@ -300,7 +300,7 @@ Status VMergeIteratorContext::_load_next_block() {
}
if (_block == nullptr) {
_block = std::make_shared<Block>();
- static_cast<void>(block_reset(_block));
+ RETURN_IF_ERROR(block_reset(_block));
}
Status st = _iter->next_batch(_block.get());
if (!st.ok()) {
diff --git a/be/src/vec/sink/async_writer_sink.h b/be/src/vec/sink/async_writer_sink.h
index 8105ff96573..337fbcf3f99 100644
--- a/be/src/vec/sink/async_writer_sink.h
+++ b/be/src/vec/sink/async_writer_sink.h
@@ -72,7 +72,7 @@ public:
// Prepare the exprs to run.
RETURN_IF_ERROR(VExpr::open(_output_vexpr_ctxs, state));
if (state->enable_pipeline_exec()) {
- _writer->start_writer(state, _profile);
+ RETURN_IF_ERROR(_writer->start_writer(state, _profile));
} else {
RETURN_IF_ERROR(_writer->open(state, _profile));
}
diff --git a/be/src/vec/sink/autoinc_buffer.cpp b/be/src/vec/sink/autoinc_buffer.cpp
index 844d8ed8524..39613159aca 100644
--- a/be/src/vec/sink/autoinc_buffer.cpp
+++ b/be/src/vec/sink/autoinc_buffer.cpp
@@ -19,6 +19,7 @@
#include <gen_cpp/HeartbeatService_types.h>
+#include "common/status.h"
#include "runtime/client_cache.h"
#include "runtime/exec_env.h"
#include "util/runtime_profile.h"
@@ -48,7 +49,7 @@ void AutoIncIDBuffer::_wait_for_prefetching() {
Status AutoIncIDBuffer::sync_request_ids(size_t length,
std::vector<std::pair<int64_t, size_t>>* result) {
std::unique_lock<std::mutex> lock(_mutex);
- _prefetch_ids(_prefetch_size());
+ RETURN_IF_ERROR(_prefetch_ids(_prefetch_size()));
if (_front_buffer.second > 0) {
auto min_length = std::min(_front_buffer.second, length);
length -= min_length;
@@ -75,13 +76,13 @@ Status AutoIncIDBuffer::sync_request_ids(size_t length,
return Status::OK();
}
-void AutoIncIDBuffer::_prefetch_ids(size_t length) {
+Status AutoIncIDBuffer::_prefetch_ids(size_t length) {
if (_front_buffer.second > _low_water_level_mark() || _is_fetching) {
- return;
+ return Status::OK();
}
TNetworkAddress master_addr = ExecEnv::GetInstance()->master_info()->network_address;
_is_fetching = true;
- static_cast<void>(_rpc_token->submit_func([=, this]() {
+ RETURN_IF_ERROR(_rpc_token->submit_func([=, this]() {
TAutoIncrementRangeRequest request;
TAutoIncrementRangeResult result;
request.__set_db_id(_db_id);
@@ -113,6 +114,7 @@ void AutoIncIDBuffer::_prefetch_ids(size_t length) {
}
_is_fetching = false;
}));
+ return Status::OK();
}
} // namespace doris::vectorized
\ No newline at end of file
diff --git a/be/src/vec/sink/autoinc_buffer.h b/be/src/vec/sink/autoinc_buffer.h
index 80a469596fd..3ec009b0960 100644
--- a/be/src/vec/sink/autoinc_buffer.h
+++ b/be/src/vec/sink/autoinc_buffer.h
@@ -64,7 +64,7 @@ public:
Status sync_request_ids(size_t length, std::vector<std::pair<int64_t, size_t>>* result);
private:
- void _prefetch_ids(size_t length);
+ Status _prefetch_ids(size_t length);
[[nodiscard]] size_t _prefetch_size() const {
return _batch_size * config::auto_inc_prefetch_size_ratio;
}
diff --git a/be/src/vec/sink/vdata_stream_sender.cpp b/be/src/vec/sink/vdata_stream_sender.cpp
index a7f1b4b8584..b6e4d0b962b 100644
--- a/be/src/vec/sink/vdata_stream_sender.cpp
+++ b/be/src/vec/sink/vdata_stream_sender.cpp
@@ -798,7 +798,7 @@ Status VDataStreamSender::close(RuntimeState* state, Status exec_status) {
if (_peak_memory_usage_counter) {
_peak_memory_usage_counter->set(_mem_tracker->peak_consumption());
}
- static_cast<void>(DataSink::close(state, exec_status));
+ RETURN_IF_ERROR(DataSink::close(state, exec_status));
return final_st;
}
diff --git a/be/src/vec/sink/vresult_file_sink.cpp b/be/src/vec/sink/vresult_file_sink.cpp
index 08dd881bf4b..035a76a2f7f 100644
--- a/be/src/vec/sink/vresult_file_sink.cpp
+++ b/be/src/vec/sink/vresult_file_sink.cpp
@@ -132,11 +132,11 @@ Status VResultFileSink::close(RuntimeState* state, Status exec_status) {
// close sender, this is normal path end
if (_sender) {
_sender->update_return_rows(_writer == nullptr ? 0 : _writer->get_written_rows());
- static_cast<void>(_sender->close(final_status));
+ RETURN_IF_ERROR(_sender->close(final_status));
}
- static_cast<void>(state->exec_env()->result_mgr()->cancel_at_time(
+ state->exec_env()->result_mgr()->cancel_at_time(
time(nullptr) + config::result_buffer_cancelled_interval_time,
- state->fragment_instance_id()));
+ state->fragment_instance_id());
} else {
if (final_status.ok()) {
auto st = _stream_sender->send(state, _output_block.get(), true);
diff --git a/be/src/vec/sink/vresult_sink.cpp b/be/src/vec/sink/vresult_sink.cpp
index 59bf82483c5..a7da5a6a5ff 100644
--- a/be/src/vec/sink/vresult_sink.cpp
+++ b/be/src/vec/sink/vresult_sink.cpp
@@ -170,11 +170,11 @@ Status VResultSink::close(RuntimeState* state, Status exec_status) {
if (_writer) {
_sender->update_return_rows(_writer->get_written_rows());
}
- static_cast<void>(_sender->close(final_status));
+ RETURN_IF_ERROR(_sender->close(final_status));
}
- static_cast<void>(state->exec_env()->result_mgr()->cancel_at_time(
+ state->exec_env()->result_mgr()->cancel_at_time(
time(nullptr) + config::result_buffer_cancelled_interval_time,
- state->fragment_instance_id()));
+ state->fragment_instance_id());
return DataSink::close(state, exec_status);
}
diff --git a/be/src/vec/sink/writer/async_result_writer.cpp b/be/src/vec/sink/writer/async_result_writer.cpp
index 0d88b9dedfa..fc431b6e863 100644
--- a/be/src/vec/sink/writer/async_result_writer.cpp
+++ b/be/src/vec/sink/writer/async_result_writer.cpp
@@ -17,6 +17,7 @@
#include "async_result_writer.h"
+#include "common/status.h"
#include "pipeline/pipeline_x/dependency.h"
#include "runtime/exec_env.h"
#include "runtime/fragment_mgr.h"
@@ -86,7 +87,7 @@ std::unique_ptr<Block> AsyncResultWriter::_get_block_from_queue() {
return block;
}
-void AsyncResultWriter::start_writer(RuntimeState* state, RuntimeProfile* profile) {
+Status AsyncResultWriter::start_writer(RuntimeState* state, RuntimeProfile* profile) {
// Should set to false here, to
_writer_thread_closed = false;
// This is a async thread, should lock the task ctx, to make sure runtimestate and profile
@@ -94,7 +95,7 @@ void AsyncResultWriter::start_writer(RuntimeState* state, RuntimeProfile* profil
auto task_ctx = state->get_task_execution_context();
if (state->get_query_ctx() && state->get_query_ctx()->get_non_pipe_exec_thread_pool()) {
ThreadPool* pool_ptr = state->get_query_ctx()->get_non_pipe_exec_thread_pool();
- static_cast<void>(pool_ptr->submit_func([this, state, profile, task_ctx]() {
+ RETURN_IF_ERROR(pool_ptr->submit_func([this, state, profile, task_ctx]() {
auto task_lock = task_ctx.lock();
if (task_lock == nullptr) {
_writer_thread_closed = true;
@@ -103,7 +104,7 @@ void AsyncResultWriter::start_writer(RuntimeState* state, RuntimeProfile* profil
this->process_block(state, profile);
}));
} else {
- static_cast<void>(ExecEnv::GetInstance()->fragment_mgr()->get_thread_pool()->submit_func(
+ RETURN_IF_ERROR(ExecEnv::GetInstance()->fragment_mgr()->get_thread_pool()->submit_func(
[this, state, profile, task_ctx]() {
auto task_lock = task_ctx.lock();
if (task_lock == nullptr) {
@@ -113,6 +114,7 @@ void AsyncResultWriter::start_writer(RuntimeState* state, RuntimeProfile* profil
this->process_block(state, profile);
}));
}
+ return Status::OK();
}
void AsyncResultWriter::process_block(RuntimeState* state, RuntimeProfile* profile) {
diff --git a/be/src/vec/sink/writer/async_result_writer.h b/be/src/vec/sink/writer/async_result_writer.h
index 8ed39aeb795..ac14616d333 100644
--- a/be/src/vec/sink/writer/async_result_writer.h
+++ b/be/src/vec/sink/writer/async_result_writer.h
@@ -76,7 +76,7 @@ public:
Status sink(Block* block, bool eos);
// Add the IO thread task process block() to thread pool to dispose the IO
- void start_writer(RuntimeState* state, RuntimeProfile* profile);
+ Status start_writer(RuntimeState* state, RuntimeProfile* profile);
Status get_writer_status() {
std::lock_guard l(_m);
diff --git a/be/src/vec/sink/writer/vfile_result_writer.cpp b/be/src/vec/sink/writer/vfile_result_writer.cpp
index c30a23ec219..661afa15935 100644
--- a/be/src/vec/sink/writer/vfile_result_writer.cpp
+++ b/be/src/vec/sink/writer/vfile_result_writer.cpp
@@ -199,7 +199,7 @@ Status VFileResultWriter::_get_next_file_name(std::string* file_name) {
// S3: {file_path}{fragment_instance_id}_
// BROKER: {file_path}{fragment_instance_id}_
-Status VFileResultWriter::_get_file_url(std::string* file_url) {
+void VFileResultWriter::_get_file_url(std::string* file_url) {
std::stringstream ss;
if (_storage_type == TStorageBackendType::LOCAL) {
ss << "file:///" << BackendOptions::get_localhost();
@@ -207,7 +207,6 @@ Status VFileResultWriter::_get_file_url(std::string* file_url) {
ss << _file_opts->file_path;
ss << print_id(_fragment_instance_id) << "_";
*file_url = ss.str();
- return Status::OK();
}
std::string VFileResultWriter::_file_format_to_name() {
@@ -306,7 +305,7 @@ Status VFileResultWriter::_send_result() {
row_buffer.push_bigint(_written_rows_counter->value()); // total rows
row_buffer.push_bigint(_written_data_bytes->value()); // file size
std::string file_url;
- static_cast<void>(_get_file_url(&file_url));
+ _get_file_url(&file_url);
std::stringstream ss;
ss << file_url << "*";
file_url = ss.str();
diff --git a/be/src/vec/sink/writer/vfile_result_writer.h b/be/src/vec/sink/writer/vfile_result_writer.h
index 864d0966a77..c52050ba6f1 100644
--- a/be/src/vec/sink/writer/vfile_result_writer.h
+++ b/be/src/vec/sink/writer/vfile_result_writer.h
@@ -83,7 +83,7 @@ private:
// get next export file name
Status _get_next_file_name(std::string* file_name);
Status _get_success_file_name(std::string* file_name);
- Status _get_file_url(std::string* file_url);
+ void _get_file_url(std::string* file_url);
std::string _file_format_to_name();
// close file writer, and if !done, it will create new writer for next file.
Status _close_file_writer(bool done);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 05/14: [Fix](nereids) Only rewrite the slots that appear both in trival-agg func and grouping sets (#31600)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 2047b9416f5dd78db7c33024afdcf106cdd8d3d2
Author: feiniaofeiafei <53...@users.noreply.github.com>
AuthorDate: Thu Feb 29 22:17:43 2024 +0800
[Fix](nereids) Only rewrite the slots that appear both in trival-agg func and grouping sets (#31600)
* [Fix](nereids) Only rewrite the slots that appear both in trival-agg func and grouping sets
* [Fix](nereids) Only rewrite the slots that appear both in trival-agg func and grouping sets
---------
Co-authored-by: feiniaofeiafei <mo...@selectdb.com>
---
.../nereids/rules/analysis/NormalizeRepeat.java | 84 ++++++++++++++++++----
...ot_both_appear_in_agg_fun_and_grouping_sets.out | 61 ++++++++++++++++
...both_appear_in_agg_fun_and_grouping_sets.groovy | 30 ++++++++
3 files changed, 160 insertions(+), 15 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeRepeat.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeRepeat.java
index 3c893ce4bec..8437dc40b04 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeRepeat.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeRepeat.java
@@ -29,8 +29,10 @@ import org.apache.doris.nereids.trees.expressions.OrderExpression;
import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.expressions.VirtualSlotReference;
+import org.apache.doris.nereids.trees.expressions.WindowExpression;
import org.apache.doris.nereids.trees.expressions.functions.agg.AggregateFunction;
import org.apache.doris.nereids.trees.expressions.functions.scalar.GroupingScalarFunction;
+import org.apache.doris.nereids.trees.expressions.visitor.DefaultExpressionRewriter;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.algebra.Repeat;
import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
@@ -47,6 +49,7 @@ import com.google.common.collect.Sets;
import com.google.common.collect.Sets.SetView;
import org.jetbrains.annotations.NotNull;
+import java.util.ArrayList;
import java.util.Collection;
import java.util.HashSet;
import java.util.List;
@@ -267,8 +270,11 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
private LogicalAggregate<Plan> dealSlotAppearBothInAggFuncAndGroupingSets(
@NotNull LogicalAggregate<Plan> aggregate) {
LogicalRepeat<Plan> repeat = (LogicalRepeat<Plan>) aggregate.child();
- Set<Slot> aggUsedSlots = aggregate.getOutputExpressions().stream()
- .flatMap(e -> e.<Set<AggregateFunction>>collect(AggregateFunction.class::isInstance).stream())
+
+ List<AggregateFunction> aggregateFunctions = Lists.newArrayList();
+ aggregate.getOutputExpressions().forEach(
+ o -> o.accept(PlanUtils.CollectNonWindowedAggFuncs.INSTANCE, aggregateFunctions));
+ Set<Slot> aggUsedSlots = aggregateFunctions.stream()
.flatMap(e -> e.<Set<SlotReference>>collect(SlotReference.class::isInstance).stream())
.collect(ImmutableSet.toImmutableSet());
Set<Slot> groupingSetsUsedSlot = repeat.getGroupingSets().stream()
@@ -308,20 +314,68 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
.build());
aggregate = aggregate.withChildren(ImmutableList.of(repeat));
- // modify aggregate functions' parameter slot reference to new copied slots
List<NamedExpression> newOutputExpressions = aggregate.getOutputExpressions().stream()
- .map(output -> (NamedExpression) output.rewriteDownShortCircuit(expr -> {
- if (expr instanceof AggregateFunction) {
- return expr.rewriteDownShortCircuit(e -> {
- if (e instanceof Slot && slotMapping.containsKey(e)) {
- return slotMapping.get(e).toSlot();
- }
- return e;
- });
- }
- return expr;
- })
- ).collect(Collectors.toList());
+ .map(e -> (NamedExpression) e.accept(RewriteAggFuncWithoutWindowAggFunc.INSTANCE,
+ slotMapping))
+ .collect(Collectors.toList());
return aggregate.withAggOutput(newOutputExpressions);
}
+
+ /**
+ * This class use the map(slotMapping) to rewrite all slots in trival-agg.
+ * The purpose of this class is to only rewrite the slots in trival-agg and not to rewrite the slots in window-agg.
+ */
+ private static class RewriteAggFuncWithoutWindowAggFunc
+ extends DefaultExpressionRewriter<Map<Slot, Alias>> {
+
+ private static final RewriteAggFuncWithoutWindowAggFunc
+ INSTANCE = new RewriteAggFuncWithoutWindowAggFunc();
+
+ private RewriteAggFuncWithoutWindowAggFunc() {}
+
+ @Override
+ public Expression visitAggregateFunction(AggregateFunction aggregateFunction, Map<Slot, Alias> slotMapping) {
+ return aggregateFunction.rewriteDownShortCircuit(e -> {
+ if (e instanceof Slot && slotMapping.containsKey(e)) {
+ return slotMapping.get(e).toSlot();
+ }
+ return e;
+ });
+ }
+
+ @Override
+ public Expression visitWindow(WindowExpression windowExpression, Map<Slot, Alias> slotMapping) {
+ List<Expression> newChildren = new ArrayList<>();
+ Expression function = windowExpression.getFunction();
+ Expression oldFuncChild = function.child(0);
+ boolean hasNewChildren = false;
+ if (oldFuncChild != null) {
+ Expression newFuncChild;
+ newFuncChild = function.child(0).accept(this, slotMapping);
+ hasNewChildren = (newFuncChild != oldFuncChild);
+ newChildren.add(hasNewChildren
+ ? function.withChildren(ImmutableList.of(newFuncChild)) : function);
+ } else {
+ newChildren.add(function);
+ }
+ for (Expression partitionKey : windowExpression.getPartitionKeys()) {
+ Expression newChild = partitionKey.accept(this, slotMapping);
+ if (newChild != partitionKey) {
+ hasNewChildren = true;
+ }
+ newChildren.add(newChild);
+ }
+ for (Expression orderKey : windowExpression.getOrderKeys()) {
+ Expression newChild = orderKey.accept(this, slotMapping);
+ if (newChild != orderKey) {
+ hasNewChildren = true;
+ }
+ newChildren.add(newChild);
+ }
+ if (windowExpression.getWindowFrame().isPresent()) {
+ newChildren.add(windowExpression.getWindowFrame().get());
+ }
+ return hasNewChildren ? windowExpression.withChildren(newChildren) : windowExpression;
+ }
+ }
}
diff --git a/regression-test/data/nereids_rules_p0/grouping_sets/slot_both_appear_in_agg_fun_and_grouping_sets.out b/regression-test/data/nereids_rules_p0/grouping_sets/slot_both_appear_in_agg_fun_and_grouping_sets.out
index 901226f8548..2c96648dac4 100644
--- a/regression-test/data/nereids_rules_p0/grouping_sets/slot_both_appear_in_agg_fun_and_grouping_sets.out
+++ b/regression-test/data/nereids_rules_p0/grouping_sets/slot_both_appear_in_agg_fun_and_grouping_sets.out
@@ -64,3 +64,64 @@ yeah
4
5
+-- !select6 --
+\N
+-86
+-48
+-12
+82
+89
+16054
+19196
+
+-- !select7 --
+\N
+-48
+-43
+82
+89
+35195
+
+-- !select8 --
+\N
+\N
+-86
+-86
+-48
+-12
+82
+89
+16054
+19196
+
+-- !select9 --
+\N
+\N
+\N
+-129
+-129
+-129
+-96
+-96
+-12
+164
+164
+178
+178
+16054
+19196
+35195
+35275
+
+-- !select10 --
+\N
+\N
+-172
+-172
+-96
+-24
+164
+178
+32108
+38392
+
diff --git a/regression-test/suites/nereids_rules_p0/grouping_sets/slot_both_appear_in_agg_fun_and_grouping_sets.groovy b/regression-test/suites/nereids_rules_p0/grouping_sets/slot_both_appear_in_agg_fun_and_grouping_sets.groovy
index 865ce3b5f50..bee63217a9f 100644
--- a/regression-test/suites/nereids_rules_p0/grouping_sets/slot_both_appear_in_agg_fun_and_grouping_sets.groovy
+++ b/regression-test/suites/nereids_rules_p0/grouping_sets/slot_both_appear_in_agg_fun_and_grouping_sets.groovy
@@ -60,4 +60,34 @@ suite("slot_both_appear_in_agg_fun_and_grouping_sets") {
select sum(rank() over (partition by col_text_undef_signed order by col_int_undef_signed))
as col1 from table_10_undef_undef4 group by grouping sets((col_int_undef_signed)) order by 1;
"""
+
+ qt_select6 """
+ select sum(sum(col_int_undef_signed)) over (partition by sum(col_int_undef_signed)
+ order by sum(col_int_undef_signed)) from table_10_undef_undef4 group by
+ grouping sets ((col_int_undef_signed)) order by 1;
+ """
+
+ qt_select7 """
+ select sum(sum(col_int_undef_signed)) over (partition by sum(col_int_undef_signed)
+ order by sum(col_int_undef_signed)) from table_10_undef_undef4 group by
+ grouping sets ((col_text_undef_signed)) order by 1;
+ """
+
+ qt_select8 """
+ select sum(sum(col_int_undef_signed)) over (partition by sum(col_int_undef_signed)
+ order by sum(col_int_undef_signed)) from table_10_undef_undef4 group by
+ grouping sets ((col_text_undef_signed,col_int_undef_signed)) order by 1;
+ """
+
+ qt_select9 """
+ select sum(sum(col_int_undef_signed)) over (partition by sum(col_int_undef_signed)
+ order by sum(col_int_undef_signed)) from table_10_undef_undef4 group by
+ grouping sets ((col_text_undef_signed,col_int_undef_signed), (col_text_undef_signed), ()) order by 1;
+ """
+
+ qt_select10 """
+ select sum(col_int_undef_signed + sum(col_int_undef_signed)) over (partition by sum(col_int_undef_signed)
+ order by sum(col_int_undef_signed)) from table_10_undef_undef4 group by
+ grouping sets ((col_text_undef_signed,col_int_undef_signed)) order by 1;
+ """
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/14: [feature](sink) support paritition tablet sink shuffle (#30914)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 819ab6fc002d5763040a781b059a45c3429038fd
Author: zhangstar333 <87...@users.noreply.github.com>
AuthorDate: Thu Feb 29 20:09:24 2024 +0800
[feature](sink) support paritition tablet sink shuffle (#30914)
Co-authored-by: morrySnow <mo...@126.com>
---
be/src/vec/sink/vdata_stream_sender.cpp | 102 ++++++++++++++++++++-
be/src/vec/sink/vdata_stream_sender.h | 41 ++++++++-
be/src/vec/sink/vrow_distribution.h | 3 +
.../glue/translator/PhysicalPlanTranslator.java | 3 +
.../DistributionSpecTabletIdShuffle.java | 35 +++++++
.../nereids/properties/PhysicalProperties.java | 3 +
.../trees/plans/commands/InsertExecutor.java | 26 +++++-
.../plans/commands/InsertIntoTableCommand.java | 15 ++-
.../plans/physical/PhysicalOlapTableSink.java | 21 +----
.../org/apache/doris/planner/DataPartition.java | 34 +++----
.../org/apache/doris/planner/DataStreamSink.java | 46 ++++++++++
.../org/apache/doris/planner/OlapTableSink.java | 22 ++++-
.../java/org/apache/doris/qe/SessionVariable.java | 9 ++
gensrc/thrift/DataSinks.thrift | 7 +-
14 files changed, 310 insertions(+), 57 deletions(-)
diff --git a/be/src/vec/sink/vdata_stream_sender.cpp b/be/src/vec/sink/vdata_stream_sender.cpp
index e690c8eb591..a7f1b4b8584 100644
--- a/be/src/vec/sink/vdata_stream_sender.cpp
+++ b/be/src/vec/sink/vdata_stream_sender.cpp
@@ -23,15 +23,19 @@
#include <gen_cpp/Metrics_types.h>
#include <gen_cpp/data.pb.h>
#include <gen_cpp/internal_service.pb.h>
+#include <glog/logging.h>
#include <stddef.h>
#include <algorithm>
+#include <cstdint>
#include <functional>
#include <map>
+#include <memory>
#include <random>
#include "common/object_pool.h"
#include "common/status.h"
+#include "exec/tablet_info.h"
#include "pipeline/exec/exchange_sink_operator.h"
#include "pipeline/exec/result_file_sink_operator.h"
#include "runtime/descriptors.h"
@@ -41,10 +45,13 @@
#include "runtime/types.h"
#include "util/proto_util.h"
#include "vec/columns/column_const.h"
+#include "vec/columns/columns_number.h"
#include "vec/common/sip_hash.h"
#include "vec/exprs/vexpr.h"
#include "vec/runtime/vdata_stream_mgr.h"
#include "vec/runtime/vdata_stream_recvr.h"
+#include "vec/sink/vrow_distribution.h"
+#include "vec/sink/writer/vtablet_writer_v2.h"
namespace doris::vectorized {
@@ -329,6 +336,7 @@ VDataStreamSender::VDataStreamSender(RuntimeState* state, ObjectPool* pool, int
sink.output_partition.type == TPartitionType::HASH_PARTITIONED ||
sink.output_partition.type == TPartitionType::RANDOM ||
sink.output_partition.type == TPartitionType::RANGE_PARTITIONED ||
+ sink.output_partition.type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED ||
sink.output_partition.type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED);
std::map<int64_t, int64_t> fragment_id_to_channel_index;
@@ -412,6 +420,34 @@ Status VDataStreamSender::init(const TDataSink& tsink) {
RETURN_IF_ERROR(_partitioner->init(t_stream_sink.output_partition.partition_exprs));
} else if (_part_type == TPartitionType::RANGE_PARTITIONED) {
return Status::InternalError("TPartitionType::RANGE_PARTITIONED should not be used");
+ } else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
+ _txn_id = t_stream_sink.tablet_sink_txn_id;
+ _schema = std::make_shared<OlapTableSchemaParam>();
+ RETURN_IF_ERROR(_schema->init(t_stream_sink.tablet_sink_schema));
+ _vpartition = std::make_unique<VOlapTablePartitionParam>(
+ _schema, t_stream_sink.tablet_sink_partition);
+ RETURN_IF_ERROR(_vpartition->init());
+ auto find_tablet_mode = OlapTabletFinder::FindTabletMode::FIND_TABLET_EVERY_ROW;
+ _tablet_finder = std::make_unique<OlapTabletFinder>(_vpartition.get(), find_tablet_mode);
+ _tablet_sink_tuple_desc =
+ _state->desc_tbl().get_tuple_descriptor(t_stream_sink.tablet_sink_tuple_id);
+ _tablet_sink_row_desc = _pool->add(new RowDescriptor(_tablet_sink_tuple_desc, false));
+ //_block_convertor no need init_autoinc_info here
+ _block_convertor =
+ std::make_unique<vectorized::OlapTableBlockConvertor>(_tablet_sink_tuple_desc);
+ _location = _pool->add(new OlapTableLocationParam(t_stream_sink.tablet_sink_location));
+ _row_distribution.init({.state = _state,
+ .block_convertor = _block_convertor.get(),
+ .tablet_finder = _tablet_finder.get(),
+ .vpartition = _vpartition.get(),
+ .add_partition_request_timer = _add_partition_request_timer,
+ .txn_id = _txn_id,
+ .pool = _pool,
+ .location = _location,
+ .vec_output_expr_ctxs = &_fake_expr_ctxs,
+ .schema = _schema,
+ .caller = (void*)this,
+ .create_partition_callback = &empty_callback_function});
} else {
// UNPARTITIONED
}
@@ -488,6 +524,8 @@ Status VDataStreamSender::open(RuntimeState* state) {
if (_part_type == TPartitionType::HASH_PARTITIONED ||
_part_type == TPartitionType::BUCKET_SHFFULE_HASH_PARTITIONED) {
RETURN_IF_ERROR(_partitioner->open(state));
+ } else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
+ RETURN_IF_ERROR(_row_distribution.open(_tablet_sink_row_desc));
}
_compression_type = state->fragement_transmission_compression_type();
@@ -502,6 +540,25 @@ void VDataStreamSender::_handle_eof_channel(RuntimeState* state, ChannelPtrType
static_cast<void>(channel->close(state, Status::OK()));
}
+Status VDataStreamSender::_send_new_partition_batch() {
+ if (_row_distribution.need_deal_batching()) { // maybe try_close more than 1 time
+ RETURN_IF_ERROR(_row_distribution.automatic_create_partition());
+ Block tmp_block = _row_distribution._batching_block->to_block(); // Borrow out, for lval ref
+
+ // these order is only.
+ // 1. clear batching stats(and flag goes true) so that we won't make a new batching process in dealing batched block.
+ // 2. deal batched block
+ // 3. now reuse the column of lval block. cuz write doesn't real adjust it. it generate a new block from that.
+ _row_distribution.clear_batching_stats();
+ RETURN_IF_ERROR(this->send(_state, &tmp_block, false));
+ // Recovery back
+ _row_distribution._batching_block->set_mutable_columns(tmp_block.mutate_columns());
+ _row_distribution._batching_block->clear_column_data();
+ _row_distribution._deal_batched = false;
+ }
+ return Status::OK();
+}
+
Status VDataStreamSender::send(RuntimeState* state, Block* block, bool eos) {
SCOPED_TIMER(_profile->total_time_counter());
SCOPED_TIMER(_exec_timer);
@@ -627,6 +684,38 @@ Status VDataStreamSender::send(RuntimeState* state, Block* block, bool eos) {
(uint32_t*)_partitioner->get_channel_ids(), rows,
block, _enable_pipeline_exec ? eos : false));
}
+ } else if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
+ // check out of limit
+ RETURN_IF_ERROR(_send_new_partition_batch());
+ if (UNLIKELY(block->rows() == 0)) {
+ return Status::OK();
+ }
+ std::shared_ptr<vectorized::Block> convert_block;
+ bool has_filtered_rows = false;
+ int64_t filtered_rows = 0;
+ _number_input_rows += block->rows();
+ RETURN_IF_ERROR(_row_distribution.generate_rows_distribution(
+ *block, convert_block, filtered_rows, has_filtered_rows, _row_part_tablet_ids,
+ _number_input_rows));
+
+ const auto& row_ids = _row_part_tablet_ids[0].row_ids;
+ const auto& tablet_ids = _row_part_tablet_ids[0].tablet_ids;
+ const auto& num_channels = _channels.size();
+ std::vector<std::vector<uint32>> channel2rows;
+ channel2rows.resize(num_channels);
+ for (int idx = 0; idx < row_ids.size(); ++idx) {
+ const auto& row = row_ids[idx];
+ const auto& tablet_id = tablet_ids[idx];
+ channel2rows[tablet_id % num_channels].emplace_back(row);
+ }
+
+ RETURN_IF_ERROR(channel_add_rows_with_idx(state, _channels, num_channels, channel2rows,
+ convert_block.get(),
+ _enable_pipeline_exec ? eos : false));
+ if (eos) {
+ _row_distribution._deal_batched = true;
+ RETURN_IF_ERROR(_send_new_partition_batch());
+ }
} else {
// Range partition
// 1. calculate range
@@ -662,6 +751,12 @@ Status VDataStreamSender::close(RuntimeState* state, Status exec_status) {
{
// send last block
SCOPED_CONSUME_MEM_TRACKER(_mem_tracker.get());
+ // non pipeline engin not pass eos in send function, and maybe have create partition at last block
+ // so at here to check again
+ if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
+ _row_distribution._deal_batched = true;
+ RETURN_IF_ERROR(_send_new_partition_batch());
+ }
if (_serializer.get_block() && _serializer.get_block()->rows() > 0) {
Block block = _serializer.get_block()->to_block();
RETURN_IF_ERROR(
@@ -694,7 +789,12 @@ Status VDataStreamSender::close(RuntimeState* state, Status exec_status) {
}
}
}
-
+ if (_part_type == TPartitionType::TABLET_SINK_SHUFFLE_PARTITIONED) {
+ _state->update_num_rows_load_filtered(_block_convertor->num_filtered_rows() +
+ _tablet_finder->num_filtered_rows());
+ _state->update_num_rows_load_unselected(
+ _tablet_finder->num_immutable_partition_filtered_rows());
+ }
if (_peak_memory_usage_counter) {
_peak_memory_usage_counter->set(_mem_tracker->peak_consumption());
}
diff --git a/be/src/vec/sink/vdata_stream_sender.h b/be/src/vec/sink/vdata_stream_sender.h
index 2834c7e2aa5..88a948ed05c 100644
--- a/be/src/vec/sink/vdata_stream_sender.h
+++ b/be/src/vec/sink/vdata_stream_sender.h
@@ -29,6 +29,7 @@
#include <atomic>
#include <cstddef>
+#include <cstdint>
#include <memory>
#include <ostream>
#include <string>
@@ -39,6 +40,7 @@
#include "common/logging.h"
#include "common/status.h"
#include "exec/data_sink.h"
+#include "exec/tablet_info.h"
#include "pipeline/exec/exchange_sink_buffer.h"
#include "service/backend_options.h"
#include "util/ref_count_closure.h"
@@ -48,6 +50,8 @@
#include "vec/exprs/vexpr_context.h"
#include "vec/runtime/partitioner.h"
#include "vec/runtime/vdata_stream_recvr.h"
+#include "vec/sink/vrow_distribution.h"
+#include "vec/sink/vtablet_finder.h"
namespace doris {
class ObjectPool;
@@ -160,10 +164,18 @@ protected:
Status channel_add_rows(RuntimeState* state, Channels& channels, int num_channels,
const HashValueType* __restrict channel_ids, int rows, Block* block,
bool eos);
+ template <typename Channels>
+ Status channel_add_rows_with_idx(RuntimeState* state, Channels& channels, int num_channels,
+ std::vector<std::vector<uint32_t>>& channel2rows, Block* block,
+ bool eos);
template <typename ChannelPtrType>
void _handle_eof_channel(RuntimeState* state, ChannelPtrType channel, Status st);
+ static Status empty_callback_function(void* sender, TCreatePartitionResult* result) {
+ return Status::OK();
+ }
+ Status _send_new_partition_batch();
// Sender instance id, unique within a fragment.
int _sender_id;
@@ -221,6 +233,22 @@ protected:
bool _enable_pipeline_exec = false;
BlockSerializer<VDataStreamSender> _serializer;
+
+ // for shuffle data by partition and tablet
+ VRowDistribution _row_distribution;
+ RuntimeProfile::Counter* _add_partition_request_timer = nullptr;
+ int64_t _txn_id = -1;
+ RowDescriptor* _tablet_sink_row_desc = nullptr;
+ TupleDescriptor* _tablet_sink_tuple_desc = nullptr;
+ OlapTableLocationParam* _location = nullptr;
+ int64_t _number_input_rows = 0;
+ // reuse to avoid frequent memory allocation and release.
+ std::vector<RowPartTabletIds> _row_part_tablet_ids;
+ vectorized::VExprContextSPtrs _fake_expr_ctxs;
+ std::unique_ptr<VOlapTablePartitionParam> _vpartition = nullptr;
+ std::unique_ptr<OlapTabletFinder> _tablet_finder = nullptr;
+ std::shared_ptr<OlapTableSchemaParam> _schema = nullptr;
+ std::unique_ptr<vectorized::OlapTableBlockConvertor> _block_convertor = nullptr;
};
template <typename Parent = VDataStreamSender>
@@ -402,12 +430,20 @@ Status VDataStreamSender::channel_add_rows(RuntimeState* state, Channels& channe
int num_channels,
const HashValueType* __restrict channel_ids, int rows,
Block* block, bool eos) {
- std::vector<uint32_t> channel2rows[num_channels];
-
+ std::vector<std::vector<uint32_t>> channel2rows;
+ channel2rows.resize(num_channels);
for (uint32_t i = 0; i < rows; i++) {
channel2rows[channel_ids[i]].emplace_back(i);
}
+ RETURN_IF_ERROR(
+ channel_add_rows_with_idx(state, channels, num_channels, channel2rows, block, eos));
+ return Status::OK();
+}
+template <typename Channels>
+Status VDataStreamSender::channel_add_rows_with_idx(
+ RuntimeState* state, Channels& channels, int num_channels,
+ std::vector<std::vector<uint32_t>>& channel2rows, Block* block, bool eos) {
Status status;
for (int i = 0; i < num_channels; ++i) {
if (!channels[i]->is_receiver_eof() && !channel2rows[i].empty()) {
@@ -424,7 +460,6 @@ Status VDataStreamSender::channel_add_rows(RuntimeState* state, Channels& channe
}
}
}
-
return Status::OK();
}
diff --git a/be/src/vec/sink/vrow_distribution.h b/be/src/vec/sink/vrow_distribution.h
index 54cd6f42ce6..12acda73ed2 100644
--- a/be/src/vec/sink/vrow_distribution.h
+++ b/be/src/vec/sink/vrow_distribution.h
@@ -161,8 +161,11 @@ private:
// for auto partitions
std::vector<std::vector<TStringLiteral>> _partitions_need_create;
+
+public:
std::unique_ptr<MutableBlock> _batching_block;
bool _deal_batched = false; // If true, send batched block before any block's append.
+private:
size_t _batching_rows = 0, _batching_bytes = 0;
OlapTableBlockConvertor* _block_convertor = nullptr;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
index c9cb534c9c2..c64965080fb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PhysicalPlanTranslator.java
@@ -70,6 +70,7 @@ import org.apache.doris.nereids.properties.DistributionSpecHash;
import org.apache.doris.nereids.properties.DistributionSpecReplicated;
import org.apache.doris.nereids.properties.DistributionSpecStorageAny;
import org.apache.doris.nereids.properties.DistributionSpecStorageGather;
+import org.apache.doris.nereids.properties.DistributionSpecTabletIdShuffle;
import org.apache.doris.nereids.properties.OrderKey;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.rules.implementation.LogicalWindowToPhysicalWindow.WindowFrameGroup;
@@ -2458,6 +2459,8 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
+ distributionSpecHash.getShuffleType());
}
return new DataPartition(partitionType, partitionExprs);
+ } else if (distributionSpec instanceof DistributionSpecTabletIdShuffle) {
+ return DataPartition.TABLET_ID;
} else {
throw new RuntimeException("Unknown DistributionSpec: " + distributionSpec);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecTabletIdShuffle.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecTabletIdShuffle.java
new file mode 100644
index 00000000000..17a84401c5b
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/DistributionSpecTabletIdShuffle.java
@@ -0,0 +1,35 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.properties;
+
+/**
+ * use for shuffle data by tablet-id before sink.
+ */
+public class DistributionSpecTabletIdShuffle extends DistributionSpec {
+
+ public static final DistributionSpecTabletIdShuffle INSTANCE = new DistributionSpecTabletIdShuffle();
+
+ private DistributionSpecTabletIdShuffle() {
+ super();
+ }
+
+ @Override
+ public boolean satisfy(DistributionSpec other) {
+ return other instanceof DistributionSpecTabletIdShuffle;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/PhysicalProperties.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/PhysicalProperties.java
index 4fa32dbfb69..cc5d7db6a08 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/PhysicalProperties.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/properties/PhysicalProperties.java
@@ -46,6 +46,9 @@ public class PhysicalProperties {
public static PhysicalProperties MUST_SHUFFLE = new PhysicalProperties(DistributionSpecMustShuffle.INSTANCE);
+ public static PhysicalProperties TABLET_ID_SHUFFLE
+ = new PhysicalProperties(DistributionSpecTabletIdShuffle.INSTANCE);
+
private final OrderSpec orderSpec;
private final DistributionSpec distributionSpec;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertExecutor.java
index 45471429bba..2c5ee559f3d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertExecutor.java
@@ -60,7 +60,10 @@ import org.apache.doris.nereids.types.DataType;
import org.apache.doris.nereids.util.RelationUtil;
import org.apache.doris.nereids.util.TypeCoercionUtils;
import org.apache.doris.planner.DataSink;
+import org.apache.doris.planner.DataStreamSink;
+import org.apache.doris.planner.ExchangeNode;
import org.apache.doris.planner.OlapTableSink;
+import org.apache.doris.planner.PlanFragment;
import org.apache.doris.proto.InternalService;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.Coordinator;
@@ -74,6 +77,8 @@ import org.apache.doris.task.LoadEtlTask;
import org.apache.doris.thrift.TFileFormatType;
import org.apache.doris.thrift.TFileType;
import org.apache.doris.thrift.TMergeType;
+import org.apache.doris.thrift.TOlapTableLocationParam;
+import org.apache.doris.thrift.TPartitionType;
import org.apache.doris.thrift.TQueryType;
import org.apache.doris.thrift.TStreamLoadPutRequest;
import org.apache.doris.thrift.TTxnParams;
@@ -153,8 +158,8 @@ public class InsertExecutor {
/**
* finalize sink to complete enough info for sink execution
*/
- public void finalizeSink(DataSink sink, boolean isPartialUpdate, boolean isFromInsert,
- boolean allowAutoPartition) {
+ public void finalizeSink(PlanFragment fragment, DataSink sink,
+ boolean isPartialUpdate, boolean isFromInsert, boolean allowAutoPartition) {
if (!(sink instanceof OlapTableSink)) {
return;
}
@@ -175,6 +180,23 @@ public class InsertExecutor {
if (!allowAutoPartition) {
olapTableSink.setAutoPartition(false);
}
+ // update
+
+ // set schema and partition info for tablet id shuffle exchange
+ if (fragment.getPlanRoot() instanceof ExchangeNode
+ && fragment.getDataPartition().getType() == TPartitionType.TABLET_SINK_SHUFFLE_PARTITIONED) {
+ DataStreamSink dataStreamSink = (DataStreamSink) (fragment.getChild(0).getSink());
+ Analyzer analyzer = new Analyzer(Env.getCurrentEnv(), ConnectContext.get());
+ dataStreamSink.setTabletSinkSchemaParam(olapTableSink.createSchema(
+ database.getId(), olapTableSink.getDstTable(), analyzer));
+ dataStreamSink.setTabletSinkPartitionParam(olapTableSink.createPartition(
+ database.getId(), olapTableSink.getDstTable(), analyzer));
+ dataStreamSink.setTabletSinkTupleDesc(olapTableSink.getTupleDescriptor());
+ List<TOlapTableLocationParam> locationParams = olapTableSink
+ .createLocation(olapTableSink.getDstTable());
+ dataStreamSink.setTabletSinkLocationParam(locationParams.get(0));
+ dataStreamSink.setTabletSinkTxnId(olapTableSink.getTxnId());
+ }
} catch (Exception e) {
throw new AnalysisException(e.getMessage(), e);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertIntoTableCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertIntoTableCommand.java
index d5101aabcda..166219dfae1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertIntoTableCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertIntoTableCommand.java
@@ -37,9 +37,12 @@ import org.apache.doris.nereids.trees.expressions.Slot;
import org.apache.doris.nereids.trees.plans.Explainable;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.PlanType;
+import org.apache.doris.nereids.trees.plans.algebra.OneRowRelation;
import org.apache.doris.nereids.trees.plans.commands.info.DMLCommandType;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
import org.apache.doris.nereids.trees.plans.physical.PhysicalOlapTableSink;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalUnion;
import org.apache.doris.nereids.trees.plans.visitor.PlanVisitor;
import org.apache.doris.planner.DataSink;
import org.apache.doris.planner.GroupCommitPlanner;
@@ -183,7 +186,7 @@ public class InsertIntoTableCommand extends Command implements ForwardWithSync,
physicalOlapTableSink.getDatabase(),
physicalOlapTableSink.getTargetTable(), label, planner);
insertExecutor.beginTransaction();
- insertExecutor.finalizeSink(sink, physicalOlapTableSink.isPartialUpdate(),
+ insertExecutor.finalizeSink(planner.getFragments().get(0), sink, physicalOlapTableSink.isPartialUpdate(),
physicalOlapTableSink.getDmlCommandType() == DMLCommandType.INSERT, this.allowAutoPartition);
} finally {
targetTableIf.readUnlock();
@@ -254,12 +257,20 @@ public class InsertIntoTableCommand extends Command implements ForwardWithSync,
}
OlapTable targetTable = physicalOlapTableSink.getTargetTable();
return ctx.getSessionVariable().getSqlMode() != SqlModeHelper.MODE_NO_BACKSLASH_ESCAPES
- && !ctx.isTxnModel() && sink.getFragment().getPlanRoot() instanceof UnionNode
+ && !ctx.isTxnModel() && isGroupCommitAvailablePlan(physicalOlapTableSink)
&& physicalOlapTableSink.getPartitionIds().isEmpty() && targetTable.getTableProperty()
.getUseSchemaLightChange() && !targetTable.getQualifiedDbName()
.equalsIgnoreCase(FeConstants.INTERNAL_DB_NAME);
}
+ private boolean isGroupCommitAvailablePlan(PhysicalOlapTableSink<? extends Plan> sink) {
+ Plan child = sink.child();
+ if (child instanceof PhysicalDistribute) {
+ child = child.child(0);
+ }
+ return child instanceof OneRowRelation || (child instanceof PhysicalUnion && child.arity() == 0);
+ }
+
@Override
public Plan getExplainPlan(ConnectContext ctx) {
if (!ctx.getSessionVariable().isEnableNereidsDML()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalOlapTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalOlapTableSink.java
index 11a85ed100f..86a6a4b8c61 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalOlapTableSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalOlapTableSink.java
@@ -25,7 +25,6 @@ import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.RandomDistributionInfo;
import org.apache.doris.nereids.exceptions.AnalysisException;
import org.apache.doris.nereids.memo.GroupExpression;
-import org.apache.doris.nereids.properties.DistributionSpecHash.ShuffleType;
import org.apache.doris.nereids.properties.LogicalProperties;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.trees.expressions.Expression;
@@ -42,13 +41,11 @@ import org.apache.doris.statistics.Statistics;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableSet;
-import com.google.common.collect.Lists;
import java.util.List;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
-import java.util.stream.Collectors;
/**
* physical olap table sink for insert command
@@ -218,23 +215,7 @@ public class PhysicalOlapTableSink<CHILD_TYPE extends Plan> extends PhysicalSink
if (targetTable.isPartitionDistributed()) {
DistributionInfo distributionInfo = targetTable.getDefaultDistributionInfo();
if (distributionInfo instanceof HashDistributionInfo) {
- HashDistributionInfo hashDistributionInfo
- = ((HashDistributionInfo) targetTable.getDefaultDistributionInfo());
- List<Column> distributedColumns = hashDistributionInfo.getDistributionColumns();
- List<Integer> columnIndexes = Lists.newArrayList();
- int idx = 0;
- for (int i = 0; i < targetTable.getFullSchema().size(); ++i) {
- if (targetTable.getFullSchema().get(i).equals(distributedColumns.get(idx))) {
- columnIndexes.add(i);
- idx++;
- if (idx == distributedColumns.size()) {
- break;
- }
- }
- }
- return PhysicalProperties.createHash(columnIndexes.stream()
- .map(colIdx -> child().getOutput().get(colIdx).getExprId())
- .collect(Collectors.toList()), ShuffleType.NATURAL);
+ return PhysicalProperties.TABLET_ID_SHUFFLE;
} else if (distributionInfo instanceof RandomDistributionInfo) {
return PhysicalProperties.ANY;
} else {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java b/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java
index 75352086940..9c5c375a35c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/DataPartition.java
@@ -32,8 +32,6 @@ import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Lists;
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
import java.util.List;
@@ -46,16 +44,14 @@ import java.util.List;
* TODO: better name? just Partitioning?
*/
public class DataPartition {
- private static final Logger LOG = LogManager.getLogger(DataPartition.class);
public static final DataPartition UNPARTITIONED = new DataPartition(TPartitionType.UNPARTITIONED);
-
public static final DataPartition RANDOM = new DataPartition(TPartitionType.RANDOM);
+ public static final DataPartition TABLET_ID = new DataPartition(TPartitionType.TABLET_SINK_SHUFFLE_PARTITIONED);
private final TPartitionType type;
-
// for hash partition: exprs used to compute hash value
- private ImmutableList<Expr> partitionExprs = ImmutableList.of();
+ private ImmutableList<Expr> partitionExprs;
public DataPartition(TPartitionType type, List<Expr> exprs) {
Preconditions.checkNotNull(exprs);
@@ -67,13 +63,10 @@ public class DataPartition {
this.partitionExprs = ImmutableList.copyOf(exprs);
}
- public void substitute(ExprSubstitutionMap smap, Analyzer analyzer) throws AnalysisException {
- List<Expr> list = Expr.trySubstituteList(partitionExprs, smap, analyzer, false);
- partitionExprs = ImmutableList.copyOf(list);
- }
-
public DataPartition(TPartitionType type) {
- Preconditions.checkState(type == TPartitionType.UNPARTITIONED || type == TPartitionType.RANDOM);
+ Preconditions.checkState(type == TPartitionType.UNPARTITIONED
+ || type == TPartitionType.RANDOM
+ || type == TPartitionType.TABLET_SINK_SHUFFLE_PARTITIONED);
this.type = type;
this.partitionExprs = ImmutableList.of();
}
@@ -82,6 +75,11 @@ public class DataPartition {
return new DataPartition(TPartitionType.HASH_PARTITIONED, exprs);
}
+ public void substitute(ExprSubstitutionMap smap, Analyzer analyzer) throws AnalysisException {
+ List<Expr> list = Expr.trySubstituteList(partitionExprs, smap, analyzer, false);
+ partitionExprs = ImmutableList.copyOf(list);
+ }
+
public boolean isPartitioned() {
return type != TPartitionType.UNPARTITIONED;
}
@@ -106,16 +104,6 @@ public class DataPartition {
return result;
}
- /**
- * Returns true if 'this' is a partition that is compatible with the
- * requirements of 's'.
- * TODO: specify more clearly and implement
- */
- public boolean isCompatible(DataPartition s) {
- // TODO: implement
- return true;
- }
-
public String getExplainString(TExplainLevel explainLevel) {
StringBuilder str = new StringBuilder();
str.append(type.toString());
@@ -127,7 +115,7 @@ public class DataPartition {
for (Expr expr : partitionExprs) {
strings.add(expr.toSql());
}
- str.append(": " + Joiner.on(", ").join(strings));
+ str.append(": ").append(Joiner.on(", ").join(strings));
}
str.append("\n");
return str.toString();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/DataStreamSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/DataStreamSink.java
index 4d4a2c641a9..b9cf516bc3d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/DataStreamSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/DataStreamSink.java
@@ -27,8 +27,12 @@ import org.apache.doris.thrift.TDataSink;
import org.apache.doris.thrift.TDataSinkType;
import org.apache.doris.thrift.TDataStreamSink;
import org.apache.doris.thrift.TExplainLevel;
+import org.apache.doris.thrift.TOlapTableLocationParam;
+import org.apache.doris.thrift.TOlapTablePartitionParam;
+import org.apache.doris.thrift.TOlapTableSchemaParam;
import com.google.common.base.Joiner;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import org.springframework.util.CollectionUtils;
@@ -52,6 +56,13 @@ public class DataStreamSink extends DataSink {
protected List<RuntimeFilter> runtimeFilters = Lists.newArrayList();
+ // use for tablet id shuffle sink only
+ protected TOlapTableSchemaParam tabletSinkSchemaParam = null;
+ protected TOlapTablePartitionParam tabletSinkPartitionParam = null;
+ protected TOlapTableLocationParam tabletSinkLocationParam = null;
+ protected TupleDescriptor tabletSinkTupleDesc = null;
+ protected long tabletSinkTxnId = -1;
+
public DataStreamSink() {
}
@@ -118,6 +129,26 @@ public class DataStreamSink extends DataSink {
this.runtimeFilters.add(runtimeFilter);
}
+ public void setTabletSinkSchemaParam(TOlapTableSchemaParam schemaParam) {
+ this.tabletSinkSchemaParam = schemaParam;
+ }
+
+ public void setTabletSinkPartitionParam(TOlapTablePartitionParam partitionParam) {
+ this.tabletSinkPartitionParam = partitionParam;
+ }
+
+ public void setTabletSinkTupleDesc(TupleDescriptor tupleDesc) {
+ this.tabletSinkTupleDesc = tupleDesc;
+ }
+
+ public void setTabletSinkLocationParam(TOlapTableLocationParam locationParam) {
+ this.tabletSinkLocationParam = locationParam;
+ }
+
+ public void setTabletSinkTxnId(long txnId) {
+ this.tabletSinkTxnId = txnId;
+ }
+
@Override
public String getExplainString(String prefix, TExplainLevel explainLevel) {
StringBuilder strBuilder = new StringBuilder();
@@ -179,6 +210,21 @@ public class DataStreamSink extends DataSink {
tStreamSink.addToRuntimeFilters(rf.toThrift());
}
}
+ Preconditions.checkState((tabletSinkSchemaParam != null) == (tabletSinkPartitionParam != null),
+ "schemaParam and partitionParam should be set together.");
+ if (tabletSinkSchemaParam != null) {
+ tStreamSink.setTabletSinkSchema(tabletSinkSchemaParam);
+ }
+ if (tabletSinkPartitionParam != null) {
+ tStreamSink.setTabletSinkPartition(tabletSinkPartitionParam);
+ }
+ if (tabletSinkTupleDesc != null) {
+ tStreamSink.setTabletSinkTupleId(tabletSinkTupleDesc.getId().asInt());
+ }
+ if (tabletSinkLocationParam != null) {
+ tStreamSink.setTabletSinkLocation(tabletSinkLocationParam);
+ }
+ tStreamSink.setTabletSinkTxnId(tabletSinkTxnId);
result.setStreamSink(tStreamSink);
return result;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java
index 485fde67fb1..924b64f87cd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapTableSink.java
@@ -110,6 +110,7 @@ public class OlapTableSink extends DataSink {
private boolean singleReplicaLoad;
private boolean isStrictMode = false;
+ private long txnId = -1;
public OlapTableSink(OlapTable dstTable, TupleDescriptor tupleDescriptor, List<Long> partitionIds,
boolean singleReplicaLoad) {
@@ -129,6 +130,7 @@ public class OlapTableSink extends DataSink {
tSink.setLoadChannelTimeoutS(loadChannelTimeoutS);
tSink.setSendBatchParallelism(sendBatchParallelism);
this.isStrictMode = isStrictMode;
+ this.txnId = txnId;
if (loadToSingleTablet && !(dstTable.getDefaultDistributionInfo() instanceof RandomDistributionInfo)) {
throw new AnalysisException(
"if load_to_single_tablet set to true," + " the olap table must be with random distribution");
@@ -237,7 +239,7 @@ public class OlapTableSink extends DataSink {
return tDataSink;
}
- private TOlapTableSchemaParam createSchema(long dbId, OlapTable table, Analyzer analyzer) throws AnalysisException {
+ public TOlapTableSchemaParam createSchema(long dbId, OlapTable table, Analyzer analyzer) throws AnalysisException {
TOlapTableSchemaParam schemaParam = new TOlapTableSchemaParam();
schemaParam.setDbId(dbId);
schemaParam.setTableId(table.getId());
@@ -321,7 +323,7 @@ public class OlapTableSink extends DataSink {
return distColumns;
}
- private TOlapTablePartitionParam createPartition(long dbId, OlapTable table, Analyzer analyzer)
+ public TOlapTablePartitionParam createPartition(long dbId, OlapTable table, Analyzer analyzer)
throws UserException {
TOlapTablePartitionParam partitionParam = new TOlapTablePartitionParam();
partitionParam.setDbId(dbId);
@@ -479,7 +481,7 @@ public class OlapTableSink extends DataSink {
}
}
- private List<TOlapTableLocationParam> createLocation(OlapTable table) throws UserException {
+ public List<TOlapTableLocationParam> createLocation(OlapTable table) throws UserException {
TOlapTableLocationParam locationParam = new TOlapTableLocationParam();
TOlapTableLocationParam slaveLocationParam = new TOlapTableLocationParam();
// BE id -> path hash
@@ -571,7 +573,7 @@ public class OlapTableSink extends DataSink {
bePathsMap.putAll(result);
}
- private TPaloNodesInfo createPaloNodesInfo() {
+ public TPaloNodesInfo createPaloNodesInfo() {
TPaloNodesInfo nodesInfo = new TPaloNodesInfo();
SystemInfoService systemInfoService = Env.getCurrentSystemInfo();
for (Long id : systemInfoService.getAllBackendIds(false)) {
@@ -584,4 +586,16 @@ public class OlapTableSink extends DataSink {
protected TDataSinkType getDataSinkType() {
return TDataSinkType.OLAP_TABLE_SINK;
}
+
+ public OlapTable getDstTable() {
+ return dstTable;
+ }
+
+ public TupleDescriptor getTupleDescriptor() {
+ return tupleDescriptor;
+ }
+
+ public long getTxnId() {
+ return txnId;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index a5bc477c3f6..e561ded693a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -3206,6 +3206,15 @@ public class SessionVariable implements Serializable, Writable {
this.dumpNereidsMemo = dumpNereidsMemo;
}
+ public void disableStrictConsistencyDmlOnce() throws DdlException {
+ if (!enableStrictConsistencyDml) {
+ return;
+ }
+ setIsSingleSetVar(true);
+ VariableMgr.setVar(this,
+ new SetVar(SessionVariable.ENABLE_STRICT_CONSISTENCY_DML, new StringLiteral("false")));
+ }
+
public void enableFallbackToOriginalPlannerOnce() throws DdlException {
if (enableFallbackToOriginalPlanner) {
return;
diff --git a/gensrc/thrift/DataSinks.thrift b/gensrc/thrift/DataSinks.thrift
index 068b5927004..602943b4207 100644
--- a/gensrc/thrift/DataSinks.thrift
+++ b/gensrc/thrift/DataSinks.thrift
@@ -172,8 +172,11 @@ struct TDataStreamSink {
7: optional list<PlanNodes.TRuntimeFilterDesc> runtime_filters
// used for partition_type = TABLET_SINK_SHUFFLE_PARTITIONED
- 8: optional Descriptors.TOlapTableSchemaParam schema
- 9: optional Descriptors.TOlapTablePartitionParam partition
+ 8: optional Descriptors.TOlapTableSchemaParam tablet_sink_schema
+ 9: optional Descriptors.TOlapTablePartitionParam tablet_sink_partition
+ 10: optional Descriptors.TOlapTableLocationParam tablet_sink_location
+ 11: optional i64 tablet_sink_txn_id
+ 12: optional Types.TTupleId tablet_sink_tuple_id
}
struct TMultiCastDataStreamSink {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 03/14: [improve](export) Support partition data consistency (#31290)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 1f825ee2d68c12f1a15f09308ed6c6cc3191f623
Author: walter <w4...@gmail.com>
AuthorDate: Thu Feb 29 22:08:29 2024 +0800
[improve](export) Support partition data consistency (#31290)
---
.../java/org/apache/doris/analysis/ExportStmt.java | 15 +
.../main/java/org/apache/doris/load/ExportJob.java | 87 ++++--
.../main/java/org/apache/doris/load/ExportMgr.java | 1 +
.../org/apache/doris/load/ExportTaskExecutor.java | 5 +-
.../trees/plans/commands/ExportCommand.java | 13 +
.../export_p0/test_export_data_consistency.out | 305 +++++++++++++++++++++
.../export_p0/test_export_data_consistency.groovy | 210 ++++++++++++++
7 files changed, 611 insertions(+), 25 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/ExportStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/ExportStmt.java
index 123bfa727fb..1a7d9dc93b5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/ExportStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/ExportStmt.java
@@ -63,6 +63,7 @@ import java.util.stream.Collectors;
public class ExportStmt extends StatementBase {
public static final String PARALLELISM = "parallelism";
public static final String LABEL = "label";
+ public static final String DATA_CONSISTENCY = "data_consistency";
private static final String DEFAULT_COLUMN_SEPARATOR = "\t";
private static final String DEFAULT_LINE_DELIMITER = "\n";
@@ -72,6 +73,7 @@ public class ExportStmt extends StatementBase {
private static final ImmutableSet<String> PROPERTIES_SET = new ImmutableSet.Builder<String>()
.add(LABEL)
.add(PARALLELISM)
+ .add(DATA_CONSISTENCY)
.add(LoadStmt.KEY_IN_PARAM_COLUMNS)
.add(OutFileClause.PROP_MAX_FILE_SIZE)
.add(OutFileClause.PROP_DELETE_EXISTING_FILES)
@@ -104,6 +106,7 @@ public class ExportStmt extends StatementBase {
private String maxFileSize;
private String deleteExistingFiles;
private String withBom;
+ private String dataConsistency;
private SessionVariable sessionVariables;
private String qualifiedUser;
@@ -230,6 +233,7 @@ public class ExportStmt extends StatementBase {
exportJob.setMaxFileSize(this.maxFileSize);
exportJob.setDeleteExistingFiles(this.deleteExistingFiles);
exportJob.setWithBom(this.withBom);
+ exportJob.setDataConsistency(this.dataConsistency);
if (columns != null) {
Splitter split = Splitter.on(',').trimResults().omitEmptyStrings();
@@ -359,6 +363,17 @@ public class ExportStmt extends StatementBase {
// with bom
this.withBom = properties.getOrDefault(OutFileClause.PROP_WITH_BOM, "false");
+
+ // data consistency
+ String dataConsistencyStr = properties.get(DATA_CONSISTENCY);
+ if (dataConsistencyStr != null) {
+ if (!dataConsistencyStr.equalsIgnoreCase(ExportJob.CONSISTENT_PARTITION)) {
+ throw new UserException("The value of data_consistency is invalid, only `partition` is allowed");
+ }
+ this.dataConsistency = ExportJob.CONSISTENT_PARTITION;
+ } else {
+ this.dataConsistency = ExportJob.CONSISTENT_ALL;
+ }
}
private void checkColumns() throws DdlException {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
index 9fb827f6d6a..fe28151a546 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportJob.java
@@ -99,6 +99,7 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Optional;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.stream.Collectors;
@Data
public class ExportJob implements Writable {
@@ -108,6 +109,9 @@ public class ExportJob implements Writable {
private static final int MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT = Config.maximum_tablets_of_outfile_in_export;
+ public static final String CONSISTENT_ALL = "all";
+ public static final String CONSISTENT_PARTITION = "partition";
+
@SerializedName("id")
private long id;
@SerializedName("label")
@@ -168,6 +172,8 @@ public class ExportJob implements Writable {
private Integer tabletsNum;
@SerializedName("withBom")
private String withBom;
+ @SerializedName("dataConsistency")
+ private String dataConsistency;
private TableRef tableRef;
@@ -222,6 +228,7 @@ public class ExportJob implements Writable {
this.lineDelimiter = "\n";
this.columns = "";
this.withBom = "false";
+ this.dataConsistency = "all";
}
public ExportJob(long jobId) {
@@ -229,6 +236,10 @@ public class ExportJob implements Writable {
this.id = jobId;
}
+ public boolean isPartitionConsistency() {
+ return dataConsistency != null && dataConsistency.equals(CONSISTENT_PARTITION);
+ }
+
public void generateOutfileStatement() throws UserException {
exportTable.readLock();
try {
@@ -302,16 +313,12 @@ public class ExportJob implements Writable {
}
// get all tablets
- List<List<Long>> tabletsListPerParallel = splitTablets();
+ List<List<List<Long>>> tabletsListPerParallel = splitTablets();
// Each Outfile clause responsible for MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT tablets
- for (List<Long> tabletsList : tabletsListPerParallel) {
+ for (List<List<Long>> tabletsList : tabletsListPerParallel) {
List<StatementBase> logicalPlanAdapters = Lists.newArrayList();
- for (int i = 0; i < tabletsList.size(); i += MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT) {
- int end = i + MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT < tabletsList.size()
- ? i + MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT : tabletsList.size();
- List<Long> tabletIds = new ArrayList<>(tabletsList.subList(i, end));
-
+ for (List<Long> tabletIds : tabletsList) {
// generate LogicalPlan
LogicalPlan plan = generateOneLogicalPlan(qualifiedTableName, tabletIds,
this.partitionNames, selectLists);
@@ -471,15 +478,12 @@ public class ExportJob implements Writable {
}
private List<List<TableRef>> getTableRefListPerParallel() throws UserException {
- List<List<Long>> tabletsListPerParallel = splitTablets();
+ List<List<List<Long>>> tabletsListPerParallel = splitTablets();
List<List<TableRef>> tableRefListPerParallel = Lists.newArrayList();
- for (List<Long> tabletsList : tabletsListPerParallel) {
+ for (List<List<Long>> tabletsList : tabletsListPerParallel) {
List<TableRef> tableRefList = Lists.newArrayList();
- for (int i = 0; i < tabletsList.size(); i += MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT) {
- int end = i + MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT < tabletsList.size()
- ? i + MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT : tabletsList.size();
- List<Long> tablets = new ArrayList<>(tabletsList.subList(i, end));
+ for (List<Long> tablets : tabletsList) {
// Since export does not support the alias, here we pass the null value.
// we can not use this.tableRef.getAlias(),
// because the constructor of `Tableref` will convert this.tableRef.getAlias()
@@ -494,11 +498,13 @@ public class ExportJob implements Writable {
return tableRefListPerParallel;
}
- private List<List<Long>> splitTablets() throws UserException {
+ private List<List<List<Long>>> splitTablets() throws UserException {
// get tablets
Database db = Env.getCurrentEnv().getInternalCatalog().getDbOrAnalysisException(this.tableName.getDb());
OlapTable table = db.getOlapTableOrAnalysisException(this.tableName.getTbl());
- List<Long> tabletIdList = Lists.newArrayList();
+
+ Integer tabletsAllNum = 0;
+ List<List<Long>> tabletIdList = Lists.newArrayList();
table.readLock();
try {
final Collection<Partition> partitions = new ArrayList<Partition>();
@@ -516,26 +522,56 @@ public class ExportJob implements Writable {
// get tablets
for (Partition partition : partitions) {
- partitionToVersion.put(partition.getName(), partition.getVisibleVersion());
+ // Partition data consistency is not need to verify partition version.
+ if (!isPartitionConsistency()) {
+ partitionToVersion.put(partition.getName(), partition.getVisibleVersion());
+ }
for (MaterializedIndex index : partition.getMaterializedIndices(IndexExtState.VISIBLE)) {
- tabletIdList.addAll(index.getTabletIdsInOrder());
+ List<Long> tablets = index.getTabletIdsInOrder();
+ tabletsAllNum += tablets.size();
+ tabletIdList.add(tablets);
}
}
} finally {
table.readUnlock();
}
+ if (isPartitionConsistency()) {
+ // Assign tablets of a partition to per parallel.
+ int totalPartitions = tabletIdList.size();
+ int numPerParallel = totalPartitions / this.parallelism;
+ int numPerQueryRemainder = totalPartitions - numPerParallel * this.parallelism;
+ int realParallelism = this.parallelism;
+ if (totalPartitions < this.parallelism) {
+ realParallelism = totalPartitions;
+ LOG.warn("Export Job [{}]: The number of partitions ({}) is smaller than parallelism ({}), "
+ + "set parallelism to partition num.", id, totalPartitions, this.parallelism);
+ }
+ int start = 0;
+ List<List<List<Long>>> tabletsListPerParallel = new ArrayList<>();
+ for (int i = 0; i < realParallelism; ++i) {
+ int partitionNum = numPerParallel;
+ if (numPerQueryRemainder > 0) {
+ partitionNum += 1;
+ --numPerQueryRemainder;
+ }
+ List<List<Long>> tablets = new ArrayList<>(tabletIdList.subList(start, start + partitionNum));
+ start += partitionNum;
+ tabletsListPerParallel.add(tablets);
+ }
+ return tabletsListPerParallel;
+ }
+
/**
* Assign tablets to per parallel, for example:
* If the number of all tablets if 10, and the real parallelism is 4,
* then, the number of tablets of per parallel should be: 3 3 2 2.
*/
- Integer tabletsAllNum = tabletIdList.size();
tabletsNum = tabletsAllNum;
Integer tabletsNumPerParallel = tabletsAllNum / this.parallelism;
Integer tabletsNumPerQueryRemainder = tabletsAllNum - tabletsNumPerParallel * this.parallelism;
- List<List<Long>> tabletsListPerParallel = Lists.newArrayList();
+ List<List<List<Long>>> tabletsListPerParallel = Lists.newArrayList();
Integer realParallelism = this.parallelism;
if (tabletsAllNum < this.parallelism) {
realParallelism = tabletsAllNum;
@@ -543,15 +579,22 @@ public class ExportJob implements Writable {
+ "set parallelism to tablets num.", id, tabletsAllNum, this.parallelism);
}
Integer start = 0;
- for (int i = 0; i < realParallelism; ++i) {
+ List<Long> flatTabletIdList = tabletIdList.stream().flatMap(List::stream).collect(Collectors.toList());
+ for (int j = 0; j < realParallelism; ++j) {
Integer tabletsNum = tabletsNumPerParallel;
if (tabletsNumPerQueryRemainder > 0) {
tabletsNum = tabletsNum + 1;
--tabletsNumPerQueryRemainder;
}
- ArrayList<Long> tablets = new ArrayList<>(tabletIdList.subList(start, start + tabletsNum));
- start += tabletsNum;
+ List<Long> tabletsList = new ArrayList<>(flatTabletIdList.subList(start, start + tabletsNum));
+ List<List<Long>> tablets = new ArrayList<>();
+ for (int i = 0; i < tabletsList.size(); i += MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT) {
+ int end = i + MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT < tabletsList.size()
+ ? i + MAXIMUM_TABLETS_OF_OUTFILE_IN_EXPORT : tabletsList.size();
+ tablets.add(new ArrayList<>(tabletsList.subList(i, end)));
+ }
+ start += tabletsNum;
tabletsListPerParallel.add(tablets);
}
return tabletsListPerParallel;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
index efbfd33966f..f72c0b44a63 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportMgr.java
@@ -355,6 +355,7 @@ public class ExportMgr {
infoMap.put("tablet_num", job.getTabletsNum());
infoMap.put("max_file_size", job.getMaxFileSize());
infoMap.put("delete_existing_files", job.getDeleteExistingFiles());
+ infoMap.put("data_consistency", job.getDataConsistency());
jobInfo.add(new Gson().toJson(infoMap));
// path
jobInfo.add(job.getExportPath());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/ExportTaskExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/load/ExportTaskExecutor.java
index 48f9b2ca1a0..f4ee84298b2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/ExportTaskExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/ExportTaskExecutor.java
@@ -89,8 +89,8 @@ public class ExportTaskExecutor implements TransientTaskExecutor {
if (isCanceled.get()) {
throw new JobException("Export executor has been canceled, task id: {}", taskId);
}
- // check the version of tablets
- if (exportJob.getExportTable().getType() == TableType.OLAP) {
+ // check the version of tablets, skip if the consistency is in partition level.
+ if (exportJob.getExportTable().getType() == TableType.OLAP && !exportJob.isPartitionConsistency()) {
try {
Database db = Env.getCurrentEnv().getInternalCatalog().getDbOrAnalysisException(
exportJob.getTableName().getDb());
@@ -136,7 +136,6 @@ public class ExportTaskExecutor implements TransientTaskExecutor {
}
try (AutoCloseConnectContext r = buildConnectContext()) {
-
StatementBase statementBase = selectStmtLists.get(idx);
OriginStatement originStatement = new OriginStatement(
StringUtils.isEmpty(statementBase.getOrigStmt().originStmt)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExportCommand.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExportCommand.java
index 713ff6b2760..e5484336491 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExportCommand.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/ExportCommand.java
@@ -73,6 +73,7 @@ import java.util.stream.Collectors;
public class ExportCommand extends Command implements ForwardWithSync {
public static final String PARALLELISM = "parallelism";
public static final String LABEL = "label";
+ public static final String DATA_CONSISTENCY = "data_consistency";
private static final String DEFAULT_COLUMN_SEPARATOR = "\t";
private static final String DEFAULT_LINE_DELIMITER = "\n";
private static final String DEFAULT_PARALLELISM = "1";
@@ -81,6 +82,7 @@ public class ExportCommand extends Command implements ForwardWithSync {
private static final ImmutableSet<String> PROPERTIES_SET = new ImmutableSet.Builder<String>()
.add(LABEL)
.add(PARALLELISM)
+ .add(DATA_CONSISTENCY)
.add(LoadStmt.KEY_IN_PARAM_COLUMNS)
.add(OutFileClause.PROP_MAX_FILE_SIZE)
.add(OutFileClause.PROP_DELETE_EXISTING_FILES)
@@ -310,6 +312,17 @@ public class ExportCommand extends Command implements ForwardWithSync {
exportJob.setQualifiedUser(ctx.getQualifiedUser());
exportJob.setUserIdentity(ctx.getCurrentUserIdentity());
+ // set data consistency
+ String dataConsistencyStr = fileProperties.get(DATA_CONSISTENCY);
+ if (dataConsistencyStr != null) {
+ if (!dataConsistencyStr.equalsIgnoreCase(ExportJob.CONSISTENT_PARTITION)) {
+ throw new AnalysisException("The value of data_consistency is invalid, only partition is allowed!");
+ }
+ exportJob.setDataConsistency(ExportJob.CONSISTENT_PARTITION);
+ } else {
+ exportJob.setDataConsistency(ExportJob.CONSISTENT_ALL);
+ }
+
// Must copy session variable, because session variable may be changed during export job running.
SessionVariable clonedSessionVariable = VariableMgr.cloneSessionVariable(Optional.ofNullable(
ConnectContext.get().getSessionVariable()).orElse(VariableMgr.getDefaultSessionVariable()));
diff --git a/regression-test/data/export_p0/test_export_data_consistency.out b/regression-test/data/export_p0/test_export_data_consistency.out
new file mode 100644
index 00000000000..97135d2bb98
--- /dev/null
+++ b/regression-test/data/export_p0/test_export_data_consistency.out
@@ -0,0 +1,305 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_export --
+1 ftw-1 19
+2 ftw-2 20
+3 ftw-3 21
+4 ftw-4 22
+5 ftw-5 23
+6 ftw-6 24
+7 ftw-7 25
+8 ftw-8 26
+9 ftw-9 27
+10 ftw-10 28
+11 ftw-11 29
+12 ftw-12 30
+13 ftw-13 31
+14 ftw-14 32
+15 ftw-15 33
+16 ftw-16 34
+17 ftw-17 35
+18 ftw-18 36
+19 ftw-19 37
+20 ftw-20 38
+21 ftw-21 39
+22 ftw-22 40
+23 ftw-23 41
+24 ftw-24 42
+25 ftw-25 43
+26 ftw-26 44
+27 ftw-27 45
+28 ftw-28 46
+29 ftw-29 47
+30 ftw-30 48
+31 ftw-31 49
+32 ftw-32 50
+33 ftw-33 51
+34 ftw-34 52
+35 ftw-35 53
+36 ftw-36 54
+37 ftw-37 55
+38 ftw-38 56
+39 ftw-39 57
+40 ftw-40 58
+41 ftw-41 59
+42 ftw-42 60
+43 ftw-43 61
+44 ftw-44 62
+45 ftw-45 63
+46 ftw-46 64
+47 ftw-47 65
+48 ftw-48 66
+49 ftw-49 67
+50 ftw-50 68
+51 ftw-51 69
+52 ftw-52 70
+53 ftw-53 71
+54 ftw-54 72
+55 ftw-55 73
+56 ftw-56 74
+57 ftw-57 75
+58 ftw-58 76
+59 ftw-59 77
+60 ftw-60 78
+61 ftw-61 79
+62 ftw-62 80
+63 ftw-63 81
+64 ftw-64 82
+65 ftw-65 83
+66 ftw-66 84
+67 ftw-67 85
+68 ftw-68 86
+69 ftw-69 87
+70 ftw-70 88
+71 ftw-71 89
+72 ftw-72 90
+73 ftw-73 91
+74 ftw-74 92
+75 ftw-75 93
+76 ftw-76 94
+77 ftw-77 95
+78 ftw-78 96
+79 ftw-79 97
+80 ftw-80 98
+81 ftw-81 99
+82 ftw-82 100
+83 ftw-83 101
+84 ftw-84 102
+85 ftw-85 103
+86 ftw-86 104
+87 ftw-87 105
+88 ftw-88 106
+89 ftw-89 107
+90 ftw-90 108
+91 ftw-91 109
+92 ftw-92 110
+93 ftw-93 111
+94 ftw-94 112
+95 ftw-95 113
+96 ftw-96 114
+97 ftw-97 115
+98 ftw-98 116
+99 ftw-99 117
+100 ftw-100 118
+101 ftw-101 119
+102 ftw-102 120
+103 ftw-103 121
+104 ftw-104 122
+105 ftw-105 123
+106 ftw-106 124
+107 ftw-107 125
+108 ftw-108 126
+109 ftw-109 127
+110 ftw-110 128
+111 ftw-111 129
+112 ftw-112 130
+113 ftw-113 131
+114 ftw-114 132
+115 ftw-115 133
+116 ftw-116 134
+117 ftw-117 135
+118 ftw-118 136
+119 ftw-119 137
+120 ftw-120 138
+121 ftw-121 139
+122 ftw-122 140
+123 ftw-123 141
+124 ftw-124 142
+125 ftw-125 143
+126 ftw-126 144
+127 ftw-127 145
+128 ftw-128 146
+129 ftw-129 147
+130 ftw-130 148
+131 ftw-131 149
+132 ftw-132 150
+133 ftw-133 151
+134 ftw-134 152
+135 ftw-135 153
+136 ftw-136 154
+137 ftw-137 155
+138 ftw-138 156
+139 ftw-139 157
+140 ftw-140 158
+141 ftw-141 159
+142 ftw-142 160
+143 ftw-143 161
+144 ftw-144 162
+145 ftw-145 163
+146 ftw-146 164
+147 ftw-147 165
+148 ftw-148 166
+149 ftw-149 167
+150 \N \N
+
+-- !select_load1 --
+1 ftw-1 19
+2 ftw-2 20
+3 ftw-3 21
+4 ftw-4 22
+5 ftw-5 23
+6 ftw-6 24
+7 ftw-7 25
+8 ftw-8 26
+9 ftw-9 27
+10 ftw-10 28
+11 ftw-11 29
+12 ftw-12 30
+13 ftw-13 31
+14 ftw-14 32
+15 ftw-15 33
+16 ftw-16 34
+17 ftw-17 35
+18 ftw-18 36
+19 ftw-19 37
+20 ftw-20 38
+21 ftw-21 39
+22 ftw-22 40
+23 ftw-23 41
+24 ftw-24 42
+25 ftw-25 43
+26 ftw-26 44
+27 ftw-27 45
+28 ftw-28 46
+29 ftw-29 47
+30 ftw-30 48
+31 ftw-31 49
+32 ftw-32 50
+33 ftw-33 51
+34 ftw-34 52
+35 ftw-35 53
+36 ftw-36 54
+37 ftw-37 55
+38 ftw-38 56
+39 ftw-39 57
+40 ftw-40 58
+41 ftw-41 59
+42 ftw-42 60
+43 ftw-43 61
+44 ftw-44 62
+45 ftw-45 63
+46 ftw-46 64
+47 ftw-47 65
+48 ftw-48 66
+49 ftw-49 67
+50 ftw-50 68
+51 ftw-51 69
+52 ftw-52 70
+53 ftw-53 71
+54 ftw-54 72
+55 ftw-55 73
+56 ftw-56 74
+57 ftw-57 75
+58 ftw-58 76
+59 ftw-59 77
+60 ftw-60 78
+61 ftw-61 79
+62 ftw-62 80
+63 ftw-63 81
+64 ftw-64 82
+65 ftw-65 83
+66 ftw-66 84
+67 ftw-67 85
+68 ftw-68 86
+69 ftw-69 87
+70 ftw-70 88
+71 ftw-71 89
+72 ftw-72 90
+73 ftw-73 91
+74 ftw-74 92
+75 ftw-75 93
+76 ftw-76 94
+77 ftw-77 95
+78 ftw-78 96
+79 ftw-79 97
+80 ftw-80 98
+81 ftw-81 99
+82 ftw-82 100
+83 ftw-83 101
+84 ftw-84 102
+85 ftw-85 103
+86 ftw-86 104
+87 ftw-87 105
+88 ftw-88 106
+89 ftw-89 107
+90 ftw-90 108
+91 ftw-91 109
+92 ftw-92 110
+93 ftw-93 111
+94 ftw-94 112
+95 ftw-95 113
+96 ftw-96 114
+97 ftw-97 115
+98 ftw-98 116
+99 ftw-99 117
+100 ftw-100 118
+101 ftw-101 119
+102 ftw-102 120
+103 ftw-103 121
+104 ftw-104 122
+105 ftw-105 123
+106 ftw-106 124
+107 ftw-107 125
+108 ftw-108 126
+109 ftw-109 127
+110 ftw-110 128
+111 ftw-111 129
+112 ftw-112 130
+113 ftw-113 131
+114 ftw-114 132
+115 ftw-115 133
+116 ftw-116 134
+117 ftw-117 135
+118 ftw-118 136
+119 ftw-119 137
+120 ftw-120 138
+121 ftw-121 139
+122 ftw-122 140
+123 ftw-123 141
+124 ftw-124 142
+125 ftw-125 143
+126 ftw-126 144
+127 ftw-127 145
+128 ftw-128 146
+129 ftw-129 147
+130 ftw-130 148
+131 ftw-131 149
+132 ftw-132 150
+133 ftw-133 151
+134 ftw-134 152
+135 ftw-135 153
+136 ftw-136 154
+137 ftw-137 155
+138 ftw-138 156
+139 ftw-139 157
+140 ftw-140 158
+141 ftw-141 159
+142 ftw-142 160
+143 ftw-143 161
+144 ftw-144 162
+145 ftw-145 163
+146 ftw-146 164
+147 ftw-147 165
+148 ftw-148 166
+149 ftw-149 167
+150 \N \N
+
diff --git a/regression-test/suites/export_p0/test_export_data_consistency.groovy b/regression-test/suites/export_p0/test_export_data_consistency.groovy
new file mode 100644
index 00000000000..87d2eb4a7b7
--- /dev/null
+++ b/regression-test/suites/export_p0/test_export_data_consistency.groovy
@@ -0,0 +1,210 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.codehaus.groovy.runtime.IOGroovyMethods
+
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.nio.file.Paths
+
+suite("test_export_data_consistency", "p0") {
+ // open nereids
+ sql """ set enable_nereids_planner=true """
+ sql """ set enable_fallback_to_original_planner=false """
+
+ def db = "regression_test_export_p0"
+
+ // check whether the FE config 'enable_outfile_to_local' is true
+ StringBuilder strBuilder = new StringBuilder()
+ strBuilder.append("curl --location-trusted -u " + context.config.jdbcUser + ":" + context.config.jdbcPassword)
+ strBuilder.append(" http://" + context.config.feHttpAddress + "/rest/v1/config/fe")
+
+ String command = strBuilder.toString()
+ def process = command.toString().execute()
+ def code = process.waitFor()
+ def err = IOGroovyMethods.getText(new BufferedReader(new InputStreamReader(process.getErrorStream())));
+ def out = process.getText()
+ logger.info("Request FE Config: code=" + code + ", out=" + out + ", err=" + err)
+ assertEquals(code, 0)
+ def response = parseJson(out.trim())
+ assertEquals(response.code, 0)
+ assertEquals(response.msg, "success")
+ def configJson = response.data.rows
+ boolean enableOutfileToLocal = false
+ for (Object conf: configJson) {
+ assert conf instanceof Map
+ if (((Map<String, String>) conf).get("Name").toLowerCase() == "enable_outfile_to_local") {
+ enableOutfileToLocal = ((Map<String, String>) conf).get("Value").toLowerCase() == "true"
+ }
+ }
+ if (!enableOutfileToLocal) {
+ logger.warn("Please set enable_outfile_to_local to true to run test_outfile")
+ return
+ }
+
+ def table_export_name = "test_export_data_consistency"
+ def table_load_name = "test_load_data_consistency"
+ def outfile_path_prefix = """/tmp/test_export_data_consistency"""
+
+ // create table and insert
+ sql """ DROP TABLE IF EXISTS ${table_export_name} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${table_export_name} (
+ `id` int(11) NULL,
+ `name` string NULL,
+ `age` int(11) NULL
+ )
+ PARTITION BY RANGE(id)
+ (
+ PARTITION less_than_20 VALUES LESS THAN ("20"),
+ PARTITION between_20_70 VALUES [("20"),("70")),
+ PARTITION more_than_70 VALUES LESS THAN ("151")
+ )
+ DISTRIBUTED BY HASH(id) BUCKETS 3
+ PROPERTIES("replication_num" = "1");
+ """
+ StringBuilder sb = new StringBuilder()
+ int i = 1
+ for (; i < 150; i ++) {
+ sb.append("""
+ (${i}, 'ftw-${i}', ${i + 18}),
+ """)
+ }
+ sb.append("""
+ (${i}, NULL, NULL)
+ """)
+ sql """ INSERT INTO ${table_export_name} VALUES
+ ${sb.toString()}
+ """
+ def insert_res = sql "show last insert;"
+ logger.info("insert result: " + insert_res.toString())
+ qt_select_export """ SELECT * FROM ${table_export_name} t ORDER BY id; """
+
+
+ def check_path_exists = { dir_path ->
+ File path = new File(dir_path)
+ if (!path.exists()) {
+ assert path.mkdirs()
+ } else {
+ throw new IllegalStateException("""${dir_path} already exists! """)
+ }
+ }
+
+ def check_file_amounts = { dir_path, amount ->
+ File path = new File(dir_path)
+ File[] files = path.listFiles()
+ assert files.length == amount
+ }
+
+ def delete_files = { dir_path ->
+ File path = new File(dir_path)
+ if (path.exists()) {
+ for (File f: path.listFiles()) {
+ f.delete();
+ }
+ path.delete();
+ }
+ }
+
+ def waiting_export = { the_db, export_label ->
+ while (true) {
+ def res = sql """ show export from ${the_db} where label = "${export_label}" """
+ logger.info("export state: " + res[0][2])
+ if (res[0][2] == "FINISHED") {
+ break;
+ } else if (res[0][2] == "CANCELLED") {
+ throw new IllegalStateException("""export failed: ${res[0][10]}""")
+ } else {
+ sleep(5000)
+ }
+ }
+ }
+
+ // 1. basic test
+ def uuid = UUID.randomUUID().toString()
+ def outFilePath = """${outfile_path_prefix}_${uuid}"""
+ def label = "label_${uuid}"
+ try {
+ // check export path
+ check_path_exists.call("${outFilePath}")
+
+ // exec export
+ sql """
+ EXPORT TABLE ${table_export_name} TO "file://${outFilePath}/"
+ PROPERTIES(
+ "label" = "${label}",
+ "format" = "csv",
+ "column_separator" = ",",
+ "data_consistency" = "partition"
+ );
+ """
+ // do insert in parallel
+ sql """INSERT INTO ${table_export_name} VALUES
+ (10, 'test', 11),
+ (20, 'test', 21),
+ (40, 'test', 51),
+ (80, 'test', 51)
+ """
+
+ // wait export
+ waiting_export.call(db, label)
+
+ // check file amounts
+ check_file_amounts.call("${outFilePath}", 3)
+
+ // check data correctness
+ sql """ DROP TABLE IF EXISTS ${table_load_name} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${table_load_name} (
+ `id` int(11) NULL,
+ `name` string NULL,
+ `age` int(11) NULL
+ )
+ DISTRIBUTED BY HASH(id) PROPERTIES("replication_num" = "1");
+ """
+
+ File[] files = new File("${outFilePath}").listFiles()
+ for (exportLoadFile in files) {
+ String file_path = exportLoadFile.getAbsolutePath()
+ streamLoad {
+ table "${table_load_name}"
+
+ set 'column_separator', ','
+ set 'columns', 'id, name, age'
+ set 'strict_mode', 'true'
+
+ file "${file_path}"
+ time 10000 // limit inflight 10s
+
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ log.info("Stream load result: ${result}".toString())
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(0, json.NumberFilteredRows)
+ }
+ }
+ }
+
+ qt_select_load1 """ SELECT * FROM ${table_load_name} t ORDER BY id; """
+ } finally {
+ try_sql("DROP TABLE IF EXISTS ${table_load_name}")
+ delete_files.call("${outFilePath}")
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 10/14: [doc](fix) add k8s operator doc for upgrade doris (#31586)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 6b2af74a63fa5ce8fe2ff7fe6ff33dc021b7152a
Author: catpineapple <42...@users.noreply.github.com>
AuthorDate: Thu Feb 29 23:33:46 2024 +0800
[doc](fix) add k8s operator doc for upgrade doris (#31586)
---
.../install/k8s-deploy/doris-cluster-upgrade.md | 286 ++++++++++++++++++++
docs/sidebars.json | 1 +
.../install/k8s-deploy/doris-cluster-upgrade.md | 291 +++++++++++++++++++++
3 files changed, 578 insertions(+)
diff --git a/docs/en/docs/install/k8s-deploy/doris-cluster-upgrade.md b/docs/en/docs/install/k8s-deploy/doris-cluster-upgrade.md
new file mode 100644
index 00000000000..43a714950ea
--- /dev/null
+++ b/docs/en/docs/install/k8s-deploy/doris-cluster-upgrade.md
@@ -0,0 +1,286 @@
+---
+{
+"title": "Upgrade Apache Doris cluster deployed by Doris-Operator",
+"language": "en"
+}
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+This document describes how to use updates to upgrade an Apache Doris cluster based on a Doris-Operator deployment.
+
+Similar to conventionally deployed cluster upgrades, Doris clusters deployed by Doris-Operator still require rolling upgrades from BE to FE nodes. Doris-Operator is based on Kubernetes' [Performing a Rolling Update](https://kubernetes.io/docs/tutorials/kubernetes-basics/update/update-intro/) provides rolling upgrade capabilities.
+
+## Things to note before upgrading
+
+- It is recommended that the upgrade operation be performed during off-peak periods.
+- During the rolling upgrade process, the connection to the closed node will fail, causing the request to fail. For this type of business, it is recommended to add retry capabilities to the client.
+- Before upgrading, you can read the [General Upgrade Manual](https://doris.apache.org/docs/dev/admin-manual/cluster-management/upgrade) to help you understand some principles and precautions during the upgrade. .
+- The compatibility of data and metadata cannot be verified before upgrading. Therefore, cluster upgrade must avoid single copy of data and single FE FOLLOWER node in the cluster.
+- Nodes will be restarted during the upgrade process, so unnecessary cluster balancing and replica repair logic may be triggered. Please shut it down first with the following command.
+```
+admin set frontend config("disable_balance" = "true");
+admin set frontend config("disable_colocate_balance" = "true");
+admin set frontend config("disable_tablet_scheduler" = "true");
+```
+- When upgrading Doris, please follow the principle of not upgrading across two or more key node versions. If you want to upgrade across multiple key node versions, upgrade to the latest key node version first, and then upgrade in sequence. If it is a non-key node version, You can ignore skipping. For details, please refer to [Upgrade Version Instructions](https://doris.apache.org/docs/dev/admin-manual/cluster-management/upgrade/#doris-release-notes)
+
+## Upgrade operation
+
+The order of node types in the upgrade process is as follows. If a certain type of node does not exist, it will be skipped:
+```
+ cn/be -> fe -> broker
+```
+It is recommended to modify the `image` of the corresponding cluster components in sequence and then apply the configuration. After the current type of component is fully upgraded and the status returns to normal, the rolling upgrade of the next type of node can be performed.
+
+### Upgrade BE
+
+If you retain the cluster's crd (Doris-Operator defines the abbreviation of `DorisCluster` type resource name) file, you can upgrade by modifying the configuration file and running the `kubectl apply` command.
+
+1. Modify `spec.beSpec.image`
+
+ Change `selectdb/doris.be-ubuntu:2.0.4` to `selectdb/doris.be-ubuntu:2.1.0`
+```
+$ vim doriscluster-sample.yaml
+```
+
+2. Save the changes and apply the changes to be upgraded:
+```
+$ kubectl apply -f doriscluster-sample.yaml -n doris
+```
+
+It can also be modified directly through `kubectl edit dcr`.
+
+1. Check the dcr list under namespace 'doris' to obtain the `cluster_name` that needs to be updated.
+```
+$ kubectl get dcr -n doris
+NAME FESTATUS BESTATUS CNSTATUS
+doriscluster-sample available available
+```
+
+2. Modify, save and take effect
+```
+$ kubectl edit dcr doriscluster-sample -n doris
+```
+ After entering the text editor, you will find `spec.beSpec.image` and change `selectdb/doris.be-ubuntu:2.0.4` to `selectdb/doris.be-ubuntu:2.1.0`
+
+3. View the upgrade process and results:
+```
+$ kubectl get pod -n doris
+```
+
+When all Pods are rebuilt and enter the Running state, the upgrade is complete.
+
+### Upgrade FE
+
+If you retain the cluster's crd (Doris-Operator defines the abbreviation of the `DorisCluster` type resource name) file, you can upgrade by modifying the configuration file and running the `kubectl apply` command.
+
+1. Modify `spec.feSpec.image`
+
+ Change `selectdb/doris.fe-ubuntu:2.0.4` to `selectdb/doris.fe-ubuntu:2.1.0`
+```
+$ vim doriscluster-sample.yaml
+```
+
+2. Save the changes and apply the changes to be upgraded:
+```
+$ kubectl apply -f doriscluster-sample.yaml -n doris
+```
+
+It can also be modified directly through `kubectl edit dcr`.
+
+1. Modify, save and take effect
+```
+$ kubectl edit dcr doriscluster-sample -n doris
+```
+ After entering the text editor, you will find `spec.feSpec.image` and change `selectdb/doris.fe-ubuntu:2.0.4` to `selectdb/doris.fe-ubuntu:2.1.0`
+
+2. View the upgrade process and results:
+```
+$ kubectl get pod -n doris
+```
+
+When all Pods are rebuilt and enter the Running state, the upgrade is complete.
+
+## After the upgrade is completed
+### Verify cluster node status
+Access Doris through `mysql-client` through the method provided in the [Access Doris Cluster](https://doris.apache.org/docs/dev/install/k8s-deploy/network) document.
+Use SQL such as `show frontends` and `show backends` to view the version and status of each component.
+```
+mysql> show frontends\G;
+*************************** 1. row ***************************
+ Name: fe_13c132aa_3281_4f4f_97e8_655d01287425
+ Host: doriscluster-sample-fe-0.doriscluster-sample-fe-internal.doris.svc.cluster.local
+ EditLogPort: 9010
+ HttpPort: 8030
+ QueryPort: 9030
+ RpcPort: 9020
+ArrowFlightSqlPort: -1
+ Role: FOLLOWER
+ IsMaster: false
+ ClusterId: 1779160761
+ Join: true
+ Alive: true
+ ReplayedJournalId: 2422
+ LastStartTime: 2024-02-19 06:38:47
+ LastHeartbeat: 2024-02-19 09:31:33
+ IsHelper: true
+ ErrMsg:
+ Version: doris-2.1.0
+ CurrentConnected: Yes
+*************************** 2. row ***************************
+ Name: fe_f1a9d008_d110_4780_8e60_13d392faa54e
+ Host: doriscluster-sample-fe-2.doriscluster-sample-fe-internal.doris.svc.cluster.local
+ EditLogPort: 9010
+ HttpPort: 8030
+ QueryPort: 9030
+ RpcPort: 9020
+ArrowFlightSqlPort: -1
+ Role: FOLLOWER
+ IsMaster: true
+ ClusterId: 1779160761
+ Join: true
+ Alive: true
+ ReplayedJournalId: 2423
+ LastStartTime: 2024-02-19 06:37:35
+ LastHeartbeat: 2024-02-19 09:31:33
+ IsHelper: true
+ ErrMsg:
+ Version: doris-2.1.0
+ CurrentConnected: No
+*************************** 3. row ***************************
+ Name: fe_e42bf9da_006f_4302_b861_770d2c955a47
+ Host: doriscluster-sample-fe-1.doriscluster-sample-fe-internal.doris.svc.cluster.local
+ EditLogPort: 9010
+ HttpPort: 8030
+ QueryPort: 9030
+ RpcPort: 9020
+ArrowFlightSqlPort: -1
+ Role: FOLLOWER
+ IsMaster: false
+ ClusterId: 1779160761
+ Join: true
+ Alive: true
+ ReplayedJournalId: 2422
+ LastStartTime: 2024-02-19 06:38:17
+ LastHeartbeat: 2024-02-19 09:31:33
+ IsHelper: true
+ ErrMsg:
+ Version: doris-2.1.0
+ CurrentConnected: No
+3 rows in set (0.02 sec)
+```
+If the `Alive` status of the FE node is true and the `Version` value is the new version, the FE node is upgraded successfully.
+
+```
+mysql> show backends\G;
+*************************** 1. row ***************************
+ BackendId: 10002
+ Host: doriscluster-sample-be-0.doriscluster-sample-be-internal.doris.svc.cluster.local
+ HeartbeatPort: 9050
+ BePort: 9060
+ HttpPort: 8040
+ BrpcPort: 8060
+ ArrowFlightSqlPort: -1
+ LastStartTime: 2024-02-19 06:37:56
+ LastHeartbeat: 2024-02-19 09:32:43
+ Alive: true
+ SystemDecommissioned: false
+ TabletNum: 14
+ DataUsedCapacity: 0.000
+ TrashUsedCapcacity: 0.000
+ AvailCapacity: 12.719 GB
+ TotalCapacity: 295.167 GB
+ UsedPct: 95.69 %
+ MaxDiskUsedPct: 95.69 %
+ RemoteUsedCapacity: 0.000
+ Tag: {"location" : "default"}
+ ErrMsg:
+ Version: doris-2.1.0
+ Status: {"lastSuccessReportTabletsTime":"2024-02-19 09:31:48","lastStreamLoadTime":-1,"isQueryDisabled":false,"isLoadDisabled":false}
+HeartbeatFailureCounter: 0
+ NodeRole: mix
+*************************** 2. row ***************************
+ BackendId: 10003
+ Host: doriscluster-sample-be-1.doriscluster-sample-be-internal.doris.svc.cluster.local
+ HeartbeatPort: 9050
+ BePort: 9060
+ HttpPort: 8040
+ BrpcPort: 8060
+ ArrowFlightSqlPort: -1
+ LastStartTime: 2024-02-19 06:37:35
+ LastHeartbeat: 2024-02-19 09:32:43
+ Alive: true
+ SystemDecommissioned: false
+ TabletNum: 8
+ DataUsedCapacity: 0.000
+ TrashUsedCapcacity: 0.000
+ AvailCapacity: 12.719 GB
+ TotalCapacity: 295.167 GB
+ UsedPct: 95.69 %
+ MaxDiskUsedPct: 95.69 %
+ RemoteUsedCapacity: 0.000
+ Tag: {"location" : "default"}
+ ErrMsg:
+ Version: doris-2.1.0
+ Status: {"lastSuccessReportTabletsTime":"2024-02-19 09:31:43","lastStreamLoadTime":-1,"isQueryDisabled":false,"isLoadDisabled":false}
+HeartbeatFailureCounter: 0
+ NodeRole: mix
+*************************** 3. row ***************************
+ BackendId: 11024
+ Host: doriscluster-sample-be-2.doriscluster-sample-be-internal.doris.svc.cluster.local
+ HeartbeatPort: 9050
+ BePort: 9060
+ HttpPort: 8040
+ BrpcPort: 8060
+ ArrowFlightSqlPort: -1
+ LastStartTime: 2024-02-19 08:50:36
+ LastHeartbeat: 2024-02-19 09:32:43
+ Alive: true
+ SystemDecommissioned: false
+ TabletNum: 0
+ DataUsedCapacity: 0.000
+ TrashUsedCapcacity: 0.000
+ AvailCapacity: 12.719 GB
+ TotalCapacity: 295.167 GB
+ UsedPct: 95.69 %
+ MaxDiskUsedPct: 95.69 %
+ RemoteUsedCapacity: 0.000
+ Tag: {"location" : "default"}
+ ErrMsg:
+ Version: doris-2.1.0
+ Status: {"lastSuccessReportTabletsTime":"2024-02-19 09:32:04","lastStreamLoadTime":-1,"isQueryDisabled":false,"isLoadDisabled":false}
+HeartbeatFailureCounter: 0
+ NodeRole: mix
+3 rows in set (0.01 sec)
+```
+
+If the `Alive` status of the BE node is true and the `Version` value is the new version, the BE node is upgraded successfully.
+
+### Restore cluster replica synchronization and balancing
+After confirming that the status of each node is correct, execute the following SQL to restore cluster balancing and replica repair:
+```
+admin set frontend config("disable_balance" = "false");
+admin set frontend config("disable_colocate_balance" = "false");
+admin set frontend config("disable_tablet_scheduler" = "false");
+```
+
+
+
diff --git a/docs/sidebars.json b/docs/sidebars.json
index b44a593e384..432d0a614e9 100644
--- a/docs/sidebars.json
+++ b/docs/sidebars.json
@@ -27,6 +27,7 @@
"items": [
"install/k8s-deploy/operator-deploy",
"install/k8s-deploy/helm-chart-deploy",
+ "install/k8s-deploy/doris-cluster-upgrade",
"install/k8s-deploy/network",
"install/k8s-deploy/persistent-volume",
"install/k8s-deploy/root-user-use",
diff --git a/docs/zh-CN/docs/install/k8s-deploy/doris-cluster-upgrade.md b/docs/zh-CN/docs/install/k8s-deploy/doris-cluster-upgrade.md
new file mode 100644
index 00000000000..7cd2690456f
--- /dev/null
+++ b/docs/zh-CN/docs/install/k8s-deploy/doris-cluster-upgrade.md
@@ -0,0 +1,291 @@
+---
+{
+"title": "升级基于 Doris-Operator 部署的 Apache Doris 集群",
+"language": "zh-CN"
+}
+---
+
+<!--
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+-->
+
+
+本文介绍如何使用 更新来升级 基于 Doris-Operator 部署的 Apache Doris 集群。
+
+和常规部署的集群升级类似，Doris-Operator 部署的 Doris 集群依然需要 BE 到 FE 节点滚动升级，Doris-Operator 基于 Kubernetes 的 [滚动更新功能](https://kubernetes.io/docs/tutorials/kubernetes-basics/update/update-intro/) 提供了滚动升级能力。
+
+## 升级前注意事项
+
+- 升级操作推荐在业务低峰期进行。
+- 滚动升级过程中，会导致连接到被关闭节点的连接失效，造成请求失败，对于这类业务，推荐在客户端添加重试能力。
+- 升级前可以阅读 [常规升级手册](https://doris.apache.org/zh-CN/docs/dev/admin-manual/cluster-management/upgrade)，便于理解升级中的一些原理和注意事项。
+- 升级前无法对数据和元数据的兼容性进行验证，因此集群升级一定要避免数据存在 单副本 情况 和 集群单 FE FOLLOWER 节点。
+- 升级过程中会有节点重启，所以可能会触发不必要的集群均衡和副本修复逻辑，先通过以下命令关闭
+```
+admin set frontend config("disable_balance" = "true");
+admin set frontend config("disable_colocate_balance" = "true");
+admin set frontend config("disable_tablet_scheduler" = "true");
+```
+- Doris 升级请遵守不要跨两个及以上关键节点版本升级的原则，若要跨多个关键节点版本升级，先升级到最近的关键节点版本，随后再依次往后升级，若是非关键节点版本，则可忽略跳过。具体参考 [升级版本说明](https://doris.apache.org/zh-CN/docs/dev/admin-manual/cluster-management/upgrade#doris-%E7%89%88%E6%9C%AC%E8%AF%B4%E6%98%8E)
+
+## 升级操作
+
+升级过程节点类型顺序如下，如果某类型节点不存在则跳过：
+```
+ cn/be -> fe -> broker
+```
+建议依次修改对应集群组件的 `image` 然后 应用该配置，待当前类型的组件完全升级成功状态恢复正常后，再进行下一个类型节点的滚动升级。
+
+### 升级 BE
+
+如果保留了集群的 crd（Doris-Operator 定义了 `DorisCluster` 类型资源名称的简写）文件，则可以通过修改该配置文件并且 `kubectl apply` 的命令来进行升级。
+
+1. 修改 `spec.beSpec.image`
+
+ 将 `selectdb/doris.be-ubuntu:2.0.4` 变为 `selectdb/doris.be-ubuntu:2.1.0`
+```
+$ vim doriscluster-sample.yaml
+```
+
+2. 保存修改后应用本次修改进行be升级:
+```
+$ kubectl apply -f doriscluster-sample.yaml -n doris
+```
+也可通过 `kubectl edit dcr` 的方式直接修改。
+
+1. 查看 namespace 为 'doris' 下的 dcr 列表，获取需要更新的 `cluster_name`
+```
+$ kubectl get dcr -n doris
+NAME FESTATUS BESTATUS CNSTATUS
+doriscluster-sample available available
+```
+
+2. 修改、保存并生效
+```
+$ kubectl edit dcr doriscluster-sample -n doris
+```
+ 进入文本编辑器后，将找到`spec.beSpec.image`，将 `selectdb/doris.be-ubuntu:2.0.4` 修改为 `selectdb/doris.be-ubuntu:2.1.0`
+
+3. 查看升级过程和结果：
+```
+$ kubectl get pod -n doris
+```
+
+当所有 Pod 都重建完毕进入 Running 状态后，升级完成。
+
+### 升级 FE
+
+如果保留了集群的 crd （ Doris-Operator 定义了 `DorisCluster` 类型资源名称的简写）文件，则可以通过修改该配置文件并且 `kubectl apply` 的命令来进行升级。
+
+1. 修改 `spec.feSpec.image`
+
+ 将 `selectdb/doris.fe-ubuntu:2.0.4` 变为 `selectdb/doris.fe-ubuntu:2.1.0`
+```
+$ vim doriscluster-sample.yaml
+```
+
+2. 保存修改后应用本次修改进行be升级:
+```
+$ kubectl apply -f doriscluster-sample.yaml -n doris
+```
+
+也可通过 `kubectl edit dcr` 的方式直接修改。
+
+1. 修改、保存并生效
+```
+$ kubectl edit dcr doriscluster-sample -n doris
+```
+ 进入文本编辑器后，将找到`spec.feSpec.image`，将 `selectdb/doris.fe-ubuntu:2.0.4` 修改为 `selectdb/doris.fe-ubuntu:2.1.0`
+
+2. 查看升级过程和结果
+```
+$ kubectl get pod -n doris
+```
+
+当所有 Pod 都重建完毕进入 Running 状态后，升级完成。
+
+## 升级完成后
+### 验证集群节点状态
+通过 [访问 Doris 集群](https://doris.apache.org/zh-CN/docs/dev/install/k8s-deploy/network) 文档提供的方式，通过 `mysql-client` 访问 Doris。
+使用 `show frontends` 和 `show backends` 等 SQL 查看各个组件的 版本 和 状态。
+```
+mysql> show frontends\G;
+*************************** 1. row ***************************
+ Name: fe_13c132aa_3281_4f4f_97e8_655d01287425
+ Host: doriscluster-sample-fe-0.doriscluster-sample-fe-internal.doris.svc.cluster.local
+ EditLogPort: 9010
+ HttpPort: 8030
+ QueryPort: 9030
+ RpcPort: 9020
+ArrowFlightSqlPort: -1
+ Role: FOLLOWER
+ IsMaster: false
+ ClusterId: 1779160761
+ Join: true
+ Alive: true
+ ReplayedJournalId: 2422
+ LastStartTime: 2024-02-19 06:38:47
+ LastHeartbeat: 2024-02-19 09:31:33
+ IsHelper: true
+ ErrMsg:
+ Version: doris-2.1.0
+ CurrentConnected: Yes
+*************************** 2. row ***************************
+ Name: fe_f1a9d008_d110_4780_8e60_13d392faa54e
+ Host: doriscluster-sample-fe-2.doriscluster-sample-fe-internal.doris.svc.cluster.local
+ EditLogPort: 9010
+ HttpPort: 8030
+ QueryPort: 9030
+ RpcPort: 9020
+ArrowFlightSqlPort: -1
+ Role: FOLLOWER
+ IsMaster: true
+ ClusterId: 1779160761
+ Join: true
+ Alive: true
+ ReplayedJournalId: 2423
+ LastStartTime: 2024-02-19 06:37:35
+ LastHeartbeat: 2024-02-19 09:31:33
+ IsHelper: true
+ ErrMsg:
+ Version: doris-2.1.0
+ CurrentConnected: No
+*************************** 3. row ***************************
+ Name: fe_e42bf9da_006f_4302_b861_770d2c955a47
+ Host: doriscluster-sample-fe-1.doriscluster-sample-fe-internal.doris.svc.cluster.local
+ EditLogPort: 9010
+ HttpPort: 8030
+ QueryPort: 9030
+ RpcPort: 9020
+ArrowFlightSqlPort: -1
+ Role: FOLLOWER
+ IsMaster: false
+ ClusterId: 1779160761
+ Join: true
+ Alive: true
+ ReplayedJournalId: 2422
+ LastStartTime: 2024-02-19 06:38:17
+ LastHeartbeat: 2024-02-19 09:31:33
+ IsHelper: true
+ ErrMsg:
+ Version: doris-2.1.0
+ CurrentConnected: No
+3 rows in set (0.02 sec)
+```
+若 FE 节点 `alive` 状态为 true，且 `Version` 值为新版本，则该 FE 节点升级成功。
+
+```
+mysql> show backends\G;
+*************************** 1. row ***************************
+ BackendId: 10002
+ Host: doriscluster-sample-be-0.doriscluster-sample-be-internal.doris.svc.cluster.local
+ HeartbeatPort: 9050
+ BePort: 9060
+ HttpPort: 8040
+ BrpcPort: 8060
+ ArrowFlightSqlPort: -1
+ LastStartTime: 2024-02-19 06:37:56
+ LastHeartbeat: 2024-02-19 09:32:43
+ Alive: true
+ SystemDecommissioned: false
+ TabletNum: 14
+ DataUsedCapacity: 0.000
+ TrashUsedCapcacity: 0.000
+ AvailCapacity: 12.719 GB
+ TotalCapacity: 295.167 GB
+ UsedPct: 95.69 %
+ MaxDiskUsedPct: 95.69 %
+ RemoteUsedCapacity: 0.000
+ Tag: {"location" : "default"}
+ ErrMsg:
+ Version: doris-2.1.0
+ Status: {"lastSuccessReportTabletsTime":"2024-02-19 09:31:48","lastStreamLoadTime":-1,"isQueryDisabled":false,"isLoadDisabled":false}
+HeartbeatFailureCounter: 0
+ NodeRole: mix
+*************************** 2. row ***************************
+ BackendId: 10003
+ Host: doriscluster-sample-be-1.doriscluster-sample-be-internal.doris.svc.cluster.local
+ HeartbeatPort: 9050
+ BePort: 9060
+ HttpPort: 8040
+ BrpcPort: 8060
+ ArrowFlightSqlPort: -1
+ LastStartTime: 2024-02-19 06:37:35
+ LastHeartbeat: 2024-02-19 09:32:43
+ Alive: true
+ SystemDecommissioned: false
+ TabletNum: 8
+ DataUsedCapacity: 0.000
+ TrashUsedCapcacity: 0.000
+ AvailCapacity: 12.719 GB
+ TotalCapacity: 295.167 GB
+ UsedPct: 95.69 %
+ MaxDiskUsedPct: 95.69 %
+ RemoteUsedCapacity: 0.000
+ Tag: {"location" : "default"}
+ ErrMsg:
+ Version: doris-2.1.0
+ Status: {"lastSuccessReportTabletsTime":"2024-02-19 09:31:43","lastStreamLoadTime":-1,"isQueryDisabled":false,"isLoadDisabled":false}
+HeartbeatFailureCounter: 0
+ NodeRole: mix
+*************************** 3. row ***************************
+ BackendId: 11024
+ Host: doriscluster-sample-be-2.doriscluster-sample-be-internal.doris.svc.cluster.local
+ HeartbeatPort: 9050
+ BePort: 9060
+ HttpPort: 8040
+ BrpcPort: 8060
+ ArrowFlightSqlPort: -1
+ LastStartTime: 2024-02-19 08:50:36
+ LastHeartbeat: 2024-02-19 09:32:43
+ Alive: true
+ SystemDecommissioned: false
+ TabletNum: 0
+ DataUsedCapacity: 0.000
+ TrashUsedCapcacity: 0.000
+ AvailCapacity: 12.719 GB
+ TotalCapacity: 295.167 GB
+ UsedPct: 95.69 %
+ MaxDiskUsedPct: 95.69 %
+ RemoteUsedCapacity: 0.000
+ Tag: {"location" : "default"}
+ ErrMsg:
+ Version: doris-2.1.0
+ Status: {"lastSuccessReportTabletsTime":"2024-02-19 09:32:04","lastStreamLoadTime":-1,"isQueryDisabled":false,"isLoadDisabled":false}
+HeartbeatFailureCounter: 0
+ NodeRole: mix
+3 rows in set (0.01 sec)
+```
+若 BE 节点 `alive` 状态为 true，且 `Version` 值为新版本，则该 BE 节点升级成功
+
+### 恢复集群副本同步和均衡
+在确认各个节点状态无误后，执行以下 SQL 恢复集群均衡和副本修复：
+```
+admin set frontend config("disable_balance" = "false");
+admin set frontend config("disable_colocate_balance" = "false");
+admin set frontend config("disable_tablet_scheduler" = "false");
+```
+
+
+
+
+
+
+
+
+
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 14/14: [nereids] explode function stats configurable (#31332)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 0d77fc484747ea26822abc868e7a11c15994efbb
Author: xzj7019 <13...@users.noreply.github.com>
AuthorDate: Thu Feb 29 23:52:19 2024 +0800
[nereids] explode function stats configurable (#31332)
---------
Co-authored-by: zhongjian.xzj <zh...@zhongjianxzjdeMacBook-Pro.local>
---
.../doris/nereids/stats/StatsCalculator.java | 3 +-
.../java/org/apache/doris/qe/SessionVariable.java | 33 ++++++++++++++++++++++
2 files changed, 35 insertions(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
index ad16d3e8f50..08d4a3e781c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/stats/StatsCalculator.java
@@ -957,7 +957,8 @@ public class StatsCalculator extends DefaultPlanVisitor<Statistics, Void> {
private Statistics computeGenerate(Generate generate) {
Statistics stats = groupExpression.childStatistics(0);
- double count = stats.getRowCount() * generate.getGeneratorOutput().size() * 5;
+ int statsFactor = ConnectContext.get().getSessionVariable().generateStatsFactor;
+ double count = stats.getRowCount() * generate.getGeneratorOutput().size() * statsFactor;
Map<Expression, ColumnStatistic> columnStatsMap = Maps.newHashMap();
for (Map.Entry<Expression, ColumnStatistic> entry : stats.columnStatistics().entrySet()) {
ColumnStatistic columnStatistic = new ColumnStatisticBuilder(entry.getValue()).setCount(count).build();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 3a6f120d501..fe9d47fefde 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -479,6 +479,8 @@ public class SessionVariable implements Serializable, Writable {
public static final String HUGE_TABLE_DEFAULT_SAMPLE_ROWS = "huge_table_default_sample_rows";
public static final String HUGE_TABLE_LOWER_BOUND_SIZE_IN_BYTES = "huge_table_lower_bound_size_in_bytes";
+ public static final String GENERATE_STATS_FACTOR = "generate_stats_factor";
+
public static final String HUGE_TABLE_AUTO_ANALYZE_INTERVAL_IN_MILLIS
= "huge_table_auto_analyze_interval_in_millis";
@@ -740,6 +742,13 @@ public class SessionVariable implements Serializable, Writable {
@VariableMgr.VarAttr(name = ENABLE_BUCKET_SHUFFLE_DOWNGRADE, needForward = true)
public boolean enableBucketShuffleDownGrade = false;
+ /**
+ * explode function row count enlarge factor.
+ */
+ @VariableMgr.VarAttr(name = GENERATE_STATS_FACTOR, checker = "checkGenerateStatsFactor",
+ setter = "setGenerateStatsFactor")
+ public int generateStatsFactor = 5;
+
@VariableMgr.VarAttr(name = PREFER_JOIN_METHOD)
public String preferJoinMethod = "broadcast";
@@ -3324,6 +3333,30 @@ public class SessionVariable implements Serializable, Writable {
}
}
+ public void checkGenerateStatsFactor(String generateStatsFactor) {
+ int value = Integer.valueOf(generateStatsFactor);
+ if (value <= 0) {
+ UnsupportedOperationException exception =
+ new UnsupportedOperationException("Generate stats factor " + value + " should greater than 0");
+ LOG.warn("Check generate stats factor failed", exception);
+ throw exception;
+ }
+ }
+
+ public void setGenerateStatsFactor(int factor) {
+ this.generateStatsFactor = factor;
+ if (factor <= 0) {
+ LOG.warn("Invalid generate stats factor: {}", factor, new RuntimeException(""));
+ }
+ }
+
+ public void setGenerateStatsFactor(String factor) {
+ this.generateStatsFactor = Integer.valueOf(factor);
+ if (generateStatsFactor <= 0) {
+ LOG.warn("Invalid generate stats factor: {}", generateStatsFactor, new RuntimeException(""));
+ }
+ }
+
public boolean getEnableDescribeExtendVariantColumn() {
return enableDescribeExtendVariantColumn;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 06/14: [fix](test)Fix mtmv switch is false in incremtn case (#31584)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 336f82e4267b50559c907fe629ffb40818938251
Author: zfr95 <87...@users.noreply.github.com>
AuthorDate: Thu Feb 29 22:18:24 2024 +0800
[fix](test)Fix mtmv switch is false in incremtn case (#31584)
---
.../mv/increment_create/cross_join_list_str_increment_create.groovy | 2 +-
.../mv/increment_create/cross_join_range_date_increment_create.groovy | 2 +-
.../mv/increment_create/cross_join_range_number_increment_create.groovy | 2 +-
.../mv/increment_create/full_join_list_str_increment_create.groovy | 2 +-
.../mv/increment_create/full_join_range_date_increment_create.groovy | 2 +-
.../mv/increment_create/full_join_range_number_increment_create.groovy | 2 +-
.../mv/increment_create/inner_join_list_str_increment_create.groovy | 2 +-
.../mv/increment_create/inner_join_range_date_increment_create.groovy | 2 +-
.../mv/increment_create/inner_join_range_number_increment_create.groovy | 2 +-
.../mv/increment_create/left_anti_join_list_str_increment_create.groovy | 2 +-
.../increment_create/left_anti_join_range_date_increment_create.groovy | 2 +-
.../left_anti_join_range_number_increment_create.groovy | 2 +-
.../mv/increment_create/left_join_list_str_increment_create.groovy | 2 +-
.../mv/increment_create/left_join_range_date_increment_create.groovy | 2 +-
.../mv/increment_create/left_join_range_number_increment_create.groovy | 2 +-
.../mv/increment_create/left_semi_join_list_str_increment_create.groovy | 2 +-
.../increment_create/left_semi_join_range_date_increment_create.groovy | 2 +-
.../left_semi_join_range_number_increment_create.groovy | 2 +-
.../increment_create/right_anti_join_list_str_increment_create.groovy | 2 +-
.../increment_create/right_anti_join_range_date_increment_create.groovy | 2 +-
.../right_anti_join_range_number_increment_create.groovy | 2 +-
.../mv/increment_create/right_join_list_str_increment_create.groovy | 2 +-
.../mv/increment_create/right_join_range_date_increment_create.groovy | 2 +-
.../mv/increment_create/right_join_range_number_increment_create.groovy | 2 +-
.../increment_create/right_semi_join_list_str_increment_create.groovy | 2 +-
.../increment_create/right_semi_join_range_date_increment_create.groovy | 2 +-
.../right_semi_join_range_number_increment_create.groovy | 2 +-
.../mv/increment_create/self_conn_list_str_increment_create.groovy | 2 +-
.../mv/increment_create/self_conn_range_date_increment_create.groovy | 2 +-
.../mv/increment_create/self_conn_range_number_increment_create.groovy | 2 +-
30 files changed, 30 insertions(+), 30 deletions(-)
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_list_str_increment_create.groovy
index ba8c38b141b..21475ce3886 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("cross_join_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_range_date_increment_create.groovy
index 8a5f03d7de2..f89040f7f7b 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("cross_join_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_range_number_increment_create.groovy
index 90b29647caf..4e12b2eaa2c 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/cross_join_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("cross_join_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_list_str_increment_create.groovy
index 09dbe4cf65c..0442f8cb218 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("full_join_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_range_date_increment_create.groovy
index c9cb2a81662..10f2cb087eb 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("full_join_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_range_number_increment_create.groovy
index c6efa3ee689..dcfac6a8d38 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/full_join_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("full_join_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_list_str_increment_create.groovy
index 878a8de6f88..9605aa99b14 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("inner_join_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_range_date_increment_create.groovy
index 0b8eec54519..b8190e76c17 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("inner_join_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_range_number_increment_create.groovy
index 487c09ff99b..4fb766cef77 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/inner_join_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("inner_join_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_list_str_increment_create.groovy
index f9aa96b463f..2c27c097f0c 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("left_anti_join_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_range_date_increment_create.groovy
index 7d8b47b2403..47c2b2f38d3 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("left_anti_join_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_range_number_increment_create.groovy
index e4baee0abf1..3d9bac82eea 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_anti_join_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("left_anti_join_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_list_str_increment_create.groovy
index 7363cd2a9ba..4e973d88012 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("left_join_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_range_date_increment_create.groovy
index bc19ab31b95..18045c51043 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("left_join_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_range_number_increment_create.groovy
index 19d636947cd..6f43675ba96 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_join_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("left_join_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_list_str_increment_create.groovy
index fc9ebb47d1a..49989225de7 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("left_semi_join_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_range_date_increment_create.groovy
index ed8c63db6c5..26977398b28 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("left_semi_join_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_range_number_increment_create.groovy
index e5fe0477eee..6439d61b2de 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/left_semi_join_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("left_semi_join_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_list_str_increment_create.groovy
index be14569b85e..f09c6565ea6 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("right_anti_join_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_range_date_increment_create.groovy
index fed6fcd51ec..4f8605fb3ea 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("right_anti_join_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_range_number_increment_create.groovy
index 14bec54480c..b878037f940 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_anti_join_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("right_anti_join_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_list_str_increment_create.groovy
index 1dd865db060..26a7a926011 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("right_join_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_range_date_increment_create.groovy
index d3fd18189ce..1cc603a24f4 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("right_join_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_range_number_increment_create.groovy
index 6994a7ace73..ca09fd1d6e7 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_join_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("right_join_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_list_str_increment_create.groovy
index 1a8bbfc077b..1b716e6af1c 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("right_semi_join_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_range_date_increment_create.groovy
index 945d0f488b3..459c8c3f51c 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("right_semi_join_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_range_number_increment_create.groovy
index 88e9cd9a12a..fc34ba25f29 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/right_semi_join_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("right_semi_join_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_list_str_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_list_str_increment_create.groovy
index 3bd0d212f3d..5548152f166 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_list_str_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_list_str_increment_create.groovy
@@ -20,7 +20,7 @@ suite("self_conn_list_str_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_range_date_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_range_date_increment_create.groovy
index 6b09554d613..a586a38c4e9 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_range_date_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_range_date_increment_create.groovy
@@ -20,7 +20,7 @@ suite("self_conn_range_date_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
diff --git a/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_range_number_increment_create.groovy b/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_range_number_increment_create.groovy
index c3b88617d14..833942fb3a3 100644
--- a/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_range_number_increment_create.groovy
+++ b/regression-test/suites/nereids_rules_p0/mv/increment_create/self_conn_range_number_increment_create.groovy
@@ -20,7 +20,7 @@ suite("self_conn_range_number_increment_create") {
sql "use ${db}"
sql "SET enable_nereids_planner=true"
sql "SET enable_fallback_to_original_planner=false"
- sql "SET enable_materialized_view_rewrite=true"
+ sql "SET enable_materialized_view_rewrite=false"
sql "SET enable_nereids_timeout = false"
sql """
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 07/14: [pipelineX](conf) enable pipelineX by default (#29894)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 07703633dd2a26e1c050a2ad58293a5898a376e6
Author: Gabriel <ga...@gmail.com>
AuthorDate: Thu Feb 29 22:19:04 2024 +0800
[pipelineX](conf) enable pipelineX by default (#29894)
* update
* update
---
be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp | 4 ++--
fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java | 2 +-
regression-test/suites/external_table_p0/hive/test_hive_other.groovy | 2 +-
3 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp b/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp
index 0d1d7784f88..77a63209cc6 100644
--- a/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp
+++ b/be/src/pipeline/pipeline_x/pipeline_x_fragment_context.cpp
@@ -225,11 +225,11 @@ Status PipelineXFragmentContext::prepare(const doris::TPipelineFragmentParams& r
request, root_pipeline->output_row_desc(), _runtime_state.get(), *_desc_tbl,
root_pipeline->id()));
RETURN_IF_ERROR(_sink->init(request.fragment.output_sink));
- static_cast<void>(root_pipeline->set_sink(_sink));
+ RETURN_IF_ERROR(root_pipeline->set_sink(_sink));
for (PipelinePtr& pipeline : _pipelines) {
DCHECK(pipeline->sink_x() != nullptr) << pipeline->operator_xs().size();
- static_cast<void>(pipeline->sink_x()->set_child(pipeline->operator_xs().back()));
+ RETURN_IF_ERROR(pipeline->sink_x()->set_child(pipeline->operator_xs().back()));
}
if (_enable_local_shuffle()) {
RETURN_IF_ERROR(_plan_local_exchange(request.num_buckets,
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index e561ded693a..3a6f120d501 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -849,7 +849,7 @@ public class SessionVariable implements Serializable, Writable {
private boolean enablePipelineEngine = true;
@VariableMgr.VarAttr(name = ENABLE_PIPELINE_X_ENGINE, fuzzy = false, varType = VariableAnnotation.EXPERIMENTAL)
- private boolean enablePipelineXEngine = false;
+ private boolean enablePipelineXEngine = true;
@VariableMgr.VarAttr(name = ENABLE_SHARED_SCAN, fuzzy = false, varType = VariableAnnotation.EXPERIMENTAL,
needForward = true)
diff --git a/regression-test/suites/external_table_p0/hive/test_hive_other.groovy b/regression-test/suites/external_table_p0/hive/test_hive_other.groovy
index 5400c94fc81..a7502ac1dc2 100644
--- a/regression-test/suites/external_table_p0/hive/test_hive_other.groovy
+++ b/regression-test/suites/external_table_p0/hive/test_hive_other.groovy
@@ -19,7 +19,7 @@ suite("test_hive_other", "p0,external,hive,external_docker,external_docker_hive"
def q01 = {
qt_q24 """ select name, count(1) as c from student group by name order by c desc;"""
- qt_q25 """ select lo_orderkey, count(1) as c from lineorder group by lo_orderkey order by c desc;"""
+ qt_q25 """ select lo_orderkey, count(1) as c from lineorder group by lo_orderkey order by lo_orderkey asc, c desc;"""
qt_q26 """ select * from test1 order by col_1;"""
qt_q27 """ select * from string_table order by p_partkey desc;"""
qt_q28 """ select * from account_fund order by batchno;"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 08/14: [improvement](docs)add faq for flink-connector-doris (#31426)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 05f525e922baba869eb690881caf4c57ca92c4fa
Author: Petrichor <31...@users.noreply.github.com>
AuthorDate: Thu Feb 29 23:19:59 2024 +0800
[improvement](docs)add faq for flink-connector-doris (#31426)
---
docs/en/docs/ecosystem/flink-doris-connector.md | 4 ++++
docs/zh-CN/docs/ecosystem/flink-doris-connector.md | 4 ++++
2 files changed, 8 insertions(+)
diff --git a/docs/en/docs/ecosystem/flink-doris-connector.md b/docs/en/docs/ecosystem/flink-doris-connector.md
index b5cfbe2def3..256b15e94a6 100644
--- a/docs/en/docs/ecosystem/flink-doris-connector.md
+++ b/docs/en/docs/ecosystem/flink-doris-connector.md
@@ -784,3 +784,7 @@ This problem is mainly caused by the conditional varchar/string type, which need
15. **Failed to connect to backend: http://host:webserver_port, and BE is still alive**
The issue may have occurred due to configuring the IP address of `be`, which is not reachable by the external Flink cluster.This is mainly because when connecting to `fe`, the address of `be` is resolved through fe. For instance, if you add a be address as '127.0.0.1', the be address obtained by the Flink cluster through fe will be '127.0.0.1:webserver_port', and Flink will connect to that address. When this issue arises, you can resolve it by adding the actual corresponding external IP [...]
+
+16. **When using Flink-connector to synchronize MySQL data to Doris, there is a time difference of several hours between the timestamp.**
+
+Flink Connector synchronizes the entire database from MySQL with a default timezone of UTC+8. If your data resides in a different timezone, you can adjust it using the following configuration, for example: `--mysql-conf debezium.date.format.timestamp.zone="UTC+3"`.
diff --git a/docs/zh-CN/docs/ecosystem/flink-doris-connector.md b/docs/zh-CN/docs/ecosystem/flink-doris-connector.md
index cf3f22ef2f5..5c51da5f0a7 100644
--- a/docs/zh-CN/docs/ecosystem/flink-doris-connector.md
+++ b/docs/zh-CN/docs/ecosystem/flink-doris-connector.md
@@ -784,3 +784,7 @@ Flink在数据导入时，如果有脏数据，比如字段格式、长度等问
15. **如果出现Failed to connect to backend: http://host:webserver_port, 并且Be还是活着的**
可能是因为你配置的be的ip，外部的Flink集群无法访问。这主要是因为当连接fe时，会通过fe解析出be的地址。例如，当你添加的be 地址为`127.0.0.1`，那么Flink通过fe获取的be地址就为`127.0.0.1:webserver_port`， 此时Flink就会去访问这个地址。当出现这个问题时，可以通过在with属性中增加实际对应的be外部ip地`'benodes' = "be_ip:webserver_port, be_ip:webserver_port..."`，整库同步则可增加`--sink-conf benodes=be_ip:webserver,be_ip:webserver...`。
+
+16. **如果使用整库同步 MySQL 数据到 Doris，出现 timestamp 类型与源数据相差多个小时**
+
+整库同步默认timezone="UTC+8"，如果你同步的数据不是该时区，可以尝试如下设置相对应的时区，例如：`--mysql-conf debezium.date.format.timestamp.zone="UTC+3"来解决。`
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 02/14: Support analyze rollup. (#31576)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 82faa7469bd72fc98fdc9e9d3612415d6f095738
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Thu Feb 29 21:54:35 2024 +0800
Support analyze rollup. (#31576)
---
.../java/org/apache/doris/catalog/OlapTable.java | 8 ++-
.../java/org/apache/doris/qe/ShowExecutor.java | 7 +--
.../apache/doris/statistics/AnalysisManager.java | 10 ++--
.../statistics/StatisticsAutoCollectorTest.java | 14 +++++
.../suites/statistics/test_analyze_mv.groovy | 64 +++++++++++++++++++++-
5 files changed, 91 insertions(+), 12 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
index 2acdf00a491..6b0a218f302 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/OlapTable.java
@@ -495,12 +495,18 @@ public class OlapTable extends Table implements MTMVRelatedTableIf {
return null;
}
+ /**
+ * This function is for statistics collection only. To get all the index ids that contains the given columnName.
+ * For base index, return -1 as its id, this is for compatibility with older version of column stats.
+ * @param columnName
+ * @return index id list that contains the given columnName.
+ */
public List<Long> getMvColumnIndexIds(String columnName) {
List<Long> ids = Lists.newArrayList();
for (MaterializedIndexMeta meta : getVisibleIndexIdToMeta().values()) {
Column target = meta.getColumnByDefineName(columnName);
if (target != null) {
- ids.add(meta.getIndexId());
+ ids.add(meta.getIndexId() == baseIndexId ? -1 : meta.getIndexId());
}
}
return ids;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
index a6748145e34..8bafeaae280 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
@@ -2597,9 +2597,8 @@ public class ShowExecutor {
for (String colName : columnNames) {
// Olap base index use -1 as index id.
List<Long> indexIds = Lists.newArrayList();
- if (StatisticsUtil.isMvColumn(tableIf, colName)) {
- OlapTable olapTable = (OlapTable) tableIf;
- indexIds = olapTable.getMvColumnIndexIds(colName);
+ if (tableIf instanceof OlapTable) {
+ indexIds = ((OlapTable) tableIf).getMvColumnIndexIds(colName);
} else {
indexIds.add(-1L);
}
@@ -3026,7 +3025,7 @@ public class ShowExecutor {
List<String> row = new ArrayList<>();
row.add(String.valueOf(analysisInfo.taskId));
row.add(analysisInfo.colName);
- if (StatisticsUtil.isMvColumn(table, analysisInfo.colName)) {
+ if (table instanceof OlapTable && analysisInfo.indexId != -1) {
row.add(((OlapTable) table).getIndexNameById(analysisInfo.indexId));
} else {
row.add("N/A");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
index eac50b40757..eafcff3e7f2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
@@ -423,9 +423,8 @@ public class AnalysisManager implements Writable {
String colName = entry.getKey();
List<Long> indexIds = Lists.newArrayList();
// Get index id this column belongs to for OlapTable. Set it to -1 for baseIndex id.
- if (StatisticsUtil.isMvColumn(table, colName)) {
- OlapTable olapTable = (OlapTable) table;
- indexIds = olapTable.getMvColumnIndexIds(colName);
+ if (table instanceof OlapTable) {
+ indexIds = ((OlapTable) table).getMvColumnIndexIds(colName);
} else {
indexIds.add(-1L);
}
@@ -702,9 +701,8 @@ public class AnalysisManager implements Writable {
for (String column : columns) {
List<Long> indexIds = Lists.newArrayList();
- if (StatisticsUtil.isMvColumn(table, column)) {
- OlapTable olapTable = (OlapTable) table;
- indexIds = olapTable.getMvColumnIndexIds(column);
+ if (table instanceof OlapTable) {
+ indexIds = ((OlapTable) table).getMvColumnIndexIds(column);
} else {
indexIds.add(-1L);
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java b/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java
index a85a1e5f1d9..466db9088e4 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java
@@ -379,6 +379,13 @@ public class StatisticsAutoCollectorTest {
public BaseAnalysisTask createAnalysisTask(AnalysisInfo info) {
return new OlapAnalysisTask(info);
}
+
+ @Mock
+ public List<Long> getMvColumnIndexIds(String columnName) {
+ ArrayList<Long> objects = new ArrayList<>();
+ objects.add(-1L);
+ return objects;
+ }
};
new MockUp<StatisticsUtil>() {
@@ -444,6 +451,13 @@ public class StatisticsAutoCollectorTest {
public BaseAnalysisTask createAnalysisTask(AnalysisInfo info) {
return new OlapAnalysisTask(info);
}
+
+ @Mock
+ public List<Long> getMvColumnIndexIds(String columnName) {
+ ArrayList<Long> objects = new ArrayList<>();
+ objects.add(-1L);
+ return objects;
+ }
};
new MockUp<StatisticsUtil>() {
diff --git a/regression-test/suites/statistics/test_analyze_mv.groovy b/regression-test/suites/statistics/test_analyze_mv.groovy
index 5c60c902bf0..7295b3e3a0e 100644
--- a/regression-test/suites/statistics/test_analyze_mv.groovy
+++ b/regression-test/suites/statistics/test_analyze_mv.groovy
@@ -240,11 +240,73 @@ suite("test_analyze_mv") {
wait_mv_finish("test_analyze_mv", "mvTestAgg")
sql """create materialized view mv6 as select key1, sum(value1) from mvTestAgg group by key1;"""
wait_mv_finish("test_analyze_mv", "mvTestAgg")
+ sql """alter table mvTestAgg ADD ROLLUP rollup1(key1, value1)"""
+ wait_mv_finish("test_analyze_mv", "mvTestAgg")
sql """insert into mvTestAgg values (1, 2, 3, 4, 5), (1, 2, 3, 4, 5), (1, 11, 22, 33, 44), (10, 20, 30, 40, 50), (10, 20, 30, 40, 50), (100, 200, 300, 400, 500), (1001, 2001, 3001, 4001, 5001);"""
sql """analyze table mvTestAgg with sync;"""
result_sample = sql """show column stats mvTestAgg"""
- assertEquals(13, result_sample.size())
+ assertEquals(15, result_sample.size())
+
+ result_sample = sql """show column stats mvTestAgg(key1)"""
+ assertEquals(2, result_sample.size())
+ if (result_sample[0][1] == "N/A") {
+ assertEquals("key1", result_sample[0][0])
+ assertEquals("N/A", result_sample[0][1])
+ assertEquals("5.0", result_sample[0][2])
+ assertEquals("4.0", result_sample[0][3])
+ assertEquals("1", result_sample[0][7])
+ assertEquals("1001", result_sample[0][8])
+ assertEquals("key1", result_sample[1][0])
+ assertEquals("rollup1", result_sample[1][1])
+ assertEquals("4.0", result_sample[1][2])
+ assertEquals("4.0", result_sample[1][3])
+ assertEquals("1", result_sample[1][7])
+ assertEquals("1001", result_sample[1][8])
+ } else {
+ assertEquals("key1", result_sample[1][0])
+ assertEquals("N/A", result_sample[1][1])
+ assertEquals("5.0", result_sample[1][2])
+ assertEquals("4.0", result_sample[1][3])
+ assertEquals("1", result_sample[1][7])
+ assertEquals("1001", result_sample[1][8])
+ assertEquals("key1", result_sample[0][0])
+ assertEquals("rollup1", result_sample[0][1])
+ assertEquals("4.0", result_sample[0][2])
+ assertEquals("4.0", result_sample[0][3])
+ assertEquals("1", result_sample[0][7])
+ assertEquals("1001", result_sample[0][8])
+ }
+
+ result_sample = sql """show column stats mvTestAgg(value1)"""
+ assertEquals(2, result_sample.size())
+ if (result_sample[0][1] == "N/A") {
+ assertEquals("value1", result_sample[0][0])
+ assertEquals("N/A", result_sample[0][1])
+ assertEquals("5.0", result_sample[0][2])
+ assertEquals("5.0", result_sample[0][3])
+ assertEquals("6", result_sample[0][7])
+ assertEquals("3001", result_sample[0][8])
+ assertEquals("value1", result_sample[1][0])
+ assertEquals("rollup1", result_sample[1][1])
+ assertEquals("4.0", result_sample[1][2])
+ assertEquals("4.0", result_sample[1][3])
+ assertEquals("28", result_sample[1][7])
+ assertEquals("3001", result_sample[1][8])
+ } else {
+ assertEquals("value1", result_sample[1][0])
+ assertEquals("N/A", result_sample[1][1])
+ assertEquals("5.0", result_sample[1][2])
+ assertEquals("5.0", result_sample[1][3])
+ assertEquals("6", result_sample[1][7])
+ assertEquals("3001", result_sample[1][8])
+ assertEquals("value1", result_sample[0][0])
+ assertEquals("rollup1", result_sample[0][1])
+ assertEquals("4.0", result_sample[0][2])
+ assertEquals("4.0", result_sample[0][3])
+ assertEquals("28", result_sample[0][7])
+ assertEquals("3001", result_sample[0][8])
+ }
result_sample = sql """show column stats mvTestAgg(key2)"""
assertEquals(1, result_sample.size())
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 04/14: [pipelineX](bug) Fix core dump if union has no children (#31611)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit ae926d0d8c1b2f5bad3951619d6b83d93860a09a
Author: Gabriel <ga...@gmail.com>
AuthorDate: Thu Feb 29 22:16:27 2024 +0800
[pipelineX](bug) Fix core dump if union has no children (#31611)
---
be/src/pipeline/exec/union_source_operator.cpp | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/be/src/pipeline/exec/union_source_operator.cpp b/be/src/pipeline/exec/union_source_operator.cpp
index dc1de2900b9..213b9bc0d3f 100644
--- a/be/src/pipeline/exec/union_source_operator.cpp
+++ b/be/src/pipeline/exec/union_source_operator.cpp
@@ -182,7 +182,8 @@ Status UnionSourceOperatorX::get_block(RuntimeState* state, vectorized::Block* b
local_state.reached_limit(block, eos);
//have executing const expr, queue have no data anymore, and child could be closed
*eos = (_child_size == 0 && !local_state._need_read_for_const_expr) ||
- (local_state._shared_state->data_queue.is_all_finish() && !_has_data(state));
+ (_child_size > 0 && local_state._shared_state->data_queue.is_all_finish() &&
+ !_has_data(state));
return Status::OK();
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 11/14: [improvement](group_commit) Empty wal should be deleted when replaying it (#31427)
Posted by yi...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-2.1
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 1398f112461b528c4925aa9c48b68fc2e57aa4d2
Author: huanghaibin <28...@qq.com>
AuthorDate: Thu Feb 29 23:35:35 2024 +0800
[improvement](group_commit) Empty wal should be deleted when replaying it (#31427)
* [improvement](group_commit) Corrupt wal should be deleted when replaying it
* edit
---
be/src/olap/wal/wal_reader.cpp | 3 +++
be/src/olap/wal/wal_table.cpp | 2 +-
2 files changed, 4 insertions(+), 1 deletion(-)
diff --git a/be/src/olap/wal/wal_reader.cpp b/be/src/olap/wal/wal_reader.cpp
index 0dfaa18a241..d1263daf1c5 100644
--- a/be/src/olap/wal/wal_reader.cpp
+++ b/be/src/olap/wal/wal_reader.cpp
@@ -86,6 +86,9 @@ Status WalReader::read_block(PBlock& block) {
}
Status WalReader::read_header(std::string& col_ids) {
+ if (file_reader->size() == 0) {
+ return Status::DataQualityError("empty file");
+ }
size_t bytes_read = 0;
std::string magic_str;
magic_str.resize(k_wal_magic_length);
diff --git a/be/src/olap/wal/wal_table.cpp b/be/src/olap/wal/wal_table.cpp
index 93cd12765f8..0389c6fed21 100644
--- a/be/src/olap/wal/wal_table.cpp
+++ b/be/src/olap/wal/wal_table.cpp
@@ -98,7 +98,7 @@ Status WalTable::_relay_wal_one_by_one() {
doris::wal_fail << 1;
LOG(WARNING) << "failed to replay wal=" << wal_info->get_wal_path()
<< ", st=" << st.to_string();
- if (!st.is<ErrorCode::NOT_FOUND>()) {
+ if (!st.is<ErrorCode::NOT_FOUND>() && !st.is<ErrorCode::DATA_QUALITY_ERROR>()) {
need_retry_wals.push_back(wal_info);
} else {
need_delete_wals.push_back(wal_info);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org