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/17 00:11:41 UTC
(doris) branch branch-2.1 updated (9ad196f1890 -> c34f5045c88)
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 9ad196f1890 Revert "[fix](cloud) ignore some case in cloud mode (#32261)"
new a90a1a76f16 [bugfix](profile) support multi execution profile for brokerload (#32280)
new f64a9a33f8e [fix](Nereids): don't pushdown project when project contains both side of join (#32214)
new 844a1b53b74 [fix](retry) Set query encounter rpc exception default retry times to 3 (#28555)
new bf820302707 [Chore](FE)Remove unused components (#32295)
new a15bf3057f4 [Fix](nereids) remove duplicate expr in grouping set (#32290)
new 83ab61ad226 Add QUEUE_START_TIME/QUEUE_END_TIME/QUERY_STATUS column for active_queries (#32259)
new c5ffeff8330 [fix](s3 client)add default ca cert list for s3 client to avoid problem:'curlCode:77' (#32285)
new 5ceccb5ba5d [fix](compatibility) should enable windown funnel mode from 2.0 (#32284)
new bf9332a2759 [docs](docs) Fix invalid link and typo of Master Branch (#32213)
new 4bf202db042 [pipelineX](exchange) Make exchange buffer size configurable (#32201)
new c34f5045c88 fix compile
The 11 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/agent/be_exec_version_manager.h | 3 +-
be/src/common/config.cpp | 7 ++
be/src/common/config.h | 4 +
.../schema_active_queries_scanner.cpp | 8 +-
be/src/pipeline/exec/exchange_sink_buffer.cpp | 6 +-
be/src/pipeline/exec/exchange_sink_buffer.h | 1 -
be/src/pipeline/exec/scan_operator.cpp | 1 +
be/src/util/s3_util.cpp | 23 ++++++
be/src/util/s3_util.h | 2 +
.../aggregate_function_window_funnel.h | 4 +-
.../main/java/org/apache/doris/common/Config.java | 2 +-
.../java/org/apache/doris/catalog/SchemaTable.java | 5 +-
.../doris/common/profile/AggregatedProfile.java | 41 ----------
.../org/apache/doris/common/profile/Profile.java | 40 ++++++----
.../apache/doris/load/loadv2/LoadLoadingTask.java | 2 +-
.../nereids/rules/analysis/NormalizeRepeat.java | 11 +++
.../join/PushDownProjectThroughSemiJoin.java | 19 +++++
.../doris/nereids/trees/plans/algebra/Repeat.java | 3 -
.../trees/plans/commands/InsertExecutor.java | 2 +-
.../nereids/trees/plans/logical/LogicalRepeat.java | 4 +
.../main/java/org/apache/doris/qe/Coordinator.java | 4 +
.../java/org/apache/doris/qe/QeProcessorImpl.java | 29 ++++++-
.../org/apache/doris/qe/QueryStatisticsItem.java | 6 +-
.../java/org/apache/doris/qe/StmtExecutor.java | 6 +-
.../doris/resource/workloadgroup/QueryQueue.java | 6 +-
.../doris/resource/workloadgroup/QueueToken.java | 33 +++++++-
.../doris/tablefunction/MetadataGenerator.java | 90 +++++++++-------------
.../join/PushDownProjectThroughSemiJoinTest.java | 32 ++++++++
fe/pom.xml | 10 +++
.../remove_duplicate_expr_in_grouping_set.out} | 21 +++--
.../remove_duplicate_expr_in_grouping_set.groovy} | 28 ++++---
.../schema_table/test_active_queries.groovy | 6 +-
32 files changed, 301 insertions(+), 158 deletions(-)
delete mode 100644 fe/fe-core/src/main/java/org/apache/doris/common/profile/AggregatedProfile.java
copy regression-test/data/{load/insert/test_publish_one_succ.out => nereids_rules_p0/grouping_sets/remove_duplicate_expr_in_grouping_set.out} (52%)
copy regression-test/suites/nereids_rules_p0/{cte/test_cte_reuse_with_window.groovy => grouping_sets/remove_duplicate_expr_in_grouping_set.groovy} (58%)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 07/11: [fix](s3 client)add default ca cert list for s3 client to avoid problem:'curlCode:77' (#32285)
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 c5ffeff833060add948eb98cea7fa538b9fcd918
Author: ryanzryu <14...@users.noreply.github.com>
AuthorDate: Fri Mar 15 22:19:58 2024 +0800
[fix](s3 client)add default ca cert list for s3 client to avoid problem:'curlCode:77' (#32285)
Co-authored-by: ryanzryu <ry...@tencent.com>
---
be/src/common/config.cpp | 6 ++++++
be/src/common/config.h | 3 +++
be/src/util/s3_util.cpp | 23 +++++++++++++++++++++++
be/src/util/s3_util.h | 2 ++
4 files changed, 34 insertions(+)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 3d44528ea36..63ed49eca24 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -1190,6 +1190,12 @@ DEFINE_mBool(check_segment_when_build_rowset_meta, "false");
DEFINE_mInt32(max_s3_client_retry, "10");
+// ca_cert_file is in this path by default, Normally no modification is required
+// ca cert default path is different from different OS
+DEFINE_mString(ca_cert_file_paths,
+ "/etc/pki/tls/certs/ca-bundle.crt;/etc/ssl/certs/ca-certificates.crt;"
+ "/etc/ssl/ca-bundle.pem");
+
// clang-format off
#ifdef BE_TEST
// test s3
diff --git a/be/src/common/config.h b/be/src/common/config.h
index a5219c42590..a0a20450d5d 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -1267,6 +1267,9 @@ DECLARE_mInt32(max_s3_client_retry);
// write as inverted index tmp directory
DECLARE_String(tmp_file_dir);
+// the file paths(one or more) of CA cert, splite using ";" aws s3 lib use it to init s3client
+DECLARE_mString(ca_cert_file_paths);
+
#ifdef BE_TEST
// test s3
DECLARE_String(test_s3_resource);
diff --git a/be/src/util/s3_util.cpp b/be/src/util/s3_util.cpp
index d09b808868b..063cc16c67b 100644
--- a/be/src/util/s3_util.cpp
+++ b/be/src/util/s3_util.cpp
@@ -30,6 +30,7 @@
#include <atomic>
#include <cstdlib>
+#include <filesystem>
#include <functional>
#include <memory>
#include <ostream>
@@ -114,6 +115,18 @@ S3ClientFactory::S3ClientFactory() {
return std::make_shared<DorisAWSLogger>(logLevel);
};
Aws::InitAPI(_aws_options);
+ _ca_cert_file_path = get_valid_ca_cert_path();
+}
+
+string S3ClientFactory::get_valid_ca_cert_path() {
+ vector<std::string> vec_ca_file_path = doris::split(config::ca_cert_file_paths, ";");
+ vector<std::string>::iterator it = vec_ca_file_path.begin();
+ for (; it != vec_ca_file_path.end(); ++it) {
+ if (std::filesystem::exists(*it)) {
+ return *it;
+ }
+ }
+ return "";
}
S3ClientFactory::~S3ClientFactory() {
@@ -157,6 +170,16 @@ std::shared_ptr<Aws::S3::S3Client> S3ClientFactory::create(const S3Conf& s3_conf
Aws::Client::ClientConfiguration aws_config = S3ClientFactory::getClientConfiguration();
aws_config.endpointOverride = s3_conf.endpoint;
aws_config.region = s3_conf.region;
+ std::string ca_cert = get_valid_ca_cert_path();
+ if ("" != _ca_cert_file_path) {
+ aws_config.caFile = _ca_cert_file_path;
+ } else {
+ // config::ca_cert_file_paths is valmutable,get newest value if file path invaild
+ _ca_cert_file_path = get_valid_ca_cert_path();
+ if ("" != _ca_cert_file_path) {
+ aws_config.caFile = _ca_cert_file_path;
+ }
+ }
if (s3_conf.max_connections > 0) {
aws_config.maxConnections = s3_conf.max_connections;
} else {
diff --git a/be/src/util/s3_util.h b/be/src/util/s3_util.h
index a47b5620548..74af0121d20 100644
--- a/be/src/util/s3_util.h
+++ b/be/src/util/s3_util.h
@@ -133,10 +133,12 @@ public:
private:
S3ClientFactory();
+ static std::string get_valid_ca_cert_path();
Aws::SDKOptions _aws_options;
std::mutex _lock;
std::unordered_map<uint64_t, std::shared_ptr<Aws::S3::S3Client>> _cache;
+ std::string _ca_cert_file_path;
};
} // end namespace doris
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 06/11: Add QUEUE_START_TIME/QUEUE_END_TIME/QUERY_STATUS column for active_queries (#32259)
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 83ab61ad2269d4d58a9de0f0e9cfac5f9bf951f7
Author: wangbo <wa...@apache.org>
AuthorDate: Fri Mar 15 22:06:47 2024 +0800
Add QUEUE_START_TIME/QUEUE_END_TIME/QUERY_STATUS column for active_queries (#32259)
---
.../schema_active_queries_scanner.cpp | 8 +-
.../java/org/apache/doris/catalog/SchemaTable.java | 5 +-
.../main/java/org/apache/doris/qe/Coordinator.java | 4 +
.../java/org/apache/doris/qe/QeProcessorImpl.java | 29 ++++++-
.../org/apache/doris/qe/QueryStatisticsItem.java | 6 +-
.../doris/resource/workloadgroup/QueryQueue.java | 6 +-
.../doris/resource/workloadgroup/QueueToken.java | 33 +++++++-
.../doris/tablefunction/MetadataGenerator.java | 90 +++++++++-------------
.../schema_table/test_active_queries.groovy | 6 +-
9 files changed, 123 insertions(+), 64 deletions(-)
diff --git a/be/src/exec/schema_scanner/schema_active_queries_scanner.cpp b/be/src/exec/schema_scanner/schema_active_queries_scanner.cpp
index f16326dc8f5..36cb145e3f5 100644
--- a/be/src/exec/schema_scanner/schema_active_queries_scanner.cpp
+++ b/be/src/exec/schema_scanner/schema_active_queries_scanner.cpp
@@ -29,11 +29,14 @@ namespace doris {
std::vector<SchemaScanner::ColumnDesc> SchemaActiveQueriesScanner::_s_tbls_columns = {
// name, type, size
{"QUERY_ID", TYPE_VARCHAR, sizeof(StringRef), true},
- {"START_TIME", TYPE_VARCHAR, sizeof(StringRef), true},
+ {"QUERY_START_TIME", TYPE_VARCHAR, sizeof(StringRef), true},
{"QUERY_TIME_MS", TYPE_BIGINT, sizeof(int64_t), true},
{"WORKLOAD_GROUP_ID", TYPE_BIGINT, sizeof(int64_t), true},
{"DATABASE", TYPE_VARCHAR, sizeof(StringRef), true},
{"FRONTEND_INSTANCE", TYPE_VARCHAR, sizeof(StringRef), true},
+ {"QUEUE_START_TIME", TYPE_VARCHAR, sizeof(StringRef), true},
+ {"QUEUE_END_TIME", TYPE_VARCHAR, sizeof(StringRef), true},
+ {"QUERY_STATUS", TYPE_VARCHAR, sizeof(StringRef), true},
{"SQL", TYPE_STRING, sizeof(StringRef), true}};
SchemaActiveQueriesScanner::SchemaActiveQueriesScanner()
@@ -127,6 +130,9 @@ Status SchemaActiveQueriesScanner::_get_active_queries_block_from_fe() {
insert_string_value(4, row.column_value[4].stringVal, _active_query_block.get());
insert_string_value(5, row.column_value[5].stringVal, _active_query_block.get());
insert_string_value(6, row.column_value[6].stringVal, _active_query_block.get());
+ insert_string_value(7, row.column_value[7].stringVal, _active_query_block.get());
+ insert_string_value(8, row.column_value[8].stringVal, _active_query_block.get());
+ insert_string_value(9, row.column_value[9].stringVal, _active_query_block.get());
}
return Status::OK();
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
index 736df74f1e6..b89b7e1ea6b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/SchemaTable.java
@@ -460,11 +460,14 @@ public class SchemaTable extends Table {
.build()))
.put("active_queries", new SchemaTable(SystemIdGenerator.getNextId(), "active_queries", TableType.SCHEMA,
builder().column("QUERY_ID", ScalarType.createVarchar(256))
- .column("START_TIME", ScalarType.createVarchar(256))
+ .column("QUERY_START_TIME", ScalarType.createVarchar(256))
.column("QUERY_TIME_MS", ScalarType.createType(PrimitiveType.BIGINT))
.column("WORKLOAD_GROUP_ID", ScalarType.createType(PrimitiveType.BIGINT))
.column("DATABASE", ScalarType.createVarchar(256))
.column("FRONTEND_INSTANCE", ScalarType.createVarchar(256))
+ .column("QUEUE_START_TIME", ScalarType.createVarchar(256))
+ .column("QUEUE_END_TIME", ScalarType.createVarchar(256))
+ .column("QUERY_STATUS", ScalarType.createVarchar(256))
.column("SQL", ScalarType.createStringType())
.build()))
.put("workload_groups", new SchemaTable(SystemIdGenerator.getNextId(), "workload_groups", TableType.SCHEMA,
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 6e016d94e2a..d0661006d1d 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
@@ -3904,6 +3904,10 @@ public class Coordinator implements CoordInterface {
}
}
+ public QueueToken getQueueToken() {
+ return queueToken;
+ }
+
// fragment instance exec param, it is used to assemble
// the per-instance TPlanFragmentExecParams, as a member of
// FragmentExecParams
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java b/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java
index 55536639bfd..03144fc797c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/QeProcessorImpl.java
@@ -24,6 +24,7 @@ import org.apache.doris.common.UserException;
import org.apache.doris.common.profile.ExecutionProfile;
import org.apache.doris.common.util.DebugUtil;
import org.apache.doris.metric.MetricRepo;
+import org.apache.doris.resource.workloadgroup.QueueToken.TokenState;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TQueryType;
import org.apache.doris.thrift.TReportExecStatusParams;
@@ -265,7 +266,6 @@ public final class QeProcessorImpl implements QeProcessor {
private final ConnectContext connectContext;
private final Coordinator coord;
private final String sql;
- private final long startExecTime;
// from Export, Pull load, Insert
public QueryInfo(Coordinator coord) {
@@ -277,7 +277,6 @@ public final class QeProcessorImpl implements QeProcessor {
this.connectContext = connectContext;
this.coord = coord;
this.sql = sql;
- this.startExecTime = System.currentTimeMillis();
}
public ConnectContext getConnectContext() {
@@ -293,7 +292,31 @@ public final class QeProcessorImpl implements QeProcessor {
}
public long getStartExecTime() {
- return startExecTime;
+ if (coord.getQueueToken() != null) {
+ return coord.getQueueToken().getQueueEndTime();
+ }
+ return -1;
+ }
+
+ public long getQueueStartTime() {
+ if (coord.getQueueToken() != null) {
+ return coord.getQueueToken().getQueueStartTime();
+ }
+ return -1;
+ }
+
+ public long getQueueEndTime() {
+ if (coord.getQueueToken() != null) {
+ return coord.getQueueToken().getQueueEndTime();
+ }
+ return -1;
+ }
+
+ public TokenState getQueueStatus() {
+ if (coord.getQueueToken() != null) {
+ return coord.getQueueToken().getTokenState();
+ }
+ return null;
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/QueryStatisticsItem.java b/fe/fe-core/src/main/java/org/apache/doris/qe/QueryStatisticsItem.java
index 79c3e083113..76b528464d6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/QueryStatisticsItem.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/QueryStatisticsItem.java
@@ -74,7 +74,11 @@ public final class QueryStatisticsItem {
public String getQueryExecTime() {
final long currentTime = System.currentTimeMillis();
- return String.valueOf(currentTime - queryStartTime);
+ if (queryStartTime <= 0) {
+ return String.valueOf(-1);
+ } else {
+ return String.valueOf(currentTime - queryStartTime);
+ }
}
public String getQueryId() {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/QueryQueue.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/QueryQueue.java
index 7ba6353e746..5953edbf66e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/QueryQueue.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/QueryQueue.java
@@ -100,7 +100,6 @@ public class QueryQueue {
}
public QueueToken getToken() throws UserException {
-
queueLock.lock();
try {
if (LOG.isDebugEnabled()) {
@@ -108,13 +107,16 @@ public class QueryQueue {
}
if (currentRunningQueryNum < maxConcurrency) {
currentRunningQueryNum++;
- return new QueueToken(TokenState.READY_TO_RUN, queueTimeout, "offer success");
+ QueueToken retToken = new QueueToken(TokenState.READY_TO_RUN, queueTimeout, "offer success");
+ retToken.setQueueTimeWhenOfferSuccess();
+ return retToken;
}
if (priorityTokenQueue.size() >= maxQueueSize) {
throw new UserException("query waiting queue is full, queue length=" + maxQueueSize);
}
QueueToken newQueryToken = new QueueToken(TokenState.ENQUEUE_SUCCESS, queueTimeout,
"query wait timeout " + queueTimeout + " ms");
+ newQueryToken.setQueueTimeWhenQueueSuccess();
this.priorityTokenQueue.offer(newQueryToken);
return newQueryToken;
} finally {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/QueueToken.java b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/QueueToken.java
index 189ba77e8de..6bf44c78828 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/QueueToken.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/resource/workloadgroup/QueueToken.java
@@ -36,7 +36,7 @@ public class QueueToken implements Comparable<QueueToken> {
return Long.compare(this.tokenId, other.getTokenId());
}
- enum TokenState {
+ public enum TokenState {
ENQUEUE_SUCCESS,
READY_TO_RUN
}
@@ -56,6 +56,9 @@ public class QueueToken implements Comparable<QueueToken> {
private final ReentrantLock tokenLock = new ReentrantLock();
private final Condition tokenCond = tokenLock.newCondition();
+ private long queueStartTime = -1;
+ private long queueEndTime = -1;
+
public QueueToken(TokenState tokenState, long queueWaitTimeout,
String offerResultDetail) {
this.tokenId = tokenIdGenerator.addAndGet(1);
@@ -94,6 +97,7 @@ public class QueueToken implements Comparable<QueueToken> {
return false;
} finally {
this.tokenLock.unlock();
+ this.setQueueTimeWhenQueueEnd();
}
}
@@ -126,6 +130,33 @@ public class QueueToken implements Comparable<QueueToken> {
return this.tokenState == TokenState.READY_TO_RUN;
}
+ public void setQueueTimeWhenOfferSuccess() {
+ long currentTime = System.currentTimeMillis();
+ this.queueStartTime = currentTime;
+ this.queueEndTime = currentTime;
+ }
+
+ public void setQueueTimeWhenQueueSuccess() {
+ long currentTime = System.currentTimeMillis();
+ this.queueStartTime = currentTime;
+ }
+
+ public void setQueueTimeWhenQueueEnd() {
+ this.queueEndTime = System.currentTimeMillis();
+ }
+
+ public long getQueueStartTime() {
+ return queueStartTime;
+ }
+
+ public long getQueueEndTime() {
+ return queueEndTime;
+ }
+
+ public TokenState getTokenState() {
+ return tokenState;
+ }
+
@Override
public boolean equals(Object obj) {
if (this == obj) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
index 8f4dbf95616..adf3a9b3ed2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/MetadataGenerator.java
@@ -42,6 +42,7 @@ import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.qe.ConnectContext;
import org.apache.doris.qe.QeProcessorImpl;
import org.apache.doris.qe.QeProcessorImpl.QueryInfo;
+import org.apache.doris.resource.workloadgroup.QueueToken.TokenState;
import org.apache.doris.system.Backend;
import org.apache.doris.system.SystemInfoService;
import org.apache.doris.thrift.FrontendService;
@@ -57,7 +58,6 @@ import org.apache.doris.thrift.TMetadataTableRequestParams;
import org.apache.doris.thrift.TMetadataType;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TPipelineWorkloadGroup;
-import org.apache.doris.thrift.TQueryStatistics;
import org.apache.doris.thrift.TRow;
import org.apache.doris.thrift.TSchemaTableRequestParams;
import org.apache.doris.thrift.TStatus;
@@ -91,11 +91,14 @@ public class MetadataGenerator {
private static final ImmutableList<Column> ACTIVE_QUERIES_SCHEMA = ImmutableList.of(
new Column("QUERY_ID", ScalarType.createStringType()),
- new Column("START_TIME", ScalarType.createStringType()),
+ new Column("QUERY_START_TIME", ScalarType.createStringType()),
new Column("QUERY_TIME_MS", PrimitiveType.BIGINT),
new Column("WORKLOAD_GROUP_ID", PrimitiveType.BIGINT),
new Column("DATABASE", ScalarType.createStringType()),
new Column("FRONTEND_INSTANCE", ScalarType.createStringType()),
+ new Column("QUEUE_START_TIME", ScalarType.createStringType()),
+ new Column("QUEUE_END_TIME", ScalarType.createStringType()),
+ new Column("QUERY_STATUS", ScalarType.createStringType()),
new Column("SQL", ScalarType.createStringType()));
private static final ImmutableMap<String, Integer> ACTIVE_QUERIES_COLUMN_TO_INDEX;
@@ -490,53 +493,6 @@ public class MetadataGenerator {
return result;
}
- private static TRow makeQueryStatisticsTRow(SimpleDateFormat sdf, String queryId, Backend be,
- String selfNode, QueryInfo queryInfo, TQueryStatistics qs) {
- TRow trow = new TRow();
- if (be != null) {
- trow.addToColumnValue(new TCell().setStringVal(be.getHost()));
- trow.addToColumnValue(new TCell().setLongVal(be.getBePort()));
- } else {
- trow.addToColumnValue(new TCell().setStringVal("invalid host"));
- trow.addToColumnValue(new TCell().setLongVal(-1));
- }
- trow.addToColumnValue(new TCell().setStringVal(queryId));
-
- String strDate = sdf.format(new Date(queryInfo.getStartExecTime()));
- trow.addToColumnValue(new TCell().setStringVal(strDate));
- trow.addToColumnValue(new TCell().setLongVal(System.currentTimeMillis() - queryInfo.getStartExecTime()));
-
- if (qs != null) {
- trow.addToColumnValue(new TCell().setLongVal(qs.workload_group_id));
- trow.addToColumnValue(new TCell().setLongVal(qs.cpu_ms));
- trow.addToColumnValue(new TCell().setLongVal(qs.scan_rows));
- trow.addToColumnValue(new TCell().setLongVal(qs.scan_bytes));
- trow.addToColumnValue(new TCell().setLongVal(qs.max_peak_memory_bytes));
- trow.addToColumnValue(new TCell().setLongVal(qs.current_used_memory_bytes));
- trow.addToColumnValue(new TCell().setLongVal(qs.shuffle_send_bytes));
- trow.addToColumnValue(new TCell().setLongVal(qs.shuffle_send_rows));
- } else {
- trow.addToColumnValue(new TCell().setLongVal(0L));
- trow.addToColumnValue(new TCell().setLongVal(0L));
- trow.addToColumnValue(new TCell().setLongVal(0L));
- trow.addToColumnValue(new TCell().setLongVal(0L));
- trow.addToColumnValue(new TCell().setLongVal(0L));
- trow.addToColumnValue(new TCell().setLongVal(0L));
- trow.addToColumnValue(new TCell().setLongVal(0L));
- trow.addToColumnValue(new TCell().setLongVal(0L));
- }
-
- if (queryInfo.getConnectContext() != null) {
- trow.addToColumnValue(new TCell().setStringVal(queryInfo.getConnectContext().getDatabase()));
- } else {
- trow.addToColumnValue(new TCell().setStringVal(""));
- }
- trow.addToColumnValue(new TCell().setStringVal(selfNode));
- trow.addToColumnValue(new TCell().setStringVal(queryInfo.getSql()));
-
- return trow;
- }
-
private static TFetchSchemaTableDataResult queriesMetadataResult(TSchemaTableRequestParams tSchemaTableParams,
TFetchSchemaTableDataRequest parentRequest) {
TFetchSchemaTableDataResult result = new TFetchSchemaTableDataResult();
@@ -557,9 +513,15 @@ public class MetadataGenerator {
TRow trow = new TRow();
trow.addToColumnValue(new TCell().setStringVal(queryId));
- String strDate = sdf.format(new Date(queryInfo.getStartExecTime()));
- trow.addToColumnValue(new TCell().setStringVal(strDate));
- trow.addToColumnValue(new TCell().setLongVal(System.currentTimeMillis() - queryInfo.getStartExecTime()));
+ long queryStartTime = queryInfo.getStartExecTime();
+ if (queryStartTime > 0) {
+ trow.addToColumnValue(new TCell().setStringVal(sdf.format(new Date(queryStartTime))));
+ trow.addToColumnValue(
+ new TCell().setLongVal(System.currentTimeMillis() - queryInfo.getStartExecTime()));
+ } else {
+ trow.addToColumnValue(new TCell());
+ trow.addToColumnValue(new TCell().setLongVal(-1));
+ }
List<TPipelineWorkloadGroup> tgroupList = queryInfo.getCoord().gettWorkloadGroups();
if (tgroupList != null && tgroupList.size() == 1) {
@@ -574,6 +536,30 @@ public class MetadataGenerator {
trow.addToColumnValue(new TCell().setStringVal(""));
}
trow.addToColumnValue(new TCell().setStringVal(selfNode));
+
+ long queueStartTime = queryInfo.getQueueStartTime();
+ if (queueStartTime > 0) {
+ trow.addToColumnValue(new TCell().setStringVal(sdf.format(new Date(queueStartTime))));
+ } else {
+ trow.addToColumnValue(new TCell());
+ }
+
+ long queueEndTime = queryInfo.getQueueEndTime();
+ if (queueEndTime > 0) {
+ trow.addToColumnValue(new TCell().setStringVal(sdf.format(new Date(queueEndTime))));
+ } else {
+ trow.addToColumnValue(new TCell());
+ }
+
+ TokenState tokenState = queryInfo.getQueueStatus();
+ if (tokenState == null) {
+ trow.addToColumnValue(new TCell());
+ } else if (tokenState == TokenState.READY_TO_RUN) {
+ trow.addToColumnValue(new TCell().setStringVal("RUNNING"));
+ } else {
+ trow.addToColumnValue(new TCell().setStringVal("QUEUED"));
+ }
+
trow.addToColumnValue(new TCell().setStringVal(queryInfo.getSql()));
dataBatch.add(trow);
}
diff --git a/regression-test/suites/query_p0/schema_table/test_active_queries.groovy b/regression-test/suites/query_p0/schema_table/test_active_queries.groovy
index eecba3d063b..64c344deec7 100644
--- a/regression-test/suites/query_p0/schema_table/test_active_queries.groovy
+++ b/regression-test/suites/query_p0/schema_table/test_active_queries.groovy
@@ -22,19 +22,19 @@ suite("test_active_queries") {
sql "set experimental_enable_pipeline_engine=false"
sql "set experimental_enable_pipeline_x_engine=false"
sql "select * from information_schema.active_queries"
- sql "select QUERY_ID,START_TIME,QUERY_TIME_MS,WORKLOAD_GROUP_ID,SQL from information_schema.active_queries"
+ sql "select QUERY_ID,QUERY_START_TIME,QUERY_TIME_MS,WORKLOAD_GROUP_ID,SQL,QUERY_STATUS from information_schema.active_queries"
// pipeline
sql "set experimental_enable_pipeline_engine=true"
sql "set experimental_enable_pipeline_x_engine=false"
sql "select * from information_schema.active_queries"
- sql "select QUERY_ID,START_TIME,QUERY_TIME_MS,WORKLOAD_GROUP_ID,SQL from information_schema.active_queries"
+ sql "select QUERY_ID,QUERY_START_TIME,QUERY_TIME_MS,WORKLOAD_GROUP_ID,SQL,QUERY_STATUS from information_schema.active_queries"
// pipelinex
sql "set experimental_enable_pipeline_engine=true"
sql "set experimental_enable_pipeline_x_engine=true"
sql "select * from information_schema.active_queries"
- sql "select QUERY_ID,START_TIME,QUERY_TIME_MS,WORKLOAD_GROUP_ID,SQL from information_schema.active_queries"
+ sql "select QUERY_ID,QUERY_START_TIME,QUERY_TIME_MS,WORKLOAD_GROUP_ID,SQL,QUERY_STATUS from information_schema.active_queries"
Thread.sleep(1000)
}
})
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 10/11: [pipelineX](exchange) Make exchange buffer size configurable (#32201)
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 4bf202db04292cd4c3972f73d167fd4ad7717188
Author: Gabriel <ga...@gmail.com>
AuthorDate: Sat Mar 16 09:04:03 2024 +0800
[pipelineX](exchange) Make exchange buffer size configurable (#32201)
---
be/src/common/config.cpp | 1 +
be/src/common/config.h | 1 +
be/src/pipeline/exec/exchange_sink_buffer.cpp | 6 ++++--
be/src/pipeline/exec/exchange_sink_buffer.h | 1 -
be/src/pipeline/exec/scan_operator.cpp | 1 +
5 files changed, 7 insertions(+), 3 deletions(-)
diff --git a/be/src/common/config.cpp b/be/src/common/config.cpp
index 63ed49eca24..dc31d2c621e 100644
--- a/be/src/common/config.cpp
+++ b/be/src/common/config.cpp
@@ -264,6 +264,7 @@ DEFINE_mInt32(doris_max_scan_key_num, "48");
DEFINE_mInt32(max_pushdown_conditions_per_column, "1024");
// (Advanced) Maximum size of per-query receive-side buffer
DEFINE_mInt32(exchg_node_buffer_size_bytes, "20485760");
+DEFINE_mInt32(exchg_buffer_queue_capacity_factor, "64");
DEFINE_mInt64(column_dictionary_key_ratio_threshold, "0");
DEFINE_mInt64(column_dictionary_key_size_threshold, "0");
diff --git a/be/src/common/config.h b/be/src/common/config.h
index a0a20450d5d..abb833f42e8 100644
--- a/be/src/common/config.h
+++ b/be/src/common/config.h
@@ -306,6 +306,7 @@ DECLARE_mInt32(doris_max_scan_key_num);
DECLARE_mInt32(max_pushdown_conditions_per_column);
// (Advanced) Maximum size of per-query receive-side buffer
DECLARE_mInt32(exchg_node_buffer_size_bytes);
+DECLARE_mInt32(exchg_buffer_queue_capacity_factor);
DECLARE_mInt64(column_dictionary_key_ratio_threshold);
DECLARE_mInt64(column_dictionary_key_size_threshold);
diff --git a/be/src/pipeline/exec/exchange_sink_buffer.cpp b/be/src/pipeline/exec/exchange_sink_buffer.cpp
index ed7f18bfcb7..2b97551d8fb 100644
--- a/be/src/pipeline/exec/exchange_sink_buffer.cpp
+++ b/be/src/pipeline/exec/exchange_sink_buffer.cpp
@@ -111,7 +111,8 @@ void ExchangeSinkBuffer<Parent>::close() {
template <typename Parent>
bool ExchangeSinkBuffer<Parent>::can_write() const {
- size_t max_package_size = QUEUE_CAPACITY_FACTOR * _instance_to_package_queue.size();
+ size_t max_package_size =
+ config::exchg_buffer_queue_capacity_factor * _instance_to_package_queue.size();
size_t total_package_size = 0;
for (auto& [_, q] : _instance_to_package_queue) {
total_package_size += q.size();
@@ -168,7 +169,8 @@ void ExchangeSinkBuffer<Parent>::register_sink(TUniqueId fragment_instance_id) {
std::queue<TransmitInfo<Parent>, std::list<TransmitInfo<Parent>>>();
_instance_to_broadcast_package_queue[low_id] =
std::queue<BroadcastTransmitInfo<Parent>, std::list<BroadcastTransmitInfo<Parent>>>();
- _queue_capacity = QUEUE_CAPACITY_FACTOR * _instance_to_package_queue.size();
+ _queue_capacity =
+ config::exchg_buffer_queue_capacity_factor * _instance_to_package_queue.size();
PUniqueId finst_id;
finst_id.set_hi(fragment_instance_id.hi);
finst_id.set_lo(fragment_instance_id.lo);
diff --git a/be/src/pipeline/exec/exchange_sink_buffer.h b/be/src/pipeline/exec/exchange_sink_buffer.h
index 0afa59bf731..8c0375499c3 100644
--- a/be/src/pipeline/exec/exchange_sink_buffer.h
+++ b/be/src/pipeline/exec/exchange_sink_buffer.h
@@ -270,7 +270,6 @@ private:
int64_t get_sum_rpc_time();
std::atomic<int> _total_queue_size = 0;
- static constexpr int QUEUE_CAPACITY_FACTOR = 64;
std::shared_ptr<Dependency> _queue_dependency;
std::shared_ptr<Dependency> _finish_dependency;
std::atomic<bool> _should_stop {false};
diff --git a/be/src/pipeline/exec/scan_operator.cpp b/be/src/pipeline/exec/scan_operator.cpp
index c10e7777bdb..8870ba619cb 100644
--- a/be/src/pipeline/exec/scan_operator.cpp
+++ b/be/src/pipeline/exec/scan_operator.cpp
@@ -1413,6 +1413,7 @@ Status ScanLocalState<Derived>::close(RuntimeState* state) {
if (_scanner_ctx) {
_scanner_ctx->stop_scanners(state);
}
+ std::list<std::shared_ptr<vectorized::ScannerDelegate>> {}.swap(_scanners);
COUNTER_SET(_wait_for_dependency_timer, _scan_dependency->watcher_elapse_time());
COUNTER_SET(_wait_for_rf_timer, _filter_dependency->watcher_elapse_time());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/11: [bugfix](profile) support multi execution profile for brokerload (#32280)
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 a90a1a76f168709d7695efaf1cbb4b49d2abc9ce
Author: yiguolei <67...@qq.com>
AuthorDate: Fri Mar 15 17:41:47 2024 +0800
[bugfix](profile) support multi execution profile for brokerload (#32280)
The bug is introduced by #27184
Profile Format is :
Summary
MergedProfile
ExecutionProfile1
ExecutionProfile2
...
There maybe multiple execution profiles for broker load.
---
.../doris/common/profile/AggregatedProfile.java | 41 -----
.../org/apache/doris/common/profile/Profile.java | 40 +++--
.../apache/doris/load/loadv2/LoadLoadingTask.java | 2 +-
.../trees/plans/commands/InsertExecutor.java | 2 +-
.../commands/insert/AbstractInsertExecutor.java | 194 +++++++++++++++++++++
.../java/org/apache/doris/qe/StmtExecutor.java | 6 +-
6 files changed, 222 insertions(+), 63 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/profile/AggregatedProfile.java b/fe/fe-core/src/main/java/org/apache/doris/common/profile/AggregatedProfile.java
deleted file mode 100644
index f8481b21e65..00000000000
--- a/fe/fe-core/src/main/java/org/apache/doris/common/profile/AggregatedProfile.java
+++ /dev/null
@@ -1,41 +0,0 @@
-// 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.common.profile;
-
-import org.apache.doris.common.util.RuntimeProfile;
-
-import java.util.Map;
-
-/**
-* AggregatedProfile is part of a query profile.
-* It contains the aggregated information of a query.
-*/
-public class AggregatedProfile {
-
- public static final String PROFILE_NAME = "MergedProfile";
- private ExecutionProfile executionProfile;
-
- public AggregatedProfile(RuntimeProfile rootProfile, ExecutionProfile executionProfile) {
- this.executionProfile = executionProfile;
- }
-
- public RuntimeProfile getAggregatedFragmentsProfile(Map<Integer, String> planNodeMap) {
- return executionProfile.getAggregatedFragmentsProfile(planNodeMap);
- }
-
-}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java b/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java
index 19a51a1aaa6..5f3ed601630 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/common/profile/Profile.java
@@ -21,11 +21,13 @@ import org.apache.doris.common.util.ProfileManager;
import org.apache.doris.common.util.RuntimeProfile;
import org.apache.doris.planner.Planner;
+import com.google.common.collect.Lists;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
+import java.util.List;
import java.util.Map;
/**
@@ -33,20 +35,23 @@ import java.util.Map;
* following structure: root profile: // summary of this profile, such as start
* time, end time, query id, etc. [SummaryProfile] // each execution profile is
* a complete execution of a query, a job may contain multiple queries.
- * [List<ExecutionProfile>]
+ * [List<ExecutionProfile>].
+ * There maybe multi execution profiles for one job, for example broker load job.
+ * It will create one execution profile for every single load task.
*
* SummaryProfile: Summary: Execution Summary:
*
*
- * ExecutionProfile: Fragment 0: Fragment 1: ...
+ * ExecutionProfile1: Fragment 0: Fragment 1: ...
+ * ExecutionProfile2: Fragment 0: Fragment 1: ...
+ *
*/
public class Profile {
private static final Logger LOG = LogManager.getLogger(Profile.class);
private static final int MergedProfileLevel = 1;
private RuntimeProfile rootProfile;
private SummaryProfile summaryProfile;
- private AggregatedProfile aggregatedProfile;
- private ExecutionProfile executionProfile;
+ private List<ExecutionProfile> executionProfiles = Lists.newArrayList();
private boolean isFinished;
private Map<Integer, String> planNodeMap;
@@ -59,14 +64,13 @@ public class Profile {
this.isFinished = !isEnable;
}
- public void setExecutionProfile(ExecutionProfile executionProfile) {
+ public void addExecutionProfile(ExecutionProfile executionProfile) {
if (executionProfile == null) {
LOG.warn("try to set a null excecution profile, it is abnormal", new Exception());
return;
}
- this.executionProfile = executionProfile;
- this.executionProfile.addToProfileAsChild(rootProfile);
- this.aggregatedProfile = new AggregatedProfile(rootProfile, executionProfile);
+ this.executionProfiles.add(executionProfile);
+ executionProfile.addToProfileAsChild(rootProfile);
}
public synchronized void update(long startTime, Map<String, String> summaryInfo, boolean isFinished,
@@ -75,12 +79,10 @@ public class Profile {
if (this.isFinished) {
return;
}
- if (executionProfile == null) {
- // Sometimes execution profile is not set
- return;
- }
summaryProfile.update(summaryInfo);
- executionProfile.update(startTime, isFinished);
+ for (ExecutionProfile executionProfile : executionProfiles) {
+ executionProfile.update(startTime, isFinished);
+ }
rootProfile.computeTimeInProfile();
// Nerids native insert not set planner, so it is null
if (planner != null) {
@@ -109,18 +111,22 @@ public class Profile {
// add summary to builder
summaryProfile.prettyPrint(builder);
LOG.info(builder.toString());
- if (this.profileLevel == MergedProfileLevel) {
+ // Only generate merged profile for select, insert into select.
+ // Not support broker load now.
+ if (this.profileLevel == MergedProfileLevel && this.executionProfiles.size() == 1) {
try {
builder.append("\n MergedProfile \n");
- aggregatedProfile.getAggregatedFragmentsProfile(planNodeMap).prettyPrint(builder, " ");
+ this.executionProfiles.get(0).getAggregatedFragmentsProfile(planNodeMap).prettyPrint(builder, " ");
} catch (Throwable aggProfileException) {
LOG.warn("build merged simple profile failed", aggProfileException);
builder.append("build merged simple profile failed");
}
}
try {
- builder.append("\n");
- executionProfile.getExecutionProfile().prettyPrint(builder, "");
+ for (ExecutionProfile executionProfile : executionProfiles) {
+ builder.append("\n");
+ executionProfile.getExecutionProfile().prettyPrint(builder, "");
+ }
} catch (Throwable aggProfileException) {
LOG.warn("build profile failed", aggProfileException);
builder.append("build profile failed");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
index 8fbabea8629..94fb49d6c85 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadLoadingTask.java
@@ -143,7 +143,7 @@ public class LoadLoadingTask extends LoadTask {
Coordinator curCoordinator = new Coordinator(callback.getCallbackId(), loadId, planner.getDescTable(),
planner.getFragments(), planner.getScanNodes(), planner.getTimezone(), loadZeroTolerance);
if (this.jobProfile != null) {
- this.jobProfile.setExecutionProfile(curCoordinator.getExecutionProfile());
+ this.jobProfile.addExecutionProfile(curCoordinator.getExecutionProfile());
}
curCoordinator.setQueryType(TQueryType.LOAD);
curCoordinator.setExecMemoryLimit(execMemLimit);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertExecutor.java
index da4a1c6ac68..295478349cf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/InsertExecutor.java
@@ -222,7 +222,7 @@ public class InsertExecutor {
try {
coordinator.setLoadZeroTolerance(ctx.getSessionVariable().getEnableInsertStrict());
coordinator.setQueryType(TQueryType.LOAD);
- executor.getProfile().setExecutionProfile(coordinator.getExecutionProfile());
+ executor.getProfile().addExecutionProfile(coordinator.getExecutionProfile());
QueryInfo queryInfo = new QueryInfo(ConnectContext.get(), executor.getOriginStmtInString(), coordinator);
QeProcessorImpl.INSTANCE.registerQuery(ctx.queryId(), queryInfo);
coordinator.exec();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
new file mode 100644
index 00000000000..77eeef22ebf
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
@@ -0,0 +1,194 @@
+// 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.trees.plans.commands.insert;
+
+import org.apache.doris.catalog.DatabaseIf;
+import org.apache.doris.catalog.EnvFactory;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.util.DebugUtil;
+import org.apache.doris.nereids.NereidsPlanner;
+import org.apache.doris.nereids.trees.plans.physical.PhysicalSink;
+import org.apache.doris.planner.DataSink;
+import org.apache.doris.planner.PlanFragment;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.Coordinator;
+import org.apache.doris.qe.QeProcessorImpl;
+import org.apache.doris.qe.QeProcessorImpl.QueryInfo;
+import org.apache.doris.qe.StmtExecutor;
+import org.apache.doris.task.LoadEtlTask;
+import org.apache.doris.thrift.TQueryType;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.Optional;
+
+/**
+ * Abstract insert executor.
+ * The derived class should implement the abstract method for certain type of target table
+ */
+public abstract class AbstractInsertExecutor {
+ private static final Logger LOG = LogManager.getLogger(AbstractInsertExecutor.class);
+ protected long jobId;
+ protected final ConnectContext ctx;
+ protected final Coordinator coordinator;
+ protected String labelName;
+ protected final DatabaseIf database;
+ protected final TableIf table;
+ protected final long createTime = System.currentTimeMillis();
+ protected long loadedRows = 0;
+ protected int filteredRows = 0;
+
+ protected String errMsg = "";
+ protected Optional<InsertCommandContext> insertCtx;
+
+ /**
+ * Constructor
+ */
+ public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String labelName, NereidsPlanner planner,
+ Optional<InsertCommandContext> insertCtx) {
+ this.ctx = ctx;
+ this.coordinator = EnvFactory.getInstance().createCoordinator(ctx, null, planner, ctx.getStatsErrorEstimator());
+ this.labelName = labelName;
+ this.table = table;
+ this.database = table.getDatabase();
+ this.insertCtx = insertCtx;
+ }
+
+ public Coordinator getCoordinator() {
+ return coordinator;
+ }
+
+ public DatabaseIf getDatabase() {
+ return database;
+ }
+
+ public TableIf getTable() {
+ return table;
+ }
+
+ public String getLabelName() {
+ return labelName;
+ }
+
+ /**
+ * begin transaction if necessary
+ */
+ public abstract void beginTransaction();
+
+ /**
+ * finalize sink to complete enough info for sink execution
+ */
+ protected abstract void finalizeSink(PlanFragment fragment, DataSink sink, PhysicalSink physicalSink);
+
+ /**
+ * Do something before exec
+ */
+ protected abstract void beforeExec();
+
+ /**
+ * Do something after exec finished
+ */
+ protected abstract void onComplete() throws UserException;
+
+ /**
+ * Do something when exec throw exception
+ */
+ protected abstract void onFail(Throwable t);
+
+ /**
+ * Do something after exec
+ */
+ protected abstract void afterExec(StmtExecutor executor);
+
+ protected final void execImpl(StmtExecutor executor, long jobId) throws Exception {
+ String queryId = DebugUtil.printId(ctx.queryId());
+ coordinator.setLoadZeroTolerance(ctx.getSessionVariable().getEnableInsertStrict());
+ coordinator.setQueryType(TQueryType.LOAD);
+ executor.getProfile().addExecutionProfile(coordinator.getExecutionProfile());
+ QueryInfo queryInfo = new QueryInfo(ConnectContext.get(), executor.getOriginStmtInString(), coordinator);
+ QeProcessorImpl.INSTANCE.registerQuery(ctx.queryId(), queryInfo);
+ coordinator.exec();
+ int execTimeout = ctx.getExecTimeout();
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("insert [{}] with query id {} execution timeout is {}", labelName, queryId, execTimeout);
+ }
+ boolean notTimeout = coordinator.join(execTimeout);
+ if (!coordinator.isDone()) {
+ coordinator.cancel();
+ if (notTimeout) {
+ errMsg = coordinator.getExecStatus().getErrorMsg();
+ ErrorReport.reportDdlException("there exists unhealthy backend. "
+ + errMsg, ErrorCode.ERR_FAILED_WHEN_INSERT);
+ } else {
+ ErrorReport.reportDdlException(ErrorCode.ERR_EXECUTE_TIMEOUT);
+ }
+ }
+ if (!coordinator.getExecStatus().ok()) {
+ errMsg = coordinator.getExecStatus().getErrorMsg();
+ LOG.warn("insert [{}] with query id {} failed, {}", labelName, queryId, errMsg);
+ ErrorReport.reportDdlException(errMsg, ErrorCode.ERR_FAILED_WHEN_INSERT);
+ }
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("insert [{}] with query id {} delta files is {}",
+ labelName, queryId, coordinator.getDeltaUrls());
+ }
+ if (coordinator.getLoadCounters().get(LoadEtlTask.DPP_NORMAL_ALL) != null) {
+ loadedRows = Long.parseLong(coordinator.getLoadCounters().get(LoadEtlTask.DPP_NORMAL_ALL));
+ }
+ if (coordinator.getLoadCounters().get(LoadEtlTask.DPP_ABNORMAL_ALL) != null) {
+ filteredRows = Integer.parseInt(coordinator.getLoadCounters().get(LoadEtlTask.DPP_ABNORMAL_ALL));
+ }
+ }
+
+ private boolean checkStrictMode() {
+ // if in strict mode, insert will fail if there are filtered rows
+ if (ctx.getSessionVariable().getEnableInsertStrict()) {
+ if (filteredRows > 0) {
+ ctx.getState().setError(ErrorCode.ERR_FAILED_WHEN_INSERT,
+ "Insert has filtered data in strict mode, tracking_url=" + coordinator.getTrackingUrl());
+ return false;
+ }
+ }
+ return true;
+ }
+
+ /**
+ * execute insert txn for insert into select command.
+ */
+ public void executeSingleInsert(StmtExecutor executor, long jobId) {
+ beforeExec();
+ try {
+ execImpl(executor, jobId);
+ if (!checkStrictMode()) {
+ return;
+ }
+ onComplete();
+ } catch (Throwable t) {
+ onFail(t);
+ return;
+ } finally {
+ executor.updateProfile(true);
+ QeProcessorImpl.INSTANCE.unregisterQuery(ctx.queryId());
+ }
+ afterExec(executor);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
index 063cbf0ad71..b6a8c882fae 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
@@ -1576,7 +1576,7 @@ public class StmtExecutor {
coord = new Coordinator(context, analyzer, planner, context.getStatsErrorEstimator());
QeProcessorImpl.INSTANCE.registerQuery(context.queryId(),
new QeProcessorImpl.QueryInfo(context, originStmt.originStmt, coord));
- profile.setExecutionProfile(coord.getExecutionProfile());
+ profile.addExecutionProfile(coord.getExecutionProfile());
coordBase = coord;
}
@@ -2053,7 +2053,7 @@ public class StmtExecutor {
coord = new Coordinator(context, analyzer, planner, context.getStatsErrorEstimator());
coord.setLoadZeroTolerance(context.getSessionVariable().getEnableInsertStrict());
coord.setQueryType(TQueryType.LOAD);
- profile.setExecutionProfile(coord.getExecutionProfile());
+ profile.addExecutionProfile(coord.getExecutionProfile());
QueryInfo queryInfo = new QueryInfo(ConnectContext.get(), this.getOriginStmtInString(), coord);
QeProcessorImpl.INSTANCE.registerQuery(context.queryId(), queryInfo);
@@ -2875,7 +2875,7 @@ public class StmtExecutor {
}
RowBatch batch;
coord = new Coordinator(context, analyzer, planner, context.getStatsErrorEstimator());
- profile.setExecutionProfile(coord.getExecutionProfile());
+ profile.addExecutionProfile(coord.getExecutionProfile());
try {
QeProcessorImpl.INSTANCE.registerQuery(context.queryId(),
new QeProcessorImpl.QueryInfo(context, originStmt.originStmt, coord));
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 03/11: [fix](retry) Set query encounter rpc exception default retry times to 3 (#28555)
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 844a1b53b74bebd3a2dc2384071e778ef7a2c793
Author: deardeng <56...@qq.com>
AuthorDate: Fri Mar 15 19:38:38 2024 +0800
[fix](retry) Set query encounter rpc exception default retry times to 3 (#28555)
---
fe/fe-common/src/main/java/org/apache/doris/common/Config.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
index 4dd10de1cea..2c5a36e9f85 100644
--- a/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
+++ b/fe/fe-common/src/main/java/org/apache/doris/common/Config.java
@@ -911,7 +911,7 @@ public class Config extends ConfigBase {
* You may reduce this number to avoid Avalanche disaster.
*/
@ConfField(mutable = true)
- public static int max_query_retry_time = 1;
+ public static int max_query_retry_time = 3;
/**
* The number of point query retries in executor.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 05/11: [Fix](nereids) remove duplicate expr in grouping set (#32290)
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 a15bf3057f446aea49be8f1570b8cb593f56c5b7
Author: feiniaofeiafei <53...@users.noreply.github.com>
AuthorDate: Fri Mar 15 21:36:16 2024 +0800
[Fix](nereids) remove duplicate expr in grouping set (#32290)
db reported a error " expression duplicate in grouping set" when there are duplicate expression in grouping set.
e.g.select a from mal_test1 group by grouping sets((a,a))
This pr removes duplicate expr in grouping set :
select a from mal_test1 group by grouping sets((a))
---
.../nereids/rules/analysis/NormalizeRepeat.java | 11 +++++++
.../doris/nereids/trees/plans/algebra/Repeat.java | 3 --
.../nereids/trees/plans/logical/LogicalRepeat.java | 4 +++
.../remove_duplicate_expr_in_grouping_set.out | 19 +++++++++++
.../remove_duplicate_expr_in_grouping_set.groovy | 37 ++++++++++++++++++++++
5 files changed, 71 insertions(+), 3 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeRepeat.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeRepeat.java
index 9451cc40bc2..e90528a9d4a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeRepeat.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/analysis/NormalizeRepeat.java
@@ -84,6 +84,7 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
return RuleType.NORMALIZE_REPEAT.build(
logicalRepeat(any()).when(LogicalRepeat::canBindVirtualSlot).then(repeat -> {
checkRepeatLegality(repeat);
+ repeat = removeDuplicateColumns(repeat);
// add virtual slot, LogicalAggregate and LogicalProject for normalize
LogicalAggregate<Plan> agg = normalizeRepeat(repeat);
return dealSlotAppearBothInAggFuncAndGroupingSets(agg);
@@ -91,6 +92,16 @@ public class NormalizeRepeat extends OneAnalysisRuleFactory {
);
}
+ private LogicalRepeat<Plan> removeDuplicateColumns(LogicalRepeat<Plan> repeat) {
+ List<List<Expression>> groupingSets = repeat.getGroupingSets();
+ ImmutableList.Builder<List<Expression>> builder = ImmutableList.builder();
+ for (List<Expression> sets : groupingSets) {
+ List<Expression> newList = ImmutableList.copyOf(ImmutableSet.copyOf(sets));
+ builder.add(newList);
+ }
+ return repeat.withGroupSets(builder.build());
+ }
+
private void checkRepeatLegality(LogicalRepeat<Plan> repeat) {
checkGroupingSetsSize(repeat);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/Repeat.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/Repeat.java
index 29258b6bbab..e729f2a7cb3 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/Repeat.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/algebra/Repeat.java
@@ -175,9 +175,6 @@ public interface Repeat<CHILD_PLAN extends Plan> extends Aggregate<CHILD_PLAN> {
if (index == null) {
throw new AnalysisException("Can not find grouping set expression in output: " + expression);
}
- if (groupingSetIndex.contains(index)) {
- throw new AnalysisException("expression duplicate in grouping set: " + expression);
- }
groupingSetIndex.add(index);
}
groupingSetsIndex.add(groupingSetIndex);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalRepeat.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalRepeat.java
index 1c8a468b44b..5e0311afe1b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalRepeat.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/logical/LogicalRepeat.java
@@ -158,6 +158,10 @@ public class LogicalRepeat<CHILD_TYPE extends Plan> extends LogicalUnary<CHILD_T
children.get(0));
}
+ public LogicalRepeat<CHILD_TYPE> withGroupSets(List<List<Expression>> groupingSets) {
+ return new LogicalRepeat<>(groupingSets, outputExpressions, child());
+ }
+
public LogicalRepeat<CHILD_TYPE> withGroupSetsAndOutput(List<List<Expression>> groupingSets,
List<NamedExpression> outputExpressionList) {
return new LogicalRepeat<>(groupingSets, outputExpressionList, child());
diff --git a/regression-test/data/nereids_rules_p0/grouping_sets/remove_duplicate_expr_in_grouping_set.out b/regression-test/data/nereids_rules_p0/grouping_sets/remove_duplicate_expr_in_grouping_set.out
new file mode 100644
index 00000000000..5956255d7bd
--- /dev/null
+++ b/regression-test/data/nereids_rules_p0/grouping_sets/remove_duplicate_expr_in_grouping_set.out
@@ -0,0 +1,19 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !test_col --
+\N 6
+1 20
+2 8
+3 20
+4 2
+5 26
+7 1
+
+-- !test_expr --
+\N 6
+2 20
+3 8
+4 20
+5 2
+6 26
+8 1
+
diff --git a/regression-test/suites/nereids_rules_p0/grouping_sets/remove_duplicate_expr_in_grouping_set.groovy b/regression-test/suites/nereids_rules_p0/grouping_sets/remove_duplicate_expr_in_grouping_set.groovy
new file mode 100644
index 00000000000..b9a10fe21d2
--- /dev/null
+++ b/regression-test/suites/nereids_rules_p0/grouping_sets/remove_duplicate_expr_in_grouping_set.groovy
@@ -0,0 +1,37 @@
+// 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("remove_duplicate_expr_in_grouping_set") {
+ sql "SET enable_nereids_planner=true"
+ sql "SET enable_fallback_to_original_planner=false"
+ sql """
+ DROP TABLE IF EXISTS mal_test1
+ """
+
+ sql """
+ create table mal_test1(pk int, a int, b int) distributed by hash(pk) buckets 10
+ properties('replication_num' = '1');
+ """
+
+ sql """
+ insert into mal_test1 values(2,1,3),(1,1,2),(3,5,6),(6,null,6),(4,5,6),(2,1,4),(2,3,5),(1,1,4)
+ ,(3,5,6),(3,5,null),(6,7,1),(2,1,7),(2,4,2),(2,3,9),(1,3,6),(3,5,8),(3,2,8);
+ """
+ sql "sync"
+ qt_test_col "select a, sum(b) from mal_test1 group by grouping sets((a,a)) order by 1,2"
+ qt_test_expr "select a+1,sum(b) from mal_test1 group by grouping sets((a+1,a+1)) order by 1,2"
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 02/11: [fix](Nereids): don't pushdown project when project contains both side of join (#32214)
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 f64a9a33f8e28d8f13af476cb8b508fd35067203
Author: 谢健 <ji...@gmail.com>
AuthorDate: Fri Mar 15 19:21:26 2024 +0800
[fix](Nereids): don't pushdown project when project contains both side of join (#32214)
---
.../join/PushDownProjectThroughSemiJoin.java | 19 +++++++++++++
.../join/PushDownProjectThroughSemiJoinTest.java | 32 ++++++++++++++++++++++
2 files changed, 51 insertions(+)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/join/PushDownProjectThroughSemiJoin.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/join/PushDownProjectThroughSemiJoin.java
index 0629092f451..66735d39b98 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/join/PushDownProjectThroughSemiJoin.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/exploration/join/PushDownProjectThroughSemiJoin.java
@@ -38,6 +38,7 @@ import java.util.stream.Collectors;
/**
* Rule for pushdown project through left-semi/anti join
* Just push down project inside join to avoid to push the top of Join-Cluster.
+ * Note this rule is only used to push down project between join for join ordering.
* <pre>
* Join Join
* | |
@@ -61,6 +62,9 @@ public class PushDownProjectThroughSemiJoin implements ExplorationRuleFactory {
.whenNot(j -> j.left().child().hasDistributeHint())
.then(topJoin -> {
LogicalProject<LogicalJoin<GroupPlan, GroupPlan>> project = topJoin.left();
+ if (projectBothJoinSide(project)) {
+ return null;
+ }
Plan newLeft = pushdownProject(project);
return topJoin.withChildren(newLeft, topJoin.right());
}).toRule(RuleType.PUSH_DOWN_PROJECT_THROUGH_SEMI_JOIN_LEFT),
@@ -72,12 +76,27 @@ public class PushDownProjectThroughSemiJoin implements ExplorationRuleFactory {
.whenNot(j -> j.right().child().hasDistributeHint())
.then(topJoin -> {
LogicalProject<LogicalJoin<GroupPlan, GroupPlan>> project = topJoin.right();
+ if (projectBothJoinSide(project)) {
+ return null;
+ }
Plan newRight = pushdownProject(project);
return topJoin.withChildren(topJoin.left(), newRight);
}).toRule(RuleType.PUSH_DOWN_PROJECT_THROUGH_SEMI_JOIN_RIGHT)
);
}
+ private boolean projectBothJoinSide(LogicalProject<LogicalJoin<GroupPlan, GroupPlan>> project) {
+ // if project contains both side of join, it can't be pushed.
+ // such as:
+ // Project(l, null as r)
+ // ------ L(l) left anti join R(r)
+ LogicalJoin<?, ?> join = project.child();
+ Set<Slot> projectOutput = project.getOutputSet();
+ boolean containLeft = join.left().getOutput().stream().anyMatch(projectOutput::contains);
+ boolean containRight = join.right().getOutput().stream().anyMatch(projectOutput::contains);
+ return containRight && containLeft;
+ }
+
private Plan pushdownProject(LogicalProject<LogicalJoin<GroupPlan, GroupPlan>> project) {
LogicalJoin<GroupPlan, GroupPlan> join = project.child();
Set<Slot> conditionLeftSlots = CBOUtils.joinChildConditionSlots(join, true);
diff --git a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/join/PushDownProjectThroughSemiJoinTest.java b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/join/PushDownProjectThroughSemiJoinTest.java
index 8c5a29f0d55..bccd3056d35 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/join/PushDownProjectThroughSemiJoinTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/nereids/rules/exploration/join/PushDownProjectThroughSemiJoinTest.java
@@ -22,6 +22,7 @@ import org.apache.doris.nereids.trees.expressions.Add;
import org.apache.doris.nereids.trees.expressions.Alias;
import org.apache.doris.nereids.trees.expressions.NamedExpression;
import org.apache.doris.nereids.trees.expressions.literal.Literal;
+import org.apache.doris.nereids.trees.expressions.literal.NullLiteral;
import org.apache.doris.nereids.trees.plans.JoinType;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalPlan;
@@ -133,4 +134,35 @@ class PushDownProjectThroughSemiJoinTest implements MemoPatternMatchSupported {
)
);
}
+
+ @Test
+ void testProjectLiteral() {
+ List<NamedExpression> projectExprs = ImmutableList.of(
+ new Alias(new Add(scan1.getOutput().get(0), Literal.of(1)), "alias"),
+ new Alias(scan2.getOutput().get(0).getExprId(), new NullLiteral(), scan2.getOutput().get(0).getName())
+ );
+ // complex projection contain ti.id, which isn't in Join Condition
+ LogicalPlan plan = new LogicalPlanBuilder(scan1)
+ .join(scan2, JoinType.LEFT_SEMI_JOIN, Pair.of(1, 1))
+ .projectExprs(projectExprs)
+ .join(scan3, JoinType.INNER_JOIN, Pair.of(1, 1))
+ .build();
+ PlanChecker.from(MemoTestUtils.createConnectContext(), plan)
+ .applyExploration(PushDownProjectThroughSemiJoin.INSTANCE.buildRules())
+ .nonMatch(logicalJoin(logicalJoin(logicalProject(), group()), group()));
+
+ projectExprs = ImmutableList.of(
+ new Alias(new Add(scan2.getOutput().get(0), Literal.of(1)), "alias"),
+ new Alias(scan1.getOutput().get(0).getExprId(), new NullLiteral(), scan2.getOutput().get(0).getName())
+ );
+ // complex projection contain ti.id, which isn't in Join Condition
+ plan = new LogicalPlanBuilder(scan1)
+ .join(scan2, JoinType.RIGHT_SEMI_JOIN, Pair.of(1, 1))
+ .projectExprs(projectExprs)
+ .join(scan3, JoinType.INNER_JOIN, Pair.of(1, 1))
+ .build();
+ PlanChecker.from(MemoTestUtils.createConnectContext(), plan)
+ .applyExploration(PushDownProjectThroughSemiJoin.INSTANCE.buildRules())
+ .nonMatch(logicalJoin(logicalJoin(logicalProject(), group()), group()));
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 04/11: [Chore](FE)Remove unused components (#32295)
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 bf820302707d10efa2d2b46cb376b49b05cb8592
Author: Calvin Kirs <ki...@apache.org>
AuthorDate: Fri Mar 15 21:01:30 2024 +0800
[Chore](FE)Remove unused components (#32295)
The tomcat-embed-el dependency is primarily used for standardizing EL functionality, which we don't require in our application. Therefore, we can safely remove it.
---
fe/pom.xml | 10 ++++++++++
1 file changed, 10 insertions(+)
diff --git a/fe/pom.xml b/fe/pom.xml
index e1d4f44d752..d96d4697880 100644
--- a/fe/pom.xml
+++ b/fe/pom.xml
@@ -467,6 +467,12 @@ under the License.
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-jetty</artifactId>
<version>${spring.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>org.apache.tomcat</groupId>
+ <artifactId>tomcat-embed-el</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
@@ -478,6 +484,10 @@ under the License.
<artifactId>spring-boot-starter-web</artifactId>
<version>${spring.version}</version>
<exclusions>
+ <exclusion>
+ <groupId>org.springframework.boot</groupId>
+ <artifactId>spring-boot-starter-tomcat</artifactId>
+ </exclusion>
<exclusion>
<groupId>validator</groupId>
<artifactId>hibernate-validator</artifactId>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 11/11: fix compile
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 c34f5045c8835412d80eb4e67d3a79242e0cddc7
Author: yiguolei <yi...@gmail.com>
AuthorDate: Sat Mar 16 21:37:02 2024 +0800
fix compile
---
.../commands/insert/AbstractInsertExecutor.java | 194 ---------------------
1 file changed, 194 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
deleted file mode 100644
index 77eeef22ebf..00000000000
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/insert/AbstractInsertExecutor.java
+++ /dev/null
@@ -1,194 +0,0 @@
-// 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.trees.plans.commands.insert;
-
-import org.apache.doris.catalog.DatabaseIf;
-import org.apache.doris.catalog.EnvFactory;
-import org.apache.doris.catalog.TableIf;
-import org.apache.doris.common.ErrorCode;
-import org.apache.doris.common.ErrorReport;
-import org.apache.doris.common.UserException;
-import org.apache.doris.common.util.DebugUtil;
-import org.apache.doris.nereids.NereidsPlanner;
-import org.apache.doris.nereids.trees.plans.physical.PhysicalSink;
-import org.apache.doris.planner.DataSink;
-import org.apache.doris.planner.PlanFragment;
-import org.apache.doris.qe.ConnectContext;
-import org.apache.doris.qe.Coordinator;
-import org.apache.doris.qe.QeProcessorImpl;
-import org.apache.doris.qe.QeProcessorImpl.QueryInfo;
-import org.apache.doris.qe.StmtExecutor;
-import org.apache.doris.task.LoadEtlTask;
-import org.apache.doris.thrift.TQueryType;
-
-import org.apache.logging.log4j.LogManager;
-import org.apache.logging.log4j.Logger;
-
-import java.util.Optional;
-
-/**
- * Abstract insert executor.
- * The derived class should implement the abstract method for certain type of target table
- */
-public abstract class AbstractInsertExecutor {
- private static final Logger LOG = LogManager.getLogger(AbstractInsertExecutor.class);
- protected long jobId;
- protected final ConnectContext ctx;
- protected final Coordinator coordinator;
- protected String labelName;
- protected final DatabaseIf database;
- protected final TableIf table;
- protected final long createTime = System.currentTimeMillis();
- protected long loadedRows = 0;
- protected int filteredRows = 0;
-
- protected String errMsg = "";
- protected Optional<InsertCommandContext> insertCtx;
-
- /**
- * Constructor
- */
- public AbstractInsertExecutor(ConnectContext ctx, TableIf table, String labelName, NereidsPlanner planner,
- Optional<InsertCommandContext> insertCtx) {
- this.ctx = ctx;
- this.coordinator = EnvFactory.getInstance().createCoordinator(ctx, null, planner, ctx.getStatsErrorEstimator());
- this.labelName = labelName;
- this.table = table;
- this.database = table.getDatabase();
- this.insertCtx = insertCtx;
- }
-
- public Coordinator getCoordinator() {
- return coordinator;
- }
-
- public DatabaseIf getDatabase() {
- return database;
- }
-
- public TableIf getTable() {
- return table;
- }
-
- public String getLabelName() {
- return labelName;
- }
-
- /**
- * begin transaction if necessary
- */
- public abstract void beginTransaction();
-
- /**
- * finalize sink to complete enough info for sink execution
- */
- protected abstract void finalizeSink(PlanFragment fragment, DataSink sink, PhysicalSink physicalSink);
-
- /**
- * Do something before exec
- */
- protected abstract void beforeExec();
-
- /**
- * Do something after exec finished
- */
- protected abstract void onComplete() throws UserException;
-
- /**
- * Do something when exec throw exception
- */
- protected abstract void onFail(Throwable t);
-
- /**
- * Do something after exec
- */
- protected abstract void afterExec(StmtExecutor executor);
-
- protected final void execImpl(StmtExecutor executor, long jobId) throws Exception {
- String queryId = DebugUtil.printId(ctx.queryId());
- coordinator.setLoadZeroTolerance(ctx.getSessionVariable().getEnableInsertStrict());
- coordinator.setQueryType(TQueryType.LOAD);
- executor.getProfile().addExecutionProfile(coordinator.getExecutionProfile());
- QueryInfo queryInfo = new QueryInfo(ConnectContext.get(), executor.getOriginStmtInString(), coordinator);
- QeProcessorImpl.INSTANCE.registerQuery(ctx.queryId(), queryInfo);
- coordinator.exec();
- int execTimeout = ctx.getExecTimeout();
- if (LOG.isDebugEnabled()) {
- LOG.debug("insert [{}] with query id {} execution timeout is {}", labelName, queryId, execTimeout);
- }
- boolean notTimeout = coordinator.join(execTimeout);
- if (!coordinator.isDone()) {
- coordinator.cancel();
- if (notTimeout) {
- errMsg = coordinator.getExecStatus().getErrorMsg();
- ErrorReport.reportDdlException("there exists unhealthy backend. "
- + errMsg, ErrorCode.ERR_FAILED_WHEN_INSERT);
- } else {
- ErrorReport.reportDdlException(ErrorCode.ERR_EXECUTE_TIMEOUT);
- }
- }
- if (!coordinator.getExecStatus().ok()) {
- errMsg = coordinator.getExecStatus().getErrorMsg();
- LOG.warn("insert [{}] with query id {} failed, {}", labelName, queryId, errMsg);
- ErrorReport.reportDdlException(errMsg, ErrorCode.ERR_FAILED_WHEN_INSERT);
- }
- if (LOG.isDebugEnabled()) {
- LOG.debug("insert [{}] with query id {} delta files is {}",
- labelName, queryId, coordinator.getDeltaUrls());
- }
- if (coordinator.getLoadCounters().get(LoadEtlTask.DPP_NORMAL_ALL) != null) {
- loadedRows = Long.parseLong(coordinator.getLoadCounters().get(LoadEtlTask.DPP_NORMAL_ALL));
- }
- if (coordinator.getLoadCounters().get(LoadEtlTask.DPP_ABNORMAL_ALL) != null) {
- filteredRows = Integer.parseInt(coordinator.getLoadCounters().get(LoadEtlTask.DPP_ABNORMAL_ALL));
- }
- }
-
- private boolean checkStrictMode() {
- // if in strict mode, insert will fail if there are filtered rows
- if (ctx.getSessionVariable().getEnableInsertStrict()) {
- if (filteredRows > 0) {
- ctx.getState().setError(ErrorCode.ERR_FAILED_WHEN_INSERT,
- "Insert has filtered data in strict mode, tracking_url=" + coordinator.getTrackingUrl());
- return false;
- }
- }
- return true;
- }
-
- /**
- * execute insert txn for insert into select command.
- */
- public void executeSingleInsert(StmtExecutor executor, long jobId) {
- beforeExec();
- try {
- execImpl(executor, jobId);
- if (!checkStrictMode()) {
- return;
- }
- onComplete();
- } catch (Throwable t) {
- onFail(t);
- return;
- } finally {
- executor.updateProfile(true);
- QeProcessorImpl.INSTANCE.unregisterQuery(ctx.queryId());
- }
- afterExec(executor);
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 08/11: [fix](compatibility) should enable windown funnel mode from 2.0 (#32284)
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 5ceccb5ba5dc01856ae6790d9f5762d40679fefe
Author: Jerry Hu <mr...@gmail.com>
AuthorDate: Sat Mar 16 08:54:33 2024 +0800
[fix](compatibility) should enable windown funnel mode from 2.0 (#32284)
---
be/src/agent/be_exec_version_manager.h | 3 ++-
be/src/vec/aggregate_functions/aggregate_function_window_funnel.h | 4 ++--
2 files changed, 4 insertions(+), 3 deletions(-)
diff --git a/be/src/agent/be_exec_version_manager.h b/be/src/agent/be_exec_version_manager.h
index 0290e3bee32..470d78ee981 100644
--- a/be/src/agent/be_exec_version_manager.h
+++ b/be/src/agent/be_exec_version_manager.h
@@ -60,8 +60,9 @@ private:
* e. add repeat_max_num in repeat function
* 3: start from doris 2.0 (by some mistakes)
* a. aggregation function do not serialize bitmap to string.
+ * b. support window funnel mode.
* 4: start from doris 2.1
- * a. support window funnel mode from 2.0
+ * a. ignore this line, window funnel mode should be enabled from 2.0.
* b. array contains/position/countequal function return nullable in less situations.
* c. cleared old version of Version 2.
* d. unix_timestamp function support timestamp with float for datetimev2, and change nullable mode.
diff --git a/be/src/vec/aggregate_functions/aggregate_function_window_funnel.h b/be/src/vec/aggregate_functions/aggregate_function_window_funnel.h
index ed4c9aed059..9356cfd4b68 100644
--- a/be/src/vec/aggregate_functions/aggregate_function_window_funnel.h
+++ b/be/src/vec/aggregate_functions/aggregate_function_window_funnel.h
@@ -270,8 +270,8 @@ public:
void create(AggregateDataPtr __restrict place) const override {
auto data = new (place) WindowFunnelState<DateValueType, NativeType>();
- /// support window funnel mode from 2.1. See `BeExecVersionManager::max_be_exec_version`
- data->enable_mode = version >= USE_NEW_SERDE;
+ /// support window funnel mode from 2.0. See `BeExecVersionManager::max_be_exec_version`
+ data->enable_mode = version >= 3;
}
String get_name() const override { return "window_funnel"; }
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 09/11: [docs](docs) Fix invalid link and typo of Master Branch (#32213)
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 bf9332a275975cc6d9f386c6a3ab4716cbbbb682
Author: KassieZ <13...@users.noreply.github.com>
AuthorDate: Sat Mar 16 09:03:35 2024 +0800
[docs](docs) Fix invalid link and typo of Master Branch (#32213)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org