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/03/12 06:21:26 UTC
(doris) branch branch-2.1 updated (3358f76a7f6 -> 61928f7df53)
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 3358f76a7f6 [feature](spill) Implement spill to disk for hash join, aggregation and sort for pipelineX (#31910)
new 24706348594 [RuntimeFilter] fix <=> runtime filter failed bug (#32003)
new b0b7161ad01 [feature](rf) add filter info profile when rf run as expr (#31822)
new ab21d85e8c4 [nereids](topn-filter) support multi-topn filter (FE part) (#31485)
new c0f2d0188bd [feature](pipelineX) add mem control in local exchange sink (#31982)
new 31ee448c87d [test](fix) Fix one missing line of output in out file (#32036)
new fe3988c6862 [fix](compile) fix BE compile failure on Mac (#32044)
new a7a85dd3300 [chore](config) support select experimental session variable (#31837)
new 68a5319da36 [fix](pipelineX) _local_channel_dependency is null in non pipelineX (#32054)
new 1509fbd5387 [feature](docker)Doris On Docker All In One (#31735)
new ccd21a6ea43 [Improve](InPredict) enhance in predict with array type (#31828)
new 4268634115f [fix](memory) Fix Allocator cancel pipelinex query #32048
new 926908ece2a [fix](hive) fix spelling mistakes for "separatorChar" #32061
new dc7d80860f6 [fix](case) fix export data consistency table key type (#32045)
new aea9ddc3cb0 [Fix](Outfiel) fix be core when the open method of vfile_result_writer failed #32042
new 2ee12b8678a [improvement](docker)Optimize Dockerfile content for FE and BE (#32027)
new a937c54b2b4 [fix](docker)Fix the registration logic of Doris On Docker (#32026)
new b41b17ad0a6 [fix](spill) fix storage engine type cast error (#32071)
new 194f3432ab8 [Improvement](executor)Routine load support workload group #31671
new da60a111d02 [refactor](nereids) rename PlanNode.distributeExprLists to childrenDistributeExprLists #32069
new 65835cdd2b8 [doc](sql-dialect) update sql convertor tool to 1.0.1 (#32077)
new ae206596c91 [DOC](Variant) add bloom filter description and correct some benchmark results (#31998)
new 2ffcfb4bc50 [fix](doc) fix wrong download link (#32082)
new ffa904c4870 [enhance](Cooldown) Skip cooldown if the tablet is dropped (#32079)
new 61928f7df53 [pipelineX](scanner) Use the actual instances num when ignore data distribution (#32081)
The 24 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/exprs/bloom_filter_func.h | 68 +++---
be/src/exprs/runtime_filter.cpp | 49 ++--
be/src/exprs/runtime_filter.h | 8 +-
be/src/olap/tablet.cpp | 4 +
be/src/pipeline/pipeline_x/pipeline_x_task.cpp | 4 +-
be/src/runtime/memory/thread_mem_tracker_mgr.cpp | 13 +-
be/src/runtime/memory/thread_mem_tracker_mgr.h | 10 +-
be/src/runtime/thread_context.cpp | 7 +-
be/src/runtime/thread_context.h | 13 +-
be/src/vec/common/allocator.cpp | 4 +-
be/src/vec/exec/runtime_filter_consumer.cpp | 9 +-
be/src/vec/exec/runtime_filter_consumer.h | 6 +-
be/src/vec/exec/scan/new_olap_scan_node.cpp | 4 +-
be/src/vec/exec/scan/scanner_context.cpp | 3 +-
be/src/vec/exprs/vdirect_in_predicate.h | 7 +
be/src/vec/exprs/vin_predicate.cpp | 5 +-
be/src/vec/exprs/vruntimefilter_wrapper.cpp | 32 ++-
be/src/vec/exprs/vruntimefilter_wrapper.h | 19 +-
...on_struct_in.cpp => function_collection_in.cpp} | 8 +-
...nction_struct_in.h => function_collection_in.h} | 65 +++---
be/src/vec/functions/simple_function_factory.h | 4 +-
be/src/vec/runtime/vdata_stream_recvr.cpp | 31 ++-
be/src/vec/runtime/vdata_stream_recvr.h | 6 +-
be/src/vec/sink/writer/vfile_result_writer.cpp | 24 +-
be/src/vec/sink/writer/vfile_result_writer.h | 2 +-
be/src/vec/spill/spill_stream.cpp | 5 +-
be/src/vec/spill/spill_stream.h | 11 +-
be/src/vec/spill/spill_stream_manager.cpp | 46 +++-
be/src/vec/spill/spill_stream_manager.h | 43 +++-
be/src/vec/spill/spill_writer.cpp | 1 -
be/src/vec/spill/spill_writer.h | 6 +-
docker/runtime/{be => all-in-one}/Dockerfile | 8 +-
docker/runtime/all-in-one/resource/entry_point.sh | 248 +++++++++++++++++++++
docker/runtime/be/Dockerfile | 7 +-
docker/runtime/be/resource/entry_point.sh | 16 +-
docker/runtime/be/resource/init_be.sh | 96 ++++----
docker/runtime/fe/Dockerfile | 11 +-
.../hive/scripts/create_preinstalled_table.hql | 2 +-
docs/en/docs/lakehouse/sql-dialect.md | 2 +-
.../sql-manual/sql-reference/Data-Types/VARIANT.md | 22 +-
docs/zh-CN/docs/lakehouse/sql-dialect.md | 2 +-
.../sql-manual/sql-reference/Data-Types/VARIANT.md | 22 +-
.../doris/analysis/AlterRoutineLoadStmt.java | 8 +
.../doris/analysis/CreateRoutineLoadStmt.java | 14 ++
.../doris/datasource/hive/source/HiveScanNode.java | 6 +-
.../doris/load/routineload/KafkaTaskInfo.java | 27 +++
.../doris/load/routineload/RoutineLoadJob.java | 14 ++
.../org/apache/doris/nereids/CascadesContext.java | 6 +
.../glue/translator/PhysicalPlanTranslator.java | 63 ++++--
.../glue/translator/PlanTranslatorContext.java | 9 +-
.../doris/nereids/processor/post/TopNScanOpt.java | 117 ++++------
.../nereids/processor/post/TopnFilterContext.java | 93 ++++++++
.../nereids/trees/expressions/InPredicate.java | 12 +
.../nereids/trees/plans/physical/PhysicalTopN.java | 39 ++--
.../doris/nereids/util/TypeCoercionUtils.java | 10 +-
.../org/apache/doris/planner/OlapScanNode.java | 25 ++-
.../java/org/apache/doris/planner/PlanNode.java | 14 +-
.../java/org/apache/doris/planner/SortNode.java | 2 +-
.../main/java/org/apache/doris/qe/Coordinator.java | 11 +-
.../main/java/org/apache/doris/qe/VariableMgr.java | 26 ++-
.../resource/workloadgroup/WorkloadGroupMgr.java | 39 +++-
.../nereids/postprocess/TopNRuntimeFilterTest.java | 24 +-
gensrc/proto/internal_service.proto | 3 +
.../jdbc/test_mysql_jdbc_driver5_catalog.out | 1 +
.../nereids_syntax_p0/inpredicate_with_list.out | 19 ++
.../data/nereids_tpch_p0/tpch/topn-filter.out | 29 +++
.../data/query_p0/sql_functions/test_in_expr.out | 3 +
regression-test/pipeline/p0/conf/fe.conf | 1 +
.../export_p0/test_export_data_consistency.groovy | 2 +
.../load_p0/routine_load/test_routine_load.groovy | 8 +-
.../nereids_syntax_p0/inpredicate_with_list.groovy | 60 +++++
.../suites/nereids_syntax_p0/system_var.groovy | 5 +
.../suites/nereids_tpch_p0/tpch/topn-filter.groovy | 120 ++++++++++
regression-test/suites/point_query_p0/load.groovy | 4 +-
.../query_p0/sql_functions/test_in_expr.groovy | 5 +-
.../workload_manager_p0/test_curd_wlg.groovy | 12 +-
76 files changed, 1341 insertions(+), 443 deletions(-)
rename be/src/vec/functions/{function_struct_in.cpp => function_collection_in.cpp} (79%)
rename be/src/vec/functions/{function_struct_in.h => function_collection_in.h} (75%)
copy docker/runtime/{be => all-in-one}/Dockerfile (89%)
create mode 100644 docker/runtime/all-in-one/resource/entry_point.sh
create mode 100644 fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterContext.java
create mode 100644 regression-test/data/nereids_syntax_p0/inpredicate_with_list.out
create mode 100644 regression-test/data/nereids_tpch_p0/tpch/topn-filter.out
create mode 100644 regression-test/suites/nereids_syntax_p0/inpredicate_with_list.groovy
create mode 100644 regression-test/suites/nereids_tpch_p0/tpch/topn-filter.groovy
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 02/24: [feature](rf) add filter info profile when rf run as expr (#31822)
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 b0b7161ad01ab69ef635ba02a53ac100ea669de7
Author: Mryange <59...@users.noreply.github.com>
AuthorDate: Mon Mar 11 10:35:56 2024 +0800
[feature](rf) add filter info profile when rf run as expr (#31822)
---
be/src/exprs/runtime_filter.cpp | 20 +++++++++++++-----
be/src/exprs/runtime_filter.h | 3 ++-
be/src/vec/exec/runtime_filter_consumer.cpp | 9 ++++----
be/src/vec/exec/runtime_filter_consumer.h | 6 +++---
be/src/vec/exec/scan/new_olap_scan_node.cpp | 4 ++--
be/src/vec/exprs/vruntimefilter_wrapper.cpp | 32 +++++++++++++++++++++--------
be/src/vec/exprs/vruntimefilter_wrapper.h | 19 ++++++++++++++---
7 files changed, 67 insertions(+), 26 deletions(-)
diff --git a/be/src/exprs/runtime_filter.cpp b/be/src/exprs/runtime_filter.cpp
index a949969ca65..6e4f57ef8e1 100644
--- a/be/src/exprs/runtime_filter.cpp
+++ b/be/src/exprs/runtime_filter.cpp
@@ -444,7 +444,8 @@ public:
}
Status get_push_exprs(std::list<vectorized::VExprContextSPtr>& probe_ctxs,
- std::vector<vectorized::VExprSPtr>& push_exprs, const TExpr& probe_expr);
+ std::vector<vectorized::VRuntimeFilterPtr>& push_exprs,
+ const TExpr& probe_expr);
Status merge(const RuntimePredicateWrapper* wrapper) {
bool can_not_merge_in_or_bloom =
@@ -1056,14 +1057,23 @@ Status IRuntimeFilter::push_to_remote(const TNetworkAddress* addr, bool opt_remo
}
Status IRuntimeFilter::get_push_expr_ctxs(std::list<vectorized::VExprContextSPtr>& probe_ctxs,
- std::vector<vectorized::VExprSPtr>& push_exprs,
+ std::vector<vectorized::VRuntimeFilterPtr>& push_exprs,
bool is_late_arrival) {
DCHECK(is_consumer());
+ auto origin_size = push_exprs.size();
if (!_wrapper->is_ignored()) {
_set_push_down(!is_late_arrival);
RETURN_IF_ERROR(_wrapper->get_push_exprs(probe_ctxs, push_exprs, _probe_expr));
}
_profile->add_info_string("Info", _format_status());
+ // The runtime filter is pushed down, adding filtering information.
+ auto* expr_filtered_rows_counter = ADD_COUNTER(_profile, "expr_filtered_rows", TUnit::UNIT);
+ auto* expr_input_rows_counter = ADD_COUNTER(_profile, "expr_input_rows", TUnit::UNIT);
+ auto* always_true_counter = ADD_COUNTER(_profile, "always_true", TUnit::UNIT);
+ for (auto i = origin_size; i < push_exprs.size(); i++) {
+ push_exprs[i]->attach_profile_counter(expr_filtered_rows_counter, expr_input_rows_counter,
+ always_true_counter);
+ }
return Status::OK();
}
@@ -1715,9 +1725,9 @@ void IRuntimeFilter::update_filter(RuntimePredicateWrapper* wrapper, int64_t mer
this->signal();
}
-Status RuntimePredicateWrapper::get_push_exprs(std::list<vectorized::VExprContextSPtr>& probe_ctxs,
- std::vector<vectorized::VExprSPtr>& container,
- const TExpr& probe_expr) {
+Status RuntimePredicateWrapper::get_push_exprs(
+ std::list<vectorized::VExprContextSPtr>& probe_ctxs,
+ std::vector<vectorized::VRuntimeFilterPtr>& container, const TExpr& probe_expr) {
vectorized::VExprContextSPtr probe_ctx;
RETURN_IF_ERROR(vectorized::VExpr::create_expr_tree(probe_expr, probe_ctx));
probe_ctxs.push_back(probe_ctx);
diff --git a/be/src/exprs/runtime_filter.h b/be/src/exprs/runtime_filter.h
index 5cfc88f4ed8..fe5ddd68da6 100644
--- a/be/src/exprs/runtime_filter.h
+++ b/be/src/exprs/runtime_filter.h
@@ -233,7 +233,8 @@ public:
PrimitiveType column_type() const;
Status get_push_expr_ctxs(std::list<vectorized::VExprContextSPtr>& probe_ctxs,
- std::vector<vectorized::VExprSPtr>& push_exprs, bool is_late_arrival);
+ std::vector<vectorized::VRuntimeFilterPtr>& push_exprs,
+ bool is_late_arrival);
bool is_broadcast_join() const { return _is_broadcast_join; }
diff --git a/be/src/vec/exec/runtime_filter_consumer.cpp b/be/src/vec/exec/runtime_filter_consumer.cpp
index 52caf84e361..097df801615 100644
--- a/be/src/vec/exec/runtime_filter_consumer.cpp
+++ b/be/src/vec/exec/runtime_filter_consumer.cpp
@@ -85,7 +85,7 @@ void RuntimeFilterConsumer::init_runtime_filter_dependency(
Status RuntimeFilterConsumer::_acquire_runtime_filter() {
SCOPED_TIMER(_acquire_runtime_filter_timer);
- VExprSPtrs vexprs;
+ std::vector<vectorized::VRuntimeFilterPtr> vexprs;
for (size_t i = 0; i < _runtime_filter_descs.size(); ++i) {
IRuntimeFilter* runtime_filter = _runtime_filter_ctxs[i].runtime_filter;
bool ready = runtime_filter->is_ready();
@@ -111,12 +111,13 @@ Status RuntimeFilterConsumer::_acquire_runtime_filter() {
return Status::OK();
}
-Status RuntimeFilterConsumer::_append_rf_into_conjuncts(const VExprSPtrs& vexprs) {
+Status RuntimeFilterConsumer::_append_rf_into_conjuncts(
+ const std::vector<vectorized::VRuntimeFilterPtr>& vexprs) {
if (vexprs.empty()) {
return Status::OK();
}
- for (auto& expr : vexprs) {
+ for (const auto& expr : vexprs) {
VExprContextSPtr conjunct = VExprContext::create_shared(expr);
RETURN_IF_ERROR(conjunct->prepare(_state, _row_descriptor_ref));
RETURN_IF_ERROR(conjunct->open(_state));
@@ -142,7 +143,7 @@ Status RuntimeFilterConsumer::try_append_late_arrival_runtime_filter(int* arrive
}
// 1. Check if are runtime filter ready but not applied.
- VExprSPtrs exprs;
+ std::vector<vectorized::VRuntimeFilterPtr> exprs;
int current_arrived_rf_num = 0;
for (size_t i = 0; i < _runtime_filter_descs.size(); ++i) {
if (_runtime_filter_ctxs[i].apply_mark) {
diff --git a/be/src/vec/exec/runtime_filter_consumer.h b/be/src/vec/exec/runtime_filter_consumer.h
index b8513e666bc..86609624be6 100644
--- a/be/src/vec/exec/runtime_filter_consumer.h
+++ b/be/src/vec/exec/runtime_filter_consumer.h
@@ -46,7 +46,7 @@ protected:
// Get all arrived runtime filters at Open phase.
Status _acquire_runtime_filter();
// Append late-arrival runtime filters to the vconjunct_ctx.
- Status _append_rf_into_conjuncts(const VExprSPtrs& vexprs);
+ Status _append_rf_into_conjuncts(const std::vector<vectorized::VRuntimeFilterPtr>& vexprs);
void _init_profile(RuntimeProfile* profile);
@@ -54,9 +54,9 @@ protected:
// For runtime filters
struct RuntimeFilterContext {
- RuntimeFilterContext(IRuntimeFilter* rf) : apply_mark(false), runtime_filter(rf) {}
+ RuntimeFilterContext(IRuntimeFilter* rf) : runtime_filter(rf) {}
// set to true if this runtime filter is already applied to vconjunct_ctx_ptr
- bool apply_mark;
+ bool apply_mark = false;
IRuntimeFilter* runtime_filter = nullptr;
};
diff --git a/be/src/vec/exec/scan/new_olap_scan_node.cpp b/be/src/vec/exec/scan/new_olap_scan_node.cpp
index 44a435ed2a2..7473820abef 100644
--- a/be/src/vec/exec/scan/new_olap_scan_node.cpp
+++ b/be/src/vec/exec/scan/new_olap_scan_node.cpp
@@ -757,8 +757,8 @@ void NewOlapScanNode::add_filter_info(int id, const PredicateFilterInfo& update_
filter_name += std::to_string(id);
std::string info_str;
info_str += "type = " + type_to_string(static_cast<PredicateType>(info.type)) + ", ";
- info_str += "input = " + std::to_string(info.input_row) + ", ";
- info_str += "filtered = " + std::to_string(info.filtered_row);
+ info_str += "predicate input = " + std::to_string(info.input_row) + ", ";
+ info_str += "predicate filtered = " + std::to_string(info.filtered_row);
info_str = "[" + info_str + "]";
// add info
diff --git a/be/src/vec/exprs/vruntimefilter_wrapper.cpp b/be/src/vec/exprs/vruntimefilter_wrapper.cpp
index 075e5194866..dcf3cba72e2 100644
--- a/be/src/vec/exprs/vruntimefilter_wrapper.cpp
+++ b/be/src/vec/exprs/vruntimefilter_wrapper.cpp
@@ -20,9 +20,12 @@
#include <fmt/format.h>
#include <stddef.h>
+#include <cstdint>
#include <memory>
#include <utility>
+#include "util/defer_op.h"
+#include "util/runtime_profile.h"
#include "util/simd/bits.h"
#include "vec/columns/column.h"
#include "vec/columns/column_const.h"
@@ -84,7 +87,19 @@ Status VRuntimeFilterWrapper::execute(VExprContext* context, Block* block, int*
*result_column_id = num_columns_without_result;
return Status::OK();
} else {
- _scan_rows += block->rows();
+ int64_t input_rows = 0, filter_rows = 0;
+ Defer statistic_filter_info {[&]() {
+ if (_expr_filtered_rows_counter) {
+ COUNTER_UPDATE(_expr_filtered_rows_counter, filter_rows);
+ }
+ if (_expr_input_rows_counter) {
+ COUNTER_UPDATE(_expr_input_rows_counter, input_rows);
+ }
+ if (_always_true_counter) {
+ COUNTER_SET(_always_true_counter, (int64_t)_always_true);
+ }
+ }};
+ input_rows += block->rows();
if (_getting_const_col) {
_impl->set_getting_const_col(true);
@@ -99,28 +114,29 @@ Status VRuntimeFilterWrapper::execute(VExprContext* context, Block* block, int*
if (is_column_const(*result_column.column)) {
auto* constant_val = const_cast<char*>(result_column.column->get_data_at(0).data);
if (constant_val == nullptr || !*reinterpret_cast<bool*>(constant_val)) {
- _filtered_rows += block->rows();
+ filter_rows += block->rows();
}
} else if (const auto* nullable =
check_and_get_column<ColumnNullable>(*result_column.column)) {
data = ((ColumnVector<UInt8>*)nullable->get_nested_column_ptr().get())
->get_data()
.data();
- _filtered_rows += doris::simd::count_zero_num(reinterpret_cast<const int8_t*>(data),
- nullable->get_null_map_data().data(),
- block->rows());
+ filter_rows += doris::simd::count_zero_num(reinterpret_cast<const int8_t*>(data),
+ nullable->get_null_map_data().data(),
+ block->rows());
} else if (const auto* res_col =
check_and_get_column<ColumnVector<UInt8>>(*result_column.column)) {
data = const_cast<uint8_t*>(res_col->get_data().data());
- _filtered_rows += doris::simd::count_zero_num(reinterpret_cast<const int8_t*>(data),
- block->rows());
+ filter_rows += doris::simd::count_zero_num(reinterpret_cast<const int8_t*>(data),
+ block->rows());
} else {
return Status::InternalError(
"Invalid type for runtime filters!, and _expr_name is: {}. _data_type is: {}. "
"result_column_id is: {}. block structure: {}.",
_expr_name, _data_type->get_name(), *result_column_id, block->dump_structure());
}
-
+ _filtered_rows += filter_rows;
+ _scan_rows += input_rows;
calculate_filter(VRuntimeFilterWrapper::EXPECTED_FILTER_RATE, _filtered_rows, _scan_rows,
_has_calculate_filter, _always_true);
return Status::OK();
diff --git a/be/src/vec/exprs/vruntimefilter_wrapper.h b/be/src/vec/exprs/vruntimefilter_wrapper.h
index 6d50b914beb..30017850028 100644
--- a/be/src/vec/exprs/vruntimefilter_wrapper.h
+++ b/be/src/vec/exprs/vruntimefilter_wrapper.h
@@ -17,9 +17,8 @@
#pragma once
-#include <stdint.h>
-
#include <atomic>
+#include <cstdint>
#include <string>
#include <vector>
@@ -27,6 +26,7 @@
#include "common/object_pool.h"
#include "common/status.h"
#include "udf/udf.h"
+#include "util/runtime_profile.h"
#include "vec/exprs/vexpr.h"
namespace doris {
@@ -57,6 +57,14 @@ public:
const VExprSPtr get_impl() const override { return _impl; }
+ void attach_profile_counter(RuntimeProfile::Counter* expr_filtered_rows_counter,
+ RuntimeProfile::Counter* expr_input_rows_counter,
+ RuntimeProfile::Counter* always_true_counter) {
+ _expr_filtered_rows_counter = expr_filtered_rows_counter;
+ _expr_input_rows_counter = expr_input_rows_counter;
+ _always_true_counter = always_true_counter;
+ }
+
// if filter rate less than this, bloom filter will set always true
constexpr static double EXPECTED_FILTER_RATE = 0.4;
@@ -74,12 +82,17 @@ private:
VExprSPtr _impl;
bool _always_true;
- /// TODO: statistic filter rate in the profile
std::atomic<int64_t> _filtered_rows;
std::atomic<int64_t> _scan_rows;
+ RuntimeProfile::Counter* _expr_filtered_rows_counter = nullptr;
+ RuntimeProfile::Counter* _expr_input_rows_counter = nullptr;
+ RuntimeProfile::Counter* _always_true_counter = nullptr;
bool _has_calculate_filter = false;
std::string _expr_name;
};
+
+using VRuntimeFilterPtr = std::shared_ptr<VRuntimeFilterWrapper>;
+
} // namespace doris::vectorized
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 13/24: [fix](case) fix export data consistency table key type (#32045)
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 dc7d80860f60705c5b19a157f2f8d50352f8fa52
Author: walter <w4...@gmail.com>
AuthorDate: Mon Mar 11 22:14:20 2024 +0800
[fix](case) fix export data consistency table key type (#32045)
---
regression-test/suites/export_p0/test_export_data_consistency.groovy | 2 ++
1 file changed, 2 insertions(+)
diff --git a/regression-test/suites/export_p0/test_export_data_consistency.groovy b/regression-test/suites/export_p0/test_export_data_consistency.groovy
index e388d869fff..354010a8582 100644
--- a/regression-test/suites/export_p0/test_export_data_consistency.groovy
+++ b/regression-test/suites/export_p0/test_export_data_consistency.groovy
@@ -68,6 +68,7 @@ suite("test_export_data_consistency", "p0") {
`name` string NULL,
`age` int(11) NULL
)
+ UNIQUE KEY(`id`)
PARTITION BY RANGE(id)
(
PARTITION less_than_20 VALUES LESS THAN ("20"),
@@ -149,6 +150,7 @@ suite("test_export_data_consistency", "p0") {
"label" = "${label}",
"format" = "csv",
"column_separator" = ",",
+ "parallelism" = "10",
"data_consistency" = "partition"
);
"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 12/24: [fix](hive) fix spelling mistakes for "separatorChar" #32061
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 926908ece2a83a03cc5c3499926debb771ffa2d0
Author: wuwenchi <wu...@hotmail.com>
AuthorDate: Mon Mar 11 22:12:36 2024 +0800
[fix](hive) fix spelling mistakes for "separatorChar" #32061
---
.../docker-compose/hive/scripts/create_preinstalled_table.hql | 2 +-
.../java/org/apache/doris/datasource/hive/source/HiveScanNode.java | 6 +++---
2 files changed, 4 insertions(+), 4 deletions(-)
diff --git a/docker/thirdparties/docker-compose/hive/scripts/create_preinstalled_table.hql b/docker/thirdparties/docker-compose/hive/scripts/create_preinstalled_table.hql
index 9baf13020f6..345d62842d5 100644
--- a/docker/thirdparties/docker-compose/hive/scripts/create_preinstalled_table.hql
+++ b/docker/thirdparties/docker-compose/hive/scripts/create_preinstalled_table.hql
@@ -1934,6 +1934,6 @@ create table employee_gz(name string,salary string)
row format serde 'org.apache.hadoop.hive.serde2.OpenCSVSerde'
with serdeproperties
('quoteChar'='\"'
-,'seperatorChar'=',');
+,'separatorChar'=',');
insert into employee_gz values ('a', '1.1'), ('b', '2.2');
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java
index c901ee654ae..d4d471a4612 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/hive/source/HiveScanNode.java
@@ -78,7 +78,7 @@ public class HiveScanNode extends FileQueryScanNode {
public static final String DEFAULT_FIELD_DELIMITER = "\1"; // "\x01"
public static final String PROP_LINE_DELIMITER = "line.delim";
public static final String DEFAULT_LINE_DELIMITER = "\n";
- public static final String PROP_SEPERATOR_CHAR = "seperatorChar";
+ public static final String PROP_SEPARATOR_CHAR = "separatorChar";
public static final String PROP_QUOTA_CHAR = "quoteChar";
public static final String PROP_COLLECTION_DELIMITER_HIVE2 = "colelction.delim";
@@ -363,8 +363,8 @@ public class HiveScanNode extends FileQueryScanNode {
java.util.Map<String, String> delimiter = hmsTable.getRemoteTable().getSd().getSerdeInfo().getParameters();
if (delimiter.containsKey(PROP_FIELD_DELIMITER)) {
textParams.setColumnSeparator(delimiter.get(PROP_FIELD_DELIMITER));
- } else if (delimiter.containsKey(PROP_SEPERATOR_CHAR)) {
- textParams.setColumnSeparator(delimiter.get(PROP_SEPERATOR_CHAR));
+ } else if (delimiter.containsKey(PROP_SEPARATOR_CHAR)) {
+ textParams.setColumnSeparator(delimiter.get(PROP_SEPARATOR_CHAR));
} else {
textParams.setColumnSeparator(DEFAULT_FIELD_DELIMITER);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 11/24: [fix](memory) Fix Allocator cancel pipelinex query #32048
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 4268634115f3acef3bb332fa6ee664b52cc3072a
Author: Xinyi Zou <zo...@gmail.com>
AuthorDate: Mon Mar 11 22:09:08 2024 +0800
[fix](memory) Fix Allocator cancel pipelinex query #32048
---
be/src/runtime/memory/thread_mem_tracker_mgr.cpp | 13 ++++++-------
be/src/runtime/memory/thread_mem_tracker_mgr.h | 10 +++++-----
be/src/runtime/thread_context.cpp | 7 +++----
be/src/runtime/thread_context.h | 13 ++++++-------
be/src/vec/common/allocator.cpp | 4 ++--
be/src/vec/runtime/vdata_stream_recvr.cpp | 2 +-
6 files changed, 23 insertions(+), 26 deletions(-)
diff --git a/be/src/runtime/memory/thread_mem_tracker_mgr.cpp b/be/src/runtime/memory/thread_mem_tracker_mgr.cpp
index 0a1686704c0..1feca6976b8 100644
--- a/be/src/runtime/memory/thread_mem_tracker_mgr.cpp
+++ b/be/src/runtime/memory/thread_mem_tracker_mgr.cpp
@@ -25,12 +25,11 @@
namespace doris {
void ThreadMemTrackerMgr::attach_limiter_tracker(
- const std::shared_ptr<MemTrackerLimiter>& mem_tracker,
- const TUniqueId& fragment_instance_id) {
+ const std::shared_ptr<MemTrackerLimiter>& mem_tracker, const TUniqueId& query_id) {
DCHECK(mem_tracker);
CHECK(init());
flush_untracked_mem();
- _fragment_instance_id = fragment_instance_id;
+ _query_id = query_id;
_limiter_tracker = mem_tracker;
_limiter_tracker_raw = mem_tracker.get();
_wait_gc = true;
@@ -40,15 +39,15 @@ void ThreadMemTrackerMgr::detach_limiter_tracker(
const std::shared_ptr<MemTrackerLimiter>& old_mem_tracker) {
CHECK(init());
flush_untracked_mem();
- _fragment_instance_id = TUniqueId();
+ _query_id = TUniqueId();
_limiter_tracker = old_mem_tracker;
_limiter_tracker_raw = old_mem_tracker.get();
_wait_gc = false;
}
-void ThreadMemTrackerMgr::cancel_instance(const std::string& exceed_msg) {
- ExecEnv::GetInstance()->fragment_mgr()->cancel_instance(
- _fragment_instance_id, PPlanFragmentCancelReason::MEMORY_LIMIT_EXCEED, exceed_msg);
+void ThreadMemTrackerMgr::cancel_query(const std::string& exceed_msg) {
+ ExecEnv::GetInstance()->fragment_mgr()->cancel_query(
+ _query_id, PPlanFragmentCancelReason::MEMORY_LIMIT_EXCEED, exceed_msg);
}
} // namespace doris
diff --git a/be/src/runtime/memory/thread_mem_tracker_mgr.h b/be/src/runtime/memory/thread_mem_tracker_mgr.h
index 4cb22b9e1ae..d54f77ce3fa 100644
--- a/be/src/runtime/memory/thread_mem_tracker_mgr.h
+++ b/be/src/runtime/memory/thread_mem_tracker_mgr.h
@@ -53,7 +53,7 @@ public:
// After attach, the current thread Memory Hook starts to consume/release task mem_tracker
void attach_limiter_tracker(const std::shared_ptr<MemTrackerLimiter>& mem_tracker,
- const TUniqueId& fragment_instance_id);
+ const TUniqueId& query_id);
void detach_limiter_tracker(const std::shared_ptr<MemTrackerLimiter>& old_mem_tracker =
ExecEnv::GetInstance()->orphan_mem_tracker());
@@ -82,7 +82,7 @@ public:
void consume(int64_t size, int skip_large_memory_check = 0);
void flush_untracked_mem();
- bool is_attach_query() { return _fragment_instance_id != TUniqueId(); }
+ bool is_attach_query() { return _query_id != TUniqueId(); }
std::shared_ptr<MemTrackerLimiter> limiter_mem_tracker() {
CHECK(init());
@@ -95,7 +95,7 @@ public:
void disable_wait_gc() { _wait_gc = false; }
[[nodiscard]] bool wait_gc() const { return _wait_gc; }
- void cancel_instance(const std::string& exceed_msg);
+ void cancel_query(const std::string& exceed_msg);
std::string print_debug_string() {
fmt::memory_buffer consumer_tracker_buf;
@@ -130,7 +130,7 @@ private:
// If there is a memory new/delete operation in the consume method, it may enter infinite recursion.
bool _stop_consume = false;
- TUniqueId _fragment_instance_id = TUniqueId();
+ TUniqueId _query_id = TUniqueId();
};
inline bool ThreadMemTrackerMgr::init() {
@@ -185,7 +185,7 @@ inline void ThreadMemTrackerMgr::consume(int64_t size, int skip_large_memory_che
"malloc or new large memory: {}, {}, this is just a warning, not prevent memory "
"alloc, stacktrace:\n{}",
size,
- is_attach_query() ? "in query or load: " + print_id(_fragment_instance_id)
+ is_attach_query() ? "in query or load: " + print_id(_query_id)
: "not in query or load",
get_stack_trace());
_stop_consume = false;
diff --git a/be/src/runtime/thread_context.cpp b/be/src/runtime/thread_context.cpp
index fca09fcabc5..03c10986bd8 100644
--- a/be/src/runtime/thread_context.cpp
+++ b/be/src/runtime/thread_context.cpp
@@ -24,18 +24,17 @@ namespace doris {
class MemTracker;
AttachTask::AttachTask(const std::shared_ptr<MemTrackerLimiter>& mem_tracker,
- const TUniqueId& task_id, const TUniqueId& fragment_instance_id) {
+ const TUniqueId& task_id) {
ThreadLocalHandle::create_thread_local_if_not_exits();
signal::set_signal_task_id(task_id);
- thread_context()->attach_task(task_id, fragment_instance_id, mem_tracker);
+ thread_context()->attach_task(task_id, mem_tracker);
}
AttachTask::AttachTask(RuntimeState* runtime_state) {
ThreadLocalHandle::create_thread_local_if_not_exits();
signal::set_signal_task_id(runtime_state->query_id());
signal::set_signal_is_nereids(runtime_state->is_nereids());
- thread_context()->attach_task(runtime_state->query_id(), runtime_state->fragment_instance_id(),
- runtime_state->query_mem_tracker());
+ thread_context()->attach_task(runtime_state->query_id(), runtime_state->query_mem_tracker());
}
AttachTask::~AttachTask() {
diff --git a/be/src/runtime/thread_context.h b/be/src/runtime/thread_context.h
index f6ca46acaba..463dbf44349 100644
--- a/be/src/runtime/thread_context.h
+++ b/be/src/runtime/thread_context.h
@@ -45,8 +45,8 @@
// This will save some info about a working thread in the thread context.
// Looking forward to tracking memory during thread execution into MemTrackerLimiter.
#define SCOPED_ATTACH_TASK(arg1) auto VARNAME_LINENUM(attach_task) = AttachTask(arg1)
-#define SCOPED_ATTACH_TASK_WITH_ID(arg1, arg2, arg3) \
- auto VARNAME_LINENUM(attach_task) = AttachTask(arg1, arg2, arg3)
+#define SCOPED_ATTACH_TASK_WITH_ID(arg1, arg2) \
+ auto VARNAME_LINENUM(attach_task) = AttachTask(arg1, arg2)
// Switch MemTrackerLimiter for count memory during thread execution.
// Used after SCOPED_ATTACH_TASK, in order to count the memory into another
@@ -60,7 +60,7 @@
auto VARNAME_LINENUM(add_mem_consumer) = doris::AddThreadMemTrackerConsumer(mem_tracker)
#else
#define SCOPED_ATTACH_TASK(arg1, ...) (void)0
-#define SCOPED_ATTACH_TASK_WITH_ID(arg1, arg2, arg3) (void)0
+#define SCOPED_ATTACH_TASK_WITH_ID(arg1, arg2) (void)0
#define SCOPED_SWITCH_THREAD_MEM_TRACKER_LIMITER(mem_tracker_limiter) (void)0
#define SCOPED_CONSUME_MEM_TRACKER(mem_tracker) (void)0
#endif
@@ -134,7 +134,7 @@ public:
~ThreadContext() = default;
- void attach_task(const TUniqueId& task_id, const TUniqueId& fragment_instance_id,
+ void attach_task(const TUniqueId& task_id,
const std::shared_ptr<MemTrackerLimiter>& mem_tracker) {
#ifndef BE_TEST
// will only attach_task at the beginning of the thread function, there should be no duplicate attach_task.
@@ -144,7 +144,7 @@ public:
<< ", attach mem tracker label: " << mem_tracker->label();
#endif
_task_id = task_id;
- thread_mem_tracker_mgr->attach_limiter_tracker(mem_tracker, fragment_instance_id);
+ thread_mem_tracker_mgr->attach_limiter_tracker(mem_tracker, _task_id);
}
void detach_task() {
@@ -292,8 +292,7 @@ private:
class AttachTask {
public:
explicit AttachTask(const std::shared_ptr<MemTrackerLimiter>& mem_tracker,
- const TUniqueId& task_id = TUniqueId(),
- const TUniqueId& fragment_instance_id = TUniqueId());
+ const TUniqueId& task_id = TUniqueId());
explicit AttachTask(RuntimeState* runtime_state);
diff --git a/be/src/vec/common/allocator.cpp b/be/src/vec/common/allocator.cpp
index 3d6d2a7a11e..f1dd2af83f7 100644
--- a/be/src/vec/common/allocator.cpp
+++ b/be/src/vec/common/allocator.cpp
@@ -110,7 +110,7 @@ void Allocator<clear_memory_, mmap_populate, use_mmap>::sys_memory_check(size_t
"Query:{} canceled asyn, after waiting for memory {}ms, {}.",
print_id(doris::thread_context()->task_id()), wait_milliseconds,
err_msg);
- doris::thread_context()->thread_mem_tracker_mgr->cancel_instance(err_msg);
+ doris::thread_context()->thread_mem_tracker_mgr->cancel_query(err_msg);
} else {
LOG(INFO) << fmt::format(
"Query:{} throw exception, after waiting for memory {}ms, {}.",
@@ -148,7 +148,7 @@ void Allocator<clear_memory_, mmap_populate, use_mmap>::memory_tracker_check(siz
if (!doris::enable_thread_catch_bad_alloc) {
LOG(INFO) << fmt::format("query/load:{} canceled asyn, {}.",
print_id(doris::thread_context()->task_id()), err_msg);
- doris::thread_context()->thread_mem_tracker_mgr->cancel_instance(err_msg);
+ doris::thread_context()->thread_mem_tracker_mgr->cancel_query(err_msg);
} else {
LOG(INFO) << fmt::format("query/load:{} throw exception, {}.",
print_id(doris::thread_context()->task_id()), err_msg);
diff --git a/be/src/vec/runtime/vdata_stream_recvr.cpp b/be/src/vec/runtime/vdata_stream_recvr.cpp
index 0bef6b6eaab..1cfc359de30 100644
--- a/be/src/vec/runtime/vdata_stream_recvr.cpp
+++ b/be/src/vec/runtime/vdata_stream_recvr.cpp
@@ -425,7 +425,7 @@ Status VDataStreamRecvr::create_merger(const VExprContextSPtrs& ordering_expr,
Status VDataStreamRecvr::add_block(const PBlock& pblock, int sender_id, int be_number,
int64_t packet_seq, ::google::protobuf::Closure** done) {
- SCOPED_ATTACH_TASK_WITH_ID(_query_mem_tracker, _query_id, _fragment_instance_id);
+ SCOPED_ATTACH_TASK_WITH_ID(_query_mem_tracker, _query_id);
int use_sender_id = _is_merging ? sender_id : 0;
return _sender_queues[use_sender_id]->add_block(pblock, be_number, packet_seq, done);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 15/24: [improvement](docker)Optimize Dockerfile content for FE and BE (#32027)
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 2ee12b8678a0fe6ed371599953606b7ad4852e8c
Author: FreeOnePlus <54...@users.noreply.github.com>
AuthorDate: Mon Mar 11 22:19:33 2024 +0800
[improvement](docker)Optimize Dockerfile content for FE and BE (#32027)
* Refactor upgrade documentation
* Optimize Dockerfile content for FE and BE.
* Optimize Dockerfile content for FE and BE.
---------
Co-authored-by: Yijia Su <su...@selectdb.com>
---
docker/runtime/be/Dockerfile | 7 +++++--
docker/runtime/fe/Dockerfile | 11 +++++++----
2 files changed, 12 insertions(+), 6 deletions(-)
diff --git a/docker/runtime/be/Dockerfile b/docker/runtime/be/Dockerfile
index 0833b44d1ff..e890f053237 100644
--- a/docker/runtime/be/Dockerfile
+++ b/docker/runtime/be/Dockerfile
@@ -31,7 +31,10 @@ FROM ubuntu:22.04
ARG TARGETARCH
-RUN apt-get update -y && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
+ARG DORIS_VERSION="x.x.x"
+
+RUN sed -i -e 's/^APT/# APT/' -e 's/^DPkg/# DPkg/' /etc/apt/apt.conf.d/docker-clean && \
+ apt-get update -y && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
patchelf gdb binutils binutils-common mysql-client \
curl wget less vim htop iproute2 numactl jq iotop sysstat \
tcpdump iputils-ping dnsutils strace lsof blktrace tzdata \
@@ -44,7 +47,7 @@ ENV JAVA_HOME=/usr/lib/jvm/java-8-openjdk-${TARGETARCH:-amd64} \
PATH="/opt/apache-doris/be/bin:${PATH}"
# apache-doris/be from doris release xxxx.tar.gz.please update the version in follows x.x.x.
-ADD resource/apache-doris-x.x.x-bin-${TARGETARCH:-amd64}/be /opt/apache-doris/be
+ADD resource/apache-doris-${DORIS_VERSION}-bin-${TARGETARCH:-amd64}/be /opt/apache-doris/be
COPY resource/be_*.sh /opt/apache-doris/
diff --git a/docker/runtime/fe/Dockerfile b/docker/runtime/fe/Dockerfile
index ce1b75c0260..6ef16b9ccd7 100644
--- a/docker/runtime/fe/Dockerfile
+++ b/docker/runtime/fe/Dockerfile
@@ -30,14 +30,17 @@ FROM ubuntu:22.04
ARG TARGETARCH
+ARG DORIS_VERSION="x.x.x"
+
# set environment variables
ENV JAVA_HOME=/usr/lib/jvm/java-8-openjdk-${TARGETARCH:-amd64} \
PATH="/opt/apache-doris/fe/bin:${PATH}"
# apache-doris/fe from doris release tar.gz, please update the version in follows x.x.x.
-COPY resource/apache-doris-x.x.x-bin-${TARGETARCH:-amd64}/fe /opt/apache-doris/fe
+COPY resource/apache-doris-${DORIS_VERSION}-bin-${TARGETARCH:-amd64}/fe /opt/apache-doris/fe
-RUN apt-get update -y && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
+RUN sed -i -e 's/^APT/# APT/' -e 's/^DPkg/# DPkg/' /etc/apt/apt.conf.d/docker-clean && \
+ apt-get update -y && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
patchelf gdb binutils binutils-common mysql-client \
curl wget less vim htop iproute2 numactl jq iotop sysstat tzdata \
tcpdump iputils-ping dnsutils strace lsof blktrace \
@@ -45,10 +48,10 @@ RUN apt-get update -y && DEBIAN_FRONTEND=noninteractive apt-get install -y --no
net-tools openjdk-8-jdk && \
rm -rf /var/lib/apt/lists/*
-ADD resource/init_fe.sh /opt/apache-doris/fe/bin/
+ADD resource/init_fe.sh /usr/local/bin/
COPY resource/fe_*.sh /opt/apache-doris/
WORKDIR /opt/apache-doris
-ENTRYPOINT ["/opt/apache-doris/fe/bin/init_fe.sh"]
+ENTRYPOINT ["bash","init_fe.sh"]
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 07/24: [chore](config) support select experimental session variable (#31837)
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 a7a85dd33005870e5d61078b959a7575cc5dbd89
Author: camby <ca...@tencent.com>
AuthorDate: Mon Mar 11 15:45:27 2024 +0800
[chore](config) support select experimental session variable (#31837)
support select experimental variables:
Before change:
Before change:
select @@experimental_enable_nereids_planner;
ERROR 1193 (HY000): errCode = 2, detailMessage = Unknown system variable 'experimental_enable_nereids_planner'
show variables like 'experimental_enable_nereids_planner';
+-------------------------------------+-------+---------------+---------+
| Variable_name | Value | Default_Value | Changed |
+-------------------------------------+-------+---------------+---------+
| experimental_enable_nereids_planner | false | true | 1 |
+-------------------------------------+-------+---------------+---------+
After change:
> select @@experimental_enable_nereids_planner;
+---------------------------------------+
| @@experimental_enable_nereids_planner |
+---------------------------------------+
| 1 |
+---------------------------------------+
---
.../main/java/org/apache/doris/qe/VariableMgr.java | 26 +++++++++++++---------
.../suites/nereids_syntax_p0/system_var.groovy | 5 +++++
2 files changed, 21 insertions(+), 10 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java b/fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java
index 93c842f0bfa..e83fd474daf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/VariableMgr.java
@@ -284,7 +284,10 @@ public class VariableMgr {
// setVar: variable information that needs to be set
public static void setVar(SessionVariable sessionVariable, SetVar setVar)
throws DdlException {
- VarContext varCtx = setVarPreCheck(setVar);
+ VarContext varCtx = getVarContext(setVar.getVariable());
+ if (varCtx == null) {
+ ErrorReport.reportDdlException(ErrorCode.ERR_UNKNOWN_SYSTEM_VARIABLE, setVar.getVariable());
+ }
checkUpdate(setVar, varCtx.getFlag());
setVarInternal(sessionVariable, setVar, varCtx);
}
@@ -299,7 +302,10 @@ public class VariableMgr {
// So in this case, we should just ignore this exception and return.
public static void setVarForNonMasterFE(SessionVariable sessionVariable, SetVar setVar)
throws DdlException {
- VarContext varCtx = setVarPreCheck(setVar);
+ VarContext varCtx = getVarContext(setVar.getVariable());
+ if (varCtx == null) {
+ ErrorReport.reportDdlException(ErrorCode.ERR_UNKNOWN_SYSTEM_VARIABLE, setVar.getVariable());
+ }
try {
checkUpdate(setVar, varCtx.getFlag());
} catch (DdlException e) {
@@ -311,9 +317,9 @@ public class VariableMgr {
setVarInternal(sessionVariable, setVar, varCtx);
}
- @NotNull
- private static VarContext setVarPreCheck(SetVar setVar) throws DdlException {
- String varName = setVar.getVariable();
+ @Nullable
+ private static VarContext getVarContext(String name) {
+ String varName = name;
boolean hasExpPrefix = false;
if (varName.startsWith(VariableAnnotation.EXPERIMENTAL.getPrefix())) {
varName = varName.substring(VariableAnnotation.EXPERIMENTAL.getPrefix().length());
@@ -325,13 +331,13 @@ public class VariableMgr {
}
VarContext ctx = ctxByVarName.get(varName);
if (ctx == null) {
- ErrorReport.reportDdlException(ErrorCode.ERR_UNKNOWN_SYSTEM_VARIABLE, setVar.getVariable());
+ return null;
}
// for non-matched prefix, report an error
VariableAnnotation varType = ctx.getField().getAnnotation(VarAttr.class).varType();
- if (hasExpPrefix && (!setVar.getVariable().startsWith(varType.getPrefix())
+ if (hasExpPrefix && (!name.startsWith(varType.getPrefix())
|| varType == VariableAnnotation.NONE)) {
- ErrorReport.reportDdlException(ErrorCode.ERR_UNKNOWN_SYSTEM_VARIABLE, setVar.getVariable());
+ return null;
}
return ctx;
}
@@ -533,7 +539,7 @@ public class VariableMgr {
// Get variable value through variable name, used to satisfy statement like `SELECT @@comment_version`
public static void fillValue(SessionVariable var, VariableExpr desc) throws AnalysisException {
- VarContext ctx = ctxByVarName.get(desc.getName());
+ VarContext ctx = getVarContext(desc.getName());
if (ctx == null) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_UNKNOWN_SYSTEM_VARIABLE, desc.getName());
}
@@ -621,7 +627,7 @@ public class VariableMgr {
// For Nereids optimizer
public static @Nullable Literal getLiteral(SessionVariable var, String name, SetType setType) {
- VarContext ctx = ctxByVarName.get(name);
+ VarContext ctx = getVarContext(name);
if (ctx == null) {
return null;
}
diff --git a/regression-test/suites/nereids_syntax_p0/system_var.groovy b/regression-test/suites/nereids_syntax_p0/system_var.groovy
index a2ab3bfa9ed..242df6ec94e 100644
--- a/regression-test/suites/nereids_syntax_p0/system_var.groovy
+++ b/regression-test/suites/nereids_syntax_p0/system_var.groovy
@@ -46,4 +46,9 @@ suite("nereids_sys_var") {
sql "select @@session.enable_nereids_planner"
result ([[true]])
}
+
+ test {
+ sql "select @@session.experimental_enable_nereids_planner"
+ result ([[true]])
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 23/24: [enhance](Cooldown) Skip cooldown if the tablet is dropped (#32079)
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 ffa904c4870220afe7997754edbdc2ef57c129e8
Author: AlexYue <yj...@gmail.com>
AuthorDate: Tue Mar 12 11:40:24 2024 +0800
[enhance](Cooldown) Skip cooldown if the tablet is dropped (#32079)
---
be/src/olap/tablet.cpp | 4 ++++
1 file changed, 4 insertions(+)
diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
index bfe31843ed1..903f169f4b9 100644
--- a/be/src/olap/tablet.cpp
+++ b/be/src/olap/tablet.cpp
@@ -223,6 +223,10 @@ void WriteCooldownMetaExecutors::WriteCooldownMetaExecutors::submit(TabletShared
VLOG_NOTICE << "tablet " << tablet_id << " is not cooldown replica";
return;
}
+ if (tablet->tablet_state() == TABLET_SHUTDOWN) [[unlikely]] {
+ LOG_INFO("tablet {} has been dropped, don't do cooldown", tablet_id);
+ return;
+ }
}
{
// one tablet could at most have one cooldown task to be done
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 08/24: [fix](pipelineX) _local_channel_dependency is null in non pipelineX (#32054)
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 68a5319da366aa7495e9caa8efaff2c9ed2aaf81
Author: Mryange <59...@users.noreply.github.com>
AuthorDate: Mon Mar 11 18:20:04 2024 +0800
[fix](pipelineX) _local_channel_dependency is null in non pipelineX (#32054)
---
be/src/vec/runtime/vdata_stream_recvr.cpp | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/be/src/vec/runtime/vdata_stream_recvr.cpp b/be/src/vec/runtime/vdata_stream_recvr.cpp
index 9708be6bde9..0bef6b6eaab 100644
--- a/be/src/vec/runtime/vdata_stream_recvr.cpp
+++ b/be/src/vec/runtime/vdata_stream_recvr.cpp
@@ -487,7 +487,7 @@ void VDataStreamRecvr::cancel_stream(Status exec_status) {
void VDataStreamRecvr::SenderQueue::update_blocks_memory_usage(int64_t size) {
_recvr->update_blocks_memory_usage(size);
- if (_recvr->exceeds_limit(0)) {
+ if (_local_channel_dependency && _recvr->exceeds_limit(0)) {
_local_channel_dependency->block();
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 03/24: [nereids](topn-filter) support multi-topn filter (FE part) (#31485)
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 ab21d85e8c488a3c8ac69533a999c0acd4ca3143
Author: minghong <en...@gmail.com>
AuthorDate: Mon Mar 11 10:43:30 2024 +0800
[nereids](topn-filter) support multi-topn filter (FE part) (#31485)
support multi-topn-filter
---
.../org/apache/doris/nereids/CascadesContext.java | 6 ++
.../glue/translator/PhysicalPlanTranslator.java | 39 +++++--
.../glue/translator/PlanTranslatorContext.java | 9 +-
.../doris/nereids/processor/post/TopNScanOpt.java | 117 ++++++++------------
.../nereids/processor/post/TopnFilterContext.java | 93 ++++++++++++++++
.../nereids/trees/plans/physical/PhysicalTopN.java | 39 +++----
.../org/apache/doris/planner/OlapScanNode.java | 25 ++++-
.../java/org/apache/doris/planner/SortNode.java | 2 +-
.../main/java/org/apache/doris/qe/Coordinator.java | 11 +-
.../nereids/postprocess/TopNRuntimeFilterTest.java | 24 ++++-
.../data/nereids_tpch_p0/tpch/topn-filter.out | 29 +++++
.../suites/nereids_tpch_p0/tpch/topn-filter.groovy | 120 +++++++++++++++++++++
regression-test/suites/point_query_p0/load.groovy | 4 +-
13 files changed, 400 insertions(+), 118 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
index 5addae25c29..18fad8d2941 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/CascadesContext.java
@@ -42,6 +42,7 @@ import org.apache.doris.nereids.jobs.scheduler.SimpleJobScheduler;
import org.apache.doris.nereids.memo.Group;
import org.apache.doris.nereids.memo.Memo;
import org.apache.doris.nereids.processor.post.RuntimeFilterContext;
+import org.apache.doris.nereids.processor.post.TopnFilterContext;
import org.apache.doris.nereids.properties.PhysicalProperties;
import org.apache.doris.nereids.rules.RuleFactory;
import org.apache.doris.nereids.rules.RuleSet;
@@ -107,6 +108,7 @@ public class CascadesContext implements ScheduleContext {
// subqueryExprIsAnalyzed: whether the subquery has been analyzed.
private final Map<SubqueryExpr, Boolean> subqueryExprIsAnalyzed;
private final RuntimeFilterContext runtimeFilterContext;
+ private final TopnFilterContext topnFilterContext = new TopnFilterContext();
private Optional<Scope> outerScope = Optional.empty();
private Map<Long, TableIf> tables = null;
@@ -280,6 +282,10 @@ public class CascadesContext implements ScheduleContext {
return runtimeFilterContext;
}
+ public TopnFilterContext getTopnFilterContext() {
+ return topnFilterContext;
+ }
+
public void setCurrentJobContext(JobContext currentJobContext) {
this.currentJobContext = currentJobContext;
}
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 81ff155f0d7..75c29b3123d 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
@@ -207,6 +207,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
+import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
import java.util.stream.Collectors;
@@ -724,6 +725,10 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
)
);
olapScanNode.setPushDownAggNoGrouping(context.getRelationPushAggOp(olapScan.getRelationId()));
+ if (context.getTopnFilterContext().isTopnFilterTarget(olapScan)) {
+ olapScanNode.setUseTopnOpt(true);
+ context.getTopnFilterContext().addLegacyTarget(olapScan, olapScanNode);
+ }
// TODO: we need to remove all finalizeForNereids
olapScanNode.finalizeForNereids();
// Create PlanFragment
@@ -747,6 +752,10 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
PhysicalDeferMaterializeOlapScan deferMaterializeOlapScan, PlanTranslatorContext context) {
PlanFragment planFragment = visitPhysicalOlapScan(deferMaterializeOlapScan.getPhysicalOlapScan(), context);
OlapScanNode olapScanNode = (OlapScanNode) planFragment.getPlanRoot();
+ if (context.getTopnFilterContext().isTopnFilterTarget(deferMaterializeOlapScan)) {
+ olapScanNode.setUseTopnOpt(true);
+ context.getTopnFilterContext().addLegacyTarget(deferMaterializeOlapScan, olapScanNode);
+ }
TupleDescriptor tupleDescriptor = context.getTupleDesc(olapScanNode.getTupleId());
for (SlotDescriptor slotDescriptor : tupleDescriptor.getSlots()) {
if (deferMaterializeOlapScan.getDeferMaterializeSlotIds()
@@ -2009,20 +2018,23 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
public PlanFragment visitPhysicalTopN(PhysicalTopN<? extends Plan> topN, PlanTranslatorContext context) {
PlanFragment inputFragment = topN.child(0).accept(this, context);
List<List<Expr>> distributeExprLists = getDistributeExprs(topN.child(0));
-
// 2. According to the type of sort, generate physical plan
if (!topN.getSortPhase().isMerge()) {
// For localSort or Gather->Sort, we just need to add TopNNode
SortNode sortNode = translateSortNode(topN, inputFragment.getPlanRoot(), context);
sortNode.setOffset(topN.getOffset());
sortNode.setLimit(topN.getLimit());
- if (topN.isEnableRuntimeFilter()) {
+ if (context.getTopnFilterContext().isTopnFilterSource(topN)) {
sortNode.setUseTopnOpt(true);
- PlanNode child = sortNode.getChild(0);
- Preconditions.checkArgument(child instanceof OlapScanNode,
- "topN opt expect OlapScanNode, but we get " + child);
- OlapScanNode scanNode = ((OlapScanNode) child);
- scanNode.setUseTopnOpt(true);
+ context.getTopnFilterContext().getTargets(topN).forEach(
+ olapScan -> {
+ Optional<OlapScanNode> legacyScan =
+ context.getTopnFilterContext().getLegacyScanNode(olapScan);
+ Preconditions.checkState(legacyScan.isPresent(),
+ "cannot find OlapScanNode for topn filter");
+ legacyScan.get().addTopnFilterSortNode(sortNode);
+ }
+ );
}
// push sort to scan opt
if (sortNode.getChild(0) instanceof OlapScanNode) {
@@ -2067,12 +2079,23 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
@Override
public PlanFragment visitPhysicalDeferMaterializeTopN(PhysicalDeferMaterializeTopN<? extends Plan> topN,
PlanTranslatorContext context) {
-
PlanFragment planFragment = visitPhysicalTopN(topN.getPhysicalTopN(), context);
if (planFragment.getPlanRoot() instanceof SortNode) {
SortNode sortNode = (SortNode) planFragment.getPlanRoot();
sortNode.setUseTwoPhaseReadOpt(true);
sortNode.getSortInfo().setUseTwoPhaseRead();
+ if (context.getTopnFilterContext().isTopnFilterSource(topN)) {
+ sortNode.setUseTopnOpt(true);
+ context.getTopnFilterContext().getTargets(topN).forEach(
+ olapScan -> {
+ Optional<OlapScanNode> legacyScan =
+ context.getTopnFilterContext().getLegacyScanNode(olapScan);
+ Preconditions.checkState(legacyScan.isPresent(),
+ "cannot find OlapScanNode for topn filter");
+ legacyScan.get().addTopnFilterSortNode(sortNode);
+ }
+ );
+ }
TupleDescriptor tupleDescriptor = sortNode.getSortInfo().getSortTupleDescriptor();
for (SlotDescriptor slotDescriptor : tupleDescriptor.getSlots()) {
if (topN.getDeferMaterializeSlotIds()
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
index 8a723b1fd1f..90539332791 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/glue/translator/PlanTranslatorContext.java
@@ -29,6 +29,7 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.IdGenerator;
import org.apache.doris.nereids.CascadesContext;
+import org.apache.doris.nereids.processor.post.TopnFilterContext;
import org.apache.doris.nereids.trees.expressions.CTEId;
import org.apache.doris.nereids.trees.expressions.ExprId;
import org.apache.doris.nereids.trees.expressions.SlotReference;
@@ -70,7 +71,7 @@ public class PlanTranslatorContext {
private final DescriptorTable descTable = new DescriptorTable();
private final RuntimeFilterTranslator translator;
-
+ private final TopnFilterContext topnFilterContext;
/**
* index from Nereids' slot to legacy slot.
*/
@@ -115,12 +116,14 @@ public class PlanTranslatorContext {
public PlanTranslatorContext(CascadesContext ctx) {
this.connectContext = ctx.getConnectContext();
this.translator = new RuntimeFilterTranslator(ctx.getRuntimeFilterContext());
+ this.topnFilterContext = ctx.getTopnFilterContext();
}
@VisibleForTesting
public PlanTranslatorContext() {
this.connectContext = null;
this.translator = null;
+ this.topnFilterContext = new TopnFilterContext();
}
/**
@@ -187,6 +190,10 @@ public class PlanTranslatorContext {
return Optional.ofNullable(translator);
}
+ public TopnFilterContext getTopnFilterContext() {
+ return topnFilterContext;
+ }
+
public PlanFragmentId nextFragmentId() {
return fragmentIdGenerator.getNextId();
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopNScanOpt.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopNScanOpt.java
index fab4e34a12f..a9425cb715b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopNScanOpt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopNScanOpt.java
@@ -19,103 +19,57 @@ package org.apache.doris.nereids.processor.post;
import org.apache.doris.nereids.CascadesContext;
import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.nereids.trees.plans.Plan;
import org.apache.doris.nereids.trees.plans.SortPhase;
-import org.apache.doris.nereids.trees.plans.algebra.Filter;
import org.apache.doris.nereids.trees.plans.algebra.OlapScan;
-import org.apache.doris.nereids.trees.plans.algebra.Project;
-import org.apache.doris.nereids.trees.plans.algebra.TopN;
-import org.apache.doris.nereids.trees.plans.physical.AbstractPhysicalSort;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalCatalogRelation;
import org.apache.doris.nereids.trees.plans.physical.PhysicalDeferMaterializeTopN;
-import org.apache.doris.nereids.trees.plans.physical.PhysicalDistribute;
-import org.apache.doris.nereids.trees.plans.physical.PhysicalSink;
import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalWindow;
import org.apache.doris.qe.ConnectContext;
+import java.util.Optional;
/**
* topN opt
* refer to:
* <a href="https://github.com/apache/doris/pull/15558">...</a>
* <a href="https://github.com/apache/doris/pull/15663">...</a>
*
- * // only support simple case: select ... from tbl [where ...] order by ... limit ...
+ * // [deprecated] only support simple case: select ... from tbl [where ...] order by ... limit ...
*/
public class TopNScanOpt extends PlanPostProcessor {
- @Override
- public Plan visit(Plan plan, CascadesContext context) {
- return plan;
- }
-
- @Override
- public Plan visitPhysicalSink(PhysicalSink<? extends Plan> physicalSink, CascadesContext context) {
- if (physicalSink.child() instanceof TopN) {
- return super.visit(physicalSink, context);
- } else if (physicalSink.child() instanceof Project && physicalSink.child().child(0) instanceof TopN) {
- PhysicalTopN<?> oldTopN = (PhysicalTopN<?>) physicalSink.child().child(0);
- PhysicalTopN<?> newTopN = (PhysicalTopN<?>) oldTopN.accept(this, context);
- if (newTopN == oldTopN) {
- return physicalSink;
- } else {
- return physicalSink.withChildren(physicalSink.child().withChildren(newTopN));
- }
- }
- return physicalSink;
- }
-
- @Override
- public Plan visitPhysicalDistribute(PhysicalDistribute<? extends Plan> distribute, CascadesContext context) {
- if (distribute.child() instanceof TopN && distribute.child() instanceof AbstractPhysicalSort
- && ((AbstractPhysicalSort<?>) distribute.child()).getSortPhase() == SortPhase.LOCAL_SORT) {
- return super.visit(distribute, context);
- }
- return distribute;
- }
-
@Override
public PhysicalTopN<? extends Plan> visitPhysicalTopN(PhysicalTopN<? extends Plan> topN, CascadesContext ctx) {
- if (topN.getSortPhase() == SortPhase.LOCAL_SORT) {
- Plan child = topN.child();
- topN = rewriteTopN(topN);
- if (child != topN.child()) {
- topN = ((PhysicalTopN<? extends Plan>) topN.withChildren(child)).copyStatsAndGroupIdFrom(topN);
- }
- return topN;
- } else if (topN.getSortPhase() == SortPhase.MERGE_SORT) {
- return (PhysicalTopN<? extends Plan>) super.visit(topN, ctx);
- }
+ Optional<OlapScan> scanOpt = findScanForTopnFilter(topN);
+ scanOpt.ifPresent(scan -> ctx.getTopnFilterContext().addTopnFilter(topN, scan));
+ topN.child().accept(this, ctx);
return topN;
}
@Override
public Plan visitPhysicalDeferMaterializeTopN(PhysicalDeferMaterializeTopN<? extends Plan> topN,
CascadesContext context) {
- if (topN.getSortPhase() == SortPhase.LOCAL_SORT) {
- PhysicalTopN<? extends Plan> rewrittenTopN = rewriteTopN(topN.getPhysicalTopN());
- if (topN.getPhysicalTopN() != rewrittenTopN) {
- topN = topN.withPhysicalTopN(rewrittenTopN).copyStatsAndGroupIdFrom(topN);
- }
- return topN;
- } else if (topN.getSortPhase() == SortPhase.MERGE_SORT) {
- return super.visit(topN, context);
- }
+ Optional<OlapScan> scanOpt = findScanForTopnFilter(topN.getPhysicalTopN());
+ scanOpt.ifPresent(scan -> context.getTopnFilterContext().addTopnFilter(topN, scan));
+ topN.child().accept(this, context);
return topN;
}
- private PhysicalTopN<? extends Plan> rewriteTopN(PhysicalTopN<? extends Plan> topN) {
- Plan child = topN.child();
+ private Optional<OlapScan> findScanForTopnFilter(PhysicalTopN<? extends Plan> topN) {
if (topN.getSortPhase() != SortPhase.LOCAL_SORT) {
- return topN;
+ return Optional.empty();
}
if (topN.getOrderKeys().isEmpty()) {
- return topN;
+ return Optional.empty();
}
// topn opt
long topNOptLimitThreshold = getTopNOptLimitThreshold();
if (topNOptLimitThreshold == -1 || topN.getLimit() > topNOptLimitThreshold) {
- return topN;
+ return Optional.empty();
}
// if firstKey's column is not present, it means the firstKey is not an original column from scan node
// for example: "select cast(k1 as INT) as id from tbl1 order by id limit 2;" the firstKey "id" is
@@ -125,27 +79,44 @@ public class TopNScanOpt extends PlanPostProcessor {
// see Alias::toSlot() method to get how column info is passed around by alias of slotReference
Expression firstKey = topN.getOrderKeys().get(0).getExpr();
if (!firstKey.isColumnFromTable()) {
- return topN;
+ return Optional.empty();
}
if (firstKey.getDataType().isFloatType()
|| firstKey.getDataType().isDoubleType()) {
- return topN;
+ return Optional.empty();
}
- OlapScan olapScan;
- while (child instanceof Project || child instanceof Filter) {
- child = child.child(0);
+ if (! (firstKey instanceof SlotReference)) {
+ return Optional.empty();
}
- if (!(child instanceof OlapScan)) {
- return topN;
+ OlapScan olapScan = findScanNodeBySlotReference(topN, (SlotReference) firstKey);
+ if (olapScan != null
+ && olapScan.getTable().isDupKeysOrMergeOnWrite()
+ && olapScan instanceof PhysicalCatalogRelation) {
+ return Optional.of(olapScan);
}
- olapScan = (OlapScan) child;
- if (olapScan.getTable().isDupKeysOrMergeOnWrite()) {
- return topN.withEnableRuntimeFilter(true).copyStatsAndGroupIdFrom(topN);
- }
+ return Optional.empty();
+ }
- return topN;
+ private OlapScan findScanNodeBySlotReference(Plan root, SlotReference slot) {
+ OlapScan target = null;
+ if (root instanceof OlapScan && root.getOutputSet().contains(slot)) {
+ return (OlapScan) root;
+ } else {
+ if (! root.children().isEmpty()) {
+ // for join and intersect, push topn-filter to their left child.
+ // TODO for union, topn-filter can be pushed down to all of its children.
+ Plan child = root.child(0);
+ if (!(child instanceof PhysicalWindow) && child.getOutputSet().contains(slot)) {
+ target = findScanNodeBySlotReference(child, slot);
+ if (target != null) {
+ return target;
+ }
+ }
+ }
+ }
+ return target;
}
private long getTopNOptLimitThreshold() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterContext.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterContext.java
new file mode 100644
index 00000000000..b5f79defef4
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/processor/post/TopnFilterContext.java
@@ -0,0 +1,93 @@
+// 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.processor.post;
+
+import org.apache.doris.nereids.trees.plans.algebra.OlapScan;
+import org.apache.doris.nereids.trees.plans.algebra.TopN;
+import org.apache.doris.planner.OlapScanNode;
+import org.apache.doris.planner.SortNode;
+
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Sets;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+/**
+ * topN runtime filter context
+ */
+public class TopnFilterContext {
+ private final Map<TopN, List<OlapScan>> filters = Maps.newHashMap();
+ private final Set<TopN> sources = Sets.newHashSet();
+ private final Set<OlapScan> targets = Sets.newHashSet();
+ private final Map<OlapScan, OlapScanNode> legacyTargetsMap = Maps.newHashMap();
+ private final Map<TopN, SortNode> legacySourceMap = Maps.newHashMap();
+
+ /**
+ * add topN filter
+ */
+ public void addTopnFilter(TopN topn, OlapScan scan) {
+ targets.add(scan);
+ sources.add(topn);
+
+ List<OlapScan> targets = filters.get(topn);
+ if (targets == null) {
+ filters.put(topn, Lists.newArrayList(scan));
+ } else {
+ targets.add(scan);
+ }
+ }
+
+ /**
+ * find the corresponding sortNode for topn filter
+ */
+ public Optional<OlapScanNode> getLegacyScanNode(OlapScan scan) {
+ return legacyTargetsMap.keySet().contains(scan)
+ ? Optional.of(legacyTargetsMap.get(scan))
+ : Optional.empty();
+ }
+
+ public Optional<SortNode> getLegacySortNode(TopN topn) {
+ return legacyTargetsMap.keySet().contains(topn)
+ ? Optional.of(legacySourceMap.get(topn))
+ : Optional.empty();
+ }
+
+ public boolean isTopnFilterSource(TopN topn) {
+ return sources.contains(topn);
+ }
+
+ public boolean isTopnFilterTarget(OlapScan scan) {
+ return targets.contains(scan);
+ }
+
+ public void addLegacySource(TopN topn, SortNode sort) {
+ legacySourceMap.put(topn, sort);
+ }
+
+ public void addLegacyTarget(OlapScan olapScan, OlapScanNode legacy) {
+ legacyTargetsMap.put(olapScan, legacy);
+ }
+
+ public List<OlapScan> getTargets(TopN topn) {
+ return filters.get(topn);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalTopN.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalTopN.java
index 6989284e0b9..96dc709bbde 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalTopN.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/physical/PhysicalTopN.java
@@ -43,36 +43,33 @@ public class PhysicalTopN<CHILD_TYPE extends Plan> extends AbstractPhysicalSort<
private final long limit;
private final long offset;
- private final boolean enableRuntimeFilter;
public PhysicalTopN(List<OrderKey> orderKeys, long limit, long offset,
SortPhase phase, LogicalProperties logicalProperties, CHILD_TYPE child) {
- this(orderKeys, limit, offset, phase, false, Optional.empty(), logicalProperties, child);
+ this(orderKeys, limit, offset, phase, Optional.empty(), logicalProperties, child);
}
/**
* Constructor of PhysicalHashJoinNode.
*/
public PhysicalTopN(List<OrderKey> orderKeys, long limit, long offset,
- SortPhase phase, boolean enableRuntimeFilter,
+ SortPhase phase,
Optional<GroupExpression> groupExpression, LogicalProperties logicalProperties, CHILD_TYPE child) {
- this(orderKeys, limit, offset, phase, enableRuntimeFilter,
- groupExpression, logicalProperties, null, null, child);
+ this(orderKeys, limit, offset, phase, groupExpression,
+ logicalProperties, null, null, child);
}
/**
* Constructor of PhysicalHashJoinNode.
*/
public PhysicalTopN(List<OrderKey> orderKeys, long limit, long offset,
- SortPhase phase, boolean enableRuntimeFilter,
- Optional<GroupExpression> groupExpression, LogicalProperties logicalProperties,
+ SortPhase phase, Optional<GroupExpression> groupExpression, LogicalProperties logicalProperties,
PhysicalProperties physicalProperties, Statistics statistics, CHILD_TYPE child) {
super(PlanType.PHYSICAL_TOP_N, orderKeys, phase, groupExpression, logicalProperties, physicalProperties,
statistics, child);
Objects.requireNonNull(orderKeys, "orderKeys should not be null in PhysicalTopN.");
this.limit = limit;
this.offset = offset;
- this.enableRuntimeFilter = enableRuntimeFilter;
}
public long getLimit() {
@@ -83,10 +80,6 @@ public class PhysicalTopN<CHILD_TYPE extends Plan> extends AbstractPhysicalSort<
return offset;
}
- public boolean isEnableRuntimeFilter() {
- return enableRuntimeFilter;
- }
-
@Override
public boolean equals(Object o) {
if (this == o) {
@@ -99,12 +92,12 @@ public class PhysicalTopN<CHILD_TYPE extends Plan> extends AbstractPhysicalSort<
return false;
}
PhysicalTopN<?> that = (PhysicalTopN<?>) o;
- return limit == that.limit && offset == that.offset && enableRuntimeFilter == that.enableRuntimeFilter;
+ return limit == that.limit && offset == that.offset;
}
@Override
public int hashCode() {
- return Objects.hash(super.hashCode(), limit, offset, enableRuntimeFilter);
+ return Objects.hash(super.hashCode(), limit, offset);
}
@Override
@@ -112,22 +105,17 @@ public class PhysicalTopN<CHILD_TYPE extends Plan> extends AbstractPhysicalSort<
return visitor.visitPhysicalTopN(this, context);
}
- public PhysicalTopN<Plan> withEnableRuntimeFilter(boolean enableRuntimeFilter) {
- return new PhysicalTopN<>(orderKeys, limit, offset, phase, enableRuntimeFilter,
- groupExpression, getLogicalProperties(), child());
- }
-
@Override
public PhysicalTopN<Plan> withChildren(List<Plan> children) {
Preconditions.checkArgument(children.size() == 1,
"PhysicalTopN's children size must be 1, but real is %s", children.size());
- return new PhysicalTopN<>(orderKeys, limit, offset, phase, enableRuntimeFilter, groupExpression,
+ return new PhysicalTopN<>(orderKeys, limit, offset, phase, groupExpression,
getLogicalProperties(), physicalProperties, statistics, children.get(0));
}
@Override
public PhysicalTopN<CHILD_TYPE> withGroupExpression(Optional<GroupExpression> groupExpression) {
- return new PhysicalTopN<>(orderKeys, limit, offset, phase, enableRuntimeFilter,
+ return new PhysicalTopN<>(orderKeys, limit, offset, phase,
groupExpression, getLogicalProperties(), child());
}
@@ -136,14 +124,14 @@ public class PhysicalTopN<CHILD_TYPE extends Plan> extends AbstractPhysicalSort<
Optional<LogicalProperties> logicalProperties, List<Plan> children) {
Preconditions.checkArgument(children.size() == 1,
"PhysicalTopN's children size must be 1, but real is %s", children.size());
- return new PhysicalTopN<>(orderKeys, limit, offset, phase, enableRuntimeFilter,
+ return new PhysicalTopN<>(orderKeys, limit, offset, phase,
groupExpression, logicalProperties.get(), children.get(0));
}
@Override
public PhysicalTopN<CHILD_TYPE> withPhysicalPropertiesAndStats(PhysicalProperties physicalProperties,
Statistics statistics) {
- return new PhysicalTopN<>(orderKeys, limit, offset, phase, enableRuntimeFilter,
+ return new PhysicalTopN<>(orderKeys, limit, offset, phase,
groupExpression, getLogicalProperties(), physicalProperties, statistics, child());
}
@@ -158,8 +146,7 @@ public class PhysicalTopN<CHILD_TYPE extends Plan> extends AbstractPhysicalSort<
"limit", limit,
"offset", offset,
"orderKeys", orderKeys,
- "phase", phase.toString(),
- "enableRuntimeFilter", enableRuntimeFilter
+ "phase", phase.toString()
);
}
@@ -170,7 +157,7 @@ public class PhysicalTopN<CHILD_TYPE extends Plan> extends AbstractPhysicalSort<
@Override
public PhysicalTopN<Plan> resetLogicalProperties() {
- return new PhysicalTopN<>(orderKeys, limit, offset, phase, enableRuntimeFilter, groupExpression,
+ return new PhysicalTopN<>(orderKeys, limit, offset, phase, groupExpression,
null, physicalProperties, statistics, child());
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
index 10e311d8d14..42f9c8a551e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/OlapScanNode.java
@@ -174,8 +174,11 @@ public class OlapScanNode extends ScanNode {
// It's limit for scanner instead of scanNode so we add a new limit.
private long sortLimit = -1;
+ // useTopnOpt is equivalent to !topnFilterSortNodes.isEmpty().
+ // keep this flag for compatibility.
private boolean useTopnOpt = false;
-
+ // support multi topn filter
+ private final List<SortNode> topnFilterSortNodes = Lists.newArrayList();
// List of tablets will be scanned by current olap_scan_node
private ArrayList<Long> scanTabletIds = Lists.newArrayList();
@@ -1304,7 +1307,10 @@ public class OlapScanNode extends ScanNode {
output.append(prefix).append("SORT LIMIT: ").append(sortLimit).append("\n");
}
if (useTopnOpt) {
- output.append(prefix).append("TOPN OPT\n");
+ String topnFilterSources = String.join(",",
+ topnFilterSortNodes.stream()
+ .map(node -> node.getId().asInt() + "").collect(Collectors.toList()));
+ output.append(prefix).append("TOPN OPT:").append(topnFilterSources).append("\n");
}
if (!conjuncts.isEmpty()) {
@@ -1478,6 +1484,13 @@ public class OlapScanNode extends ScanNode {
msg.olap_scan_node.setSortLimit(sortLimit);
}
msg.olap_scan_node.setUseTopnOpt(useTopnOpt);
+ List<Integer> topnFilterSourceNodeIds = getTopnFilterSortNodes()
+ .stream()
+ .map(sortNode -> sortNode.getId().asInt())
+ .collect(Collectors.toList());
+ if (!topnFilterSourceNodeIds.isEmpty()) {
+ msg.olap_scan_node.setTopnFilterSourceNodeIds(topnFilterSourceNodeIds);
+ }
msg.olap_scan_node.setKeyType(olapTable.getKeysType().toThrift());
msg.olap_scan_node.setTableName(olapTable.getName());
msg.olap_scan_node.setEnableUniqueKeyMergeOnWrite(olapTable.getEnableUniqueKeyMergeOnWrite());
@@ -1750,4 +1763,12 @@ public class OlapScanNode extends ScanNode {
public int getScanRangeNum() {
return getScanTabletIds().size();
}
+
+ public void addTopnFilterSortNode(SortNode sortNode) {
+ topnFilterSortNodes.add(sortNode);
+ }
+
+ public List<SortNode> getTopnFilterSortNodes() {
+ return topnFilterSortNodes;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
index 33a04c5dfa1..24b384d4453 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/SortNode.java
@@ -61,7 +61,7 @@ public class SortNode extends PlanNode {
List<Expr> resolvedTupleExprs;
private final SortInfo info;
private final boolean useTopN;
- private boolean useTopnOpt;
+ private boolean useTopnOpt = false;
private boolean useTwoPhaseReadOpt;
// If mergeByexchange is set to true, the sort information is pushed to the
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index 85a65161537..8bf1b3813ba 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -3702,7 +3702,10 @@ public class Coordinator implements CoordInterface {
int rate = Math.min(Config.query_colocate_join_memory_limit_penalty_factor, instanceExecParams.size());
memLimit = queryOptions.getMemLimit() / rate;
}
-
+ Set<Integer> topnFilterSources = scanNodes.stream()
+ .filter(scanNode -> scanNode instanceof OlapScanNode)
+ .flatMap(scanNode -> ((OlapScanNode) scanNode).getTopnFilterSortNodes().stream())
+ .map(sort -> sort.getId().asInt()).collect(Collectors.toSet());
Map<TNetworkAddress, TPipelineFragmentParams> res = new HashMap();
Map<TNetworkAddress, Integer> instanceIdx = new HashMap();
TPlanFragment fragmentThrift = fragment.toThrift();
@@ -3771,6 +3774,12 @@ public class Coordinator implements CoordInterface {
localParams.setBackendNum(backendNum++);
localParams.setRuntimeFilterParams(new TRuntimeFilterParams());
localParams.runtime_filter_params.setRuntimeFilterMergeAddr(runtimeFilterMergeAddr);
+ if (!topnFilterSources.isEmpty()) {
+ // topn_filter_source_node_ids is used by nereids not by legacy planner.
+ // if there is no topnFilterSources, do not set it.
+ // topn_filter_source_node_ids=null means legacy planner
+ localParams.topn_filter_source_node_ids = Lists.newArrayList(topnFilterSources);
+ }
if (instanceExecParam.instanceId.equals(runtimeFilterMergeInstanceId)) {
Set<Integer> broadCastRf = assignedRuntimeFilters.stream().filter(RuntimeFilter::isBroadcast)
.map(r -> r.getFilterId().asInt()).collect(Collectors.toSet());
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
index 95a407d1342..57c6c045141 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/postprocess/TopNRuntimeFilterTest.java
@@ -24,12 +24,13 @@ import org.apache.doris.nereids.trees.plans.SortPhase;
import org.apache.doris.nereids.trees.plans.physical.PhysicalDeferMaterializeTopN;
import org.apache.doris.nereids.trees.plans.physical.PhysicalPlan;
import org.apache.doris.nereids.trees.plans.physical.PhysicalTopN;
+import org.apache.doris.nereids.util.MemoPatternMatchSupported;
import org.apache.doris.nereids.util.PlanChecker;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-public class TopNRuntimeFilterTest extends SSBTestBase {
+public class TopNRuntimeFilterTest extends SSBTestBase implements MemoPatternMatchSupported {
@Override
public void runBeforeAll() throws Exception {
super.runBeforeAll();
@@ -46,11 +47,11 @@ public class TopNRuntimeFilterTest extends SSBTestBase {
Assertions.assertInstanceOf(PhysicalDeferMaterializeTopN.class, plan.children().get(0).child(0));
PhysicalDeferMaterializeTopN<? extends Plan> localTopN
= (PhysicalDeferMaterializeTopN<? extends Plan>) plan.child(0).child(0);
- Assertions.assertTrue(localTopN.getPhysicalTopN().isEnableRuntimeFilter());
+ Assertions.assertTrue(checker.getCascadesContext().getTopnFilterContext().isTopnFilterSource(localTopN));
}
@Test
- public void testNotUseTopNRfForComplexCase() {
+ public void testUseTopNRfForComplexCase() {
String sql = "select * from (select 1) tl join (select * from customer order by c_custkey limit 5) tb";
PlanChecker checker = PlanChecker.from(connectContext).analyze(sql)
.rewrite()
@@ -62,6 +63,21 @@ public class TopNRuntimeFilterTest extends SSBTestBase {
.child(0).child(0).child(1).child(0)).getSortPhase());
PhysicalTopN<? extends Plan> localTopN = (PhysicalTopN<? extends Plan>) plan
.child(0).child(0).child(1).child(0);
- Assertions.assertFalse(localTopN.isEnableRuntimeFilter());
+ Assertions.assertTrue(checker.getCascadesContext().getTopnFilterContext().isTopnFilterSource(localTopN));
+ }
+
+ @Test
+ public void testNotUseTopNRfOnWindow() {
+ String sql = "select rank() over (partition by c_nation order by c_custkey) "
+ + "from customer order by c_custkey limit 3";
+ PlanChecker checker = PlanChecker.from(connectContext).analyze(sql)
+ .rewrite().implement();
+ PhysicalPlan plan = checker.getPhysicalPlan();
+ plan = new PlanPostProcessors(checker.getCascadesContext()).process(plan);
+ System.out.println(plan.treeString());
+ PhysicalTopN<? extends Plan> localTopN =
+ (PhysicalTopN<? extends Plan>) plan.child(0).child(0).child(0);
+ Assertions.assertTrue(localTopN.getSortPhase().isLocal());
+ Assertions.assertFalse(checker.getCascadesContext().getTopnFilterContext().isTopnFilterSource(localTopN));
}
}
diff --git a/regression-test/data/nereids_tpch_p0/tpch/topn-filter.out b/regression-test/data/nereids_tpch_p0/tpch/topn-filter.out
new file mode 100644
index 00000000000..be88d829f25
--- /dev/null
+++ b/regression-test/data/nereids_tpch_p0/tpch/topn-filter.out
@@ -0,0 +1,29 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !simpleTopn --
+1 3691 O 194029.55 1996-01-02 5-LOW Clerk#000000951 0 nstructions sleep furiously among
+2 7801 O 60951.63 1996-12-01 1-URGENT Clerk#000000880 0 foxes. pending accounts at the pending, silent asymptot
+3 12332 F 247296.05 1993-10-14 5-LOW Clerk#000000955 0 sly final accounts boost. carefully regular ideas cajole carefully. depos
+4 13678 O 53829.87 1995-10-11 5-LOW Clerk#000000124 0 sits. slyly regular warthogs cajole. regular, regular theodolites acro
+5 4450 F 139660.54 1994-07-30 5-LOW Clerk#000000925 0 quickly. bold deposits sleep slyly. packages use slyly
+6 5563 F 65843.52 1992-02-21 4-NOT SPECIFIED Clerk#000000058 0 ggle. special, final requests are against the furiously specia
+7 3914 O 231037.28 1996-01-10 2-HIGH Clerk#000000470 0 ly special requests
+32 13006 O 166802.63 1995-07-16 2-HIGH Clerk#000000616 0 ise blithely bold, regular requests. quickly unusual dep
+33 6697 F 118518.56 1993-10-27 3-MEDIUM Clerk#000000409 0 uriously. furiously final request
+34 6101 O 75662.77 1998-07-21 3-MEDIUM Clerk#000000223 0 ly final packages. fluffily final deposits wake blithely ideas. spe
+
+-- !complexTopn --
+1 3691 10
+2 7801 1
+
+-- !check_result --
+67 5662 0
+102 73 0
+
+-- !check_result2 --
+33 6697 24
+551 8962 24
+
+-- !groupingsets --
+0 50
+1 47
+
diff --git a/regression-test/suites/nereids_tpch_p0/tpch/topn-filter.groovy b/regression-test/suites/nereids_tpch_p0/tpch/topn-filter.groovy
new file mode 100644
index 00000000000..23e742fba15
--- /dev/null
+++ b/regression-test/suites/nereids_tpch_p0/tpch/topn-filter.groovy
@@ -0,0 +1,120 @@
+/*
+ * 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.
+ */
+
+suite("topn-filter") {
+ String db = context.config.getDbNameByFile(new File(context.file.parent))
+ sql "use ${db}"
+ sql 'set enable_nereids_planner=true'
+ sql 'set enable_fallback_to_original_planner=false'
+ sql 'set disable_join_reorder=true;'
+ sql 'set topn_opt_limit_threshold=1024'
+ def String simpleTopn = """
+ select *
+ from orders
+ order by o_orderkey
+ limit 10;"""
+
+ explain {
+ sql "${simpleTopn}"
+ contains "TOPN OPT:1"
+ }
+
+ qt_simpleTopn "${simpleTopn}"
+
+ def String complexTopn = """
+ select o_orderkey, c_custkey, n_nationkey
+ from orders
+ join[broadcast] customer on o_custkey = c_custkey
+ join[broadcast] nation on c_nationkey=n_nationkey
+ order by o_orderkey limit 2;
+ """
+ explain{
+ sql "${complexTopn}"
+ contains "TOPN OPT:7"
+ }
+ qt_complexTopn "${complexTopn}"
+
+ def multi_topn_asc = """
+ select o_orderkey, c_custkey, n_nationkey
+ from orders
+ join[broadcast] customer on o_custkey = c_custkey
+ join[broadcast] ( select * from nation order by n_nationkey asc limit 1) as n on c_nationkey=n_nationkey
+ order by o_orderkey limit 2;
+ """
+ qt_check_result "${multi_topn_asc}"
+ explain{
+ sql "${multi_topn_asc}"
+ contains "TOPN OPT:9"
+ contains "TOPN OPT:1"
+ }
+
+ def multi_topn_desc = """
+ select o_orderkey, c_custkey, n_nationkey
+ from orders
+ join[broadcast] customer on o_custkey = c_custkey
+ join[broadcast] (select * from nation order by n_nationkey desc limit 1) as n on c_nationkey=n_nationkey
+ order by o_orderkey limit 2;
+ """
+ explain {
+ sql "${multi_topn_desc}"
+ contains "TOPN OPT:9"
+ contains "TOPN OPT:1"
+ }
+
+ qt_check_result2 "${multi_topn_desc}"
+
+ // do not use topn-filter
+ explain {
+ sql """
+ select o_orderkey, c_custkey
+ from orders
+ join[broadcast] customer on o_custkey = c_custkey
+ order by c_custkey limit 2;
+ """
+ notContains "TOPN OPT:"
+ }
+
+ // push topn filter down through AGG
+ explain {
+ sql """
+ select s_nationkey, count(1) from supplier group by s_nationkey order by s_nationkey limit 1;
+ """
+ contains "TOPN OPT:"
+ }
+
+ // push topn filter down through AGG + Join
+ explain {
+ sql """
+ select *
+ from
+ (select s_nationkey, count(1) as total from supplier group by s_nationkey having total > 10 ) T
+ join nation on s_nationkey = n_nationkey
+ order by s_nationkey limit 1;
+ """
+ contains "TOPN OPT:"
+ }
+
+ explain {
+ sql "select n_regionkey, sum(n_nationkey) from nation group by grouping sets((n_regionkey)) order by n_regionkey limit 2;"
+ contains "TOPN OPT"
+ }
+
+ qt_groupingsets "select n_regionkey, sum(n_nationkey) from nation group by grouping sets((n_regionkey)) order by n_regionkey limit 2;"
+
+}
\ No newline at end of file
diff --git a/regression-test/suites/point_query_p0/load.groovy b/regression-test/suites/point_query_p0/load.groovy
index d5cf8074540..772a7130363 100644
--- a/regression-test/suites/point_query_p0/load.groovy
+++ b/regression-test/suites/point_query_p0/load.groovy
@@ -18,8 +18,7 @@
import org.codehaus.groovy.runtime.IOGroovyMethods
suite("test_point_query_load", "p0") {
-
- // nereids do not support point query now
+ //test legacy planner
sql """set enable_nereids_planner=false"""
def dataFile = """${getS3Url()}/regression/datatypes/test_scalar_types_10w.csv"""
@@ -101,6 +100,7 @@ suite("test_point_query_load", "p0") {
}
sql "INSERT INTO ${testTable} SELECT * from ${testTable}"
+ // test nereids planner
sql """set enable_nereids_planner=true;"""
explain {
sql("""SELECT
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 14/24: [Fix](Outfiel) fix be core when the open method of vfile_result_writer failed #32042
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 aea9ddc3cb04c03c34fca7631223cef006a52461
Author: Tiewei Fang <43...@users.noreply.github.com>
AuthorDate: Mon Mar 11 22:16:58 2024 +0800
[Fix](Outfiel) fix be core when the open method of vfile_result_writer failed #32042
---
be/src/vec/sink/writer/vfile_result_writer.cpp | 24 ++++++++++++++----------
be/src/vec/sink/writer/vfile_result_writer.h | 2 +-
2 files changed, 15 insertions(+), 11 deletions(-)
diff --git a/be/src/vec/sink/writer/vfile_result_writer.cpp b/be/src/vec/sink/writer/vfile_result_writer.cpp
index 661afa15935..8952aa769ae 100644
--- a/be/src/vec/sink/writer/vfile_result_writer.cpp
+++ b/be/src/vec/sink/writer/vfile_result_writer.cpp
@@ -420,17 +420,21 @@ Status VFileResultWriter::_delete_dir() {
return Status::OK();
}
-Status VFileResultWriter::close(Status) {
- // the following 2 profile "_written_rows_counter" and "_writer_close_timer"
- // must be outside the `_close_file_writer()`.
- // because `_close_file_writer()` may be called in deconstructor,
- // at that time, the RuntimeState may already been deconstructed,
- // so does the profile in RuntimeState.
- if (_written_rows_counter) {
- COUNTER_SET(_written_rows_counter, _written_rows);
- SCOPED_TIMER(_writer_close_timer);
+Status VFileResultWriter::close(Status exec_status) {
+ Status st = exec_status;
+ if (st.ok()) {
+ // the following 2 profile "_written_rows_counter" and "_writer_close_timer"
+ // must be outside the `_close_file_writer()`.
+ // because `_close_file_writer()` may be called in deconstructor,
+ // at that time, the RuntimeState may already been deconstructed,
+ // so does the profile in RuntimeState.
+ if (_written_rows_counter) {
+ COUNTER_SET(_written_rows_counter, _written_rows);
+ SCOPED_TIMER(_writer_close_timer);
+ }
+ st = _close_file_writer(true);
}
- return _close_file_writer(true);
+ return st;
}
} // namespace doris::vectorized
diff --git a/be/src/vec/sink/writer/vfile_result_writer.h b/be/src/vec/sink/writer/vfile_result_writer.h
index c52050ba6f1..29fd6d89cd3 100644
--- a/be/src/vec/sink/writer/vfile_result_writer.h
+++ b/be/src/vec/sink/writer/vfile_result_writer.h
@@ -60,7 +60,7 @@ public:
Status write(Block& block) override;
- Status close(Status s = Status::OK()) override;
+ Status close(Status exec_status) override;
Status open(RuntimeState* state, RuntimeProfile* profile) override;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 18/24: [Improvement](executor)Routine load support workload group #31671
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 194f3432ab86a42b80825a17258b08ebf7e47067
Author: wangbo <wa...@apache.org>
AuthorDate: Mon Mar 11 22:51:15 2024 +0800
[Improvement](executor)Routine load support workload group #31671
---
.../doris/analysis/AlterRoutineLoadStmt.java | 8 +++++
.../doris/analysis/CreateRoutineLoadStmt.java | 14 ++++++++
.../doris/load/routineload/KafkaTaskInfo.java | 27 +++++++++++++++
.../doris/load/routineload/RoutineLoadJob.java | 14 ++++++++
.../resource/workloadgroup/WorkloadGroupMgr.java | 39 ++++++++++++++++++++--
regression-test/pipeline/p0/conf/fe.conf | 1 +
.../load_p0/routine_load/test_routine_load.groovy | 8 ++++-
.../workload_manager_p0/test_curd_wlg.groovy | 12 +++----
8 files changed, 114 insertions(+), 9 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterRoutineLoadStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterRoutineLoadStmt.java
index 2df891fbb3c..5a1f1ba56aa 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterRoutineLoadStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/AlterRoutineLoadStmt.java
@@ -29,6 +29,7 @@ import org.apache.doris.common.util.Util;
import org.apache.doris.load.routineload.AbstractDataSourceProperties;
import org.apache.doris.load.routineload.RoutineLoadDataSourcePropertyFactory;
import org.apache.doris.load.routineload.RoutineLoadJob;
+import org.apache.doris.qe.ConnectContext;
import com.google.common.collect.ImmutableSet;
import com.google.common.collect.Maps;
@@ -66,6 +67,7 @@ public class AlterRoutineLoadStmt extends DdlStmt {
.add(CreateRoutineLoadStmt.PARTIAL_COLUMNS)
.add(LoadStmt.STRICT_MODE)
.add(LoadStmt.TIMEZONE)
+ .add(CreateRoutineLoadStmt.WORKLOAD_GROUP)
.build();
private final LabelName labelName;
@@ -242,6 +244,12 @@ public class AlterRoutineLoadStmt extends DdlStmt {
analyzedJobProperties.put(CreateRoutineLoadStmt.PARTIAL_COLUMNS,
String.valueOf(isPartialUpdate));
}
+ if (jobProperties.containsKey(CreateRoutineLoadStmt.WORKLOAD_GROUP)) {
+ String workloadGroup = jobProperties.get(CreateRoutineLoadStmt.WORKLOAD_GROUP);
+ long wgId = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getWorkloadGroup(ConnectContext.get().getCurrentUserIdentity(), workloadGroup);
+ analyzedJobProperties.put(CreateRoutineLoadStmt.WORKLOAD_GROUP, String.valueOf(wgId));
+ }
}
private void checkDataSourceProperties() throws UserException {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java
index f859d7d8f05..d58b25195c7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRoutineLoadStmt.java
@@ -111,6 +111,8 @@ public class CreateRoutineLoadStmt extends DdlStmt {
public static final String PARTIAL_COLUMNS = "partial_columns";
+ public static final String WORKLOAD_GROUP = "workload_group";
+
private static final String NAME_TYPE = "ROUTINE LOAD NAME";
public static final String ENDPOINT_REGEX = "[-A-Za-z0-9+&@#/%?=~_|!:,.;]+[-A-Za-z0-9+&@#/%=~_|]";
public static final String SEND_BATCH_PARALLELISM = "send_batch_parallelism";
@@ -138,6 +140,7 @@ public class CreateRoutineLoadStmt extends DdlStmt {
.add(SEND_BATCH_PARALLELISM)
.add(LOAD_TO_SINGLE_TABLET)
.add(PARTIAL_COLUMNS)
+ .add(WORKLOAD_GROUP)
.build();
private final LabelName labelName;
@@ -179,6 +182,8 @@ public class CreateRoutineLoadStmt extends DdlStmt {
private String escape;
+ private long workloadGroupId = -1;
+
/**
* support partial columns load(Only Unique Key Columns)
*/
@@ -330,6 +335,10 @@ public class CreateRoutineLoadStmt extends DdlStmt {
return comment;
}
+ public long getWorkloadGroupId() {
+ return workloadGroupId;
+ }
+
@Override
public void analyze(Analyzer analyzer) throws UserException {
super.analyze(analyzer);
@@ -506,6 +515,11 @@ public class CreateRoutineLoadStmt extends DdlStmt {
if (escape != null && escape.length() != 1) {
throw new AnalysisException("escape must be single-char");
}
+ String inputWorkloadGroupStr = jobProperties.get(WORKLOAD_GROUP);
+ if (!StringUtils.isEmpty(inputWorkloadGroupStr)) {
+ this.workloadGroupId = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getWorkloadGroup(ConnectContext.get().getCurrentUserIdentity(), inputWorkloadGroupStr);
+ }
if (ConnectContext.get() != null) {
timezone = ConnectContext.get().getSessionVariable().getTimeZone();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
index a8d387a2f6d..d8b79d9bdce 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/KafkaTaskInfo.java
@@ -28,6 +28,7 @@ import org.apache.doris.thrift.TFileFormatType;
import org.apache.doris.thrift.TKafkaLoadInfo;
import org.apache.doris.thrift.TLoadSourceType;
import org.apache.doris.thrift.TPipelineFragmentParams;
+import org.apache.doris.thrift.TPipelineWorkloadGroup;
import org.apache.doris.thrift.TPlanFragment;
import org.apache.doris.thrift.TRoutineLoadTask;
import org.apache.doris.thrift.TUniqueId;
@@ -130,6 +131,19 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
TExecPlanFragmentParams tExecPlanFragmentParams = routineLoadJob.plan(loadId, txnId);
TPlanFragment tPlanFragment = tExecPlanFragmentParams.getFragment();
tPlanFragment.getOutputSink().getOlapTableSink().setTxnId(txnId);
+
+ long wgId = routineLoadJob.getWorkloadId();
+ List<TPipelineWorkloadGroup> tWgList = new ArrayList<>();
+ if (wgId > 0) {
+ tWgList = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getTWorkloadGroupById(wgId);
+ }
+ if (tWgList.size() == 0) {
+ tWgList = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getTWorkloadGroupByUserIdentity(routineLoadJob.getUserIdentity());
+ }
+ tExecPlanFragmentParams.setWorkloadGroups(tWgList);
+
return tExecPlanFragmentParams;
}
@@ -139,6 +153,19 @@ public class KafkaTaskInfo extends RoutineLoadTaskInfo {
TPipelineFragmentParams tExecPlanFragmentParams = routineLoadJob.planForPipeline(loadId, txnId);
TPlanFragment tPlanFragment = tExecPlanFragmentParams.getFragment();
tPlanFragment.getOutputSink().getOlapTableSink().setTxnId(txnId);
+
+ long wgId = routineLoadJob.getWorkloadId();
+ List<TPipelineWorkloadGroup> tWgList = new ArrayList<>();
+ if (wgId > 0) {
+ tWgList = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getTWorkloadGroupById(wgId);
+ }
+ if (tWgList.size() == 0) {
+ tWgList = Env.getCurrentEnv().getWorkloadGroupMgr()
+ .getTWorkloadGroupByUserIdentity(routineLoadJob.getUserIdentity());
+ }
+ tExecPlanFragmentParams.setWorkloadGroups(tWgList);
+
return tExecPlanFragmentParams;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
index 8760dc4b71c..f9be2014e30 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/routineload/RoutineLoadJob.java
@@ -69,6 +69,7 @@ import org.apache.doris.transaction.TransactionException;
import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.TransactionStatus;
+import com.aliyuncs.utils.StringUtils;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
@@ -117,6 +118,8 @@ public abstract class RoutineLoadJob extends AbstractTxnStateChangeCallback impl
protected static final String STAR_STRING = "*";
+ public static final String WORKLOAD_GROUP = "workload_group";
+
@Getter
@Setter
private boolean isMultiTable = false;
@@ -394,6 +397,9 @@ public abstract class RoutineLoadJob extends AbstractTxnStateChangeCallback impl
if (stmt.getEscape() != null) {
jobProperties.put(LoadStmt.KEY_ESCAPE, stmt.getEscape());
}
+ if (stmt.getWorkloadGroupId() > 0) {
+ jobProperties.put(WORKLOAD_GROUP, String.valueOf(stmt.getWorkloadGroupId()));
+ }
}
private void setRoutineLoadDesc(RoutineLoadDesc routineLoadDesc) {
@@ -479,6 +485,14 @@ public abstract class RoutineLoadJob extends AbstractTxnStateChangeCallback impl
return database.getTableOrMetaException(tableId).getName();
}
+ public long getWorkloadId() {
+ String workloadIdStr = jobProperties.get(WORKLOAD_GROUP);
+ if (!StringUtils.isEmpty(workloadIdStr)) {
+ return Long.parseLong(workloadIdStr);
+ }
+ return -1;
+ }
+
public JobState getState() {
return state;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
index a7c26f7cec5..1bd1a357127 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/WorkloadGroupMgr.java
@@ -183,13 +183,48 @@ public class WorkloadGroupMgr implements Writable, GsonPostProcessable {
return workloadGroups;
}
- public WorkloadGroup getWorkloadGroupById(long wgId) {
+ public long getWorkloadGroup(UserIdentity currentUser, String groupName) throws UserException {
+ Long workloadId = getWorkloadGroupIdByName(groupName);
+ if (workloadId == null) {
+ throw new UserException("Workload group " + groupName + " does not exist");
+ }
+ if (!Env.getCurrentEnv().getAccessManager()
+ .checkWorkloadGroupPriv(currentUser, groupName, PrivPredicate.USAGE)) {
+ ErrorReport.reportAnalysisException(
+ "Access denied; you need (at least one of) the %s privilege(s) to use workload group '%s'.",
+ ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "USAGE/ADMIN", groupName);
+ }
+ return workloadId.longValue();
+ }
+
+ public List<TPipelineWorkloadGroup> getTWorkloadGroupById(long wgId) {
+ List<TPipelineWorkloadGroup> tWorkloadGroups = Lists.newArrayList();
+ readLock();
+ try {
+ WorkloadGroup wg = idToWorkloadGroup.get(wgId);
+ if (wg != null) {
+ tWorkloadGroups.add(wg.toThrift());
+ }
+ } finally {
+ readUnlock();
+ }
+ return tWorkloadGroups;
+ }
+
+ public List<TPipelineWorkloadGroup> getTWorkloadGroupByUserIdentity(UserIdentity user) throws UserException {
+ String groupName = Env.getCurrentEnv().getAuth().getWorkloadGroup(user.getQualifiedUser());
+ List<TPipelineWorkloadGroup> ret = new ArrayList<>();
readLock();
try {
- return idToWorkloadGroup.get(wgId);
+ WorkloadGroup wg = nameToWorkloadGroup.get(groupName);
+ if (wg == null) {
+ throw new UserException("can not find workload group " + groupName);
+ }
+ ret.add(wg.toThrift());
} finally {
readUnlock();
}
+ return ret;
}
public List<TopicInfo> getPublishTopicInfo() {
diff --git a/regression-test/pipeline/p0/conf/fe.conf b/regression-test/pipeline/p0/conf/fe.conf
index 22f58cb22f5..28f1972c701 100644
--- a/regression-test/pipeline/p0/conf/fe.conf
+++ b/regression-test/pipeline/p0/conf/fe.conf
@@ -109,6 +109,7 @@ label_keep_max_second = 300
enable_job_schedule_second_for_test = true
enable_workload_group = true
+publish_topic_info_interval_ms = 1000
master_sync_policy = WRITE_NO_SYNC
replica_sync_policy = WRITE_NO_SYNC
diff --git a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy
index 963d6ebc110..87eae26a48a 100644
--- a/regression-test/suites/load_p0/routine_load/test_routine_load.groovy
+++ b/regression-test/suites/load_p0/routine_load/test_routine_load.groovy
@@ -22,6 +22,10 @@ import org.apache.kafka.clients.producer.ProducerConfig
suite("test_routine_load","p0") {
+ sql "create workload group if not exists create_routine_load_group properties ( 'cpu_share'='123');"
+ sql "create workload group if not exists alter_routine_load_group properties ( 'cpu_share'='123');"
+ Thread.sleep(5000) // wait publish workload group to be
+
def tables = [
"dup_tbl_basic",
"uniq_tbl_basic",
@@ -226,7 +230,8 @@ suite("test_routine_load","p0") {
"send_batch_parallelism" = "2",
"max_batch_interval" = "5",
"max_batch_rows" = "300000",
- "max_batch_size" = "209715200"
+ "max_batch_size" = "209715200",
+ "workload_group" = "create_routine_load_group"
)
FROM KAFKA
(
@@ -1932,6 +1937,7 @@ suite("test_routine_load","p0") {
sql "ALTER ROUTINE LOAD FOR ${jobs[i]} PROPERTIES(\"timezone\" = \"Asia/Shanghai\");"
sql "ALTER ROUTINE LOAD FOR ${jobs[i]} PROPERTIES(\"num_as_string\" = \"true\");"
sql "ALTER ROUTINE LOAD FOR ${jobs[i]} PROPERTIES(\"fuzzy_parse\" = \"true\");"
+ sql "ALTER ROUTINE LOAD FOR ${jobs[i]} PROPERTIES(\"workload_group\" = \"alter_routine_load_group\");"
res = sql "show routine load for ${jobs[i]}"
log.info("routine load job properties: ${res[0][11].toString()}".toString())
diff --git a/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy b/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy
index 4b4fa51486e..864d1ab5b21 100644
--- a/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy
+++ b/regression-test/suites/workload_manager_p0/test_curd_wlg.groovy
@@ -117,7 +117,7 @@ suite("test_crud_wlg") {
");"
sql "set workload_group=test_group;"
- qt_show_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
+ qt_show_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() where name in ('normal','test_group') order by name;"
// test memory_limit
test {
@@ -128,7 +128,7 @@ suite("test_crud_wlg") {
sql "alter workload group test_group properties ( 'memory_limit'='11%' );"
qt_mem_limit_1 """ select count(1) from ${table_name} """
- qt_mem_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
+ qt_mem_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() where name in ('normal','test_group') order by name;"
// test enable_memory_overcommit
test {
@@ -141,7 +141,7 @@ suite("test_crud_wlg") {
qt_mem_overcommit_1 """ select count(1) from ${table_name} """
sql "alter workload group test_group properties ( 'enable_memory_overcommit'='false' );"
qt_mem_overcommit_2 """ select count(1) from ${table_name} """
- qt_mem_overcommit_3 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
+ qt_mem_overcommit_3 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() where name in ('normal','test_group') order by name;"
// test cpu_hard_limit
test {
@@ -160,7 +160,7 @@ suite("test_crud_wlg") {
sql "alter workload group test_group properties ( 'cpu_hard_limit'='20%' );"
qt_cpu_hard_limit_1 """ select count(1) from ${table_name} """
- qt_cpu_hard_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
+ qt_cpu_hard_limit_2 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() where name in ('normal','test_group') order by name;"
// test query queue
test {
@@ -183,7 +183,7 @@ suite("test_crud_wlg") {
sql "alter workload group test_group properties ( 'max_concurrency'='100' );"
qt_queue_1 """ select count(1) from ${table_name} """
- qt_show_queue "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
+ qt_show_queue "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() where name in ('normal','test_group') order by name;"
// test create group failed
// failed for cpu_share
@@ -261,7 +261,7 @@ suite("test_crud_wlg") {
}
// test show workload groups
- qt_select_tvf_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() order by name;"
+ qt_select_tvf_1 "select name,cpu_share,memory_limit,enable_memory_overcommit,max_concurrency,max_queue_size,queue_timeout,cpu_hard_limit,scan_thread_num from workload_groups() where name in ('normal','test_group') order by name;"
// test auth
sql """drop user if exists test_wlg_user"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 20/24: [doc](sql-dialect) update sql convertor tool to 1.0.1 (#32077)
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 65835cdd2b8e1794012a76b46758110d5dfbc59d
Author: Mingyu Chen <mo...@163.com>
AuthorDate: Tue Mar 12 10:00:20 2024 +0800
[doc](sql-dialect) update sql convertor tool to 1.0.1 (#32077)
---
docs/en/docs/lakehouse/sql-dialect.md | 2 +-
docs/zh-CN/docs/lakehouse/sql-dialect.md | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/en/docs/lakehouse/sql-dialect.md b/docs/en/docs/lakehouse/sql-dialect.md
index e3b16299d2d..a513c93c3c2 100644
--- a/docs/en/docs/lakehouse/sql-dialect.md
+++ b/docs/en/docs/lakehouse/sql-dialect.md
@@ -34,7 +34,7 @@ Starting from version 2.1, Doris can support multiple SQL dialects, such as Pres
## Deploy service
-1. Download latest [Doris SQL Convertor](https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/transform-doris-tool/transform-doris-tool-1.0.0-bin-x86).
+1. Download latest [Doris SQL Convertor(1.0.1)](https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/transform-doris-tool/transform-doris-tool-1.0.0-bin-x86)
2. On any FE node, start the service through the following command:
`nohup ./doris-sql-convertor-1.0.1-bin-x86 run --host=0.0.0.0 --port=5001 &`
diff --git a/docs/zh-CN/docs/lakehouse/sql-dialect.md b/docs/zh-CN/docs/lakehouse/sql-dialect.md
index 3475c8dfe4f..b8216562c28 100644
--- a/docs/zh-CN/docs/lakehouse/sql-dialect.md
+++ b/docs/zh-CN/docs/lakehouse/sql-dialect.md
@@ -34,7 +34,7 @@ under the License.
## 部署服务
-1. 下载最新版本的 [SQL 方言转换工具](https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/transform-doris-tool/transform-doris-tool-1.0.0-bin-x86)。
+1. 下载最新版本的 [SQL 方言转换工具(1.0.1)](https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/transform-doris-tool/transform-doris-tool-1.0.0-bin-x86)
2. 在任意 FE 节点，通过以下命令启动服务：
`nohup ./doris-sql-convertor-1.0.1-bin-x86 run --host=0.0.0.0 --port=5001 &`
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 05/24: [test](fix) Fix one missing line of output in out file (#32036)
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 31ee448c87db2f7e2332753d9813d0a215908d24
Author: zy-kkk <zh...@gmail.com>
AuthorDate: Mon Mar 11 11:25:52 2024 +0800
[test](fix) Fix one missing line of output in out file (#32036)
---
.../data/external_table_p0/jdbc/test_mysql_jdbc_driver5_catalog.out | 1 +
1 file changed, 1 insertion(+)
diff --git a/regression-test/data/external_table_p0/jdbc/test_mysql_jdbc_driver5_catalog.out b/regression-test/data/external_table_p0/jdbc/test_mysql_jdbc_driver5_catalog.out
index 339ddaf9eba..85b5ed13939 100644
--- a/regression-test/data/external_table_p0/jdbc/test_mysql_jdbc_driver5_catalog.out
+++ b/regression-test/data/external_table_p0/jdbc/test_mysql_jdbc_driver5_catalog.out
@@ -192,6 +192,7 @@ bca 2022-11-02 2022-11-02 8012 vivo
2 2
-- !information_schema --
+backend_active_tasks
character_sets
collations
column_privileges
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 09/24: [feature](docker)Doris On Docker All In One (#31735)
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 1509fbd538765d549d3cc5cad40e46e64ab90192
Author: FreeOnePlus <54...@users.noreply.github.com>
AuthorDate: Mon Mar 11 19:09:52 2024 +0800
[feature](docker)Doris On Docker All In One (#31735)
* Refactor upgrade documentation
* Doris On Docker All In One
---------
Co-authored-by: Yijia Su <su...@selectdb.com>
---
docker/runtime/all-in-one/Dockerfile | 55 +++++
docker/runtime/all-in-one/resource/entry_point.sh | 248 ++++++++++++++++++++++
2 files changed, 303 insertions(+)
diff --git a/docker/runtime/all-in-one/Dockerfile b/docker/runtime/all-in-one/Dockerfile
new file mode 100644
index 00000000000..a7bcd98cfed
--- /dev/null
+++ b/docker/runtime/all-in-one/Dockerfile
@@ -0,0 +1,55 @@
+#!/bin/bash
+# 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.
+
+# how to use Dockerfile.
+# this is dockerfile for build doris fe image on amd64.
+# when build youself image.
+# 1. pull binary from official website and decompress into resource directory that the level equals with Dockerfile_be_ubuntu.
+# 2. untar xxxx.tar.gz in resource directory, update the dockerfile field `apache-doris-xxx`, replace with real version.
+# 3. run commad docker build -t xxx.doris.be:xx -f Dockerfile_be_ubuntu.
+
+# we have support buildx for amd64 and arm64 architecture image build.
+# get the binary from doris github and utar into resource, update the directory as apache-`version(example:2.0.1)`-bin-`architecture(amd64/arm64)` mode.
+
+# choose a base image
+FROM ubuntu:22.04
+
+ARG TARGETARCH
+
+RUN apt-get update -y && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
+ patchelf gdb binutils binutils-common mysql-client \
+ curl wget less vim htop iproute2 numactl jq iotop sysstat \
+ tcpdump iputils-ping dnsutils strace lsof blktrace tzdata \
+ bpfcc-tools linux-headers-realtime linux-tools-realtime silversearcher-ag \
+ net-tools openjdk-8-jdk && \
+ rm -rf /var/lib/apt/lists/*
+
+# set environment variables
+ENV JAVA_HOME=/usr/lib/jvm/java-8-openjdk-${TARGETARCH:-amd64} \
+ PATH="/opt/apache-doris/fe/bin:/opt/apache-doris/be/bin:/opt/apache-doris/broker/bin:${PATH}"
+
+# apache-doris/be from doris release xxxx.tar.gz.please update the version in follows x.x.x.
+ADD resource/apache-doris-x.x.x-bin-${TARGETARCH:-amd64}/fe /opt/apache-doris/fe
+ADD resource/apache-doris-x.x.x-bin-${TARGETARCH:-amd64}/be /opt/apache-doris/be
+ADD resource/apache-doris-x.x.x-bin-${TARGETARCH:-amd64}/broker /opt/apache-doris/broker
+
+COPY resource/entry_point.sh /usr/local/bin/
+
+WORKDIR /opt/apache-doris
+
+ENTRYPOINT ["bash","entry_point.sh"]
diff --git a/docker/runtime/all-in-one/resource/entry_point.sh b/docker/runtime/all-in-one/resource/entry_point.sh
new file mode 100644
index 00000000000..b0ec4882977
--- /dev/null
+++ b/docker/runtime/all-in-one/resource/entry_point.sh
@@ -0,0 +1,248 @@
+#!/bin/bash
+# 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.
+
+set -eo pipefail
+shopt -s nullglob
+
+DORIS_HOME="/opt/apache-doris"
+
+# Obtain necessary and basic information to complete initialization
+
+# logging functions
+# usage: doris_[note|warn|error] $log_meg
+# ie: doris_warn "task may be risky!"
+# out: 2023-01-08T19:08:16+08:00 [Warn] [Entrypoint]: task may be risky!
+doris_log() {
+ local type="$1"
+ shift
+ # accept argument string or stdin
+ local text="$*"
+ if [ "$#" -eq 0 ]; then text="$(cat)"; fi
+ local dt="$(date -Iseconds)"
+ printf '%s [%s] [Entrypoint]: %s\n' "$dt" "$type" "$text"
+}
+doris_note() {
+ doris_log Note "$@"
+}
+doris_warn() {
+ doris_log Warn "$@" >&2
+}
+doris_error() {
+ doris_log ERROR "$@" >&2
+ exit 1
+}
+
+# check to see if this file is being run or sourced from another script
+_is_sourced() {
+ [ "${#FUNCNAME[@]}" -ge 2 ] &&
+ [ "${FUNCNAME[0]}" = '_is_sourced' ] &&
+ [ "${FUNCNAME[1]}" = 'source' ]
+}
+
+docker_setup_env() {
+ declare -g METADATA_FAILURE_RECOVERY MASTER_FE_IP CURRENT_BE_IP \
+ CURRENT_BE_PORT CURRENT_BROKER_IP CURRENT_BROKER_PORT DATABASE_ALREADY_EXISTS
+ MASTER_FE_IP="127.0.0.1"
+ CURRENT_BE_IP="127.0.0.1"
+ CURRENT_BE_PORT=9050
+ CURRENT_BROKER_IP="127.0.0.1"
+ CURRENT_BROKER_PORT=8000
+ if [[ $RECOVERY == "true" ]]; then
+ METADATA_FAILURE_RECOVERY='true'
+ fi
+ if [ -d "${DORIS_HOME}/fe/doris-meta/image" ]; then
+ DATABASE_ALREADY_EXISTS='true'
+ fi
+}
+
+# Execute sql script, passed via stdin
+docker_process_sql() {
+ set +e
+ mysql -uroot -P9030 -h127.0.0.1 --comments "$@" 2>&1
+}
+
+check_be_status() {
+ set +e
+ declare -g BE_ALREADY_EXISTS
+ for i in {1..300}; do
+ if [[ $1 == true ]]; then
+ docker_process_sql <<<"show frontends" | grep "[[:space:]]${MASTER_FE_IP}[[:space:]]"
+ else
+ docker_process_sql <<<"show backends" | grep "[[:space:]]${CURRENT_BE_IP}[[:space:]]" | grep "[[:space:]]${CURRENT_BE_PORT}[[:space:]]" | grep "[[:space:]]true[[:space:]]"
+ fi
+ be_join_status=$?
+ if [[ "${be_join_status}" == 0 ]]; then
+ if [[ $1 == true ]]; then
+ doris_note "MASTER FE is started!"
+ else
+ doris_note "EntryPoint Check - Verify that BE is registered to FE successfully"
+ BE_ALREADY_EXISTS=true
+ fi
+ return
+ fi
+ if [[ $(( $i % 20 )) == 1 ]]; then
+ if [[ $1 == true ]]; then
+ doris_note "MASTER FE is not started. retry."
+ else
+ doris_note "BE is not register. retry."
+ fi
+ fi
+ sleep 1
+ done
+}
+
+check_broker_status() {
+ set +e
+ declare -g BROKER_ALREADY_EXISTS
+ if [[ $1 == true ]]; then
+ docker_process_sql <<<"show broker" | grep "[[:space:]]${MASTER_FE_IP}[[:space:]]"
+ else
+ docker_process_sql <<<"show broker" | grep "[[:space:]]${CURRENT_BROKER_IP}[[:space:]]" | grep "[[:space:]]${CURRENT_BROKER_PORT}[[:space:]]" | grep "[[:space:]]true[[:space:]]"
+ fi
+ broker_join_status=$?
+ if [[ "${broker_join_status}" == 0 ]]; then
+ if [[ $1 == true ]]; then
+ doris_note "MASTER FE is started!"
+ else
+ doris_note "EntryPoint Check - Verify that Broker is registered to FE successfully"
+ BROKER_ALREADY_EXISTS=true
+ fi
+ return
+ fi
+ sleep 1
+}
+
+add_priority_networks() {
+ doris_note "add priority_networks ‘127.0.0.1/24’ to ${DORIS_HOME}/be/conf/be.conf"
+ echo "priority_networks = 127.0.0.1/24" >>${DORIS_HOME}/be/conf/be.conf
+}
+
+register_be_to_fe() {
+ set +e
+ # check fe status
+ local is_fe_start=false
+ if [ -n "$DATABASE_ALREADY_EXISTS" ]; then
+ check_be_status
+ if [ -n "$BE_ALREADY_EXISTS" ]; then
+ doris_warn "Same backend already exists! No need to register again！"
+ return
+ fi
+ fi
+ for i in {1..300}; do
+ docker_process_sql <<<"alter system add backend '${CURRENT_BE_IP}:${CURRENT_BE_PORT}'"
+ register_be_status=$?
+ if [[ $register_be_status == 0 ]]; then
+ doris_note "BE successfully registered to FE！"
+ is_fe_start=true
+ return
+ fi
+ if [[ $(( $i % 20 )) == 1 ]]; then
+ doris_note "Register BE to FE is failed. retry."
+ fi
+ sleep 1
+ done
+ if ! [[ $is_fe_start ]]; then
+ doris_error "Failed to register BE to FE！Tried 30 times！Maybe FE Start Failed！"
+ fi
+}
+
+register_broker_to_fe() {
+ set +e
+ # check fe status
+ local is_fe_start=false
+ for i in {1..300}; do
+ if [ -n "$BROKER_ALREADY_EXISTS" ]; then
+ doris_warn "Same Broker already exists! No need to register again！"
+ break
+ fi
+ docker_process_sql <<<"alter system add broker test '${CURRENT_BROKER_IP}:${CURRENT_BROKER_PORT}'"
+ register_broker_status=$?
+ if [[ $register_broker_status == 0 ]]; then
+ doris_note "Broker successfully registered to FE！"
+ is_fe_start=true
+ break
+ fi
+ if [[ $(( $i % 20 )) == 1 ]]; then
+ doris_note "Register Broker to FE is failed. retry."
+ fi
+ sleep 1
+ done
+ if ! [[ $is_fe_start ]]; then
+ doris_error "Failed to register Broker to FE！Tried 30 times！Maybe FE Start Failed！"
+ fi
+}
+
+start_doris() {
+ declare -g child_pid
+ if [[ $METADATA_FAILURE_RECOVERY == "true" ]]; then
+ doris_warn "Because \$RECOVERY = True, So Doris FE start metadata_failure_recovery model."
+ start_fe.sh --metadata_failure_recovery --console
+ else
+ doris_note "Start Doris FE."
+ {
+ set +e
+ bash start_fe.sh --console 2>/dev/null
+ } &
+ fi
+ if [[ $BROKER = "true" ]]; then
+ sleep 20
+ {
+ set +e
+ bash start_broker.sh 2>/dev/null
+ } &
+ check_broker_status
+ register_broker_to_fe
+ fi
+ sleep 20
+ doris_note "Start Doris BE."
+ {
+ set +e
+ bash start_be.sh --console 2>/dev/null
+ } &
+ child_pid=$!
+}
+
+stop_doris() {
+ doris_note "Container stopped, running stop_fe & stop_be script"
+ stop_broker.sh
+ sleep 2
+ stop_be.sh
+ sleep 10
+ stop_fe.sh
+}
+
+_main() {
+ trap 'stop_doris' SIGTERM SIGINT
+ docker_setup_env
+ # Check Already Exists
+ if [ -z "$DATABASE_ALREADY_EXISTS" ]; then
+ add_priority_networks
+ fi
+ # Start Doris
+ start_doris
+ # register BE
+ register_be_to_fe
+ # keep BE started status
+ wait $child_pid
+ doris_note "Apache Doris is Start Successfully!"
+ exec "$@"
+}
+
+if ! _is_sourced; then
+ _main "$@"
+fi
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 06/24: [fix](compile) fix BE compile failure on Mac (#32044)
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 fe3988c6862c3428dbd4f03604bef6af98d8b3ad
Author: zy-kkk <zh...@gmail.com>
AuthorDate: Mon Mar 11 14:02:14 2024 +0800
[fix](compile) fix BE compile failure on Mac (#32044)
---
be/src/pipeline/pipeline_x/pipeline_x_task.cpp | 4 +++-
1 file changed, 3 insertions(+), 1 deletion(-)
diff --git a/be/src/pipeline/pipeline_x/pipeline_x_task.cpp b/be/src/pipeline/pipeline_x/pipeline_x_task.cpp
index 9620976f8ae..f80302d6b48 100644
--- a/be/src/pipeline/pipeline_x/pipeline_x_task.cpp
+++ b/be/src/pipeline/pipeline_x/pipeline_x_task.cpp
@@ -283,7 +283,9 @@ Status PipelineXTask::execute(bool* eos) {
<< ",\nmin revocable mem: "
<< PrettyPrinter::print(_state->min_revocable_mem(), TUnit::BYTES)
<< ",\nrevocable mem: "
- << PrettyPrinter::print(_sink->revocable_mem_size(_state), TUnit::BYTES);
+ << PrettyPrinter::print(
+ static_cast<uint64_t>(_sink->revocable_mem_size(_state)),
+ TUnit::BYTES);
}
if (sys_mem_available < sys_mem_warning_water_mark * config::spill_mem_warning_water_mark_multiplier /*&&
(double)query_mem >= (double)_state->query_mem_limit() * 0.8*/) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 04/24: [feature](pipelineX) add mem control in local exchange sink (#31982)
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 c0f2d0188bd9df794ba42c711c0b345ac9ba2438
Author: Mryange <59...@users.noreply.github.com>
AuthorDate: Mon Mar 11 10:56:12 2024 +0800
[feature](pipelineX) add mem control in local exchange sink (#31982)
---
be/src/vec/runtime/vdata_stream_recvr.cpp | 29 ++++++++++++++++++-----------
be/src/vec/runtime/vdata_stream_recvr.h | 6 ++++--
2 files changed, 22 insertions(+), 13 deletions(-)
diff --git a/be/src/vec/runtime/vdata_stream_recvr.cpp b/be/src/vec/runtime/vdata_stream_recvr.cpp
index 7ee7f419fff..9708be6bde9 100644
--- a/be/src/vec/runtime/vdata_stream_recvr.cpp
+++ b/be/src/vec/runtime/vdata_stream_recvr.cpp
@@ -98,17 +98,17 @@ Status VDataStreamRecvr::SenderQueue::_inner_get_batch_without_lock(Block* block
DCHECK(!_block_queue.empty());
auto [next_block, block_byte_size] = std::move(_block_queue.front());
- _recvr->update_blocks_memory_usage(-block_byte_size);
+ update_blocks_memory_usage(-block_byte_size);
_block_queue.pop_front();
_record_debug_info();
- if (_block_queue.empty() && _dependency) {
+ if (_block_queue.empty() && _source_dependency) {
if (!_is_cancelled && _num_remaining_senders > 0) {
- _dependency->block();
- }
- if (_local_channel_dependency) {
- _local_channel_dependency->set_ready();
+ _source_dependency->block();
}
}
+ if (_local_channel_dependency) {
+ _local_channel_dependency->set_ready();
+ }
if (!_pending_closures.empty()) {
auto closure_pair = _pending_closures.front();
@@ -124,12 +124,12 @@ Status VDataStreamRecvr::SenderQueue::_inner_get_batch_without_lock(Block* block
}
void VDataStreamRecvr::SenderQueue::try_set_dep_ready_without_lock() {
- if (!_dependency) {
+ if (!_source_dependency) {
return;
}
const bool should_wait = !_is_cancelled && _block_queue.empty() && _num_remaining_senders > 0;
if (!should_wait) {
- _dependency->set_ready();
+ _source_dependency->set_ready();
}
}
@@ -202,7 +202,7 @@ Status VDataStreamRecvr::SenderQueue::add_block(const PBlock& pblock, int be_num
_pending_closures.emplace_back(*done, monotonicStopWatch);
*done = nullptr;
}
- _recvr->update_blocks_memory_usage(block_byte_size);
+ update_blocks_memory_usage(block_byte_size);
_data_arrival_cv.notify_one();
return Status::OK();
}
@@ -249,7 +249,7 @@ void VDataStreamRecvr::SenderQueue::add_block(Block* block, bool use_move) {
// should be done before the following logic, because the _lock will be released
// by `iter->second->wait(l)`, after `iter->second->wait(l)` returns, _recvr may
// have been closed and resouces in _recvr are released;
- _recvr->update_blocks_memory_usage(block_mem_size);
+ update_blocks_memory_usage(block_mem_size);
if (_recvr->exceeds_limit(0)) {
// yiguolei
// It is too tricky here, if the running thread is bthread then the tid may be wrong.
@@ -485,6 +485,13 @@ void VDataStreamRecvr::cancel_stream(Status exec_status) {
}
}
+void VDataStreamRecvr::SenderQueue::update_blocks_memory_usage(int64_t size) {
+ _recvr->update_blocks_memory_usage(size);
+ if (_recvr->exceeds_limit(0)) {
+ _local_channel_dependency->block();
+ }
+}
+
void VDataStreamRecvr::update_blocks_memory_usage(int64_t size) {
_blocks_memory_usage->add(size);
_blocks_memory_usage_current_value.fetch_add(size);
@@ -546,7 +553,7 @@ void VDataStreamRecvr::PipSenderQueue::add_block(Block* block, bool use_move) {
_record_debug_info();
try_set_dep_ready_without_lock();
COUNTER_UPDATE(_recvr->_local_bytes_received_counter, block_mem_size);
- _recvr->update_blocks_memory_usage(block_mem_size);
+ update_blocks_memory_usage(block_mem_size);
_data_arrival_cv.notify_one();
}
}
diff --git a/be/src/vec/runtime/vdata_stream_recvr.h b/be/src/vec/runtime/vdata_stream_recvr.h
index fa13666533a..06b05d2a577 100644
--- a/be/src/vec/runtime/vdata_stream_recvr.h
+++ b/be/src/vec/runtime/vdata_stream_recvr.h
@@ -214,9 +214,11 @@ public:
}
void set_dependency(std::shared_ptr<pipeline::Dependency> dependency) {
- _dependency = dependency;
+ _source_dependency = dependency;
}
+ void update_blocks_memory_usage(int64_t size);
+
protected:
friend class pipeline::ExchangeLocalState;
Status _inner_get_batch_without_lock(Block* block, bool* eos);
@@ -285,7 +287,7 @@ protected:
std::deque<std::pair<google::protobuf::Closure*, MonotonicStopWatch>> _pending_closures;
std::unordered_map<std::thread::id, std::unique_ptr<ThreadClosure>> _local_closure;
- std::shared_ptr<pipeline::Dependency> _dependency;
+ std::shared_ptr<pipeline::Dependency> _source_dependency;
std::shared_ptr<pipeline::Dependency> _local_channel_dependency;
};
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/24: [RuntimeFilter] fix <=> runtime filter failed bug (#32003)
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 247063485944e727975b75c32b9bb519581b4c43
Author: HappenLee <ha...@hotmail.com>
AuthorDate: Mon Mar 11 10:16:56 2024 +0800
[RuntimeFilter] fix <=> runtime filter failed bug (#32003)
---
be/src/exprs/bloom_filter_func.h | 68 +++++++++++++++++----------------
be/src/exprs/runtime_filter.cpp | 29 +++++++-------
be/src/exprs/runtime_filter.h | 5 ++-
be/src/vec/exprs/vdirect_in_predicate.h | 7 ++++
gensrc/proto/internal_service.proto | 3 ++
5 files changed, 64 insertions(+), 48 deletions(-)
diff --git a/be/src/exprs/bloom_filter_func.h b/be/src/exprs/bloom_filter_func.h
index cc9d8a71390..f5e822e2572 100644
--- a/be/src/exprs/bloom_filter_func.h
+++ b/be/src/exprs/bloom_filter_func.h
@@ -98,8 +98,6 @@ public:
void set_build_bf_exactly(bool build_bf_exactly) { _build_bf_exactly = build_bf_exactly; }
- void set_null_aware(bool null_aware) { _null_aware = null_aware; }
-
Status init_with_fixed_length() { return init_with_fixed_length(_bloom_filter_length); }
Status init_with_cardinality(const size_t build_bf_cardinality) {
@@ -139,44 +137,39 @@ public:
// If `_inited` is false, there is no memory allocated in bloom filter and this is the first
// call for `merge` function. So we just reuse this bloom filter, and we don't need to
// allocate memory again.
- if (_inited) {
- DCHECK(bloomfilter_func != nullptr);
+ std::lock_guard<std::mutex> l(_lock);
+ if (!_inited) {
auto* other_func = static_cast<BloomFilterFuncBase*>(bloomfilter_func);
- if (_bloom_filter_alloced != other_func->_bloom_filter_alloced) {
- return Status::InvalidArgument(
- "bloom filter size not the same: already allocated bytes {}, expected "
- "allocated bytes {}",
- _bloom_filter_alloced, other_func->_bloom_filter_alloced);
- }
- return _bloom_filter->merge(other_func->_bloom_filter.get());
- }
- {
- std::lock_guard<std::mutex> l(_lock);
- if (!_inited) {
- auto* other_func = static_cast<BloomFilterFuncBase*>(bloomfilter_func);
- DCHECK(_bloom_filter == nullptr);
- DCHECK(bloomfilter_func != nullptr);
- _bloom_filter = bloomfilter_func->_bloom_filter;
- _bloom_filter_alloced = other_func->_bloom_filter_alloced;
- _inited = true;
- return Status::OK();
- }
+ DCHECK(_bloom_filter == nullptr);
DCHECK(bloomfilter_func != nullptr);
- auto* other_func = static_cast<BloomFilterFuncBase*>(bloomfilter_func);
- if (_bloom_filter_alloced != other_func->_bloom_filter_alloced) {
- return Status::InvalidArgument(
- "bloom filter size not the same: already allocated bytes {}, expected "
- "allocated bytes {}",
- _bloom_filter_alloced, other_func->_bloom_filter_alloced);
- }
- return _bloom_filter->merge(other_func->_bloom_filter.get());
+ _bloom_filter = bloomfilter_func->_bloom_filter;
+ _bloom_filter_alloced = other_func->_bloom_filter_alloced;
+ _inited = true;
+ return Status::OK();
}
+ DCHECK(bloomfilter_func != nullptr);
+ auto* other_func = static_cast<BloomFilterFuncBase*>(bloomfilter_func);
+ if (_bloom_filter_alloced != other_func->_bloom_filter_alloced) {
+ return Status::InvalidArgument(
+ "bloom filter size not the same: already allocated bytes {}, expected "
+ "allocated bytes {}",
+ _bloom_filter_alloced, other_func->_bloom_filter_alloced);
+ }
+ if (other_func->_bloom_filter->contain_null()) {
+ _bloom_filter->set_contain_null();
+ }
+ return _bloom_filter->merge(other_func->_bloom_filter.get());
}
- Status assign(butil::IOBufAsZeroCopyInputStream* data, const size_t data_size) {
+ Status assign(butil::IOBufAsZeroCopyInputStream* data, const size_t data_size,
+ bool contain_null) {
if (_bloom_filter == nullptr) {
+ _null_aware = contain_null;
_bloom_filter.reset(BloomFilterAdaptor::create(_null_aware));
}
+ if (contain_null) {
+ _bloom_filter->set_contain_null();
+ }
_bloom_filter_alloced = data_size;
return _bloom_filter->init(data, data_size);
@@ -187,6 +180,16 @@ public:
*len = _bloom_filter->size();
}
+ bool contain_null() const {
+ DCHECK(_bloom_filter);
+ return _bloom_filter->contain_null();
+ }
+
+ void set_contain_null() {
+ DCHECK(_bloom_filter);
+ _bloom_filter->set_contain_null();
+ }
+
size_t get_size() const { return _bloom_filter ? _bloom_filter->size() : 0; }
void light_copy(BloomFilterFuncBase* bloomfilter_func) {
@@ -214,7 +217,6 @@ protected:
std::mutex _lock;
int64_t _bloom_filter_length;
bool _build_bf_exactly = false;
- bool _null_aware = false;
};
template <typename T, bool need_trim = false>
diff --git a/be/src/exprs/runtime_filter.cpp b/be/src/exprs/runtime_filter.cpp
index 5cd195de515..a949969ca65 100644
--- a/be/src/exprs/runtime_filter.cpp
+++ b/be/src/exprs/runtime_filter.cpp
@@ -31,7 +31,6 @@
#include <memory>
#include <mutex>
#include <ostream>
-#include <utility>
#include "common/logging.h"
#include "common/object_pool.h"
@@ -40,7 +39,6 @@
#include "exprs/bloom_filter_func.h"
#include "exprs/create_predicate_function.h"
#include "exprs/hybrid_set.h"
-#include "exprs/minmax_predicate.h"
#include "gutil/strings/substitute.h"
#include "pipeline/pipeline_x/dependency.h"
#include "runtime/define_primitive_type.h"
@@ -720,14 +718,18 @@ public:
// used by shuffle runtime filter
// assign this filter by protobuf
- Status assign(const PBloomFilter* bloom_filter, butil::IOBufAsZeroCopyInputStream* data) {
+ Status assign(const PBloomFilter* bloom_filter, butil::IOBufAsZeroCopyInputStream* data,
+ bool contain_null) {
_is_bloomfilter = true;
// we won't use this class to insert or find any data
// so any type is ok
_context.bloom_filter_func.reset(create_bloom_filter(_column_return_type == INVALID_TYPE
? PrimitiveType::TYPE_INT
: _column_return_type));
- return _context.bloom_filter_func->assign(data, bloom_filter->filter_length());
+ RETURN_IF_ERROR(_context.bloom_filter_func->assign(data, bloom_filter->filter_length(),
+ contain_null));
+
+ return Status::OK();
}
// used by shuffle runtime filter
@@ -877,6 +879,10 @@ public:
bool is_bloomfilter() const { return _is_bloomfilter; }
+ bool contain_null() const {
+ return _is_bloomfilter && _context.bloom_filter_func->contain_null();
+ }
+
bool is_ignored() const { return _ignored; }
const std::string& ignored_msg() const { return _ignored_msg; }
@@ -1302,7 +1308,8 @@ Status IRuntimeFilter::create_wrapper(const UpdateRuntimeFilterParamsV2* param,
}
case PFilterType::BLOOM_FILTER: {
DCHECK(param->request->has_bloom_filter());
- return (*wrapper)->assign(¶m->request->bloom_filter(), param->data);
+ return (*wrapper)->assign(¶m->request->bloom_filter(), param->data,
+ param->request->contain_null());
}
case PFilterType::MIN_FILTER:
case PFilterType::MAX_FILTER:
@@ -1344,7 +1351,8 @@ Status IRuntimeFilter::_create_wrapper(const T* param, ObjectPool* pool,
}
case PFilterType::BLOOM_FILTER: {
DCHECK(param->request->has_bloom_filter());
- return (*wrapper)->assign(¶m->request->bloom_filter(), param->data);
+ return (*wrapper)->assign(¶m->request->bloom_filter(), param->data,
+ param->request->contain_null());
}
case PFilterType::MIN_FILTER:
case PFilterType::MAX_FILTER:
@@ -1400,6 +1408,7 @@ Status IRuntimeFilter::serialize_impl(T* request, void** data, int* len) {
}
request->set_filter_type(get_type(real_runtime_filter_type));
+ request->set_contain_null(_wrapper->contain_null());
if (real_runtime_filter_type == RuntimeFilterType::IN_FILTER) {
auto in_filter = request->mutable_in_filter();
@@ -1732,7 +1741,6 @@ Status RuntimePredicateWrapper::get_push_exprs(std::list<vectorized::VExprContex
node.in_predicate.__set_is_not_in(false);
node.__set_opcode(TExprOpcode::FILTER_IN);
node.__set_is_nullable(false);
-
auto in_pred = vectorized::VDirectInPredicate::create_shared(node);
in_pred->set_filter(_context.hybrid_set);
in_pred->add_child(probe_ctx->root());
@@ -1753,9 +1761,6 @@ Status RuntimePredicateWrapper::get_push_exprs(std::list<vectorized::VExprContex
min_pred->add_child(min_literal);
container.push_back(
vectorized::VRuntimeFilterWrapper::create_shared(min_pred_node, min_pred));
- vectorized::VExprContextSPtr new_probe_ctx;
- RETURN_IF_ERROR(vectorized::VExpr::create_expr_tree(probe_expr, new_probe_ctx));
- probe_ctxs.push_back(new_probe_ctx);
break;
}
case RuntimeFilterType::MAX_FILTER: {
@@ -1771,10 +1776,6 @@ Status RuntimePredicateWrapper::get_push_exprs(std::list<vectorized::VExprContex
max_pred->add_child(max_literal);
container.push_back(
vectorized::VRuntimeFilterWrapper::create_shared(max_pred_node, max_pred));
-
- vectorized::VExprContextSPtr new_probe_ctx;
- RETURN_IF_ERROR(vectorized::VExpr::create_expr_tree(probe_expr, new_probe_ctx));
- probe_ctxs.push_back(new_probe_ctx);
break;
}
case RuntimeFilterType::MINMAX_FILTER: {
diff --git a/be/src/exprs/runtime_filter.h b/be/src/exprs/runtime_filter.h
index be334ee939a..5cfc88f4ed8 100644
--- a/be/src/exprs/runtime_filter.h
+++ b/be/src/exprs/runtime_filter.h
@@ -146,8 +146,11 @@ public:
bool is_runtime_filter() const { return _filter_id != -1; }
-private:
+ void set_null_aware(bool null_aware) { _null_aware = null_aware; }
+
+protected:
int _filter_id = -1;
+ bool _null_aware = false;
};
struct UpdateRuntimeFilterParams {
diff --git a/be/src/vec/exprs/vdirect_in_predicate.h b/be/src/vec/exprs/vdirect_in_predicate.h
index a68a6c3121a..fbba76de61d 100644
--- a/be/src/vec/exprs/vdirect_in_predicate.h
+++ b/be/src/vec/exprs/vdirect_in_predicate.h
@@ -67,6 +67,12 @@ public:
const auto& null_map =
static_cast<const ColumnNullable*>(argument_column.get())->get_null_map_data();
_filter->find_batch_nullable(*column_nested, sz, null_map, res_data_column->get_data());
+ if (_null_aware) {
+ auto* __restrict res_data = res_data_column->get_data().data();
+ for (size_t i = 0; i < sz; ++i) {
+ res_data[i] |= null_map[i];
+ }
+ }
} else {
_filter->find_batch(*argument_column, sz, res_data_column->get_data());
}
@@ -87,6 +93,7 @@ public:
private:
std::shared_ptr<HybridSetBase> _filter;
+ bool _null_aware = false;
std::string _expr_name;
};
} // namespace doris::vectorized
\ No newline at end of file
diff --git a/gensrc/proto/internal_service.proto b/gensrc/proto/internal_service.proto
index 87e793aa750..8b9cce2acb9 100644
--- a/gensrc/proto/internal_service.proto
+++ b/gensrc/proto/internal_service.proto
@@ -514,6 +514,7 @@ message PMergeFilterRequest {
optional bool is_pipeline = 8;
optional bool opt_remote_rf = 9;
optional PColumnType column_type = 10;
+ optional bool contain_null = 11;
};
message PMergeFilterResponse {
@@ -532,6 +533,7 @@ message PPublishFilterRequest {
optional bool is_pipeline = 8;
optional int64 merge_time = 9;
optional PColumnType column_type = 10;
+ optional bool contain_null = 11;
};
message PPublishFilterRequestV2 {
@@ -544,6 +546,7 @@ message PPublishFilterRequestV2 {
optional PInFilter in_filter = 7;
optional bool is_pipeline = 8;
optional int64 merge_time = 9;
+ optional bool contain_null = 10;
};
message PPublishFilterResponse {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 10/24: [Improve](InPredict) enhance in predict with array type (#31828)
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 ccd21a6ea437338371bf026517421a8f16402bf6
Author: Benjaminwei <33...@users.noreply.github.com>
AuthorDate: Mon Mar 11 21:22:32 2024 +0800
[Improve](InPredict) enhance in predict with array type (#31828)
---
be/src/vec/exprs/vin_predicate.cpp | 5 +-
...on_struct_in.cpp => function_collection_in.cpp} | 8 +--
...nction_struct_in.h => function_collection_in.h} | 65 +++++++++++++---------
be/src/vec/functions/simple_function_factory.h | 4 +-
.../nereids/trees/expressions/InPredicate.java | 12 ++++
.../doris/nereids/util/TypeCoercionUtils.java | 10 +++-
.../nereids_syntax_p0/inpredicate_with_list.out | 19 +++++++
.../data/query_p0/sql_functions/test_in_expr.out | 3 +
.../nereids_syntax_p0/inpredicate_with_list.groovy | 60 ++++++++++++++++++++
.../query_p0/sql_functions/test_in_expr.groovy | 5 +-
10 files changed, 150 insertions(+), 41 deletions(-)
diff --git a/be/src/vec/exprs/vin_predicate.cpp b/be/src/vec/exprs/vin_predicate.cpp
index 6f57828ef2d..838e12dc9c2 100644
--- a/be/src/vec/exprs/vin_predicate.cpp
+++ b/be/src/vec/exprs/vin_predicate.cpp
@@ -66,8 +66,9 @@ Status VInPredicate::prepare(RuntimeState* state, const RowDescriptor& desc,
// construct the proper function_name
std::string head(_is_not_in ? "not_" : "");
std::string real_function_name = head + std::string(function_name);
- if (is_struct(remove_nullable(argument_template[0].type))) {
- real_function_name = "struct_" + real_function_name;
+ auto arg_type = remove_nullable(argument_template[0].type);
+ if (is_struct(arg_type) || is_array(arg_type) || is_map(arg_type)) {
+ real_function_name = "collection_" + real_function_name;
}
_function = SimpleFunctionFactory::instance().get_function(real_function_name,
argument_template, _data_type);
diff --git a/be/src/vec/functions/function_struct_in.cpp b/be/src/vec/functions/function_collection_in.cpp
similarity index 79%
rename from be/src/vec/functions/function_struct_in.cpp
rename to be/src/vec/functions/function_collection_in.cpp
index 943e3a80f47..e7d6b56d5c5 100644
--- a/be/src/vec/functions/function_struct_in.cpp
+++ b/be/src/vec/functions/function_collection_in.cpp
@@ -17,15 +17,15 @@
// This file is copied from
// and modified by Doris
-#include "vec/functions/function_struct_in.h"
+#include "vec/functions/function_collection_in.h"
#include "vec/functions/simple_function_factory.h"
namespace doris::vectorized {
-void register_function_struct_in(SimpleFunctionFactory& factory) {
- factory.register_function<FunctionStructIn<false>>();
- factory.register_function<FunctionStructIn<true>>();
+void register_function_collection_in(SimpleFunctionFactory& factory) {
+ factory.register_function<FunctionCollectionIn<false>>();
+ factory.register_function<FunctionCollectionIn<true>>();
}
} // namespace doris::vectorized
diff --git a/be/src/vec/functions/function_struct_in.h b/be/src/vec/functions/function_collection_in.h
similarity index 75%
rename from be/src/vec/functions/function_struct_in.h
rename to be/src/vec/functions/function_collection_in.h
index e3fb9dbe70e..1e86ce25b34 100644
--- a/be/src/vec/functions/function_struct_in.h
+++ b/be/src/vec/functions/function_collection_in.h
@@ -48,11 +48,11 @@ struct ColumnRowRef {
// equals when call set insert, this operator will be used
bool operator==(const ColumnRowRef& other) const {
- return column->compare_at(row_idx, other.row_idx, *column, 0) == 0;
+ return column->compare_at(row_idx, other.row_idx, *other.column, 0) == 0;
}
// compare
bool operator<(const ColumnRowRef& other) const {
- return column->compare_at(row_idx, other.row_idx, *column, 0) < 0;
+ return column->compare_at(row_idx, other.row_idx, *other.column, 0) < 0;
}
// when call set find, will use hash to find
@@ -63,18 +63,18 @@ struct ColumnRowRef {
}
};
-struct StructInState {
- ENABLE_FACTORY_CREATOR(StructInState)
+struct CollectionInState {
+ ENABLE_FACTORY_CREATOR(CollectionInState)
std::unordered_set<ColumnRowRef, ColumnRowRef> args_set;
bool null_in_set = false;
};
template <bool negative>
-class FunctionStructIn : public IFunction {
+class FunctionCollectionIn : public IFunction {
public:
- static constexpr auto name = negative ? "struct_not_in" : "struct_in";
+ static constexpr auto name = negative ? "collection_not_in" : "collection_in";
- static FunctionPtr create() { return std::make_shared<FunctionStructIn>(); }
+ static FunctionPtr create() { return std::make_shared<FunctionCollectionIn>(); }
String get_name() const override { return name; }
@@ -98,14 +98,17 @@ public:
if (scope == FunctionContext::THREAD_LOCAL) {
return Status::OK();
}
- std::shared_ptr<StructInState> state = std::make_shared<StructInState>();
+ int num_args = context->get_num_args();
+ DCHECK(num_args >= 1);
+
+ std::shared_ptr<CollectionInState> state = std::make_shared<CollectionInState>();
context->set_function_state(scope, state);
- DCHECK(context->get_num_args() >= 1);
+
auto* col_desc = context->get_arg_type(0);
- DataTypePtr args_type = DataTypeFactory::instance().create_data_type(*col_desc);
- MutableColumnPtr column_struct_ptr_args = remove_nullable(args_type)->create_column();
- NullMap null_map(context->get_num_args(), false);
- for (int i = 1; i < context->get_num_args(); ++i) {
+ DataTypePtr args_type = DataTypeFactory::instance().create_data_type(*col_desc, false);
+ MutableColumnPtr args_column_ptr = args_type->create_column();
+
+ for (int i = 1; i < num_args; i++) {
// FE should make element type consistent and
// equalize the length of the elements in struct
const auto& const_column_ptr = context->get_constant_col(i);
@@ -117,34 +120,33 @@ public:
auto* null_col = vectorized::check_and_get_column<vectorized::ColumnNullable>(col);
if (null_col->has_null()) {
state->null_in_set = true;
- null_map[i - 1] = true;
} else {
- column_struct_ptr_args->insert_from(null_col->get_nested_column(), 0);
+ args_column_ptr->insert_from(null_col->get_nested_column(), 0);
}
} else {
- column_struct_ptr_args->insert_from(*col, 0);
+ args_column_ptr->insert_from(*col, 0);
}
}
- ColumnPtr column_ptr = std::move(column_struct_ptr_args);
- // make StructRef into set
- for (size_t i = 1; i < context->get_num_args(); ++i) {
- if (state->null_in_set && null_map[i - 1]) {
- continue;
- }
- state->args_set.insert({column_ptr, i - 1});
+ ColumnPtr column_ptr = std::move(args_column_ptr);
+ // make collection ref into set
+ int col_size = column_ptr->size();
+ for (size_t i = 0; i < col_size; i++) {
+ state->args_set.insert({column_ptr, i});
}
+
return Status::OK();
}
Status execute_impl(FunctionContext* context, Block& block, const ColumnNumbers& arguments,
size_t result, size_t input_rows_count) const override {
- auto in_state = reinterpret_cast<StructInState*>(
+ auto in_state = reinterpret_cast<CollectionInState*>(
context->get_function_state(FunctionContext::FRAGMENT_LOCAL));
if (!in_state) {
- return Status::RuntimeError("funciton context for function '{}' must have Set;",
+ return Status::RuntimeError("function context for function '{}' must have Set;",
get_name());
}
const auto& args_set = in_state->args_set;
+ const bool null_in_set = in_state->null_in_set;
auto res = ColumnUInt8::create();
ColumnUInt8::Container& vec_res = res->get_data();
vec_res.resize(input_rows_count);
@@ -155,15 +157,24 @@ public:
const ColumnWithTypeAndName& left_arg = block.get_by_position(arguments[0]);
const auto& [materialized_column, col_const] = unpack_if_const(left_arg.column);
+ auto materialized_column_not_null = materialized_column;
+ if (materialized_column_not_null->is_nullable()) {
+ materialized_column_not_null = assert_cast<ColumnPtr>(
+ vectorized::check_and_get_column<vectorized::ColumnNullable>(
+ materialized_column_not_null)
+ ->get_nested_column_ptr());
+ }
for (size_t i = 0; i < input_rows_count; ++i) {
- bool find = args_set.find({materialized_column, i}) != args_set.end();
+ bool find = args_set.find({materialized_column_not_null, i}) != args_set.end();
+
if constexpr (negative) {
vec_res[i] = !find;
} else {
vec_res[i] = find;
}
- if (in_state->null_in_set) {
+
+ if (null_in_set) {
vec_null_map_to[i] = negative == vec_res[i];
} else {
vec_null_map_to[i] = false;
diff --git a/be/src/vec/functions/simple_function_factory.h b/be/src/vec/functions/simple_function_factory.h
index b1c1b394bff..a44861a2683 100644
--- a/be/src/vec/functions/simple_function_factory.h
+++ b/be/src/vec/functions/simple_function_factory.h
@@ -65,7 +65,7 @@ void register_function_running_difference(SimpleFunctionFactory& factory);
void register_function_date_time_to_string(SimpleFunctionFactory& factory);
void register_function_date_time_string_to_string(SimpleFunctionFactory& factory);
void register_function_in(SimpleFunctionFactory& factory);
-void register_function_struct_in(SimpleFunctionFactory& factory);
+void register_function_collection_in(SimpleFunctionFactory& factory);
void register_function_if(SimpleFunctionFactory& factory);
void register_function_nullif(SimpleFunctionFactory& factory);
void register_function_date_time_computation(SimpleFunctionFactory& factory);
@@ -246,7 +246,7 @@ public:
register_function_time_of_function(instance);
register_function_string(instance);
register_function_in(instance);
- register_function_struct_in(instance);
+ register_function_collection_in(instance);
register_function_if(instance);
register_function_nullif(instance);
register_function_date_time_computation(instance);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/InPredicate.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/InPredicate.java
index f3ae9ce5b27..bcebdca4f5b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/InPredicate.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/InPredicate.java
@@ -86,6 +86,18 @@ public class InPredicate extends Expression {
}
return;
}
+
+ if (children().get(0).getDataType().isArrayType()) {
+ // we should check in value list is all list type
+ for (int i = 1; i < children().size(); i++) {
+ if (!children().get(i).getDataType().isArrayType() && !children().get(i).getDataType().isNullType()) {
+ throw new AnalysisException("in predicate list should compare with struct type list, but got : "
+ + children().get(i).getDataType().toSql());
+ }
+ }
+ return;
+ }
+
children().forEach(c -> {
if (c.getDataType().isObjectType()) {
throw new AnalysisException("in predicate could not contains object type: " + this.toSql());
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java
index 239972fc1d6..34d23aecf56 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java
@@ -954,7 +954,8 @@ public class TypeCoercionUtils {
if (inPredicate.getOptions().stream().map(Expression::getDataType)
.allMatch(dt -> dt.equals(inPredicate.getCompareExpr().getDataType()))) {
if (!supportCompare(inPredicate.getCompareExpr().getDataType())
- && !inPredicate.getCompareExpr().getDataType().isStructType()) {
+ && !inPredicate.getCompareExpr().getDataType().isStructType() && !inPredicate.getCompareExpr()
+ .getDataType().isArrayType()) {
throw new AnalysisException("data type " + inPredicate.getCompareExpr().getDataType()
+ " could not used in InPredicate " + inPredicate.toSql());
}
@@ -970,8 +971,13 @@ public class TypeCoercionUtils {
throw new AnalysisException("data type " + optionalCommonType.get()
+ " is not match " + inPredicate.getCompareExpr().getDataType() + " used in InPredicate");
}
+ if (inPredicate.getCompareExpr().getDataType().isArrayType() && optionalCommonType.isPresent()
+ && !optionalCommonType.get().isArrayType()) {
+ throw new AnalysisException("data type " + optionalCommonType.get()
+ + " is not match " + inPredicate.getCompareExpr().getDataType() + " used in InPredicate");
+ }
if (optionalCommonType.isPresent() && !supportCompare(optionalCommonType.get())
- && !optionalCommonType.get().isStructType()) {
+ && !optionalCommonType.get().isStructType() && !optionalCommonType.get().isArrayType()) {
throw new AnalysisException("data type " + optionalCommonType.get()
+ " could not used in InPredicate " + inPredicate.toSql());
}
diff --git a/regression-test/data/nereids_syntax_p0/inpredicate_with_list.out b/regression-test/data/nereids_syntax_p0/inpredicate_with_list.out
new file mode 100644
index 00000000000..2aaf57f0cd5
--- /dev/null
+++ b/regression-test/data/nereids_syntax_p0/inpredicate_with_list.out
@@ -0,0 +1,19 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_default --
+1 [1, 2, 3]
+2 [4, 5, 6]
+
+-- !in_predicate_17 --
+
+-- !in_predicate_18 --
+
+-- !in_predicate_19 --
+2 [4, 5, 6]
+
+-- !in_predicate_20 --
+2 [4, 5, 6]
+
+-- !in_predicate_21 --
+
+-- !in_predicate_22 --
+
diff --git a/regression-test/data/query_p0/sql_functions/test_in_expr.out b/regression-test/data/query_p0/sql_functions/test_in_expr.out
index 6cb6bd195f6..43bb4b43bfb 100644
--- a/regression-test/data/query_p0/sql_functions/test_in_expr.out
+++ b/regression-test/data/query_p0/sql_functions/test_in_expr.out
@@ -73,3 +73,6 @@ d
-- !select --
+-- !select --
+[1, 2, 3, 4, 5] \N
+
diff --git a/regression-test/suites/nereids_syntax_p0/inpredicate_with_list.groovy b/regression-test/suites/nereids_syntax_p0/inpredicate_with_list.groovy
new file mode 100644
index 00000000000..58f98621529
--- /dev/null
+++ b/regression-test/suites/nereids_syntax_p0/inpredicate_with_list.groovy
@@ -0,0 +1,60 @@
+// 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.
+
+suite("inpredicate_with_list") {
+ def tableName = "test_array"
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE IF NOT EXISTS ${tableName} (
+ `id` INT NULL,
+ `c_array` ARRAY<INT> NULL
+ ) DISTRIBUTED BY HASH(id) PROPERTIES("replication_num" = "1");
+ """
+
+ sql """ INSERT INTO ${tableName} (`id`,`c_array`) VALUES
+ (1, [1, 2, 3]),
+ (2, [4, 5, 6])
+ """
+ try {
+ qt_select_default """ SELECT * FROM ${tableName} t ORDER BY id; """
+
+ sql """ set enable_nereids_planner = true;"""
+ sql """ set enable_fallback_to_original_planner=false;"""
+ order_qt_in_predicate_17 """
+ SELECT * FROM ${tableName} where c_array in ([1,2], [1,3]);
+ """
+ order_qt_in_predicate_18 """
+ SELECT * FROM ${tableName} where c_array in (null, [1,3]);
+ """
+ order_qt_in_predicate_19 """
+ SELECT * FROM ${tableName} where c_array in ([1,2], [1,3], [4,5,6]);
+ """
+ order_qt_in_predicate_20 """
+ SELECT * FROM ${tableName} where c_array in ([1,2], null, [4,5,6]);
+ """
+ order_qt_in_predicate_21 """
+ SELECT * FROM ${tableName} where c_array in ([1,2], null, [4,5,3]);
+ """
+ order_qt_in_predicate_22 """
+ SELECT * FROM ${tableName} where c_array in ([1,2], [6,5,4], [4,5,3]);
+ """
+ } finally {
+ try_sql("DROP TABLE IF EXISTS ${tableName}")
+ }
+}
+
diff --git a/regression-test/suites/query_p0/sql_functions/test_in_expr.groovy b/regression-test/suites/query_p0/sql_functions/test_in_expr.groovy
index e3839a65c48..807fdd91513 100644
--- a/regression-test/suites/query_p0/sql_functions/test_in_expr.groovy
+++ b/regression-test/suites/query_p0/sql_functions/test_in_expr.groovy
@@ -145,10 +145,7 @@ suite("test_in_expr", "query,arrow_flight_sql") {
sql """ INSERT INTO `array_in_test` VALUES (1, [1,2,3,4,5]); """
- test {
- sql """ select c_array, c_array in (null) from array_in_test; """
- exception "errCode"
- }
+ qt_select """ select c_array, c_array in (null) from array_in_test; """
sql " drop table if exists `json_in_test` "
sql """
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 17/24: [fix](spill) fix storage engine type cast error (#32071)
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 b41b17ad0a6bd542767698937ff712c4d4e542b0
Author: TengJianPing <18...@users.noreply.github.com>
AuthorDate: Mon Mar 11 22:48:16 2024 +0800
[fix](spill) fix storage engine type cast error (#32071)
---
be/src/vec/spill/spill_stream.cpp | 5 +++-
be/src/vec/spill/spill_stream.h | 11 ++++----
be/src/vec/spill/spill_stream_manager.cpp | 46 +++++++++++++++++++++++++------
be/src/vec/spill/spill_stream_manager.h | 43 +++++++++++++++++++++++++++--
be/src/vec/spill/spill_writer.cpp | 1 -
be/src/vec/spill/spill_writer.h | 6 ++--
6 files changed, 90 insertions(+), 22 deletions(-)
diff --git a/be/src/vec/spill/spill_stream.cpp b/be/src/vec/spill/spill_stream.cpp
index d770e44a147..d08b63df40b 100644
--- a/be/src/vec/spill/spill_stream.cpp
+++ b/be/src/vec/spill/spill_stream.cpp
@@ -32,7 +32,7 @@
#include "vec/spill/spill_writer.h"
namespace doris::vectorized {
-SpillStream::SpillStream(RuntimeState* state, int64_t stream_id, doris::DataDir* data_dir,
+SpillStream::SpillStream(RuntimeState* state, int64_t stream_id, SpillDataDir* data_dir,
std::string spill_dir, size_t batch_rows, size_t batch_bytes,
RuntimeProfile* profile)
: state_(state),
@@ -72,6 +72,9 @@ void SpillStream::close() {
(void)reader_->close();
}
+const std::string& SpillStream::get_spill_root_dir() const {
+ return data_dir_->path();
+}
Status SpillStream::prepare_spill() {
DCHECK(!spill_promise_);
RETURN_IF_ERROR(writer_->open());
diff --git a/be/src/vec/spill/spill_stream.h b/be/src/vec/spill/spill_stream.h
index 579449e503c..9f328240d75 100644
--- a/be/src/vec/spill/spill_stream.h
+++ b/be/src/vec/spill/spill_stream.h
@@ -20,29 +20,28 @@
#include <future>
#include <memory>
-#include "olap/data_dir.h"
#include "vec/spill/spill_reader.h"
#include "vec/spill/spill_writer.h"
namespace doris {
class RuntimeProfile;
-class DataDir;
class ThreadPool;
namespace vectorized {
class Block;
+class SpillDataDir;
class SpillStream {
public:
- SpillStream(RuntimeState* state, int64_t stream_id, doris::DataDir* data_dir,
+ SpillStream(RuntimeState* state, int64_t stream_id, SpillDataDir* data_dir,
std::string spill_dir, size_t batch_rows, size_t batch_bytes,
RuntimeProfile* profile);
int64_t id() const { return stream_id_; }
- DataDir* get_data_dir() const { return data_dir_; }
- const std::string& get_spill_root_dir() const { return data_dir_->path(); }
+ SpillDataDir* get_data_dir() const { return data_dir_; }
+ const std::string& get_spill_root_dir() const;
const std::string& get_spill_dir() const { return spill_dir_; }
@@ -85,7 +84,7 @@ private:
ThreadPool* io_thread_pool_;
int64_t stream_id_;
std::atomic_bool closed_ = false;
- doris::DataDir* data_dir_ = nullptr;
+ SpillDataDir* data_dir_ = nullptr;
std::string spill_dir_;
size_t batch_rows_;
size_t batch_bytes_;
diff --git a/be/src/vec/spill/spill_stream_manager.cpp b/be/src/vec/spill/spill_stream_manager.cpp
index 4dfd847482d..ecbee7fb858 100644
--- a/be/src/vec/spill/spill_stream_manager.cpp
+++ b/be/src/vec/spill/spill_stream_manager.cpp
@@ -29,9 +29,8 @@
#include "io/fs/file_system.h"
#include "io/fs/local_file_system.h"
-#include "olap/data_dir.h"
#include "olap/olap_define.h"
-#include "olap/storage_engine.h"
+#include "runtime/runtime_state.h"
#include "util/runtime_profile.h"
#include "util/time.h"
#include "vec/spill/spill_stream.h"
@@ -112,9 +111,9 @@ void SpillStreamManager::_spill_gc_thread_callback() {
Status SpillStreamManager::_init_spill_store_map() {
for (const auto& path : _spill_store_paths) {
- auto store = std::make_unique<DataDir>(ExecEnv::GetInstance()->storage_engine().to_local(),
- path.path, path.capacity_bytes, path.storage_medium);
- auto st = store->init(false);
+ auto store =
+ std::make_unique<SpillDataDir>(path.path, path.capacity_bytes, path.storage_medium);
+ auto st = store->init();
if (!st.ok()) {
LOG(WARNING) << "Store load failed, status=" << st.to_string()
<< ", path=" << store->path();
@@ -126,9 +125,9 @@ Status SpillStreamManager::_init_spill_store_map() {
return Status::OK();
}
-std::vector<DataDir*> SpillStreamManager::_get_stores_for_spill(
+std::vector<SpillDataDir*> SpillStreamManager::_get_stores_for_spill(
TStorageMedium::type storage_medium) {
- std::vector<DataDir*> stores;
+ std::vector<SpillDataDir*> stores;
for (auto&& [_, store] : _spill_store_map) {
if (store->storage_medium() == storage_medium && !store->reach_capacity_limit(0)) {
stores.push_back(store.get());
@@ -136,7 +135,7 @@ std::vector<DataDir*> SpillStreamManager::_get_stores_for_spill(
}
std::sort(stores.begin(), stores.end(),
- [](DataDir* a, DataDir* b) { return a->get_usage(0) < b->get_usage(0); });
+ [](SpillDataDir* a, SpillDataDir* b) { return a->get_usage(0) < b->get_usage(0); });
size_t seventy_percent_index = stores.size();
size_t eighty_five_percent_index = stores.size();
@@ -176,7 +175,7 @@ Status SpillStreamManager::register_spill_stream(RuntimeState* state, SpillStrea
int64_t id = id_++;
std::string spill_dir;
- doris::DataDir* data_dir = nullptr;
+ SpillDataDir* data_dir = nullptr;
for (auto& dir : data_dirs) {
data_dir = dir;
std::string spill_root_dir = fmt::format("{}/{}", data_dir->path(), SPILL_DIR_PREFIX);
@@ -259,4 +258,33 @@ void SpillStreamManager::gc(int64_t max_file_count) {
}
}
}
+
+SpillDataDir::SpillDataDir(const std::string& path, int64_t capacity_bytes,
+ TStorageMedium::type storage_medium)
+ : _path(path),
+ _available_bytes(0),
+ _disk_capacity_bytes(0),
+ _storage_medium(storage_medium) {}
+
+Status SpillDataDir::init() {
+ bool exists = false;
+ RETURN_IF_ERROR(io::global_local_filesystem()->exists(_path, &exists));
+ if (!exists) {
+ RETURN_NOT_OK_STATUS_WITH_WARN(Status::IOError("opendir failed, path={}", _path),
+ "check file exist failed");
+ }
+
+ return Status::OK();
+}
+bool SpillDataDir::reach_capacity_limit(int64_t incoming_data_size) {
+ double used_pct = get_usage(incoming_data_size);
+ int64_t left_bytes = _available_bytes - incoming_data_size;
+ if (used_pct >= config::storage_flood_stage_usage_percent / 100.0 &&
+ left_bytes <= config::storage_flood_stage_left_capacity_bytes) {
+ LOG(WARNING) << "reach capacity limit. used pct: " << used_pct
+ << ", left bytes: " << left_bytes << ", path: " << _path;
+ return true;
+ }
+ return false;
+}
} // namespace doris::vectorized
diff --git a/be/src/vec/spill/spill_stream_manager.h b/be/src/vec/spill/spill_stream_manager.h
index 2b412072038..f73f840458e 100644
--- a/be/src/vec/spill/spill_stream_manager.h
+++ b/be/src/vec/spill/spill_stream_manager.h
@@ -31,6 +31,45 @@ class RuntimeProfile;
namespace vectorized {
+class SpillDataDir {
+public:
+ SpillDataDir(const std::string& path, int64_t capacity_bytes = -1,
+ TStorageMedium::type storage_medium = TStorageMedium::HDD);
+
+ Status init();
+
+ const std::string& path() const { return _path; }
+
+ bool is_ssd_disk() const { return _storage_medium == TStorageMedium::SSD; }
+
+ TStorageMedium::type storage_medium() const { return _storage_medium; }
+
+ // check if the capacity reach the limit after adding the incoming data
+ // return true if limit reached, otherwise, return false.
+ // TODO(cmy): for now we can not precisely calculate the capacity Doris used,
+ // so in order to avoid running out of disk capacity, we currently use the actual
+ // disk available capacity and total capacity to do the calculation.
+ // So that the capacity Doris actually used may exceeds the user specified capacity.
+ bool reach_capacity_limit(int64_t incoming_data_size);
+
+ Status update_capacity();
+
+ double get_usage(int64_t incoming_data_size) const {
+ return _disk_capacity_bytes == 0
+ ? 0
+ : (_disk_capacity_bytes - _available_bytes + incoming_data_size) /
+ (double)_disk_capacity_bytes;
+ }
+
+private:
+ std::string _path;
+
+ // the actual available capacity of the disk of this data dir
+ size_t _available_bytes;
+ // the actual capacity of the disk of this data dir
+ size_t _disk_capacity_bytes;
+ TStorageMedium::type _storage_medium;
+};
class SpillStreamManager {
public:
SpillStreamManager(const std::vector<StorePath>& paths);
@@ -74,10 +113,10 @@ public:
private:
Status _init_spill_store_map();
void _spill_gc_thread_callback();
- std::vector<DataDir*> _get_stores_for_spill(TStorageMedium::type storage_medium);
+ std::vector<SpillDataDir*> _get_stores_for_spill(TStorageMedium::type storage_medium);
std::vector<StorePath> _spill_store_paths;
- std::unordered_map<std::string, std::unique_ptr<DataDir>> _spill_store_map;
+ std::unordered_map<std::string, std::unique_ptr<SpillDataDir>> _spill_store_map;
CountDownLatch _stop_background_threads_latch;
std::unique_ptr<ThreadPool> async_task_thread_pool_;
diff --git a/be/src/vec/spill/spill_writer.cpp b/be/src/vec/spill/spill_writer.cpp
index 6e048ecd0dd..d51dcf8f1ec 100644
--- a/be/src/vec/spill/spill_writer.cpp
+++ b/be/src/vec/spill/spill_writer.cpp
@@ -21,7 +21,6 @@
#include "io/file_factory.h"
#include "io/fs/local_file_system.h"
#include "io/fs/local_file_writer.h"
-#include "olap/data_dir.h"
#include "runtime/exec_env.h"
#include "runtime/thread_context.h"
#include "vec/spill/spill_stream_manager.h"
diff --git a/be/src/vec/spill/spill_writer.h b/be/src/vec/spill/spill_writer.h
index 1ecda1aff9c..14e8120f775 100644
--- a/be/src/vec/spill/spill_writer.h
+++ b/be/src/vec/spill/spill_writer.h
@@ -25,12 +25,12 @@
#include "util/runtime_profile.h"
#include "vec/core/block.h"
namespace doris {
-class DataDir;
namespace vectorized {
+class SpillDataDir;
class SpillWriter {
public:
- SpillWriter(int64_t id, size_t batch_size, doris::DataDir* data_dir, const std::string& dir)
+ SpillWriter(int64_t id, size_t batch_size, SpillDataDir* data_dir, const std::string& dir)
: data_dir_(data_dir), stream_id_(id), batch_size_(batch_size) {
file_path_ = dir + "/" + std::to_string(file_index_);
}
@@ -66,7 +66,7 @@ private:
// not owned, point to the data dir of this rowset
// for checking disk capacity when write data to disk.
- doris::DataDir* data_dir_ = nullptr;
+ SpillDataDir* data_dir_ = nullptr;
std::atomic_bool closed_ = false;
int64_t stream_id_;
size_t batch_size_;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 22/24: [fix](doc) fix wrong download link (#32082)
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 2ffcfb4bc503a67aad7f3c27cf1ff1ac3e581566
Author: Mingyu Chen <mo...@163.com>
AuthorDate: Tue Mar 12 11:12:37 2024 +0800
[fix](doc) fix wrong download link (#32082)
---
docs/en/docs/lakehouse/sql-dialect.md | 2 +-
docs/zh-CN/docs/lakehouse/sql-dialect.md | 2 +-
2 files changed, 2 insertions(+), 2 deletions(-)
diff --git a/docs/en/docs/lakehouse/sql-dialect.md b/docs/en/docs/lakehouse/sql-dialect.md
index a513c93c3c2..f9ff35610aa 100644
--- a/docs/en/docs/lakehouse/sql-dialect.md
+++ b/docs/en/docs/lakehouse/sql-dialect.md
@@ -34,7 +34,7 @@ Starting from version 2.1, Doris can support multiple SQL dialects, such as Pres
## Deploy service
-1. Download latest [Doris SQL Convertor(1.0.1)](https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/transform-doris-tool/transform-doris-tool-1.0.0-bin-x86)
+1. Download latest [Doris SQL Convertor(1.0.1)](https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/transform-doris-tool/doris-sql-convertor-1.0.1-bin-x86)
2. On any FE node, start the service through the following command:
`nohup ./doris-sql-convertor-1.0.1-bin-x86 run --host=0.0.0.0 --port=5001 &`
diff --git a/docs/zh-CN/docs/lakehouse/sql-dialect.md b/docs/zh-CN/docs/lakehouse/sql-dialect.md
index b8216562c28..0539905f57c 100644
--- a/docs/zh-CN/docs/lakehouse/sql-dialect.md
+++ b/docs/zh-CN/docs/lakehouse/sql-dialect.md
@@ -34,7 +34,7 @@ under the License.
## 部署服务
-1. 下载最新版本的 [SQL 方言转换工具(1.0.1)](https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/transform-doris-tool/transform-doris-tool-1.0.0-bin-x86)
+1. 下载最新版本的 [SQL 方言转换工具(1.0.1)](https://doris-build-1308700295.cos.ap-beijing.myqcloud.com/transform-doris-tool/doris-sql-convertor-1.0.1-bin-x86)
2. 在任意 FE 节点，通过以下命令启动服务：
`nohup ./doris-sql-convertor-1.0.1-bin-x86 run --host=0.0.0.0 --port=5001 &`
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 24/24: [pipelineX](scanner) Use the actual instances num when ignore data distribution (#32081)
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 61928f7df53f5ffbaf2794cee8b2fc56100b90f7
Author: Gabriel <ga...@gmail.com>
AuthorDate: Tue Mar 12 12:11:06 2024 +0800
[pipelineX](scanner) Use the actual instances num when ignore data distribution (#32081)
---
be/src/vec/exec/scan/scanner_context.cpp | 3 ++-
1 file changed, 2 insertions(+), 1 deletion(-)
diff --git a/be/src/vec/exec/scan/scanner_context.cpp b/be/src/vec/exec/scan/scanner_context.cpp
index 82e95708ef8..3058708b2d9 100644
--- a/be/src/vec/exec/scan/scanner_context.cpp
+++ b/be/src/vec/exec/scan/scanner_context.cpp
@@ -76,7 +76,8 @@ ScannerContext::ScannerContext(RuntimeState* state, const TupleDescriptor* outpu
_max_thread_num = _state->num_scanner_threads() > 0
? _state->num_scanner_threads()
: config::doris_scanner_thread_pool_thread_num /
- state->query_parallel_instance_num();
+ (_local_state ? num_parallel_instances
+ : state->query_parallel_instance_num());
_max_thread_num *= num_parallel_instances;
_max_thread_num = _max_thread_num == 0 ? 1 : _max_thread_num;
_max_thread_num = std::min(_max_thread_num, (int32_t)scanners.size());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 21/24: [DOC](Variant) add bloom filter description and correct some benchmark results (#31998)
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 ae206596c9152956e1534951c8969706ced81c42
Author: lihangyu <15...@163.com>
AuthorDate: Tue Mar 12 10:44:32 2024 +0800
[DOC](Variant) add bloom filter description and correct some benchmark results (#31998)
---
.../sql-manual/sql-reference/Data-Types/VARIANT.md | 22 +++++++++++++++-------
.../sql-manual/sql-reference/Data-Types/VARIANT.md | 22 +++++++++++++++-------
2 files changed, 30 insertions(+), 14 deletions(-)
diff --git a/docs/en/docs/sql-manual/sql-reference/Data-Types/VARIANT.md b/docs/en/docs/sql-manual/sql-reference/Data-Types/VARIANT.md
index 9382b7dbda0..f61fa2f3395 100644
--- a/docs/en/docs/sql-manual/sql-reference/Data-Types/VARIANT.md
+++ b/docs/en/docs/sql-manual/sql-reference/Data-Types/VARIANT.md
@@ -45,11 +45,11 @@ Below are test results based on clickbench data:
| | Storage Space |
|--------------|------------|
-| Predefined Static Columns | 24.329 GB |
-| VARIANT Type | 24.296 GB |
-| JSON Type | 46.730 GB |
+| Predefined Static Columns | 12.618 GB |
+| VARIANT Type | 12.718 GB |
+| JSON Type | 35.711 GB |
-**Saves approximately 50% storage capacity**
+**Saves approximately 65% storage capacity**
| Query Counts | Predefined Static Columns | VARIANT Type | JSON Type |
|---------------------|---------------------------|--------------|-----------------|
@@ -84,12 +84,20 @@ CREATE TABLE IF NOT EXISTS ${table_name} (
)
table_properties;
+-- Create an bloom filter on v column, to enhance query seed on sub columns
+CREATE TABLE IF NOT EXISTS ${table_name} (
+ k BIGINT,
+ v VARIANT
+)
+...
+properties("replication_num" = "1", "bloom_filter_columns" = "v");
+
```
**Query Syntax**
``` sql
--- use v['a']['b'] format for example
+-- use v['a']['b'] format for example, v['properties']['title'] type is Variant
SELECT v['properties']['title'] from ${table_name}
```
@@ -351,8 +359,8 @@ When the above types cannot be compatible, they will be transformed into JSON ty
**Other limitations include:**
- Aggregate models are currently not supported.
-- VARIANT columns can only create inverted indexes.
-- Using the **RANDOM** mode is recommended for higher write performance.
+- VARIANT columns can only create inverted indexes or bloom filter to speed up query.
+- Using the **RANDOM** mode or [group commit](https://doris.apache.org/docs/dev/data-operate/import/import-way/group-commit-manual/) mode is recommended for higher write performance.
- Non-standard JSON types such as date and decimal should ideally use static types for better performance, since these types are infered to text type.
- Arrays with dimensions of 2 or higher will be stored as JSONB encoding, which might perform less efficiently than native arrays.
- Not supported as primary or sort keys.
diff --git a/docs/zh-CN/docs/sql-manual/sql-reference/Data-Types/VARIANT.md b/docs/zh-CN/docs/sql-manual/sql-reference/Data-Types/VARIANT.md
index c784a1ada36..9b7f012c2ee 100644
--- a/docs/zh-CN/docs/sql-manual/sql-reference/Data-Types/VARIANT.md
+++ b/docs/zh-CN/docs/sql-manual/sql-reference/Data-Types/VARIANT.md
@@ -45,13 +45,13 @@ VARIANT类型
| | 存储空间 |
|--------------|------------|
-| 预定义静态列 | 24.329 GB |
-| variant 类型 | 24.296 GB |
-| json 类型 | 46.730 GB |
+| 预定义静态列 | 12.618 GB |
+| variant 类型 | 12.718 GB |
+| json 类型 | 35.711 GB |
-**节省约 50%存储容量**
+**节省约 65%存储容量**
| 查询次数 | 预定义静态列 | variant 类型 | json 类型 |
|----------------|--------------|--------------|-----------------|
@@ -88,12 +88,20 @@ CREATE TABLE IF NOT EXISTS ${table_name} (
INDEX idx_var(v) USING INVERTED [PROPERTIES("parser" = "english|unicode|chinese")] [COMMENT 'your comment']
)
table_properties;
+
+-- 在v列创建bloom filter
+CREATE TABLE IF NOT EXISTS ${table_name} (
+ k BIGINT,
+ v VARIANT
+)
+...
+properties("replication_num" = "1", "bloom_filter_columns" = "v");
```
**查询语法**
``` sql
--- 使用 v['a']['b'] 形式例如
+-- 使用 v['a']['b'] 形式如下，v['properties']['title']类型是Variant
SELECT v['properties']['title'] from ${table_name}
```
@@ -359,8 +367,8 @@ VARIANT 动态列与预定义静态列几乎一样高效。处理诸如日志之
其它限制如下：
- 目前不支持 Aggregate 模型
-- VARIANT 列只能创建倒排索引
-- **推荐使用 RANDOM 模式， 写入性能更高效**
+- VARIANT 列只能创建倒排索引或者bloom filter来加速过滤
+- **推荐使用 RANDOM 模式和[Group Commit](https://doris.apache.org/zh-CN/docs/dev/data-operate/import/import-way/group-commit-manual/)模式， 写入性能更高效**
- 日期、decimal 等非标准 JSON 类型会被默认推断成字符串类型，所以尽可能从 VARIANT 中提取出来，用静态类型，性能更好
- 2 维及其以上的数组列存化会被存成 JSONB 编码，性能不如原生数组
- 不支持作为主键或者排序键
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 19/24: [refactor](nereids) rename PlanNode.distributeExprLists to childrenDistributeExprLists #32069
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 da60a111d02b03089b65a6a5763a80bc52c93604
Author: minghong <en...@gmail.com>
AuthorDate: Tue Mar 12 09:50:52 2024 +0800
[refactor](nereids) rename PlanNode.distributeExprLists to childrenDistributeExprLists #32069
---
.../glue/translator/PhysicalPlanTranslator.java | 24 +++++++++++-----------
.../java/org/apache/doris/planner/PlanNode.java | 14 ++++++-------
2 files changed, 19 insertions(+), 19 deletions(-)
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 75c29b3123d..af146628aed 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
@@ -325,7 +325,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
}
DataPartition dataPartition = toDataPartition(distribute.getDistributionSpec(), validOutputIds, context);
exchangeNode.setPartitionType(dataPartition.getType());
- exchangeNode.setDistributeExprLists(distributeExprLists);
+ exchangeNode.setChildrenDistributeExprLists(distributeExprLists);
PlanFragment parentFragment = new PlanFragment(context.nextFragmentId(), exchangeNode, dataPartition);
if (distribute.getDistributionSpec() instanceof DistributionSpecGather) {
// gather to one instance
@@ -917,7 +917,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
AggregationNode aggregationNode = new AggregationNode(context.nextPlanNodeId(),
inputPlanFragment.getPlanRoot(), aggInfo);
- aggregationNode.setDistributeExprLists(distributeExprLists);
+ aggregationNode.setChildrenDistributeExprLists(distributeExprLists);
aggregationNode.setNereidsId(aggregate.getId());
if (!aggregate.getAggMode().isFinalPhase) {
@@ -1017,7 +1017,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
AssertNumRowsNode assertNumRowsNode = new AssertNumRowsNode(context.nextPlanNodeId(),
currentFragment.getPlanRoot(),
ExpressionTranslator.translateAssert(assertNumRows.getAssertNumRowsElement()));
- assertNumRowsNode.setDistributeExprLists(distributeExprLists);
+ assertNumRowsNode.setChildrenDistributeExprLists(distributeExprLists);
assertNumRowsNode.setNereidsId(assertNumRows.getId());
addPlanRoot(currentFragment, assertNumRowsNode, assertNumRows);
return currentFragment;
@@ -1263,7 +1263,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
rightPlanRoot, joinOperator, execEqConjuncts, Lists.newArrayList(), markConjuncts,
null, null, null, hashJoin.isMarkJoin());
hashJoinNode.setNereidsId(hashJoin.getId());
- hashJoinNode.setDistributeExprLists(distributeExprLists);
+ hashJoinNode.setChildrenDistributeExprLists(distributeExprLists);
PlanFragment currentFragment = connectJoinNode(hashJoinNode, leftFragment, rightFragment, context, hashJoin);
if (JoinUtils.shouldColocateJoin(physicalHashJoin)) {
@@ -1527,7 +1527,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
leftFragmentPlanRoot, rightFragmentPlanRoot, tupleIds, JoinType.toJoinOperator(joinType),
null, null, null, nestedLoopJoin.isMarkJoin());
nestedLoopJoinNode.setNereidsId(nestedLoopJoin.getId());
- nestedLoopJoinNode.setDistributeExprLists(distributeExprLists);
+ nestedLoopJoinNode.setChildrenDistributeExprLists(distributeExprLists);
if (nestedLoopJoin.getStats() != null) {
nestedLoopJoinNode.setCardinality((long) nestedLoopJoin.getStats().getRowCount());
}
@@ -1703,7 +1703,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
List<List<Expr>> distributeExprLists = getDistributeExprs(partitionTopN.child(0));
PartitionSortNode partitionSortNode = translatePartitionSortNode(
partitionTopN, inputFragment.getPlanRoot(), context);
- partitionSortNode.setDistributeExprLists(distributeExprLists);
+ partitionSortNode.setChildrenDistributeExprLists(distributeExprLists);
addPlanRoot(inputFragment, partitionSortNode, partitionTopN);
// in pipeline engine, we use parallel scan by default, but it broke the rule of data distribution
// we need turn of parallel scan to ensure to get correct result.
@@ -1996,7 +1996,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
if (!sort.getSortPhase().isMerge()) {
// For localSort or Gather->Sort, we just need to add sortNode
SortNode sortNode = translateSortNode(sort, inputFragment.getPlanRoot(), context);
- sortNode.setDistributeExprLists(distributeExprLists);
+ sortNode.setChildrenDistributeExprLists(distributeExprLists);
addPlanRoot(inputFragment, sortNode, sort);
} else {
// For mergeSort, we need to push sortInfo to exchangeNode
@@ -2009,7 +2009,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
SortNode sortNode = (SortNode) inputFragment.getPlanRoot().getChild(0);
((ExchangeNode) inputFragment.getPlanRoot()).setMergeInfo(sortNode.getSortInfo());
sortNode.setMergeByExchange();
- sortNode.setDistributeExprLists(distributeExprLists);
+ sortNode.setChildrenDistributeExprLists(distributeExprLists);
}
return inputFragment;
}
@@ -2053,7 +2053,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
}
}
}
- sortNode.setDistributeExprLists(distributeExprLists);
+ sortNode.setChildrenDistributeExprLists(distributeExprLists);
addPlanRoot(inputFragment, sortNode, topN);
} else {
// For mergeSort, we need to push sortInfo to exchangeNode
@@ -2066,7 +2066,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
return inputFragment;
}
ExchangeNode exchangeNode = (ExchangeNode) inputFragment.getPlanRoot();
- exchangeNode.setDistributeExprLists(distributeExprLists);
+ exchangeNode.setChildrenDistributeExprLists(distributeExprLists);
exchangeNode.setMergeInfo(((SortNode) exchangeNode.getChild(0)).getSortInfo());
exchangeNode.setLimit(topN.getLimit());
exchangeNode.setOffset(topN.getOffset());
@@ -2159,7 +2159,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
inputPlanFragment.getPlanRoot(), groupingInfo, repeatSlotIdList,
allSlotId, repeat.computeVirtualSlotValues(sortedVirtualSlots));
repeatNode.setNereidsId(repeat.getId());
- repeatNode.setDistributeExprLists(distributeExprLists);
+ repeatNode.setChildrenDistributeExprLists(distributeExprLists);
addPlanRoot(inputPlanFragment, repeatNode, repeat);
updateLegacyPlanIdToPhysicalPlan(inputPlanFragment.getPlanRoot(), repeat);
return inputPlanFragment;
@@ -2246,7 +2246,7 @@ public class PhysicalPlanTranslator extends DefaultPlanVisitor<PlanFragment, Pla
bufferedTupleDesc
);
analyticEvalNode.setNereidsId(physicalWindow.getId());
- analyticEvalNode.setDistributeExprLists(distributeExprLists);
+ analyticEvalNode.setChildrenDistributeExprLists(distributeExprLists);
PlanNode root = inputPlanFragment.getPlanRoot();
if (root instanceof SortNode) {
((SortNode) root).setIsAnalyticSort(true);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
index a88925f9293..ee8965c18a5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
@@ -154,7 +154,7 @@ public abstract class PlanNode extends TreeNode<PlanNode> implements PlanStats {
protected int nereidsId = -1;
- private List<List<Expr>> distributeExprLists = new ArrayList<>();
+ private List<List<Expr>> childrenDistributeExprLists = new ArrayList<>();
protected PlanNode(PlanNodeId id, ArrayList<TupleId> tupleIds, String planNodeName,
StatisticalType statisticalType) {
@@ -540,8 +540,8 @@ public abstract class PlanNode extends TreeNode<PlanNode> implements PlanStats {
expBuilder.append(detailPrefix).append("project output tuple id: ")
.append(outputTupleDesc.getId().asInt()).append("\n");
}
- if (!CollectionUtils.isEmpty(distributeExprLists)) {
- for (List<Expr> distributeExprList : distributeExprLists) {
+ if (!CollectionUtils.isEmpty(childrenDistributeExprLists)) {
+ for (List<Expr> distributeExprList : childrenDistributeExprLists) {
expBuilder.append(detailPrefix).append("distribute expr lists: ")
.append(getExplainString(distributeExprList)).append("\n");
}
@@ -638,8 +638,8 @@ public abstract class PlanNode extends TreeNode<PlanNode> implements PlanStats {
msg.addToOutputSlotIds(slotId.asInt());
}
}
- if (!CollectionUtils.isEmpty(distributeExprLists)) {
- for (List<Expr> exprList : distributeExprLists) {
+ if (!CollectionUtils.isEmpty(childrenDistributeExprLists)) {
+ for (List<Expr> exprList : childrenDistributeExprLists) {
msg.addToDistributeExprLists(new ArrayList<>());
for (Expr expr : exprList) {
msg.distribute_expr_lists.get(msg.distribute_expr_lists.size() - 1).add(expr.treeToThrift());
@@ -1198,8 +1198,8 @@ public abstract class PlanNode extends TreeNode<PlanNode> implements PlanStats {
this.pushDownAggNoGroupingOp = pushDownAggNoGroupingOp;
}
- public void setDistributeExprLists(List<List<Expr>> distributeExprLists) {
- this.distributeExprLists = distributeExprLists;
+ public void setChildrenDistributeExprLists(List<List<Expr>> childrenDistributeExprLists) {
+ this.childrenDistributeExprLists = childrenDistributeExprLists;
}
public TPushAggOp getPushDownAggNoGroupingOp() {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 16/24: [fix](docker)Fix the registration logic of Doris On Docker (#32026)
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 a937c54b2b4a1b3da537f642ad3d27b4f2201c44
Author: FreeOnePlus <54...@users.noreply.github.com>
AuthorDate: Mon Mar 11 22:19:53 2024 +0800
[fix](docker)Fix the registration logic of Doris On Docker (#32026)
* Refactor upgrade documentation
* fix docker be entry point shell
---------
Co-authored-by: Yijia Su <su...@selectdb.com>
---
docker/runtime/be/resource/entry_point.sh | 16 +-----
docker/runtime/be/resource/init_be.sh | 96 ++++++++++++++-----------------
2 files changed, 45 insertions(+), 67 deletions(-)
diff --git a/docker/runtime/be/resource/entry_point.sh b/docker/runtime/be/resource/entry_point.sh
index 4012ce1c2fa..1ae418163fa 100644
--- a/docker/runtime/be/resource/entry_point.sh
+++ b/docker/runtime/be/resource/entry_point.sh
@@ -62,17 +62,6 @@ docker_setup_env() {
# Check the variables required for startup
docker_required_variables_env() {
declare -g RUN_TYPE
- if [ -n "$BUILD_TYPE" ]; then
- RUN_TYPE="K8S"
- if [[ $BUILD_TYPE =~ ^([kK]8[sS])$ ]]; then
- doris_warn "BUILD_TYPE" $BUILD_TYPE
- else
- doris_error "BUILD_TYPE rule error！example: [k8s], Default Value: docker"
- fi
- export RUN_TYPE=${RUN_TYPE}
- return
- fi
-
if [[ -n "$FE_SERVERS" && -n "$BE_ADDR" ]]; then
RUN_TYPE="ELECTION"
if [[ $FE_SERVERS =~ ^.+:[1-2]{0,1}[0-9]{0,1}[0-9]{1}(\.[1-2]{0,1}[0-9]{0,1}[0-9]{1}){3}:[1-6]{0,1}[0-9]{1,4}(,.+:[1-2]{0,1}[0-9]{0,1}[0-9]{1}(\.[1-2]{0,1}[0-9]{0,1}[0-9]{1}){3}:[1-6]{0,1}[0-9]{1,4})*$ || $FE_SERVERS =~ ^.+:([0-9a-fA-F]{1,4}:){7,7}([0-9a-fA-F]{1,4}|:)|([0-9a-fA-F]{1,4}:){1,6}(:[0-9a-fA-F]{1,4}|:)|([0-9a-fA-F]{1,4}:){1,5}((:[0-9a-fA-F]{1,4}){1,2}|:)|([0-9a-fA-F]{1,4}:){1,4}((:[0-9a-fA-F]{1,4}){1,3}|:)|([0-9a-fA-F]{1,4}:){1,3}((:[0-9a-fA-F]{1,4}){1,4}|:)|([0-9a-fA-F] [...]
@@ -110,14 +99,11 @@ docker_required_variables_env() {
return
fi
-
doris_error EOF "
Note that you did not configure the required parameters!
plan 1:
- BUILD_TYPE
- plan 2:
FE_SERVERS & BE_ADDR
- plan 3:
+ plan 2:
FE_MASTER_IP & FE_MASTER_PORT & BE_IP & BE_PORT"
EOF
}
diff --git a/docker/runtime/be/resource/init_be.sh b/docker/runtime/be/resource/init_be.sh
index 22817a39cfc..5c8a4dd0563 100644
--- a/docker/runtime/be/resource/init_be.sh
+++ b/docker/runtime/be/resource/init_be.sh
@@ -92,56 +92,53 @@ node_role_conf(){
}
register_be_to_fe() {
- set +e
- # check fe status
- local is_fe_start=false
- for i in {1..300}; do
- docker_process_sql <<<"alter system add backend '${CURRENT_BE_IP}:${CURRENT_BE_PORT}'"
- register_be_status=$?
- if [[ $register_be_status == 0 ]]; then
- doris_note "BE successfully registered to FE！"
- is_fe_start=true
- break
- else
+ set +e
+ # check fe status
+ local is_fe_start=false
+ if [ -n "$DATABASE_ALREADY_EXISTS" ]; then
check_be_status
if [ -n "$BE_ALREADY_EXISTS" ]; then
doris_warn "Same backend already exists! No need to register again！"
- break
+ return
+ fi
+ fi
+ for i in {1..300}; do
+ docker_process_sql <<<"alter system add backend '${CURRENT_BE_IP}:${CURRENT_BE_PORT}'"
+ register_be_status=$?
+ if [[ $register_be_status == 0 ]]; then
+ doris_note "BE successfully registered to FE！"
+ is_fe_start=true
+ return
fi
if [[ $(( $i % 20 )) == 1 ]]; then
- doris_warn "register_be_status: ${register_be_status}"
- doris_warn "BE failed registered to FE!"
+ doris_note "Register BE to FE is failed. retry."
fi
+ sleep 1
+ done
+ if ! [[ $is_fe_start ]]; then
+ doris_error "Failed to register BE to FE！Tried 30 times！Maybe FE Start Failed！"
fi
- if [[ $(( $i % 20 )) == 1 ]]; then
- doris_note "Register BE to FE is failed. retry."
- fi
- sleep 1
- done
- if ! [[ $is_fe_start ]]; then
- doris_error "Failed to register BE to FE！Tried 30 times！Maybe FE Start Failed！"
- fi
}
check_be_status() {
set +e
local is_fe_start=false
for i in {1..300}; do
- if [[ $(($i % 20)) == 1 ]]; then
- doris_warn "start check be register status~"
- fi
- docker_process_sql <<<"show backends;" | grep "[[:space:]]${CURRENT_BE_IP}[[:space:]]" | grep "[[:space:]]${CURRENT_BE_PORT}[[:space:]]"
- be_join_status=$?
- if [[ "${be_join_status}" == 0 ]]; then
- doris_note "Verify that BE is registered to FE successfully"
- is_fe_start=true
- return
- else
- if [[ $(($i % 20)) == 1 ]]; then
- doris_note "register is failed, wait next~"
- fi
- fi
- sleep 1
+ if [[ $(($i % 20)) == 1 ]]; then
+ doris_warn "start check be register status~"
+ fi
+ docker_process_sql <<<"show backends;" | grep "[[:space:]]${CURRENT_BE_IP}[[:space:]]" | grep "[[:space:]]${CURRENT_BE_PORT}[[:space:]]"
+ be_join_status=$?
+ if [[ "${be_join_status}" == 0 ]]; then
+ doris_note "Verify that BE is registered to FE successfully"
+ is_fe_start=true
+ return
+ else
+ if [[ $(($i % 20)) == 1 ]]; then
+ doris_note "register is failed, wait next~"
+ fi
+ fi
+ sleep 1
done
if [[ ! $is_fe_start ]]; then
doris_error "Failed to register BE to FE！Tried 30 times！Maybe FE Start Failed！"
@@ -155,22 +152,17 @@ cleanup() {
_main() {
trap 'cleanup' SIGTERM SIGINT
- if [[ $RUN_TYPE == "K8S" ]]; then
- start_be.sh --console &
- child_pid=$!
- else
- docker_setup_env
- if [ -z "$DATABASE_ALREADY_EXISTS" ]; then
- add_priority_networks $PRIORITY_NETWORKS
- node_role_conf
- show_be_args
- register_be_to_fe
- fi
- check_be_status
- doris_note "Ready to start BE！"
- start_be.sh --console &
- child_pid=$!
+ docker_setup_env
+ if [ -z "$DATABASE_ALREADY_EXISTS" ]; then
+ add_priority_networks $PRIORITY_NETWORKS
+ node_role_conf
+ show_be_args
+ register_be_to_fe
fi
+ check_be_status
+ doris_note "Ready to start BE！"
+ start_be.sh --console &
+ child_pid=$!
wait $child_pid
exec "$@"
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org