You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by li...@apache.org on 2024/03/12 05:28:26 UTC
(doris) branch high-priority-column updated (c82acf03157 -> 5944b418dfa)
This is an automated email from the ASF dual-hosted git repository.
lijibing pushed a change to branch high-priority-column
in repository https://gitbox.apache.org/repos/asf/doris.git
discard c82acf03157 support window (#32094)
omit 87494cbb462 Check column health value earlier, show job priority. (#32064)
omit 5155651c79a Support show auto analyze pending jobs. (#31926)
omit f5c037514b2 Support follower sync query columns to master. (#31859)
omit efe9517d960 Support column level health value. (#31794)
omit 624b27f8258 High priority queue and map. (#31509)
omit ec3657244f0 Collect high priority columns. (#31235)
add 1303c9b9b1a [feature](pipelineX) add mem control in local exchange sink (#31982)
add 917e4259d99 [improvement](binlog)Support inverted index in CCR (#31743)
add ad472eb4843 [test](fix) Fix one missing line of output in out file (#32036)
add b7b38434413 [fix](compile) fix BE compile failure on Mac (#32044)
add b14880664bc [fix](cloud) ignore set replica version in cloud mode (#32040)
add afa783fe032 [chore](config) support select experimental session variable (#31837)
add 01f23dd3997 [feature](Cloud) Support restore HDFS meta data in Meta Service && Add storage vault property (#31415)
add 635463ab1f2 [test](cloud)add plugin for test run cloud mode cluster cases (#31960)
add d4be56bb76f [Feature](TrinoConnector-Catalog) Be supports trino-connector catalog (#31879)
add 9c579f30f18 [fix](pipelineX) _local_channel_dependency is null in non pipelineX (#32054)
add 7edbffb7f1c [docs](docs) Update 2.1.0 release note (#32053)
add a79e5208d70 [feature](docker)Doris On Docker All In One (#31735)
add 3bb5f665b4d [Improve](InPredict) enhance in predict with array type (#31828)
add 4a18b3bd368 [thirdparty](lib) Add Patch for base64 lib (#32059)
add 023313d00b8 [fix](memory) Fix Allocator cancel pipelinex query #32048
add cd59192b07f [fix](hive) fix spelling mistakes for "separatorChar" #32061
add d68535fc70d [fix](case) fix export data consistency table key type (#32045)
add 6817141d14f [Fix](Outfiel) fix be core when the open method of vfile_result_writer failed #32042
add f04e1a47175 [improvement](docker)Optimize Dockerfile content for FE and BE (#32027)
add 3c600a056a3 [fix](docker)Fix the registration logic of Doris On Docker (#32026)
add f8a9b025479 [fix](spill) fix storage engine type cast error (#32071)
add 5d4a949840b [Improvement](executor)Routine load support workload group #31671
add b54143fdd7d [fix](merge-cloud) Fix s3 load replay after restart fe (#32073)
add 34d49c147c9 [refactor](nereids) rename PlanNode.distributeExprLists to childrenDistributeExprLists #32069
add 5754b6757eb [doc](sql-dialect) update sql convertor tool to 1.0.1 (#32077)
add b64a8a0652f [DOC](Variant) add bloom filter description and correct some benchmark results (#31998)
add 15dfa831067 [fix](doc) fix wrong download link (#32082)
add cf53b2ea22c [enhance](Cooldown) Skip cooldown if the tablet is dropped (#32079)
add db2f2608a1d [pipelineX](scanner) Use the actual instances num when ignore data distribution (#32081)
add 7f665f381ac [docs](docs) Update release note 2.0.6 and readme of Master branch (#32087)
new a816e7c646c Collect high priority columns. (#31235)
new 404d160e81e High priority queue and map. (#31509)
new 9a3470cbb4d Support column level health value. (#31794)
new 27f92e518c5 Support follower sync query columns to master. (#31859)
new f21a600b7ee Support show auto analyze pending jobs. (#31926)
new 8d62e3d55bb Check column health value earlier, show job priority. (#32064)
new 5944b418dfa support window (#32094)
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (c82acf03157)
\
N -- N -- N refs/heads/high-priority-column (5944b418dfa)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 7 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:
README.md | 9 +-
be/src/common/config.cpp | 2 +
be/src/common/config.h | 3 +
be/src/http/action/download_binlog_action.cpp | 39 +
be/src/olap/rowset/beta_rowset.cpp | 44 +-
be/src/olap/snapshot_manager.cpp | 53 +-
be/src/olap/tablet.cpp | 30 +-
be/src/olap/tablet.h | 5 +
be/src/olap/tablet_manager.cpp | 17 +-
be/src/olap/task/engine_clone_task.cpp | 14 +-
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/service/backend_service.cpp | 132 ++-
be/src/util/jni-util.cpp | 1 +
be/src/vec/common/allocator.cpp | 4 +-
.../format/table/trino_connector_jni_reader.cpp | 133 +++
.../exec/format/table/trino_connector_jni_reader.h | 71 ++
be/src/vec/exec/jni_connector.cpp | 5 +
be/src/vec/exec/scan/scanner_context.cpp | 3 +-
be/src/vec/exec/scan/vfile_scanner.cpp | 7 +
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 +-
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 +-
build.sh | 8 +-
cloud/src/meta-service/keys.cpp | 11 +
cloud/src/meta-service/keys.h | 6 +
cloud/src/meta-service/meta_service_resource.cpp | 311 +++++--
cloud/src/recycler/recycler.cpp | 5 +
cloud/test/meta_service_test.cpp | 162 +++-
conf/be.conf | 2 +-
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 +-
docs/en/docs/releasenotes/release-1.2.8.md | 47 ++
docs/en/docs/releasenotes/release-2.0.6.md | 59 ++
docs/en/docs/releasenotes/release-2.1.0.md | 159 ++++
.../sql-manual/sql-reference/Data-Types/VARIANT.md | 22 +-
docs/images/2.1-Doris-TPC-DS-best-performance.png | Bin 0 -> 82015 bytes
docs/images/2.1-comparied-to-Json-2.png | Bin 0 -> 149145 bytes
docs/images/2.1-comparied-to-Json.png | Bin 0 -> 40676 bytes
docs/images/2.1-doris-INSERT-INTO-SELECT.png | Bin 0 -> 29842 bytes
docs/images/2.1-doris-TPC-DS.png | Bin 0 -> 18176 bytes
docs/images/2.1-doris-arrow-flight-sql.png | Bin 0 -> 64781 bytes
docs/images/2.1-doris-arrow-flight.png | Bin 0 -> 102524 bytes
docs/images/2.1-doris-clickbench-tpch.png | Bin 0 -> 53975 bytes
docs/images/2.1-doris-group-commit-2.png | Bin 0 -> 167142 bytes
docs/images/2.1-doris-group-commit.png | Bin 0 -> 164155 bytes
docs/images/2.1-doris-local-shuffle.png | Bin 0 -> 290278 bytes
docs/images/2.1-doris-parallel-adaptive-scan.png | Bin 0 -> 72181 bytes
docs/images/2.1-insert-into-table.png | Bin 0 -> 84689 bytes
docs/sidebars.json | 3 +
docs/zh-CN/docs/lakehouse/sql-dialect.md | 2 +-
docs/zh-CN/docs/releasenotes/release-1.2.8.md | 48 ++
docs/zh-CN/docs/releasenotes/release-2.0.6.md | 54 ++
docs/zh-CN/docs/releasenotes/release-2.1.0.md | 892 +++++++++++++++++++++
.../sql-manual/sql-reference/Data-Types/VARIANT.md | 22 +-
fe/be-java-extensions/pom.xml | 1 +
fe/be-java-extensions/preload-extensions/pom.xml | 4 +
.../trino-connector-scanner/pom.xml | 80 ++
.../doris/trinoconnector/TrinoConnectorCache.java | 268 +++++++
.../trinoconnector/TrinoConnectorColumnValue.java | 237 ++++++
.../trinoconnector/TrinoConnectorJniScanner.java | 373 +++++++++
.../trinoconnector/TrinoConnectorPluginLoader.java | 75 ++
.../trinoconnector/TrinoConnectorScannerUtils.java | 47 ++
.../TrinoTypeToHiveTypeTranslator.java | 115 +++
.../src/main/resources/package.xml | 39 +
.../doris/analysis/AlterRoutineLoadStmt.java | 8 +
.../doris/analysis/CreateRoutineLoadStmt.java | 14 +
.../main/java/org/apache/doris/catalog/Env.java | 54 +-
.../doris/cloud/load/CloudBrokerLoadJob.java | 9 +
.../doris/datasource/hive/source/HiveScanNode.java | 6 +-
.../constants/TrinoConnectorProperties.java | 22 -
.../TrinoConnectorExternalCatalog.java | 36 +-
.../TrinoConnectorExternalCatalogFactory.java | 1 -
.../TrinoConnectorExternalDatabase.java | 1 -
.../TrinoConnectorExternalTable.java | 1 -
.../trinoconnector/TrinoConnectorPluginLoader.java | 81 ++
.../source/TrinoConnectorScanNode.java | 10 +-
.../java/org/apache/doris/load/loadv2/LoadJob.java | 2 +-
.../doris/load/routineload/KafkaTaskInfo.java | 27 +
.../doris/load/routineload/RoutineLoadJob.java | 14 +
.../glue/translator/PhysicalPlanTranslator.java | 24 +-
.../nereids/trees/expressions/InPredicate.java | 12 +
.../doris/nereids/util/TypeCoercionUtils.java | 10 +-
.../java/org/apache/doris/planner/PlanNode.java | 14 +-
.../apache/doris/planner/SingleNodePlanner.java | 4 +
.../main/java/org/apache/doris/qe/VariableMgr.java | 26 +-
.../resource/workloadgroup/WorkloadGroupMgr.java | 39 +-
.../jdbc/test_mysql_jdbc_driver5_catalog.out | 1 +
.../nereids_syntax_p0/inpredicate_with_list.out | 19 +
.../data/query_p0/sql_functions/test_in_expr.out | 3 +
regression-test/pipeline/p0/conf/fe.conf | 1 +
.../plugins/cloud_abnormal_plugin.groovy | 371 +++++++++
.../plugins/cloud_cluster_op_plugin.groovy | 86 ++
.../plugins/cloud_compaction_plugin.groovy | 112 +++
.../plugins/cloud_recycler_plugin.groovy | 195 +++++
.../plugins/cloud_resource_plugin.groovy | 76 ++
.../plugins/cloud_smoke_plugin.groovy | 35 +-
.../plugins/cloud_smooth_upgrade.groovy | 65 ++
.../plugins/cloud_stage_ram_plugin.groovy | 424 ++++++++++
.../suites/control_p0/set_replica_status.groovy | 3 +
.../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 +
.../query_p0/sql_functions/test_in_expr.groovy | 5 +-
.../version_p0/test_set_replica_version.groovy | 3 +
.../workload_manager_p0/test_curd_wlg.groovy | 12 +-
thirdparty/download-thirdparty.sh | 11 +
thirdparty/patches/base64-0.5.2.patch | 57 ++
129 files changed, 5805 insertions(+), 482 deletions(-)
create mode 100644 be/src/vec/exec/format/table/trino_connector_jni_reader.cpp
create mode 100644 be/src/vec/exec/format/table/trino_connector_jni_reader.h
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 docs/en/docs/releasenotes/release-1.2.8.md
create mode 100644 docs/en/docs/releasenotes/release-2.0.6.md
create mode 100644 docs/en/docs/releasenotes/release-2.1.0.md
create mode 100644 docs/images/2.1-Doris-TPC-DS-best-performance.png
create mode 100644 docs/images/2.1-comparied-to-Json-2.png
create mode 100644 docs/images/2.1-comparied-to-Json.png
create mode 100644 docs/images/2.1-doris-INSERT-INTO-SELECT.png
create mode 100644 docs/images/2.1-doris-TPC-DS.png
create mode 100644 docs/images/2.1-doris-arrow-flight-sql.png
create mode 100644 docs/images/2.1-doris-arrow-flight.png
create mode 100644 docs/images/2.1-doris-clickbench-tpch.png
create mode 100644 docs/images/2.1-doris-group-commit-2.png
create mode 100644 docs/images/2.1-doris-group-commit.png
create mode 100644 docs/images/2.1-doris-local-shuffle.png
create mode 100644 docs/images/2.1-doris-parallel-adaptive-scan.png
create mode 100644 docs/images/2.1-insert-into-table.png
create mode 100644 docs/zh-CN/docs/releasenotes/release-1.2.8.md
create mode 100644 docs/zh-CN/docs/releasenotes/release-2.0.6.md
create mode 100644 docs/zh-CN/docs/releasenotes/release-2.1.0.md
create mode 100644 fe/be-java-extensions/trino-connector-scanner/pom.xml
create mode 100644 fe/be-java-extensions/trino-connector-scanner/src/main/java/org/apache/doris/trinoconnector/TrinoConnectorCache.java
create mode 100644 fe/be-java-extensions/trino-connector-scanner/src/main/java/org/apache/doris/trinoconnector/TrinoConnectorColumnValue.java
create mode 100644 fe/be-java-extensions/trino-connector-scanner/src/main/java/org/apache/doris/trinoconnector/TrinoConnectorJniScanner.java
create mode 100644 fe/be-java-extensions/trino-connector-scanner/src/main/java/org/apache/doris/trinoconnector/TrinoConnectorPluginLoader.java
create mode 100644 fe/be-java-extensions/trino-connector-scanner/src/main/java/org/apache/doris/trinoconnector/TrinoConnectorScannerUtils.java
create mode 100644 fe/be-java-extensions/trino-connector-scanner/src/main/java/org/apache/doris/trinoconnector/TrinoTypeToHiveTypeTranslator.java
create mode 100644 fe/be-java-extensions/trino-connector-scanner/src/main/resources/package.xml
create mode 100644 fe/fe-core/src/main/java/org/apache/doris/datasource/trinoconnector/TrinoConnectorPluginLoader.java
create mode 100644 regression-test/data/nereids_syntax_p0/inpredicate_with_list.out
create mode 100644 regression-test/plugins/cloud_abnormal_plugin.groovy
create mode 100644 regression-test/plugins/cloud_cluster_op_plugin.groovy
create mode 100644 regression-test/plugins/cloud_compaction_plugin.groovy
create mode 100644 regression-test/plugins/cloud_recycler_plugin.groovy
create mode 100644 regression-test/plugins/cloud_resource_plugin.groovy
copy fe/fe-core/src/main/java/org/apache/doris/datasource/trinoconnector/TrinoConnectorExternalCatalogFactory.java => regression-test/plugins/cloud_smoke_plugin.groovy (53%)
create mode 100644 regression-test/plugins/cloud_smooth_upgrade.groovy
create mode 100644 regression-test/plugins/cloud_stage_ram_plugin.groovy
create mode 100644 regression-test/suites/nereids_syntax_p0/inpredicate_with_list.groovy
create mode 100644 thirdparty/patches/base64-0.5.2.patch
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 05/07: Support show auto analyze pending jobs. (#31926)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lijibing pushed a commit to branch high-priority-column
in repository https://gitbox.apache.org/repos/asf/doris.git
commit f21a600b7eea39b985bcf910605388f51840f7c8
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Thu Mar 7 15:02:02 2024 +0800
Support show auto analyze pending jobs. (#31926)
---
fe/fe-core/src/main/cup/sql_parser.cup | 4 +
.../doris/analysis/ShowAutoAnalyzeJobsStmt.java | 210 +++++++++++++++++++++
.../java/org/apache/doris/qe/ShowExecutor.java | 34 ++++
.../apache/doris/statistics/AnalysisManager.java | 40 +++-
.../doris/statistics/AutoAnalysisPendingJob.java | 50 +++++
.../org/apache/doris/statistics/JobPriority.java | 24 +++
.../doris/statistics/StatisticsAutoCollector.java | 16 +-
.../doris/statistics/StatisticsJobAppender.java | 25 ++-
8 files changed, 383 insertions(+), 20 deletions(-)
diff --git a/fe/fe-core/src/main/cup/sql_parser.cup b/fe/fe-core/src/main/cup/sql_parser.cup
index 23efc77cc05..799c9eeb6b7 100644
--- a/fe/fe-core/src/main/cup/sql_parser.cup
+++ b/fe/fe-core/src/main/cup/sql_parser.cup
@@ -4375,6 +4375,10 @@ show_param ::=
{:
RESULT = new ShowAnalyzeStmt(tbl, parser.where, true);
:}
+ | KW_AUTO KW_JOBS opt_table_name:tbl opt_wild_where
+ {:
+ RESULT = new ShowAutoAnalyzeJobsStmt(tbl, parser.where);
+ :}
| KW_ANALYZE KW_TASK KW_STATUS INTEGER_LITERAL:jobId
{:
RESULT = new ShowAnalyzeTaskStatus(jobId);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowAutoAnalyzeJobsStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowAutoAnalyzeJobsStmt.java
new file mode 100644
index 00000000000..560387fa5bc
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowAutoAnalyzeJobsStmt.java
@@ -0,0 +1,210 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.analysis;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.ScalarType;
+import org.apache.doris.common.AnalysisException;
+import org.apache.doris.common.ErrorCode;
+import org.apache.doris.common.ErrorReport;
+import org.apache.doris.common.UserException;
+import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.qe.ShowResultSetMetaData;
+import org.apache.doris.statistics.JobPriority;
+
+import com.google.common.base.Preconditions;
+import com.google.common.base.Strings;
+import com.google.common.collect.ImmutableList;
+
+/**
+ * ShowAutoAnalyzeJobsStmt is used to show pending auto analysis jobs.
+ * syntax:
+ * SHOW AUTO ANALYZE JOBS
+ * [TABLE]
+ * [
+ * WHERE
+ * [PRIORITY = ["HIGH"|"MID"|"LOW"]]
+ * ]
+ */
+public class ShowAutoAnalyzeJobsStmt extends ShowStmt {
+ private static final String PRIORITY = "priority";
+ private static final ImmutableList<String> TITLE_NAMES = new ImmutableList.Builder<String>()
+ .add("catalog_name")
+ .add("db_name")
+ .add("tbl_name")
+ .add("col_list")
+ .add("priority")
+ .build();
+
+ private final TableName tableName;
+ private final Expr whereClause;
+
+ public ShowAutoAnalyzeJobsStmt(TableName tableName, Expr whereClause) {
+ this.tableName = tableName;
+ this.whereClause = whereClause;
+ }
+
+ // extract from predicate
+ private String jobPriority;
+
+ public String getPriority() {
+ Preconditions.checkArgument(isAnalyzed(),
+ "The stateValue must be obtained after the parsing is complete");
+ return jobPriority;
+ }
+
+ public Expr getWhereClause() {
+ Preconditions.checkArgument(isAnalyzed(),
+ "The whereClause must be obtained after the parsing is complete");
+ return whereClause;
+ }
+
+ @Override
+ public void analyze(Analyzer analyzer) throws UserException {
+ if (!ConnectContext.get().getSessionVariable().enableStats) {
+ throw new UserException("Analyze function is forbidden, you should add `enable_stats=true`"
+ + "in your FE conf file");
+ }
+ super.analyze(analyzer);
+ if (tableName != null) {
+ tableName.analyze(analyzer);
+ String catalogName = tableName.getCtl();
+ String dbName = tableName.getDb();
+ String tblName = tableName.getTbl();
+ checkShowAnalyzePriv(catalogName, dbName, tblName);
+ }
+
+ // analyze where clause if not null
+ if (whereClause != null) {
+ analyzeSubPredicate(whereClause);
+ }
+ }
+
+ @Override
+ public ShowResultSetMetaData getMetaData() {
+ ShowResultSetMetaData.Builder builder = ShowResultSetMetaData.builder();
+ for (String title : TITLE_NAMES) {
+ builder.addColumn(new Column(title, ScalarType.createVarchar(128)));
+ }
+ return builder.build();
+ }
+
+ @Override
+ public RedirectStatus getRedirectStatus() {
+ return RedirectStatus.FORWARD_NO_SYNC;
+ }
+
+ private void checkShowAnalyzePriv(String catalogName, String dbName, String tblName) throws AnalysisException {
+ if (!Env.getCurrentEnv().getAccessManager()
+ .checkTblPriv(ConnectContext.get(), catalogName, dbName, tblName, PrivPredicate.SHOW)) {
+ ErrorReport.reportAnalysisException(
+ ErrorCode.ERR_TABLEACCESS_DENIED_ERROR,
+ "SHOW ANALYZE",
+ ConnectContext.get().getQualifiedUser(),
+ ConnectContext.get().getRemoteIP(),
+ dbName + ": " + tblName);
+ }
+ }
+
+ private void analyzeSubPredicate(Expr subExpr) throws AnalysisException {
+ if (subExpr == null) {
+ return;
+ }
+
+ boolean valid = true;
+
+ CHECK: {
+ if (subExpr instanceof BinaryPredicate) {
+ BinaryPredicate binaryPredicate = (BinaryPredicate) subExpr;
+ if (binaryPredicate.getOp() != BinaryPredicate.Operator.EQ) {
+ valid = false;
+ break CHECK;
+ }
+ } else {
+ valid = false;
+ break CHECK;
+ }
+
+ // left child
+ if (!(subExpr.getChild(0) instanceof SlotRef)) {
+ valid = false;
+ break CHECK;
+ }
+ String leftKey = ((SlotRef) subExpr.getChild(0)).getColumnName();
+ if (!PRIORITY.equalsIgnoreCase(leftKey)) {
+ valid = false;
+ break CHECK;
+ }
+
+ // right child
+ if (!(subExpr.getChild(1) instanceof StringLiteral)) {
+ valid = false;
+ break CHECK;
+ }
+
+ String value = subExpr.getChild(1).getStringValue();
+ if (Strings.isNullOrEmpty(value)) {
+ valid = false;
+ break CHECK;
+ }
+
+ jobPriority = value.toUpperCase();
+ try {
+ JobPriority.valueOf(jobPriority);
+ } catch (Exception e) {
+ valid = false;
+ }
+ }
+
+ if (!valid) {
+ throw new AnalysisException("Where clause should looks like: "
+ + "PRIORITY = \"HIGH|MID|LOW\"");
+ }
+ }
+
+ @Override
+ public String toSql() {
+ StringBuilder sb = new StringBuilder();
+ sb.append("SHOW AUTO ANALYZE");
+
+ if (tableName != null) {
+ sb.append(" ");
+ sb.append(tableName.toSql());
+ }
+
+ if (whereClause != null) {
+ sb.append(" ");
+ sb.append("WHERE");
+ sb.append(" ");
+ sb.append(whereClause.toSql());
+ }
+
+ return sb.toString();
+ }
+
+ @Override
+ public String toString() {
+ return toSql();
+ }
+
+ public TableName getTableName() {
+ return tableName;
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
index 79a2dca8946..dda7045eabb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
@@ -30,6 +30,7 @@ import org.apache.doris.analysis.ShowAlterStmt;
import org.apache.doris.analysis.ShowAnalyzeStmt;
import org.apache.doris.analysis.ShowAnalyzeTaskStatus;
import org.apache.doris.analysis.ShowAuthorStmt;
+import org.apache.doris.analysis.ShowAutoAnalyzeJobsStmt;
import org.apache.doris.analysis.ShowBackendsStmt;
import org.apache.doris.analysis.ShowBackupStmt;
import org.apache.doris.analysis.ShowBrokerStmt;
@@ -203,6 +204,7 @@ import org.apache.doris.mysql.privilege.PrivBitSet;
import org.apache.doris.mysql.privilege.PrivPredicate;
import org.apache.doris.mysql.privilege.Privilege;
import org.apache.doris.statistics.AnalysisInfo;
+import org.apache.doris.statistics.AutoAnalysisPendingJob;
import org.apache.doris.statistics.ColumnStatistic;
import org.apache.doris.statistics.Histogram;
import org.apache.doris.statistics.ResultRow;
@@ -440,6 +442,8 @@ public class ShowExecutor {
handleShowCreateCatalog();
} else if (stmt instanceof ShowAnalyzeStmt) {
handleShowAnalyze();
+ } else if (stmt instanceof ShowAutoAnalyzeJobsStmt) {
+ handleShowAutoAnalyzePendingJobs();
} else if (stmt instanceof ShowTabletsBelongStmt) {
handleShowTabletsBelong();
} else if (stmt instanceof AdminCopyTabletStmt) {
@@ -2865,6 +2869,36 @@ public class ShowExecutor {
resultSet = new ShowResultSet(showStmt.getMetaData(), resultRows);
}
+ private void handleShowAutoAnalyzePendingJobs() {
+ ShowAutoAnalyzeJobsStmt showStmt = (ShowAutoAnalyzeJobsStmt) stmt;
+ List<AutoAnalysisPendingJob> jobs = Env.getCurrentEnv().getAnalysisManager().showAutoPendingJobs(showStmt);
+ List<List<String>> resultRows = Lists.newArrayList();
+ for (AutoAnalysisPendingJob job : jobs) {
+ try {
+ List<String> row = new ArrayList<>();
+ CatalogIf<? extends DatabaseIf<? extends TableIf>> c
+ = StatisticsUtil.findCatalog(job.catalogName);
+ row.add(c.getName());
+ Optional<? extends DatabaseIf<? extends TableIf>> databaseIf = c.getDb(job.dbName);
+ row.add(databaseIf.isPresent() ? databaseIf.get().getFullName() : "DB may get deleted");
+ if (databaseIf.isPresent()) {
+ Optional<? extends TableIf> table = databaseIf.get().getTable(job.tableName);
+ row.add(table.isPresent() ? table.get().getName() : "Table may get deleted");
+ } else {
+ row.add("DB may get deleted");
+ }
+ row.add(job.getColumnNames());
+ row.add(String.valueOf(job.priority));
+ resultRows.add(row);
+ } catch (Exception e) {
+ LOG.warn("Failed to get pending jobs for table {}.{}.{}, reason: {}",
+ job.catalogName, job.dbName, job.tableName, e.getMessage());
+ continue;
+ }
+ }
+ resultSet = new ShowResultSet(showStmt.getMetaData(), resultRows);
+ }
+
private void handleShowTabletsBelong() {
ShowTabletsBelongStmt showStmt = (ShowTabletsBelongStmt) stmt;
List<List<String>> rows = new ArrayList<>();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
index e2bd85a0f43..c6713b3a7c1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
@@ -25,6 +25,7 @@ import org.apache.doris.analysis.DropAnalyzeJobStmt;
import org.apache.doris.analysis.DropStatsStmt;
import org.apache.doris.analysis.KillAnalysisJobStmt;
import org.apache.doris.analysis.ShowAnalyzeStmt;
+import org.apache.doris.analysis.ShowAutoAnalyzeJobsStmt;
import org.apache.doris.analysis.TableName;
import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.DatabaseIf;
@@ -113,9 +114,9 @@ public class AnalysisManager implements Writable {
private static final int COLUMN_QUEUE_SIZE = 1000;
public final Queue<HighPriorityColumn> highPriorityColumns = new ArrayBlockingQueue<>(COLUMN_QUEUE_SIZE);
public final Queue<HighPriorityColumn> midPriorityColumns = new ArrayBlockingQueue<>(COLUMN_QUEUE_SIZE);
- public final Map<TableIf, Set<String>> highPriorityJobs = new LinkedHashMap<>();
- public final Map<TableIf, Set<String>> midPriorityJobs = new LinkedHashMap<>();
- public final Map<TableIf, Set<String>> lowPriorityJobs = new LinkedHashMap<>();
+ public final Map<TableName, Set<String>> highPriorityJobs = new LinkedHashMap<>();
+ public final Map<TableName, Set<String>> midPriorityJobs = new LinkedHashMap<>();
+ public final Map<TableName, Set<String>> lowPriorityJobs = new LinkedHashMap<>();
// Tracking running manually submitted async tasks, keep in mem only
protected final ConcurrentMap<Long, Map<Long, BaseAnalysisTask>> analysisJobIdToTaskMap = new ConcurrentHashMap<>();
@@ -598,6 +599,39 @@ public class AnalysisManager implements Writable {
}
}
+ public List<AutoAnalysisPendingJob> showAutoPendingJobs(ShowAutoAnalyzeJobsStmt stmt) {
+ TableName tblName = stmt.getTableName();
+ String priority = stmt.getPriority();
+ List<AutoAnalysisPendingJob> result = Lists.newArrayList();
+ if (priority == null || priority.isEmpty()) {
+ result.addAll(getPendingJobs(highPriorityJobs, JobPriority.HIGH, tblName));
+ result.addAll(getPendingJobs(midPriorityJobs, JobPriority.MID, tblName));
+ result.addAll(getPendingJobs(lowPriorityJobs, JobPriority.LOW, tblName));
+ } else if (priority.equals(JobPriority.HIGH.name())) {
+ result.addAll(getPendingJobs(highPriorityJobs, JobPriority.HIGH, tblName));
+ } else if (priority.equals(JobPriority.MID.name())) {
+ result.addAll(getPendingJobs(midPriorityJobs, JobPriority.MID, tblName));
+ } else if (priority.equals(JobPriority.LOW.name())) {
+ result.addAll(getPendingJobs(lowPriorityJobs, JobPriority.LOW, tblName));
+ }
+ return result;
+ }
+
+ protected List<AutoAnalysisPendingJob> getPendingJobs(Map<TableName, Set<String>> jobMap,
+ JobPriority priority, TableName tblName) {
+ List<AutoAnalysisPendingJob> result = Lists.newArrayList();
+ synchronized (jobMap) {
+ for (Entry<TableName, Set<String>> entry : jobMap.entrySet()) {
+ TableName table = entry.getKey();
+ if (tblName == null || tblName.equals(table)) {
+ result.add(new AutoAnalysisPendingJob(table.getCtl(),
+ table.getDb(), table.getTbl(), entry.getValue(), priority));
+ }
+ }
+ }
+ return result;
+ }
+
public List<AnalysisInfo> showAnalysisJob(ShowAnalyzeStmt stmt) {
return findShowAnalyzeResult(analysisJobInfoMap.values(), stmt);
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AutoAnalysisPendingJob.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AutoAnalysisPendingJob.java
new file mode 100644
index 00000000000..ddd06d17c81
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AutoAnalysisPendingJob.java
@@ -0,0 +1,50 @@
+// 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.statistics;
+
+import java.util.Set;
+import java.util.StringJoiner;
+
+public class AutoAnalysisPendingJob {
+
+ public final String catalogName;
+ public final String dbName;
+ public final String tableName;
+ public final Set<String> columnNames;
+ public final JobPriority priority;
+
+ public AutoAnalysisPendingJob(String catalogName, String dbName, String tableName,
+ Set<String> columnNames, JobPriority priority) {
+ this.catalogName = catalogName;
+ this.dbName = dbName;
+ this.tableName = tableName;
+ this.columnNames = columnNames;
+ this.priority = priority;
+ }
+
+ public String getColumnNames() {
+ if (columnNames == null) {
+ return "";
+ }
+ StringJoiner stringJoiner = new StringJoiner(",");
+ for (String colName : columnNames) {
+ stringJoiner.add(colName);
+ }
+ return stringJoiner.toString();
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/JobPriority.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/JobPriority.java
new file mode 100644
index 00000000000..2786b063563
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/JobPriority.java
@@ -0,0 +1,24 @@
+// 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.statistics;
+
+public enum JobPriority {
+ HIGH,
+ MID,
+ LOW;
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
index e0df94b5cb0..227074dbb5c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
@@ -17,6 +17,7 @@
package org.apache.doris.statistics;
+import org.apache.doris.analysis.TableName;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.TableIf;
@@ -57,13 +58,14 @@ public class StatisticsAutoCollector extends StatisticsCollector {
@Override
protected void collect() {
while (canCollect()) {
- Map.Entry<TableIf, Set<String>> job = getJob();
+ Map.Entry<TableName, Set<String>> job = getJob();
if (job == null) {
// No more job to process, break and sleep.
break;
}
try {
- TableIf table = job.getKey();
+ TableName tblName = job.getKey();
+ TableIf table = StatisticsUtil.findTable(tblName.getCtl(), tblName.getDb(), tblName.getTbl());
if (!supportAutoAnalyze(table)) {
continue;
}
@@ -78,7 +80,7 @@ public class StatisticsAutoCollector extends StatisticsCollector {
processOneJob(table, columns);
} catch (Exception e) {
LOG.warn("Failed to analyze table {} with columns [{}]",
- job.getKey().getName(), job.getValue().stream().collect(Collectors.joining(",")), e);
+ job.getKey().getTbl(), job.getValue().stream().collect(Collectors.joining(",")), e);
}
}
}
@@ -88,9 +90,9 @@ public class StatisticsAutoCollector extends StatisticsCollector {
&& StatisticsUtil.inAnalyzeTime(LocalTime.now(TimeUtils.getTimeZone().toZoneId()));
}
- protected Map.Entry<TableIf, Set<String>> getJob() {
+ protected Map.Entry<TableName, Set<String>> getJob() {
AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
- Optional<Map.Entry<TableIf, Set<String>>> job = fetchJobFromMap(manager.highPriorityJobs);
+ Optional<Map.Entry<TableName, Set<String>>> job = fetchJobFromMap(manager.highPriorityJobs);
if (job.isPresent()) {
return job.get();
}
@@ -102,9 +104,9 @@ public class StatisticsAutoCollector extends StatisticsCollector {
return job.isPresent() ? job.get() : null;
}
- protected Optional<Map.Entry<TableIf, Set<String>>> fetchJobFromMap(Map<TableIf, Set<String>> jobMap) {
+ protected Optional<Map.Entry<TableName, Set<String>>> fetchJobFromMap(Map<TableName, Set<String>> jobMap) {
synchronized (jobMap) {
- Optional<Map.Entry<TableIf, Set<String>>> first = jobMap.entrySet().stream().findFirst();
+ Optional<Map.Entry<TableName, Set<String>>> first = jobMap.entrySet().stream().findFirst();
first.ifPresent(entry -> jobMap.remove(entry.getKey()));
return first;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
index 71bb71d3cda..93d03a3fdb8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
@@ -17,6 +17,7 @@
package org.apache.doris.statistics;
+import org.apache.doris.analysis.TableName;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
@@ -81,31 +82,33 @@ public class StatisticsJobAppender extends MasterDaemon {
}
}
- protected void appendColumnsToJobs(Queue<HighPriorityColumn> columnQueue, Map<TableIf, Set<String>> jobsMap) {
+ protected void appendColumnsToJobs(Queue<HighPriorityColumn> columnQueue, Map<TableName, Set<String>> jobsMap) {
int size = columnQueue.size();
for (int i = 0; i < size; i++) {
HighPriorityColumn column = columnQueue.poll();
LOG.info("Process column " + column.tblId + "." + column.colName);
TableIf table = StatisticsUtil.findTable(column.catalogId, column.dbId, column.tblId);
+ TableName tableName = new TableName(table.getDatabase().getCatalog().getName(),
+ table.getDatabase().getFullName(), table.getName());
synchronized (jobsMap) {
// If job map reach the upper limit, stop putting new jobs.
- if (!jobsMap.containsKey(table) && jobsMap.size() >= JOB_MAP_SIZE) {
+ if (!jobsMap.containsKey(tableName) && jobsMap.size() >= JOB_MAP_SIZE) {
LOG.info("Job map full.");
break;
}
- if (jobsMap.containsKey(table)) {
- jobsMap.get(table).add(column.colName);
+ if (jobsMap.containsKey(tableName)) {
+ jobsMap.get(tableName).add(column.colName);
} else {
HashSet<String> columns = new HashSet<>();
columns.add(column.colName);
- jobsMap.put(table, columns);
+ jobsMap.put(tableName, columns);
}
LOG.info("Column " + column.tblId + "." + column.colName + " added");
}
}
}
- protected void appendToLowQueue(Map<TableIf, Set<String>> jobsMap) {
+ protected void appendToLowQueue(Map<TableName, Set<String>> jobsMap) {
InternalCatalog catalog = Env.getCurrentInternalCatalog();
List<Long> sortedDbs = catalog.getDbIds().stream().sorted().collect(Collectors.toList());
int batchSize = 100;
@@ -122,18 +125,20 @@ public class StatisticsJobAppender extends MasterDaemon {
if (!(t instanceof OlapTable) || t.getId() <= currentTableId) {
continue;
}
+ TableName tableName = new TableName(t.getDatabase().getCatalog().getName(),
+ t.getDatabase().getFullName(), t.getName());
synchronized (jobsMap) {
// If job map reach the upper limit, stop adding new jobs.
- if (!jobsMap.containsKey(t) && jobsMap.size() >= JOB_MAP_SIZE) {
+ if (!jobsMap.containsKey(tableName) && jobsMap.size() >= JOB_MAP_SIZE) {
return;
}
Set<String> columns
= t.getColumns().stream().filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
.map(c -> c.getName()).collect(Collectors.toSet());
- if (jobsMap.containsKey(t)) {
- jobsMap.get(t).addAll(columns);
+ if (jobsMap.containsKey(tableName)) {
+ jobsMap.get(tableName).addAll(columns);
} else {
- jobsMap.put(t, columns);
+ jobsMap.put(tableName, columns);
}
}
currentTableId = t.getId();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 04/07: Support follower sync query columns to master. (#31859)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lijibing pushed a commit to branch high-priority-column
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 27f92e518c5d6681ba377b2cce3995b9d2426c14
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Wed Mar 6 17:05:21 2024 +0800
Support follower sync query columns to master. (#31859)
---
.../main/java/org/apache/doris/catalog/Env.java | 8 ++
.../apache/doris/service/FrontendServiceImpl.java | 8 ++
.../apache/doris/statistics/AnalysisManager.java | 15 +++
.../doris/statistics/FollowerColumnSender.java | 120 +++++++++++++++++++++
.../doris/statistics/HighPriorityColumn.java | 11 ++
.../doris/statistics/StatisticsAutoCollector.java | 21 +++-
gensrc/thrift/FrontendService.thrift | 13 +++
7 files changed, 194 insertions(+), 2 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index 1974a316ec5..0c600bb5fe8 100755
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -240,6 +240,7 @@ import org.apache.doris.scheduler.registry.ExportTaskRegister;
import org.apache.doris.service.ExecuteEnv;
import org.apache.doris.service.FrontendOptions;
import org.apache.doris.statistics.AnalysisManager;
+import org.apache.doris.statistics.FollowerColumnSender;
import org.apache.doris.statistics.StatisticsAutoCollector;
import org.apache.doris.statistics.StatisticsCache;
import org.apache.doris.statistics.StatisticsCleaner;
@@ -522,6 +523,8 @@ public class Env {
private StatisticsJobAppender statisticsJobAppender;
+ private FollowerColumnSender followerColumnSender;
+
private HiveTransactionMgr hiveTransactionMgr;
private TopicPublisherThread topicPublisherThread;
@@ -1732,6 +1735,11 @@ public class Env {
if (analysisManager != null) {
analysisManager.getStatisticsCache().preHeat();
}
+
+ if (followerColumnSender == null) {
+ followerColumnSender = new FollowerColumnSender();
+ followerColumnSender.start();
+ }
}
// Set global variable 'lower_case_table_names' only when the cluster is initialized.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index eb62cd9c75a..ec3638870b7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -220,6 +220,7 @@ import org.apache.doris.thrift.TStreamLoadMultiTablePutResult;
import org.apache.doris.thrift.TStreamLoadPutRequest;
import org.apache.doris.thrift.TStreamLoadPutResult;
import org.apache.doris.thrift.TStringLiteral;
+import org.apache.doris.thrift.TSyncQueryColumns;
import org.apache.doris.thrift.TTableIndexQueryStats;
import org.apache.doris.thrift.TTableMetadataNameIds;
import org.apache.doris.thrift.TTableQueryStats;
@@ -3717,4 +3718,11 @@ public class FrontendServiceImpl implements FrontendService.Iface {
return result;
}
+ @Override
+ public TStatus syncQueryColumns(TSyncQueryColumns request) throws TException {
+ Env.getCurrentEnv().getAnalysisManager().mergeFollowerQueryColumns(request.highPriorityColumns,
+ request.midPriorityColumns);
+ return new TStatus(TStatusCode.OK);
+ }
+
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
index bac84996dd1..e2bd85a0f43 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
@@ -64,6 +64,7 @@ import org.apache.doris.statistics.util.StatisticsUtil;
import org.apache.doris.system.Frontend;
import org.apache.doris.system.SystemInfoService;
import org.apache.doris.thrift.TInvalidateFollowerStatsCacheRequest;
+import org.apache.doris.thrift.TQueryColumn;
import com.google.common.annotations.VisibleForTesting;
import com.google.common.collect.ImmutableList;
@@ -1191,4 +1192,18 @@ public class AnalysisManager implements Writable {
}
}
}
+
+ public void mergeFollowerQueryColumns(Collection<TQueryColumn> highColumns,
+ Collection<TQueryColumn> midColumns) {
+ for (TQueryColumn c : highColumns) {
+ if (!highPriorityColumns.offer(new HighPriorityColumn(c.catalogId, c.dbId, c.tblId, c.colName))) {
+ break;
+ }
+ }
+ for (TQueryColumn c : midColumns) {
+ if (!midPriorityColumns.offer(new HighPriorityColumn(c.catalogId, c.dbId, c.tblId, c.colName))) {
+ break;
+ }
+ }
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/FollowerColumnSender.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/FollowerColumnSender.java
new file mode 100644
index 00000000000..181000c1ef2
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/FollowerColumnSender.java
@@ -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.
+
+package org.apache.doris.statistics;
+
+import org.apache.doris.catalog.Env;
+import org.apache.doris.common.ClientPool;
+import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.ha.FrontendNodeType;
+import org.apache.doris.statistics.util.StatisticsUtil;
+import org.apache.doris.system.Frontend;
+import org.apache.doris.thrift.FrontendService;
+import org.apache.doris.thrift.TNetworkAddress;
+import org.apache.doris.thrift.TQueryColumn;
+import org.apache.doris.thrift.TSyncQueryColumns;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.net.InetSocketAddress;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class FollowerColumnSender extends MasterDaemon {
+
+ private static final Logger LOG = LogManager.getLogger(FollowerColumnSender.class);
+
+ public static final long INTERVAL = 5000;
+
+ public FollowerColumnSender() {
+ super("Follower Column Sender", INTERVAL);
+ }
+
+ @Override
+ protected void runAfterCatalogReady() {
+ if (!StatisticsUtil.enableAutoAnalyze()) {
+ return;
+ }
+ if (Env.getCurrentEnv().isMaster()) {
+ return;
+ }
+ if (Env.isCheckpointThread()) {
+ return;
+ }
+ send();
+ }
+
+ protected void send() {
+ if (Env.getCurrentEnv().isMaster()) {
+ return;
+ }
+ Env currentEnv = Env.getCurrentEnv();
+ AnalysisManager analysisManager = currentEnv.getAnalysisManager();
+ if (analysisManager.highPriorityColumns.isEmpty() && analysisManager.midPriorityColumns.isEmpty()) {
+ return;
+ }
+ List<TQueryColumn> highPriorityColumns
+ = analysisManager.highPriorityColumns
+ .stream()
+ .map(HighPriorityColumn::toThrift)
+ .collect(Collectors.toList());
+ List<TQueryColumn> midPriorityColumns
+ = analysisManager.midPriorityColumns
+ .stream()
+ .map(HighPriorityColumn::toThrift)
+ .collect(Collectors.toList());
+ analysisManager.highPriorityColumns.clear();
+ analysisManager.midPriorityColumns.clear();
+ TSyncQueryColumns queryColumns = new TSyncQueryColumns();
+ queryColumns.highPriorityColumns = highPriorityColumns;
+ queryColumns.midPriorityColumns = midPriorityColumns;
+ Frontend master = null;
+ try {
+ InetSocketAddress masterAddress = currentEnv.getHaProtocol().getLeader();
+ for (Frontend fe : currentEnv.getFrontends(FrontendNodeType.FOLLOWER)) {
+ InetSocketAddress socketAddress = new InetSocketAddress(fe.getHost(), fe.getEditLogPort());
+ if (socketAddress.equals(masterAddress)) {
+ master = fe;
+ break;
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("Failed to find master FE.", e);
+ return;
+ }
+
+ if (master == null) {
+ LOG.warn("No master found in cluster.");
+ return;
+ }
+ TNetworkAddress address = new TNetworkAddress(master.getHost(), master.getRpcPort());
+ FrontendService.Client client = null;
+ try {
+ client = ClientPool.frontendPool.borrowObject(address);
+ client.syncQueryColumns(queryColumns);
+ LOG.info("Send {} high priority columns and {} mid priority columns to master.",
+ highPriorityColumns.size(), midPriorityColumns.size());
+ } catch (Throwable t) {
+ LOG.warn("Failed to sync stats to master: {}", address, t);
+ } finally {
+ if (client != null) {
+ ClientPool.frontendPool.returnObject(address, client);
+ }
+ }
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java
index c4bc20c399a..b2292ef725d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java
@@ -17,6 +17,8 @@
package org.apache.doris.statistics;
+import org.apache.doris.thrift.TQueryColumn;
+
import java.util.Objects;
public class HighPriorityColumn {
@@ -52,4 +54,13 @@ public class HighPriorityColumn {
&& this.tblId == otherCriticalColumn.tblId
&& this.colName.equals(otherCriticalColumn.colName);
}
+
+ public TQueryColumn toThrift() {
+ TQueryColumn tQueryColumn = new TQueryColumn();
+ tQueryColumn.catalogId = catalogId;
+ tQueryColumn.dbId = dbId;
+ tQueryColumn.tblId = tblId;
+ tQueryColumn.colName = colName;
+ return tQueryColumn;
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
index c498881bfbf..e0df94b5cb0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
@@ -132,6 +132,7 @@ public class StatisticsAutoCollector extends StatisticsCollector {
}
}
+ // TODO: Need refactor, hard to understand now.
protected boolean needAnalyzeColumn(TableIf table, String column) {
AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
TableStatsMeta tableStatsStatus = manager.findTableStatsStatus(table.getId());
@@ -151,11 +152,17 @@ public class StatisticsAutoCollector extends StatisticsCollector {
if (lastAnalyzeUpdateRows == 0 && currentUpdatedRows > 0) {
return true;
}
+ if (lastAnalyzeUpdateRows > currentUpdatedRows) {
+ // Shouldn't happen. Just in case.
+ return true;
+ }
OlapTable olapTable = (OlapTable) table;
+ long currentRowCount = olapTable.getRowCount();
+ long lastAnalyzeRowCount = columnStatsMeta.rowCount;
if (tableStatsStatus.newPartitionLoaded.get() && olapTable.isPartitionColumn(column)) {
return true;
}
- if (columnStatsMeta.rowCount == 0 && olapTable.getRowCount() > 0) {
+ if (lastAnalyzeRowCount == 0 && currentRowCount > 0) {
return true;
}
if (currentUpdatedRows == lastAnalyzeUpdateRows) {
@@ -163,7 +170,17 @@ public class StatisticsAutoCollector extends StatisticsCollector {
}
double healthValue = ((double) (currentUpdatedRows - lastAnalyzeUpdateRows)
/ (double) currentUpdatedRows) * 100.0;
- LOG.info("Column " + column + " health value is " + healthValue);
+ LOG.info("Column " + column + " update rows health value is " + healthValue);
+ if (healthValue < StatisticsUtil.getTableStatsHealthThreshold()) {
+ return true;
+ }
+ if (currentRowCount == 0 && lastAnalyzeRowCount != 0) {
+ return true;
+ }
+ if (currentRowCount == 0 && lastAnalyzeRowCount == 0) {
+ return false;
+ }
+ healthValue = ((double) (currentRowCount - lastAnalyzeRowCount) / (double) currentRowCount) * 100.0;
return healthValue < StatisticsUtil.getTableStatsHealthThreshold();
} else {
if (!(table instanceof HMSExternalTable)) {
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index a20d3668100..d2cdecbbe56 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -1410,6 +1410,18 @@ struct TReportCommitTxnResultRequest {
4: optional binary payload
}
+struct TQueryColumn {
+ 1: optional i64 catalogId
+ 2: optional i64 dbId
+ 3: optional i64 tblId
+ 4: optional string colName
+}
+
+struct TSyncQueryColumns {
+ 1: optional list<TQueryColumn> highPriorityColumns;
+ 2: optional list<TQueryColumn> midPriorityColumns;
+}
+
service FrontendService {
TGetDbsResult getDbNames(1: TGetDbsParams params)
TGetTablesResult getTableNames(1: TGetTablesParams params)
@@ -1495,4 +1507,5 @@ service FrontendService {
TShowProcessListResult showProcessList(1: TShowProcessListRequest request)
Status.TStatus reportCommitTxnResult(1: TReportCommitTxnResultRequest request)
+ Status.TStatus syncQueryColumns(1: TSyncQueryColumns request)
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 03/07: Support column level health value. (#31794)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lijibing pushed a commit to branch high-priority-column
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 9a3470cbb4d0ab49566e059e1a663cfe3b9ddb68
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Tue Mar 5 16:19:08 2024 +0800
Support column level health value. (#31794)
---
.../apache/doris/analysis/ShowColumnStatsStmt.java | 4 +
.../apache/doris/datasource/InternalCatalog.java | 6 +-
.../java/org/apache/doris/qe/SessionVariable.java | 7 ++
.../org/apache/doris/statistics/AnalysisInfo.java | 17 ++--
.../doris/statistics/AnalysisInfoBuilder.java | 17 ++--
.../org/apache/doris/statistics/AnalysisJob.java | 8 +-
.../apache/doris/statistics/AnalysisManager.java | 13 ++-
.../org/apache/doris/statistics/ColStatsMeta.java | 16 ++--
.../apache/doris/statistics/OlapAnalysisTask.java | 2 +-
.../doris/statistics/StatisticsAutoCollector.java | 95 ++++++++++++++++++++--
.../doris/statistics/StatisticsJobAppender.java | 28 +++++--
.../apache/doris/statistics/TableStatsMeta.java | 33 ++++----
.../doris/statistics/util/StatisticsUtil.java | 10 +++
.../apache/doris/statistics/AnalysisJobTest.java | 4 +-
.../doris/statistics/AnalysisManagerTest.java | 2 +-
15 files changed, 198 insertions(+), 64 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowColumnStatsStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowColumnStatsStmt.java
index 37be76b20df..749bfa7d360 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowColumnStatsStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowColumnStatsStmt.java
@@ -61,6 +61,8 @@ public class ShowColumnStatsStmt extends ShowStmt {
.add("trigger")
.add("query_times")
.add("updated_time")
+ .add("update_rows")
+ .add("last_analyze_row_count")
.build();
private final TableName tableName;
@@ -160,6 +162,8 @@ public class ShowColumnStatsStmt extends ShowStmt {
row.add(String.valueOf(colStatsMeta == null ? "N/A" : colStatsMeta.jobType));
row.add(String.valueOf(colStatsMeta == null ? "N/A" : colStatsMeta.queriedTimes));
row.add(String.valueOf(p.second.updatedTime));
+ row.add(String.valueOf(colStatsMeta == null ? "N/A" : colStatsMeta.updatedRows));
+ row.add(String.valueOf(colStatsMeta == null ? "N/A" : colStatsMeta.rowCount));
result.add(row);
});
return new ShowResultSet(getMetaData(), result);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
index 90d152985ac..e0b5e5ecb2e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/datasource/InternalCatalog.java
@@ -3134,7 +3134,6 @@ public class InternalCatalog implements CatalogIf<Database> {
rowsToTruncate += partition.getBaseIndex().getRowCount();
}
} else {
- rowsToTruncate = olapTable.getRowCount();
for (Partition partition : olapTable.getPartitions()) {
// If need absolutely correct, should check running txn here.
// But if the txn is in prepare state, cann't known which partitions had load data.
@@ -3143,6 +3142,7 @@ public class InternalCatalog implements CatalogIf<Database> {
}
origPartitions.put(partition.getName(), partition.getId());
partitionsDistributionInfo.put(partition.getId(), partition.getDistributionInfo());
+ rowsToTruncate += partition.getBaseIndex().getRowCount();
}
}
// if table currently has no partitions, this sql like empty command and do nothing, should return directly.
@@ -3303,10 +3303,8 @@ public class InternalCatalog implements CatalogIf<Database> {
if (truncateEntireTable) {
// Drop the whole table stats after truncate the entire table
Env.getCurrentEnv().getAnalysisManager().dropStats(olapTable);
- } else {
- // Update the updated rows in table stats after truncate some partitions.
- Env.getCurrentEnv().getAnalysisManager().updateUpdatedRows(updateRecords);
}
+ Env.getCurrentEnv().getAnalysisManager().updateUpdatedRows(updateRecords);
LOG.info("finished to truncate table {}, partitions: {}", tblRef.getName().toSql(), tblRef.getPartitionNames());
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 7a03865d1dc..13c3e8baf0c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -469,6 +469,8 @@ public class SessionVariable implements Serializable, Writable {
public static final String ENABLE_AUTO_ANALYZE = "enable_auto_analyze";
+ public static final String ENABLE_AUTO_ANALYZE_INTERNAL_CATALOG = "enable_auto_analyze_internal_catalog";
+
public static final String AUTO_ANALYZE_TABLE_WIDTH_THRESHOLD = "auto_analyze_table_width_threshold";
public static final String FASTER_FLOAT_CONVERT = "faster_float_convert";
@@ -1498,6 +1500,11 @@ public class SessionVariable implements Serializable, Writable {
flag = VariableMgr.GLOBAL)
public boolean enableAutoAnalyze = true;
+ @VariableMgr.VarAttr(name = ENABLE_AUTO_ANALYZE_INTERNAL_CATALOG,
+ description = {"临时参数，收否自动收集所有内表", "Temp variable， enable to auto collect all OlapTable."},
+ flag = VariableMgr.GLOBAL)
+ public boolean enableAutoAnalyzeInternalCatalog = false;
+
@VariableMgr.VarAttr(name = AUTO_ANALYZE_TABLE_WIDTH_THRESHOLD,
description = {"参与自动收集的最大表宽度，列数多于这个参数的表不参与自动收集",
"Maximum table width to enable auto analyze, "
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfo.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfo.java
index c707107e0e0..0c5047a53c5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfo.java
@@ -188,8 +188,11 @@ public class AnalysisInfo implements Writable {
@SerializedName("endTime")
public long endTime;
- @SerializedName("emptyJob")
- public final boolean emptyJob;
+ @SerializedName("rowCount")
+ public final long rowCount;
+
+ @SerializedName("updateRows")
+ public final long updateRows;
/**
*
* Used to store the newest partition version of tbl when creating this job.
@@ -206,7 +209,8 @@ public class AnalysisInfo implements Writable {
long lastExecTimeInMs, long timeCostInMs, AnalysisState state, ScheduleType scheduleType,
boolean isExternalTableLevelTask, boolean partitionOnly, boolean samplingPartition,
boolean isAllPartition, long partitionCount, CronExpression cronExpression, boolean forceFull,
- boolean usingSqlForPartitionColumn, long tblUpdateTime, boolean emptyJob, boolean userInject) {
+ boolean usingSqlForPartitionColumn, long tblUpdateTime, long rowCount, boolean userInject,
+ long updateRows) {
this.jobId = jobId;
this.taskId = taskId;
this.taskIds = taskIds;
@@ -242,8 +246,9 @@ public class AnalysisInfo implements Writable {
this.forceFull = forceFull;
this.usingSqlForPartitionColumn = usingSqlForPartitionColumn;
this.tblUpdateTime = tblUpdateTime;
- this.emptyJob = emptyJob;
+ this.rowCount = rowCount;
this.userInject = userInject;
+ this.updateRows = updateRows;
}
@Override
@@ -285,7 +290,9 @@ public class AnalysisInfo implements Writable {
}
sj.add("forceFull: " + forceFull);
sj.add("usingSqlForPartitionColumn: " + usingSqlForPartitionColumn);
- sj.add("emptyJob: " + emptyJob);
+ sj.add("rowCount: " + rowCount);
+ sj.add("userInject: " + userInject);
+ sj.add("updateRows: " + updateRows);
return sj.toString();
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfoBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfoBuilder.java
index 22f3d22b3ce..527d503fd52 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfoBuilder.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfoBuilder.java
@@ -62,8 +62,9 @@ public class AnalysisInfoBuilder {
private boolean forceFull;
private boolean usingSqlForPartitionColumn;
private long tblUpdateTime;
- private boolean emptyJob;
+ private long rowCount;
private boolean userInject;
+ private long updateRows;
public AnalysisInfoBuilder() {
}
@@ -101,8 +102,9 @@ public class AnalysisInfoBuilder {
forceFull = info.forceFull;
usingSqlForPartitionColumn = info.usingSqlForPartitionColumn;
tblUpdateTime = info.tblUpdateTime;
- emptyJob = info.emptyJob;
+ rowCount = info.rowCount;
userInject = info.userInject;
+ updateRows = info.updateRows;
}
public AnalysisInfoBuilder setJobId(long jobId) {
@@ -265,8 +267,8 @@ public class AnalysisInfoBuilder {
return this;
}
- public AnalysisInfoBuilder setEmptyJob(boolean emptyJob) {
- this.emptyJob = emptyJob;
+ public AnalysisInfoBuilder setRowCount(long rowCount) {
+ this.rowCount = rowCount;
return this;
}
@@ -275,12 +277,17 @@ public class AnalysisInfoBuilder {
return this;
}
+ public AnalysisInfoBuilder setUpdateRows(long updateRows) {
+ this.updateRows = updateRows;
+ return this;
+ }
+
public AnalysisInfo build() {
return new AnalysisInfo(jobId, taskId, taskIds, catalogId, dbId, tblId, colToPartitions, partitionNames,
colName, indexId, jobType, analysisMode, analysisMethod, analysisType, samplePercent,
sampleRows, maxBucketNum, periodTimeInMs, message, lastExecTimeInMs, timeCostInMs, state, scheduleType,
externalTableLevelTask, partitionOnly, samplingPartition, isAllPartition, partitionCount,
- cronExpression, forceFull, usingSqlForPartitionColumn, tblUpdateTime, emptyJob, userInject);
+ cronExpression, forceFull, usingSqlForPartitionColumn, tblUpdateTime, rowCount, userInject, updateRows);
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisJob.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisJob.java
index 22eab37f920..6846abea431 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisJob.java
@@ -84,14 +84,12 @@ public class AnalysisJob {
protected void markOneTaskDone() {
if (queryingTask.isEmpty()) {
try {
- writeBuf();
- updateTaskState(AnalysisState.FINISHED, "Cost time in sec: "
- + (System.currentTimeMillis() - start) / 1000);
+ flushBuffer();
} finally {
deregisterJob();
}
} else if (buf.size() >= StatisticsUtil.getInsertMergeCount()) {
- writeBuf();
+ flushBuffer();
}
}
@@ -115,7 +113,7 @@ public class AnalysisJob {
}
}
- protected void writeBuf() {
+ protected void flushBuffer() {
if (killed) {
return;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
index ab883c79f8f..bac84996dd1 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
@@ -416,7 +416,10 @@ public class AnalysisManager implements Writable {
infoBuilder.setColToPartitions(colToPartitions);
infoBuilder.setTaskIds(Lists.newArrayList());
infoBuilder.setTblUpdateTime(table.getUpdateTime());
- infoBuilder.setEmptyJob(table instanceof OlapTable && table.getRowCount() == 0);
+ long rowCount = table.getRowCount();
+ infoBuilder.setRowCount(rowCount);
+ TableStatsMeta tableStatsStatus = findTableStatsStatus(table.getId());
+ infoBuilder.setUpdateRows(tableStatsStatus == null ? 0 : tableStatsStatus.updatedRows.get());
return infoBuilder.build();
}
@@ -572,7 +575,7 @@ public class AnalysisManager implements Writable {
}
TableStatsMeta tableStats = findTableStatsStatus(tbl.getId());
if (tableStats == null) {
- updateTableStatsStatus(new TableStatsMeta(jobInfo.emptyJob ? 0 : tbl.getRowCount(), jobInfo, tbl));
+ updateTableStatsStatus(new TableStatsMeta(jobInfo.rowCount, jobInfo, tbl));
} else {
tableStats.update(jobInfo, tbl);
logCreateTableStats(tableStats);
@@ -807,7 +810,7 @@ public class AnalysisManager implements Writable {
analysisInfo.dbId, analysisInfo.tblId);
return table.createAnalysisTask(analysisInfo);
} catch (Throwable t) {
- LOG.warn("Failed to find table", t);
+ LOG.warn("Failed to create task.", t);
throw new DdlException("Failed to create task", t);
}
}
@@ -1156,10 +1159,12 @@ public class AnalysisManager implements Writable {
public void updateColumnUsedInPredicate(Set<Slot> slotReferences) {
+ LOG.info("Add slots to high priority queues.");
updateColumn(slotReferences, highPriorityColumns);
}
public void updateQueriedColumn(Collection<Slot> slotReferences) {
+ LOG.info("Add slots to mid priority queues.");
updateColumn(slotReferences, midPriorityColumns);
}
@@ -1179,6 +1184,8 @@ public class AnalysisManager implements Writable {
if (catalog != null) {
queue.offer(new HighPriorityColumn(catalog.getId(), database.getId(),
table.getId(), optionalColumn.get().getName()));
+ LOG.info("Offer column " + table.getName() + "(" + table.getId() + ")."
+ + optionalColumn.get().getName());
}
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/ColStatsMeta.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/ColStatsMeta.java
index 445641b2505..7e317d67bd7 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/ColStatsMeta.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/ColStatsMeta.java
@@ -43,16 +43,20 @@ public class ColStatsMeta {
@SerializedName("trigger")
public JobType jobType;
- public ColStatsMeta(long updatedTime, AnalysisMethod analysisMethod,
- AnalysisType analysisType, JobType jobType, long queriedTimes) {
+ @SerializedName("updatedRows")
+ public long updatedRows;
+
+ @SerializedName("rowCount")
+ public long rowCount;
+
+ public ColStatsMeta(long updatedTime, AnalysisMethod analysisMethod, AnalysisType analysisType, JobType jobType,
+ long queriedTimes, long rowCount, long updatedRows) {
this.updatedTime = updatedTime;
this.analysisMethod = analysisMethod;
this.analysisType = analysisType;
this.jobType = jobType;
this.queriedTimes.addAndGet(queriedTimes);
- }
-
- public void clear() {
- updatedTime = 0;
+ this.updatedRows = updatedRows;
+ this.rowCount = rowCount;
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/OlapAnalysisTask.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/OlapAnalysisTask.java
index 5a1d5829c8c..179766d1af8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/OlapAnalysisTask.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/OlapAnalysisTask.java
@@ -65,7 +65,7 @@ public class OlapAnalysisTask extends BaseAnalysisTask {
public void doExecute() throws Exception {
Set<String> partitionNames = info.colToPartitions.get(info.colName);
- if ((info.emptyJob && info.analysisMethod.equals(AnalysisInfo.AnalysisMethod.SAMPLE))
+ if ((info.rowCount == 0 && info.analysisMethod.equals(AnalysisInfo.AnalysisMethod.SAMPLE))
|| partitionNames == null || partitionNames.isEmpty()) {
if (partitionNames == null) {
LOG.warn("Table {}.{}.{}, partitionNames for column {} is null. ColToPartitions:[{}]",
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
index 19b9c69db08..c498881bfbf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
@@ -23,6 +23,8 @@ import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.util.TimeUtils;
+import org.apache.doris.datasource.hive.HMSExternalTable;
+import org.apache.doris.datasource.hive.HMSExternalTable.DLAType;
import org.apache.doris.statistics.AnalysisInfo.AnalysisMethod;
import org.apache.doris.statistics.AnalysisInfo.JobType;
import org.apache.doris.statistics.AnalysisInfo.ScheduleType;
@@ -33,6 +35,8 @@ import org.apache.logging.log4j.Logger;
import java.time.LocalTime;
import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.HashSet;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
@@ -60,9 +64,16 @@ public class StatisticsAutoCollector extends StatisticsCollector {
}
try {
TableIf table = job.getKey();
+ if (!supportAutoAnalyze(table)) {
+ continue;
+ }
Set<String> columns = job.getValue()
.stream()
- .filter(c -> needAnalyzeColumn(table, c))
+ .filter(c -> {
+ boolean needAnalyzeColumn = needAnalyzeColumn(table, c);
+ LOG.info("Need analyze column " + c + " ? " + needAnalyzeColumn);
+ return needAnalyzeColumn;
+ })
.collect(Collectors.toSet());
processOneJob(table, columns);
} catch (Exception e) {
@@ -100,22 +111,92 @@ public class StatisticsAutoCollector extends StatisticsCollector {
}
protected void processOneJob(TableIf table, Set<String> columns) throws DdlException {
- Set<String> collect = columns.stream().filter(c -> needAnalyzeColumn(table, c)).collect(Collectors.toSet());
- if (collect.isEmpty()) {
+ appendPartitionColumns(table, columns);
+ if (columns.isEmpty()) {
return;
}
AnalysisInfo analyzeJob = createAnalyzeJobForTbl(table, columns);
+ LOG.info("Analyze job : {}", analyzeJob.toString());
createSystemAnalysisJob(analyzeJob);
}
+ protected void appendPartitionColumns(TableIf table, Set<String> columns) {
+ if (!(table instanceof OlapTable)) {
+ return;
+ }
+ AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
+ TableStatsMeta tableStatsStatus = manager.findTableStatsStatus(table.getId());
+ if (tableStatsStatus != null && tableStatsStatus.newPartitionLoaded.get()) {
+ OlapTable olapTable = (OlapTable) table;
+ columns.addAll(olapTable.getPartitionNames());
+ }
+ }
+
protected boolean needAnalyzeColumn(TableIf table, String column) {
- //TODO: Calculate column health value.
- return true;
+ AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
+ TableStatsMeta tableStatsStatus = manager.findTableStatsStatus(table.getId());
+ if (tableStatsStatus == null) {
+ return true;
+ }
+ if (tableStatsStatus.userInjected) {
+ return false;
+ }
+ ColStatsMeta columnStatsMeta = tableStatsStatus.findColumnStatsMeta(column);
+ if (columnStatsMeta == null) {
+ return true;
+ }
+ if (table instanceof OlapTable) {
+ long currentUpdatedRows = tableStatsStatus.updatedRows.get();
+ long lastAnalyzeUpdateRows = columnStatsMeta.updatedRows;
+ if (lastAnalyzeUpdateRows == 0 && currentUpdatedRows > 0) {
+ return true;
+ }
+ OlapTable olapTable = (OlapTable) table;
+ if (tableStatsStatus.newPartitionLoaded.get() && olapTable.isPartitionColumn(column)) {
+ return true;
+ }
+ if (columnStatsMeta.rowCount == 0 && olapTable.getRowCount() > 0) {
+ return true;
+ }
+ if (currentUpdatedRows == lastAnalyzeUpdateRows) {
+ return false;
+ }
+ double healthValue = ((double) (currentUpdatedRows - lastAnalyzeUpdateRows)
+ / (double) currentUpdatedRows) * 100.0;
+ LOG.info("Column " + column + " health value is " + healthValue);
+ return healthValue < StatisticsUtil.getTableStatsHealthThreshold();
+ } else {
+ if (!(table instanceof HMSExternalTable)) {
+ return false;
+ }
+ HMSExternalTable hmsTable = (HMSExternalTable) table;
+ if (!hmsTable.getDlaType().equals(DLAType.HIVE)) {
+ return false;
+ }
+ return System.currentTimeMillis()
+ - tableStatsStatus.updatedTime > StatisticsUtil.getExternalTableAutoAnalyzeIntervalInMillis();
+ }
+ }
+
+ protected boolean supportAutoAnalyze(TableIf tableIf) {
+ if (tableIf == null) {
+ return false;
+ }
+ return tableIf instanceof OlapTable
+ || tableIf instanceof HMSExternalTable
+ && ((HMSExternalTable) tableIf).getDlaType().equals(HMSExternalTable.DLAType.HIVE);
}
protected AnalysisInfo createAnalyzeJobForTbl(TableIf table, Set<String> columns) {
AnalysisMethod analysisMethod = table.getDataSize(true) >= StatisticsUtil.getHugeTableLowerBoundSizeInBytes()
? AnalysisMethod.SAMPLE : AnalysisMethod.FULL;
+ AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
+ TableStatsMeta tableStatsStatus = manager.findTableStatsStatus(table.getId());
+ long rowCount = table.getRowCount();
+ Map<String, Set<String>> colToPartitions = new HashMap<>();
+ Set<String> dummyPartition = new HashSet<>();
+ dummyPartition.add("dummy partition");
+ columns.stream().forEach(c -> colToPartitions.put(c, dummyPartition));
return new AnalysisInfoBuilder()
.setJobId(Env.getCurrentEnv().getNextId())
.setCatalogId(table.getDatabase().getCatalog().getId())
@@ -133,7 +214,9 @@ public class StatisticsAutoCollector extends StatisticsCollector {
.setLastExecTimeInMs(System.currentTimeMillis())
.setJobType(JobType.SYSTEM)
.setTblUpdateTime(table.getUpdateTime())
- .setEmptyJob(table instanceof OlapTable && table.getRowCount() == 0)
+ .setRowCount(rowCount)
+ .setUpdateRows(tableStatsStatus == null ? 0 : tableStatsStatus.updatedRows.get())
+ .setColToPartitions(colToPartitions)
.build();
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
index 73d0d1340ad..71bb71d3cda 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
@@ -19,6 +19,7 @@ package org.apache.doris.statistics;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Table;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.util.MasterDaemon;
@@ -28,7 +29,7 @@ import org.apache.doris.statistics.util.StatisticsUtil;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
-import java.util.Collections;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -70,34 +71,44 @@ public class StatisticsJobAppender extends MasterDaemon {
protected void appendJobs() {
AnalysisManager manager = Env.getCurrentEnv().getAnalysisManager();
+ // LOG.info("Append column to high priority job map.");
appendColumnsToJobs(manager.highPriorityColumns, manager.highPriorityJobs);
+ // LOG.info("Append column to mid priority job map.");
appendColumnsToJobs(manager.midPriorityColumns, manager.midPriorityJobs);
- appendToLowQueue(manager.lowPriorityJobs);
+ if (StatisticsUtil.enableAutoAnalyzeInternalCatalog()) {
+ // LOG.info("Append column to low priority job map.");
+ appendToLowQueue(manager.lowPriorityJobs);
+ }
}
protected void appendColumnsToJobs(Queue<HighPriorityColumn> columnQueue, Map<TableIf, Set<String>> jobsMap) {
int size = columnQueue.size();
for (int i = 0; i < size; i++) {
HighPriorityColumn column = columnQueue.poll();
+ LOG.info("Process column " + column.tblId + "." + column.colName);
TableIf table = StatisticsUtil.findTable(column.catalogId, column.dbId, column.tblId);
synchronized (jobsMap) {
// If job map reach the upper limit, stop putting new jobs.
if (!jobsMap.containsKey(table) && jobsMap.size() >= JOB_MAP_SIZE) {
+ LOG.info("Job map full.");
break;
}
if (jobsMap.containsKey(table)) {
jobsMap.get(table).add(column.colName);
} else {
- jobsMap.put(table, Collections.singleton(column.colName));
+ HashSet<String> columns = new HashSet<>();
+ columns.add(column.colName);
+ jobsMap.put(table, columns);
}
+ LOG.info("Column " + column.tblId + "." + column.colName + " added");
}
}
}
- protected void appendToLowQueue(Map<TableIf, Set<String>> jobsMap) {
-
+ protected void appendToLowQueue(Map<TableIf, Set<String>> jobsMap) {
InternalCatalog catalog = Env.getCurrentInternalCatalog();
List<Long> sortedDbs = catalog.getDbIds().stream().sorted().collect(Collectors.toList());
+ int batchSize = 100;
for (long dbId : sortedDbs) {
if (dbId < currentDbId
|| StatisticConstants.SYSTEM_DBS.contains(catalog.getDbNullable(dbId).getFullName())) {
@@ -108,11 +119,11 @@ public class StatisticsJobAppender extends MasterDaemon {
List<Table> tables = db.get().getTables().stream()
.sorted((t1, t2) -> (int) (t1.getId() - t2.getId())).collect(Collectors.toList());
for (Table t : tables) {
- if (t.getId() <= currentTableId) {
+ if (!(t instanceof OlapTable) || t.getId() <= currentTableId) {
continue;
}
synchronized (jobsMap) {
- // If job map reach the upper limit, stop putting new jobs.
+ // If job map reach the upper limit, stop adding new jobs.
if (!jobsMap.containsKey(t) && jobsMap.size() >= JOB_MAP_SIZE) {
return;
}
@@ -126,6 +137,9 @@ public class StatisticsJobAppender extends MasterDaemon {
}
}
currentTableId = t.getId();
+ if (--batchSize <= 0) {
+ return;
+ }
}
}
// All tables have been processed once, reset for the next loop.
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java
index 9231c6a2bc7..96ca0aab54c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/TableStatsMeta.java
@@ -126,11 +126,6 @@ public class TableStatsMeta implements Writable {
return colNameToColStatsMeta.keySet();
}
- public void reset() {
- updatedTime = 0;
- colNameToColStatsMeta.values().forEach(ColStatsMeta::clear);
- }
-
public void update(AnalysisInfo analyzedJob, TableIf tableIf) {
updatedTime = analyzedJob.tblUpdateTime;
userInjected = analyzedJob.userInject;
@@ -145,34 +140,34 @@ public class TableStatsMeta implements Writable {
for (String col : cols) {
ColStatsMeta colStatsMeta = colNameToColStatsMeta.get(col);
if (colStatsMeta == null) {
- colNameToColStatsMeta.put(col, new ColStatsMeta(updatedTime,
- analyzedJob.analysisMethod, analyzedJob.analysisType, analyzedJob.jobType, 0));
+ colNameToColStatsMeta.put(col, new ColStatsMeta(updatedTime, analyzedJob.analysisMethod,
+ analyzedJob.analysisType, analyzedJob.jobType, 0, analyzedJob.rowCount,
+ analyzedJob.updateRows));
} else {
colStatsMeta.updatedTime = updatedTime;
colStatsMeta.analysisType = analyzedJob.analysisType;
colStatsMeta.analysisMethod = analyzedJob.analysisMethod;
colStatsMeta.jobType = analyzedJob.jobType;
+ colStatsMeta.updatedRows = analyzedJob.updateRows;
+ colStatsMeta.rowCount = analyzedJob.rowCount;
}
}
jobType = analyzedJob.jobType;
if (tableIf != null) {
if (tableIf instanceof OlapTable) {
- rowCount = analyzedJob.emptyJob ? 0 : tableIf.getRowCount();
- }
- if (!analyzedJob.emptyJob && analyzedJob.colToPartitions.keySet()
- .containsAll(tableIf.getBaseSchema().stream()
- .filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
- .map(Column::getName).collect(Collectors.toSet()))) {
- updatedRows.set(0);
- newPartitionLoaded.set(false);
- }
- if (tableIf instanceof OlapTable) {
+ rowCount = analyzedJob.rowCount;
PartitionInfo partitionInfo = ((OlapTable) tableIf).getPartitionInfo();
- if (partitionInfo != null && analyzedJob.colToPartitions.keySet()
+ if (analyzedJob.rowCount != 0 && partitionInfo != null && analyzedJob.colToPartitions.keySet()
.containsAll(partitionInfo.getPartitionColumns().stream()
- .map(Column::getName).collect(Collectors.toSet()))) {
+ .map(Column::getName).collect(Collectors.toSet()))) {
newPartitionLoaded.set(false);
}
+ if (analyzedJob.rowCount != 0 && analyzedJob.colToPartitions.keySet()
+ .containsAll(tableIf.getBaseSchema().stream()
+ .filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
+ .map(Column::getName).collect(Collectors.toSet()))) {
+ userInjected = false;
+ }
}
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
index 8ee08d57e69..65c2ee9e6da 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
@@ -867,6 +867,16 @@ public class StatisticsUtil {
return false;
}
+ public static boolean enableAutoAnalyzeInternalCatalog() {
+ try {
+ return findConfigFromGlobalSessionVar(
+ SessionVariable.ENABLE_AUTO_ANALYZE_INTERNAL_CATALOG).enableAutoAnalyzeInternalCatalog;
+ } catch (Exception e) {
+ LOG.warn("Fail to get value of enable auto analyze internal catalog, return false by default", e);
+ }
+ return false;
+ }
+
public static int getInsertMergeCount() {
try {
return findConfigFromGlobalSessionVar(SessionVariable.STATS_INSERT_MERGE_ITEM_COUNT)
diff --git a/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisJobTest.java b/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisJobTest.java
index 1bf2041bb4f..cb2637d5cf6 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisJobTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisJobTest.java
@@ -184,7 +184,7 @@ public class AnalysisJobTest {
protected void syncLoadStats() {
}
};
- job.writeBuf();
+ job.flushBuffer();
Assertions.assertEquals(0, job.queryFinished.size());
}
@@ -210,7 +210,7 @@ public class AnalysisJobTest {
job.buf.add(new ColStatsData());
job.queryFinished = new HashSet<>();
job.queryFinished.add(task2);
- job.writeBuf();
+ job.flushBuffer();
Assertions.assertEquals(0, job.queryFinished.size());
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java b/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java
index f8a77fe06db..4e8bbfe5aff 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/statistics/AnalysisManagerTest.java
@@ -306,7 +306,7 @@ public class AnalysisManagerTest {
Assertions.assertFalse(olapTable.needReAnalyzeTable(stats2));
TableStatsMeta stats3 = new TableStatsMeta(0, new AnalysisInfoBuilder()
- .setColToPartitions(new HashMap<>()).setEmptyJob(true).setColName("col1").build(), olapTable);
+ .setColToPartitions(new HashMap<>()).setRowCount(0).setColName("col1").build(), olapTable);
Assertions.assertTrue(olapTable.needReAnalyzeTable(stats3));
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 02/07: High priority queue and map. (#31509)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lijibing pushed a commit to branch high-priority-column
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 404d160e81e3eae8cbcf9190f6d60fcdd97e7cb6
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Wed Feb 28 12:01:55 2024 +0800
High priority queue and map. (#31509)
---
.../main/java/org/apache/doris/catalog/Env.java | 20 +-
.../apache/doris/statistics/AnalysisManager.java | 33 +-
.../doris/statistics/StatisticsAutoCollector.java | 202 ++------
.../doris/statistics/StatisticsCollector.java | 4 -
.../doris/statistics/StatisticsJobAppender.java | 135 +++++
.../statistics/StatisticsAutoCollectorTest.java | 560 ---------------------
6 files changed, 213 insertions(+), 741 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
index 2c5c90ee895..1974a316ec5 100755
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Env.java
@@ -243,6 +243,7 @@ import org.apache.doris.statistics.AnalysisManager;
import org.apache.doris.statistics.StatisticsAutoCollector;
import org.apache.doris.statistics.StatisticsCache;
import org.apache.doris.statistics.StatisticsCleaner;
+import org.apache.doris.statistics.StatisticsJobAppender;
import org.apache.doris.statistics.query.QueryStats;
import org.apache.doris.system.Backend;
import org.apache.doris.system.Frontend;
@@ -519,6 +520,8 @@ public class Env {
private StatisticsAutoCollector statisticsAutoCollector;
+ private StatisticsJobAppender statisticsJobAppender;
+
private HiveTransactionMgr hiveTransactionMgr;
private TopicPublisherThread topicPublisherThread;
@@ -748,6 +751,7 @@ public class Env {
this.analysisManager = new AnalysisManager();
this.statisticsCleaner = new StatisticsCleaner();
this.statisticsAutoCollector = new StatisticsAutoCollector();
+ this.statisticsJobAppender = new StatisticsJobAppender();
this.globalFunctionMgr = new GlobalFunctionMgr();
this.workloadGroupMgr = new WorkloadGroupMgr();
this.workloadSchedPolicyMgr = new WorkloadSchedPolicyMgr();
@@ -1028,13 +1032,6 @@ public class Env {
// If not using bdb, we need to notify the FE type transfer manually.
notifyNewFETypeTransfer(FrontendNodeType.MASTER);
}
- if (statisticsCleaner != null) {
- statisticsCleaner.start();
- }
- if (statisticsAutoCollector != null) {
- statisticsAutoCollector.start();
- }
-
queryCancelWorker.start();
TopicPublisher wgPublisher = new WorkloadGroupPublisher(this);
@@ -1672,6 +1669,11 @@ public class Env {
binlogGcer.start();
columnIdFlusher.start();
insertOverwriteManager.start();
+
+ // auto analyze related threads.
+ statisticsCleaner.start();
+ statisticsAutoCollector.start();
+ statisticsJobAppender.start();
}
// start threads that should running on all FE
@@ -6023,6 +6025,10 @@ public class Env {
return statisticsAutoCollector;
}
+ public StatisticsJobAppender getStatisticsJobAppender() {
+ return statisticsJobAppender;
+ }
+
public void alterMTMVRefreshInfo(AlterMTMVRefreshInfo info) {
AlterMTMV alter = new AlterMTMV(info.getMvName(), info.getRefreshInfo(), MTMVAlterOpType.ALTER_REFRESH_INFO);
this.alter.processAlterMTMV(alter, false);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
index cf3764719ba..ab883c79f8f 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
@@ -85,6 +85,7 @@ import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.HashSet;
+import java.util.LinkedHashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
@@ -108,15 +109,12 @@ public class AnalysisManager implements Writable {
private static final Logger LOG = LogManager.getLogger(AnalysisManager.class);
- /**
- * Mem only.
- */
- public final Queue<HighPriorityColumn> predicateColumns = new ArrayBlockingQueue<>(100);
-
- /**
- * Mem only.
- */
- public final Queue<HighPriorityColumn> queryColumns = new ArrayBlockingQueue<>(100);
+ private static final int COLUMN_QUEUE_SIZE = 1000;
+ public final Queue<HighPriorityColumn> highPriorityColumns = new ArrayBlockingQueue<>(COLUMN_QUEUE_SIZE);
+ public final Queue<HighPriorityColumn> midPriorityColumns = new ArrayBlockingQueue<>(COLUMN_QUEUE_SIZE);
+ public final Map<TableIf, Set<String>> highPriorityJobs = new LinkedHashMap<>();
+ public final Map<TableIf, Set<String>> midPriorityJobs = new LinkedHashMap<>();
+ public final Map<TableIf, Set<String>> lowPriorityJobs = new LinkedHashMap<>();
// Tracking running manually submitted async tasks, keep in mem only
protected final ConcurrentMap<Long, Map<Long, BaseAnalysisTask>> analysisJobIdToTaskMap = new ConcurrentHashMap<>();
@@ -170,11 +168,6 @@ public class AnalysisManager implements Writable {
public void createAnalysisJobs(AnalyzeDBStmt analyzeDBStmt, boolean proxy) throws DdlException, AnalysisException {
DatabaseIf<TableIf> db = analyzeDBStmt.getDb();
- // Using auto analyzer if user specifies.
- if (analyzeDBStmt.getAnalyzeProperties().getProperties().containsKey("use.auto.analyzer")) {
- Env.getCurrentEnv().getStatisticsAutoCollector().analyzeDb(db);
- return;
- }
List<AnalysisInfo> analysisInfos = buildAnalysisInfosForDB(db, analyzeDBStmt.getAnalyzeProperties());
if (!analyzeDBStmt.isSync()) {
sendJobId(analysisInfos, proxy);
@@ -222,6 +215,11 @@ public class AnalysisManager implements Writable {
// Each analyze stmt corresponding to an analysis job.
public void createAnalysisJob(AnalyzeTblStmt stmt, boolean proxy) throws DdlException {
+ // Using auto analyzer if user specifies.
+ if (stmt.getAnalyzeProperties().getProperties().containsKey("use.auto.analyzer")) {
+ Env.getCurrentEnv().getStatisticsAutoCollector().processOneJob(stmt.getTable(), stmt.getColumnNames());
+ return;
+ }
AnalysisInfo jobInfo = buildAndAssignJob(stmt);
if (jobInfo == null) {
return;
@@ -1158,11 +1156,11 @@ public class AnalysisManager implements Writable {
public void updateColumnUsedInPredicate(Set<Slot> slotReferences) {
- updateColumn(slotReferences, predicateColumns);
+ updateColumn(slotReferences, highPriorityColumns);
}
public void updateQueriedColumn(Collection<Slot> slotReferences) {
- updateColumn(slotReferences, queryColumns);
+ updateColumn(slotReferences, midPriorityColumns);
}
protected void updateColumn(Collection<Slot> slotReferences, Queue<HighPriorityColumn> queue) {
@@ -1172,7 +1170,8 @@ public class AnalysisManager implements Writable {
}
Optional<Column> optionalColumn = ((SlotReference) s).getColumn();
Optional<TableIf> optionalTable = ((SlotReference) s).getTable();
- if (optionalColumn.isPresent() && optionalTable.isPresent()) {
+ if (optionalColumn.isPresent() && optionalTable.isPresent()
+ && !StatisticsUtil.isUnsupportedType(optionalColumn.get().getType())) {
TableIf table = optionalTable.get();
DatabaseIf database = table.getDatabase();
if (database != null) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
index dbb7046467a..19b9c69db08 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
@@ -17,31 +17,24 @@
package org.apache.doris.statistics;
-import org.apache.doris.catalog.Column;
-import org.apache.doris.catalog.DatabaseIf;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
-import org.apache.doris.catalog.Partition;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
import org.apache.doris.common.util.TimeUtils;
-import org.apache.doris.datasource.CatalogIf;
-import org.apache.doris.datasource.hive.HMSExternalTable;
import org.apache.doris.statistics.AnalysisInfo.AnalysisMethod;
import org.apache.doris.statistics.AnalysisInfo.JobType;
import org.apache.doris.statistics.AnalysisInfo.ScheduleType;
import org.apache.doris.statistics.util.StatisticsUtil;
-import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.time.LocalTime;
import java.util.ArrayList;
-import java.util.HashMap;
-import java.util.List;
import java.util.Map;
+import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
@@ -59,8 +52,23 @@ public class StatisticsAutoCollector extends StatisticsCollector {
@Override
protected void collect() {
- if (canCollect()) {
- analyzeAll();
+ while (canCollect()) {
+ Map.Entry<TableIf, Set<String>> job = getJob();
+ if (job == null) {
+ // No more job to process, break and sleep.
+ break;
+ }
+ try {
+ TableIf table = job.getKey();
+ Set<String> columns = job.getValue()
+ .stream()
+ .filter(c -> needAnalyzeColumn(table, c))
+ .collect(Collectors.toSet());
+ processOneJob(table, columns);
+ } catch (Exception e) {
+ LOG.warn("Failed to analyze table {} with columns [{}]",
+ job.getKey().getName(), job.getValue().stream().collect(Collectors.joining(",")), e);
+ }
}
}
@@ -69,133 +77,56 @@ public class StatisticsAutoCollector extends StatisticsCollector {
&& StatisticsUtil.inAnalyzeTime(LocalTime.now(TimeUtils.getTimeZone().toZoneId()));
}
- protected void analyzeAll() {
- List<CatalogIf> catalogs = getCatalogsInOrder();
- for (CatalogIf ctl : catalogs) {
- if (!canCollect()) {
- analysisTaskExecutor.clear();
- break;
- }
- if (!ctl.enableAutoAnalyze()) {
- continue;
- }
- List<DatabaseIf> dbs = getDatabasesInOrder(ctl);
- for (DatabaseIf<TableIf> databaseIf : dbs) {
- if (!canCollect()) {
- analysisTaskExecutor.clear();
- break;
- }
- if (StatisticConstants.SYSTEM_DBS.contains(databaseIf.getFullName())) {
- continue;
- }
- try {
- analyzeDb(databaseIf);
- } catch (Throwable t) {
- LOG.warn("Failed to analyze database {}.{}", ctl.getName(), databaseIf.getFullName(), t);
- continue;
- }
- }
+ protected Map.Entry<TableIf, Set<String>> getJob() {
+ AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
+ Optional<Map.Entry<TableIf, Set<String>>> job = fetchJobFromMap(manager.highPriorityJobs);
+ if (job.isPresent()) {
+ return job.get();
}
+ job = fetchJobFromMap(manager.midPriorityJobs);
+ if (job.isPresent()) {
+ return job.get();
+ }
+ job = fetchJobFromMap(manager.lowPriorityJobs);
+ return job.isPresent() ? job.get() : null;
}
- public List<CatalogIf> getCatalogsInOrder() {
- return Env.getCurrentEnv().getCatalogMgr().getCopyOfCatalog().stream()
- .sorted((c1, c2) -> (int) (c1.getId() - c2.getId())).collect(Collectors.toList());
- }
-
- public List<DatabaseIf<? extends TableIf>> getDatabasesInOrder(CatalogIf<DatabaseIf> catalog) {
- return catalog.getAllDbs().stream()
- .sorted((d1, d2) -> (int) (d1.getId() - d2.getId())).collect(Collectors.toList());
- }
-
- public List<TableIf> getTablesInOrder(DatabaseIf<? extends TableIf> db) {
- return db.getTables().stream()
- .sorted((t1, t2) -> (int) (t1.getId() - t2.getId())).collect(Collectors.toList());
- }
-
- public void analyzeDb(DatabaseIf<TableIf> databaseIf) throws DdlException {
- List<AnalysisInfo> analysisInfos = constructAnalysisInfo(databaseIf);
- for (AnalysisInfo analysisInfo : analysisInfos) {
- try {
- if (!canCollect()) {
- analysisTaskExecutor.clear();
- break;
- }
- analysisInfo = getReAnalyzeRequiredPart(analysisInfo);
- if (analysisInfo == null) {
- continue;
- }
- createSystemAnalysisJob(analysisInfo);
- } catch (Throwable t) {
- analysisInfo.message = t.getMessage();
- LOG.warn("Failed to auto analyze table {}.{}, reason {}",
- databaseIf.getFullName(), analysisInfo.tblId, analysisInfo.message, t);
- continue;
- }
+ protected Optional<Map.Entry<TableIf, Set<String>>> fetchJobFromMap(Map<TableIf, Set<String>> jobMap) {
+ synchronized (jobMap) {
+ Optional<Map.Entry<TableIf, Set<String>>> first = jobMap.entrySet().stream().findFirst();
+ first.ifPresent(entry -> jobMap.remove(entry.getKey()));
+ return first;
}
}
- protected List<AnalysisInfo> constructAnalysisInfo(DatabaseIf<? extends TableIf> db) {
- List<AnalysisInfo> analysisInfos = new ArrayList<>();
- for (TableIf table : getTablesInOrder(db)) {
- try {
- if (skip(table)) {
- continue;
- }
- createAnalyzeJobForTbl(db, analysisInfos, table);
- } catch (Throwable t) {
- LOG.warn("Failed to analyze table {}.{}.{}",
- db.getCatalog().getName(), db.getFullName(), table.getName(), t);
- continue;
- }
+ protected void processOneJob(TableIf table, Set<String> columns) throws DdlException {
+ Set<String> collect = columns.stream().filter(c -> needAnalyzeColumn(table, c)).collect(Collectors.toSet());
+ if (collect.isEmpty()) {
+ return;
}
- return analysisInfos;
+ AnalysisInfo analyzeJob = createAnalyzeJobForTbl(table, columns);
+ createSystemAnalysisJob(analyzeJob);
}
- // return true if skip auto analyze this time.
- protected boolean skip(TableIf table) {
- if (!(table instanceof OlapTable || table instanceof HMSExternalTable)) {
- return true;
- }
- // For now, only support Hive HMS table auto collection.
- if (table instanceof HMSExternalTable
- && !((HMSExternalTable) table).getDlaType().equals(HMSExternalTable.DLAType.HIVE)) {
- return true;
- }
- if (table.getDataSize(true) < StatisticsUtil.getHugeTableLowerBoundSizeInBytes() * 5) {
- return false;
- }
- TableStatsMeta tableStats = Env.getCurrentEnv().getAnalysisManager().findTableStatsStatus(table.getId());
- // means it's never got analyzed or new partition loaded data.
- if (tableStats == null || tableStats.newPartitionLoaded.get()) {
- return false;
- }
- if (tableStats.userInjected) {
- return true;
- }
- return System.currentTimeMillis()
- - tableStats.updatedTime < StatisticsUtil.getHugeTableAutoAnalyzeIntervalInMillis();
+ protected boolean needAnalyzeColumn(TableIf table, String column) {
+ //TODO: Calculate column health value.
+ return true;
}
- protected void createAnalyzeJobForTbl(DatabaseIf<? extends TableIf> db,
- List<AnalysisInfo> analysisInfos, TableIf table) {
+ protected AnalysisInfo createAnalyzeJobForTbl(TableIf table, Set<String> columns) {
AnalysisMethod analysisMethod = table.getDataSize(true) >= StatisticsUtil.getHugeTableLowerBoundSizeInBytes()
? AnalysisMethod.SAMPLE : AnalysisMethod.FULL;
- AnalysisInfo jobInfo = new AnalysisInfoBuilder()
+ return new AnalysisInfoBuilder()
.setJobId(Env.getCurrentEnv().getNextId())
- .setCatalogId(db.getCatalog().getId())
- .setDBId(db.getId())
+ .setCatalogId(table.getDatabase().getCatalog().getId())
+ .setDBId(table.getDatabase().getId())
.setTblId(table.getId())
- .setColName(
- table.getSchemaAllIndexes(false).stream()
- .filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
- .map(Column::getName).collect(Collectors.joining(","))
- )
+ .setColName(columns.stream().collect(Collectors.joining(",")))
.setAnalysisType(AnalysisInfo.AnalysisType.FUNDAMENTALS)
.setAnalysisMode(AnalysisInfo.AnalysisMode.INCREMENTAL)
.setAnalysisMethod(analysisMethod)
.setSampleRows(analysisMethod.equals(AnalysisMethod.SAMPLE)
- ? StatisticsUtil.getHugeTableSampleRows() : -1)
+ ? StatisticsUtil.getHugeTableSampleRows() : -1)
.setScheduleType(ScheduleType.AUTOMATIC)
.setState(AnalysisState.PENDING)
.setTaskIds(new ArrayList<>())
@@ -204,40 +135,5 @@ public class StatisticsAutoCollector extends StatisticsCollector {
.setTblUpdateTime(table.getUpdateTime())
.setEmptyJob(table instanceof OlapTable && table.getRowCount() == 0)
.build();
- analysisInfos.add(jobInfo);
- }
-
- @VisibleForTesting
- protected AnalysisInfo getReAnalyzeRequiredPart(AnalysisInfo jobInfo) {
- TableIf table = StatisticsUtil.findTable(jobInfo.catalogId, jobInfo.dbId, jobInfo.tblId);
- // Skip tables that are too wide.
- if (table.getBaseSchema().size() > StatisticsUtil.getAutoAnalyzeTableWidthThreshold()) {
- return null;
- }
-
- AnalysisManager analysisManager = Env.getServingEnv().getAnalysisManager();
- TableStatsMeta tblStats = analysisManager.findTableStatsStatus(table.getId());
-
- Map<String, Set<String>> needRunPartitions = null;
- String colNames = jobInfo.colName;
- if (table.needReAnalyzeTable(tblStats)) {
- needRunPartitions = table.findReAnalyzeNeededPartitions();
- } else if (table instanceof OlapTable && tblStats.newPartitionLoaded.get()) {
- OlapTable olapTable = (OlapTable) table;
- needRunPartitions = new HashMap<>();
- Set<String> partitionColumnNames = olapTable.getPartitionInfo().getPartitionColumns().stream()
- .map(Column::getName).collect(Collectors.toSet());
- colNames = partitionColumnNames.stream().collect(Collectors.joining(","));
- Set<String> partitionNames = olapTable.getAllPartitions().stream()
- .map(Partition::getName).collect(Collectors.toSet());
- for (String column : partitionColumnNames) {
- needRunPartitions.put(column, partitionNames);
- }
- }
-
- if (needRunPartitions == null || needRunPartitions.isEmpty()) {
- return null;
- }
- return new AnalysisInfoBuilder(jobInfo).setColName(colNames).setColToPartitions(needRunPartitions).build();
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsCollector.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsCollector.java
index 0985b9b2b95..e26d3170178 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsCollector.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsCollector.java
@@ -61,10 +61,6 @@ public abstract class StatisticsCollector extends MasterDaemon {
@VisibleForTesting
protected void createSystemAnalysisJob(AnalysisInfo jobInfo)
throws DdlException {
- if (jobInfo.colToPartitions.isEmpty()) {
- // No statistics need to be collected or updated
- return;
- }
Map<Long, BaseAnalysisTask> analysisTasks = new HashMap<>();
AnalysisManager analysisManager = Env.getCurrentEnv().getAnalysisManager();
analysisManager.createTaskForEachColumns(jobInfo, analysisTasks, false);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
new file mode 100644
index 00000000000..73d0d1340ad
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
@@ -0,0 +1,135 @@
+// 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.statistics;
+
+import org.apache.doris.catalog.Database;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.Table;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.common.util.MasterDaemon;
+import org.apache.doris.datasource.InternalCatalog;
+import org.apache.doris.statistics.util.StatisticsUtil;
+
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.Logger;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Queue;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+public class StatisticsJobAppender extends MasterDaemon {
+
+ private static final Logger LOG = LogManager.getLogger(StatisticsJobAppender.class);
+
+ public static final long INTERVAL = 1000;
+ public static final int JOB_MAP_SIZE = 1000;
+
+ private long currentDbId;
+ private long currentTableId;
+
+ public StatisticsJobAppender() {
+ super("Statistics Job Appender", INTERVAL);
+ }
+
+ @Override
+ protected void runAfterCatalogReady() {
+ if (!StatisticsUtil.enableAutoAnalyze()) {
+ return;
+ }
+ if (!Env.getCurrentEnv().isMaster()) {
+ return;
+ }
+ if (!StatisticsUtil.statsTblAvailable()) {
+ LOG.info("Stats table not available, skip");
+ return;
+ }
+ if (Env.isCheckpointThread()) {
+ return;
+ }
+ appendJobs();
+ }
+
+ protected void appendJobs() {
+ AnalysisManager manager = Env.getCurrentEnv().getAnalysisManager();
+ appendColumnsToJobs(manager.highPriorityColumns, manager.highPriorityJobs);
+ appendColumnsToJobs(manager.midPriorityColumns, manager.midPriorityJobs);
+ appendToLowQueue(manager.lowPriorityJobs);
+ }
+
+ protected void appendColumnsToJobs(Queue<HighPriorityColumn> columnQueue, Map<TableIf, Set<String>> jobsMap) {
+ int size = columnQueue.size();
+ for (int i = 0; i < size; i++) {
+ HighPriorityColumn column = columnQueue.poll();
+ TableIf table = StatisticsUtil.findTable(column.catalogId, column.dbId, column.tblId);
+ synchronized (jobsMap) {
+ // If job map reach the upper limit, stop putting new jobs.
+ if (!jobsMap.containsKey(table) && jobsMap.size() >= JOB_MAP_SIZE) {
+ break;
+ }
+ if (jobsMap.containsKey(table)) {
+ jobsMap.get(table).add(column.colName);
+ } else {
+ jobsMap.put(table, Collections.singleton(column.colName));
+ }
+ }
+ }
+ }
+
+ protected void appendToLowQueue(Map<TableIf, Set<String>> jobsMap) {
+
+ InternalCatalog catalog = Env.getCurrentInternalCatalog();
+ List<Long> sortedDbs = catalog.getDbIds().stream().sorted().collect(Collectors.toList());
+ for (long dbId : sortedDbs) {
+ if (dbId < currentDbId
+ || StatisticConstants.SYSTEM_DBS.contains(catalog.getDbNullable(dbId).getFullName())) {
+ continue;
+ }
+ currentDbId = dbId;
+ Optional<Database> db = catalog.getDb(dbId);
+ List<Table> tables = db.get().getTables().stream()
+ .sorted((t1, t2) -> (int) (t1.getId() - t2.getId())).collect(Collectors.toList());
+ for (Table t : tables) {
+ if (t.getId() <= currentTableId) {
+ continue;
+ }
+ synchronized (jobsMap) {
+ // If job map reach the upper limit, stop putting new jobs.
+ if (!jobsMap.containsKey(t) && jobsMap.size() >= JOB_MAP_SIZE) {
+ return;
+ }
+ Set<String> columns
+ = t.getColumns().stream().filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
+ .map(c -> c.getName()).collect(Collectors.toSet());
+ if (jobsMap.containsKey(t)) {
+ jobsMap.get(t).addAll(columns);
+ } else {
+ jobsMap.put(t, columns);
+ }
+ }
+ currentTableId = t.getId();
+ }
+ }
+ // All tables have been processed once, reset for the next loop.
+ currentDbId = 0;
+ currentTableId = 0;
+ }
+}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java b/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java
deleted file mode 100644
index 678e7580f8e..00000000000
--- a/fe/fe-core/src/test/java/org/apache/doris/statistics/StatisticsAutoCollectorTest.java
+++ /dev/null
@@ -1,560 +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.statistics;
-
-import org.apache.doris.catalog.Column;
-import org.apache.doris.catalog.Database;
-import org.apache.doris.catalog.DatabaseIf;
-import org.apache.doris.catalog.Env;
-import org.apache.doris.catalog.EnvFactory;
-import org.apache.doris.catalog.OlapTable;
-import org.apache.doris.catalog.PrimitiveType;
-import org.apache.doris.catalog.Table;
-import org.apache.doris.catalog.TableIf;
-import org.apache.doris.catalog.Type;
-import org.apache.doris.catalog.View;
-import org.apache.doris.common.Config;
-import org.apache.doris.common.DdlException;
-import org.apache.doris.common.FeConstants;
-import org.apache.doris.datasource.CatalogIf;
-import org.apache.doris.datasource.InternalCatalog;
-import org.apache.doris.statistics.AnalysisInfo.AnalysisMethod;
-import org.apache.doris.statistics.AnalysisInfo.AnalysisType;
-import org.apache.doris.statistics.AnalysisInfo.JobType;
-import org.apache.doris.statistics.util.StatisticsUtil;
-
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.collect.Sets;
-import mockit.Expectations;
-import mockit.Injectable;
-import mockit.Mock;
-import mockit.MockUp;
-import mockit.Mocked;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Test;
-
-import java.time.LocalTime;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.HashMap;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Map;
-import java.util.Set;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-public class StatisticsAutoCollectorTest {
-
- @Test
- public void testAnalyzeAll(@Injectable AnalysisInfo analysisInfo) {
- new MockUp<CatalogIf>() {
- @Mock
- public Collection<DatabaseIf> getAllDbs() {
- Database db1 = new Database(1, FeConstants.INTERNAL_DB_NAME);
- Database db2 = new Database(2, "anyDB");
- List<DatabaseIf> databaseIfs = new ArrayList<>();
- databaseIfs.add(db1);
- databaseIfs.add(db2);
- return databaseIfs;
- }
- };
- new MockUp<StatisticsAutoCollector>() {
- @Mock
- public List<AnalysisInfo> constructAnalysisInfo(DatabaseIf<TableIf> db) {
- return Arrays.asList(analysisInfo, analysisInfo);
- }
-
- int count = 0;
-
- @Mock
- public AnalysisInfo getReAnalyzeRequiredPart(AnalysisInfo jobInfo) {
- return count++ == 0 ? null : jobInfo;
- }
-
- @Mock
- public void createSystemAnalysisJob(AnalysisInfo jobInfo)
- throws DdlException {
-
- }
- };
-
- StatisticsAutoCollector saa = new StatisticsAutoCollector();
- saa.runAfterCatalogReady();
- new Expectations() {
- {
- try {
- saa.createSystemAnalysisJob((AnalysisInfo) any);
- times = 1;
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
- };
- }
-
- @Test
- public void testConstructAnalysisInfo(
- @Injectable OlapTable o2, @Injectable View v) {
- new MockUp<Database>() {
- @Mock
- public List<Table> getTables() {
- List<Table> tableIfs = new ArrayList<>();
- tableIfs.add(o2);
- tableIfs.add(v);
- return tableIfs;
- }
-
- @Mock
- public String getFullName() {
- return "anyDb";
- }
- };
-
- new MockUp<OlapTable>() {
- @Mock
- public String getName() {
- return "anytable";
- }
-
- @Mock
- public List<Column> getSchemaAllIndexes(boolean full) {
- List<Column> columns = new ArrayList<>();
- columns.add(new Column("c1", PrimitiveType.INT));
- columns.add(new Column("c2", PrimitiveType.HLL));
- return columns;
- }
- };
- StatisticsAutoCollector saa = new StatisticsAutoCollector();
- List<AnalysisInfo> analysisInfoList = saa.constructAnalysisInfo(new Database(1, "anydb"));
- Assertions.assertEquals(1, analysisInfoList.size());
- Assertions.assertEquals("c1", analysisInfoList.get(0).colName.split(",")[0]);
- }
-
- @Test
- public void testGetReAnalyzeRequiredPart0() {
-
- TableIf tableIf = new OlapTable();
-
- new MockUp<OlapTable>() {
- @Mock
- protected Map<String, Set<String>> findReAnalyzeNeededPartitions() {
- Set<String> partitionNames = new HashSet<>();
- partitionNames.add("p1");
- partitionNames.add("p2");
- Map<String, Set<String>> map = new HashMap<>();
- map.put("col1", partitionNames);
- return map;
- }
-
- @Mock
- public long getRowCount() {
- return 100;
- }
-
- @Mock
- public List<Column> getBaseSchema() {
- return Lists.newArrayList(new Column("col1", Type.INT), new Column("col2", Type.INT));
- }
- };
-
- new MockUp<StatisticsUtil>() {
- @Mock
- public TableIf findTable(long catalogName, long dbName, long tblName) {
- return tableIf;
- }
- };
- AnalysisInfo analysisInfo = new AnalysisInfoBuilder().setAnalysisMethod(AnalysisMethod.FULL)
- .setColToPartitions(new HashMap<>()).setAnalysisType(
- AnalysisType.FUNDAMENTALS).setColName("col1").setJobType(JobType.SYSTEM).build();
- new MockUp<AnalysisManager>() {
-
- int count = 0;
-
- TableStatsMeta[] tableStatsArr =
- new TableStatsMeta[] {new TableStatsMeta(0, analysisInfo, tableIf),
- new TableStatsMeta(0, analysisInfo, tableIf), null};
-
- {
- tableStatsArr[0].updatedRows.addAndGet(100);
- tableStatsArr[1].updatedRows.addAndGet(0);
- }
-
- @Mock
- public TableStatsMeta findTableStatsStatus(long tblId) {
- return tableStatsArr[count++];
- }
- };
-
- new MockUp<StatisticsAutoCollector>() {
- @Mock
- public AnalysisInfo getAnalysisJobInfo(AnalysisInfo jobInfo, TableIf table,
- Set<String> needRunPartitions) {
- return new AnalysisInfoBuilder().build();
- }
- };
- StatisticsAutoCollector statisticsAutoCollector = new StatisticsAutoCollector();
- AnalysisInfo analysisInfo2 = new AnalysisInfoBuilder()
- .setCatalogId(0)
- .setDBId(0)
- .setTblId(0).build();
- Assertions.assertNotNull(statisticsAutoCollector.getReAnalyzeRequiredPart(analysisInfo2));
- // uncomment it when updatedRows gets ready
- // Assertions.assertNull(statisticsAutoCollector.getReAnalyzeRequiredPart(analysisInfo2));
- Assertions.assertNotNull(statisticsAutoCollector.getReAnalyzeRequiredPart(analysisInfo2));
- }
-
- @Test
- public void testSkipWideTable() {
-
- TableIf tableIf = new OlapTable();
-
- new MockUp<OlapTable>() {
- @Mock
- public List<Column> getBaseSchema() {
- return Lists.newArrayList(new Column("col1", Type.INT), new Column("col2", Type.INT));
- }
- };
-
- new MockUp<StatisticsUtil>() {
- int count = 0;
- int [] thresholds = {1, 10};
- @Mock
- public TableIf findTable(long catalogName, long dbName, long tblName) {
- return tableIf;
- }
-
- @Mock
- public int getAutoAnalyzeTableWidthThreshold() {
- return thresholds[count++];
- }
- };
-
- new MockUp<OlapTable>() {
- @Mock
- public Map<String, Set<String>> findReAnalyzeNeededPartitions() {
- HashMap<String, Set<String>> ret = Maps.newHashMap();
- ret.put("key1", Sets.newHashSet());
- return ret;
- }
- };
-
- AnalysisInfo analysisInfo = new AnalysisInfoBuilder().build();
- StatisticsAutoCollector statisticsAutoCollector = new StatisticsAutoCollector();
- Assertions.assertNull(statisticsAutoCollector.getReAnalyzeRequiredPart(analysisInfo));
- Assertions.assertNotNull(statisticsAutoCollector.getReAnalyzeRequiredPart(analysisInfo));
- }
-
- @Test
- public void testLoop() {
- AtomicBoolean timeChecked = new AtomicBoolean();
- AtomicBoolean switchChecked = new AtomicBoolean();
- new MockUp<StatisticsUtil>() {
-
- @Mock
- public boolean inAnalyzeTime(LocalTime now) {
- timeChecked.set(true);
- return true;
- }
-
- @Mock
- public boolean enableAutoAnalyze() {
- switchChecked.set(true);
- return true;
- }
- };
- StatisticsAutoCollector autoCollector = new StatisticsAutoCollector();
- autoCollector.collect();
- Assertions.assertTrue(timeChecked.get() && switchChecked.get());
-
- }
-
- @Test
- public void checkAvailableThread() {
- StatisticsAutoCollector autoCollector = new StatisticsAutoCollector();
- Assertions.assertEquals(Config.auto_analyze_simultaneously_running_task_num,
- autoCollector.analysisTaskExecutor.executors.getMaximumPoolSize());
- }
-
- @Test
- public void testSkip(@Mocked OlapTable olapTable, @Mocked TableStatsMeta stats, @Mocked TableIf anyOtherTable) {
- new MockUp<OlapTable>() {
-
- @Mock
- public long getDataSize(boolean singleReplica) {
- return StatisticsUtil.getHugeTableLowerBoundSizeInBytes() * 5 + 1000000000;
- }
- };
-
- new MockUp<AnalysisManager>() {
-
- @Mock
- public TableStatsMeta findTableStatsStatus(long tblId) {
- return stats;
- }
- };
- // A very huge table has been updated recently, so we should skip it this time
- stats.updatedTime = System.currentTimeMillis() - 1000;
- stats.newPartitionLoaded = new AtomicBoolean();
- stats.newPartitionLoaded.set(true);
- StatisticsAutoCollector autoCollector = new StatisticsAutoCollector();
- // Test new partition loaded data for the first time. Not skip.
- Assertions.assertFalse(autoCollector.skip(olapTable));
- stats.newPartitionLoaded.set(false);
- // Assertions.assertTrue(autoCollector.skip(olapTable));
- // The update of this huge table is long time ago, so we shouldn't skip it this time
- stats.updatedTime = System.currentTimeMillis()
- - StatisticsUtil.getHugeTableAutoAnalyzeIntervalInMillis() - 10000;
- Assertions.assertFalse(autoCollector.skip(olapTable));
- new MockUp<AnalysisManager>() {
-
- @Mock
- public TableStatsMeta findTableStatsStatus(long tblId) {
- return null;
- }
- };
- // can't find table stats meta, which means this table never get analyzed, so we shouldn't skip it this time
- Assertions.assertFalse(autoCollector.skip(olapTable));
- new MockUp<AnalysisManager>() {
-
- @Mock
- public TableStatsMeta findTableStatsStatus(long tblId) {
- return stats;
- }
- };
- stats.userInjected = true;
- Assertions.assertTrue(autoCollector.skip(olapTable));
- // this is not olap table nor external table, so we should skip it this time
- Assertions.assertTrue(autoCollector.skip(anyOtherTable));
- }
-
- // For small table, use full
- @Test
- public void testCreateAnalyzeJobForTbl1(
- @Injectable OlapTable t1,
- @Injectable Database db
- ) throws Exception {
- new MockUp<Database>() {
-
- @Mock
- public CatalogIf getCatalog() {
- return Env.getCurrentInternalCatalog();
- }
-
- @Mock
- public long getId() {
- return 0;
- }
- };
- new MockUp<OlapTable>() {
-
- int count = 0;
-
- @Mock
- public List<Column> getBaseSchema() {
- return Lists.newArrayList(new Column("test", PrimitiveType.INT));
- }
-
- @Mock
- public long getDataSize(boolean singleReplica) {
- return StatisticsUtil.getHugeTableLowerBoundSizeInBytes() - 1;
- }
-
- @Mock
- public BaseAnalysisTask createAnalysisTask(AnalysisInfo info) {
- return new OlapAnalysisTask(info);
- }
-
- @Mock
- public List<Long> getMvColumnIndexIds(String columnName) {
- ArrayList<Long> objects = new ArrayList<>();
- objects.add(-1L);
- return objects;
- }
- };
-
- new MockUp<StatisticsUtil>() {
- @Mock
- public TableIf findTable(long catalogId, long dbId, long tblId) {
- return t1;
- }
- };
-
- StatisticsAutoCollector sac = new StatisticsAutoCollector();
- List<AnalysisInfo> jobInfos = new ArrayList<>();
- sac.createAnalyzeJobForTbl(db, jobInfos, t1);
- AnalysisInfo jobInfo = jobInfos.get(0);
- Map<String, Set<String>> colToPartitions = new HashMap<>();
- colToPartitions.put("test", new HashSet<String>() {
- {
- add("p1");
- }
- });
- jobInfo = new AnalysisInfoBuilder(jobInfo).setColToPartitions(colToPartitions).build();
- Map<Long, BaseAnalysisTask> analysisTasks = new HashMap<>();
- AnalysisManager analysisManager = Env.getCurrentEnv().getAnalysisManager();
- analysisManager.createTaskForEachColumns(jobInfo, analysisTasks, false);
- Assertions.assertEquals(1, analysisTasks.size());
- for (BaseAnalysisTask task : analysisTasks.values()) {
- Assertions.assertNull(task.getTableSample());
- }
- }
-
- // for big table, use sample
- @Test
- public void testCreateAnalyzeJobForTbl2(
- @Injectable OlapTable t1,
- @Injectable Database db
- ) throws Exception {
- new MockUp<Database>() {
-
- @Mock
- public CatalogIf getCatalog() {
- return Env.getCurrentInternalCatalog();
- }
-
- @Mock
- public long getId() {
- return 0;
- }
- };
- new MockUp<OlapTable>() {
-
- int count = 0;
-
- @Mock
- public List<Column> getBaseSchema() {
- return Lists.newArrayList(new Column("test", PrimitiveType.INT));
- }
-
- @Mock
- public long getDataSize(boolean singleReplica) {
- return StatisticsUtil.getHugeTableLowerBoundSizeInBytes() * 2;
- }
-
- @Mock
- public BaseAnalysisTask createAnalysisTask(AnalysisInfo info) {
- return new OlapAnalysisTask(info);
- }
-
- @Mock
- public List<Long> getMvColumnIndexIds(String columnName) {
- ArrayList<Long> objects = new ArrayList<>();
- objects.add(-1L);
- return objects;
- }
- };
-
- new MockUp<StatisticsUtil>() {
- @Mock
- public TableIf findTable(long catalogId, long dbId, long tblId) {
- return t1;
- }
- };
-
- StatisticsAutoCollector sac = new StatisticsAutoCollector();
- List<AnalysisInfo> jobInfos = new ArrayList<>();
- sac.createAnalyzeJobForTbl(db, jobInfos, t1);
- AnalysisInfo jobInfo = jobInfos.get(0);
- Map<String, Set<String>> colToPartitions = new HashMap<>();
- colToPartitions.put("test", new HashSet<String>() {
- {
- add("p1");
- }
- });
- jobInfo = new AnalysisInfoBuilder(jobInfo).setColToPartitions(colToPartitions).build();
- Map<Long, BaseAnalysisTask> analysisTasks = new HashMap<>();
- AnalysisManager analysisManager = Env.getCurrentEnv().getAnalysisManager();
- analysisManager.createTaskForEachColumns(jobInfo, analysisTasks, false);
- Assertions.assertEquals(1, analysisTasks.size());
- for (BaseAnalysisTask task : analysisTasks.values()) {
- Assertions.assertNotNull(task.getTableSample());
- }
- }
-
- @Test
- public void testDisableAuto1() throws Exception {
- InternalCatalog catalog1 = EnvFactory.getInstance().createInternalCatalog();
- List<CatalogIf> catalogs = Lists.newArrayList();
- catalogs.add(catalog1);
-
- new MockUp<StatisticsAutoCollector>() {
- @Mock
- public List<CatalogIf> getCatalogsInOrder() {
- return catalogs;
- }
-
- @Mock
- protected boolean canCollect() {
- return false;
- }
-
- };
-
- StatisticsAutoCollector sac = new StatisticsAutoCollector();
- new Expectations(catalog1) {{
- catalog1.enableAutoAnalyze();
- times = 0;
- }};
-
- sac.analyzeAll();
- }
-
- @Test
- public void testDisableAuto2() throws Exception {
- InternalCatalog catalog1 = EnvFactory.getInstance().createInternalCatalog();
- List<CatalogIf> catalogs = Lists.newArrayList();
- catalogs.add(catalog1);
-
- Database db1 = new Database();
- List<DatabaseIf<? extends TableIf>> dbs = Lists.newArrayList();
- dbs.add(db1);
-
- new MockUp<StatisticsAutoCollector>() {
- int count = 0;
- boolean[] canCollectReturn = {true, false};
- @Mock
- public List<CatalogIf> getCatalogsInOrder() {
- return catalogs;
- }
-
- @Mock
- public List<DatabaseIf<? extends TableIf>> getDatabasesInOrder(CatalogIf<DatabaseIf> catalog) {
- return dbs;
- }
-
- @Mock
- protected boolean canCollect() {
- return canCollectReturn[count++];
- }
-
- };
-
- StatisticsAutoCollector sac = new StatisticsAutoCollector();
- new Expectations(catalog1, db1) {{
- catalog1.enableAutoAnalyze();
- result = true;
- times = 1;
- db1.getFullName();
- times = 0;
- }};
-
- sac.analyzeAll();
- }
-}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 06/07: Check column health value earlier, show job priority. (#32064)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lijibing pushed a commit to branch high-priority-column
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 8d62e3d55bb2516c168f56c3327793e211f24c9b
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Mon Mar 11 19:46:20 2024 +0800
Check column health value earlier, show job priority. (#32064)
---
.../org/apache/doris/analysis/ShowAnalyzeStmt.java | 1 +
.../java/org/apache/doris/qe/SessionVariable.java | 2 +-
.../java/org/apache/doris/qe/ShowExecutor.java | 4 +-
.../org/apache/doris/statistics/AnalysisInfo.java | 8 +-
.../doris/statistics/AnalysisInfoBuilder.java | 10 +-
.../apache/doris/statistics/AnalysisManager.java | 10 +-
.../doris/statistics/FollowerColumnSender.java | 23 +++--
.../doris/statistics/HighPriorityColumn.java | 6 +-
.../org/apache/doris/statistics/JobPriority.java | 3 +-
.../doris/statistics/StatisticsAutoCollector.java | 102 ++++-----------------
.../doris/statistics/StatisticsJobAppender.java | 8 +-
.../doris/statistics/util/StatisticsUtil.java | 88 +++++++++++++++++-
gensrc/thrift/FrontendService.thrift | 6 +-
13 files changed, 161 insertions(+), 110 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowAnalyzeStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowAnalyzeStmt.java
index efcfc517024..734073901fe 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowAnalyzeStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/ShowAnalyzeStmt.java
@@ -62,6 +62,7 @@ public class ShowAnalyzeStmt extends ShowStmt {
.add("schedule_type")
.add("start_time")
.add("end_time")
+ .add("priority")
.build();
private long jobId;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
index 13c3e8baf0c..8c1c0326faf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/SessionVariable.java
@@ -1503,7 +1503,7 @@ public class SessionVariable implements Serializable, Writable {
@VariableMgr.VarAttr(name = ENABLE_AUTO_ANALYZE_INTERNAL_CATALOG,
description = {"临时参数，收否自动收集所有内表", "Temp variable， enable to auto collect all OlapTable."},
flag = VariableMgr.GLOBAL)
- public boolean enableAutoAnalyzeInternalCatalog = false;
+ public boolean enableAutoAnalyzeInternalCatalog = true;
@VariableMgr.VarAttr(name = AUTO_ANALYZE_TABLE_WIDTH_THRESHOLD,
description = {"参与自动收集的最大表宽度，列数多于这个参数的表不参与自动收集",
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
index dda7045eabb..fa2046e5467 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ShowExecutor.java
@@ -2859,6 +2859,7 @@ public class ShowExecutor {
java.time.ZoneId.systemDefault());
row.add(startTime.format(formatter));
row.add(endTime.format(formatter));
+ row.add(analysisInfo.priority.name());
resultRows.add(row);
} catch (Exception e) {
LOG.warn("Failed to get analyze info for table {}.{}.{}, reason: {}",
@@ -2876,8 +2877,7 @@ public class ShowExecutor {
for (AutoAnalysisPendingJob job : jobs) {
try {
List<String> row = new ArrayList<>();
- CatalogIf<? extends DatabaseIf<? extends TableIf>> c
- = StatisticsUtil.findCatalog(job.catalogName);
+ CatalogIf<? extends DatabaseIf<? extends TableIf>> c = StatisticsUtil.findCatalog(job.catalogName);
row.add(c.getName());
Optional<? extends DatabaseIf<? extends TableIf>> databaseIf = c.getDb(job.dbName);
row.add(databaseIf.isPresent() ? databaseIf.get().getFullName() : "DB may get deleted");
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfo.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfo.java
index 0c5047a53c5..24cf6f38d68 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfo.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfo.java
@@ -200,8 +200,12 @@ public class AnalysisInfo implements Writable {
*/
public final long tblUpdateTime;
+ @SerializedName("userInject")
public final boolean userInject;
+ @SerializedName("priority")
+ public final JobPriority priority;
+
public AnalysisInfo(long jobId, long taskId, List<Long> taskIds, long catalogId, long dbId, long tblId,
Map<String, Set<String>> colToPartitions, Set<String> partitionNames, String colName, Long indexId,
JobType jobType, AnalysisMode analysisMode, AnalysisMethod analysisMethod, AnalysisType analysisType,
@@ -210,7 +214,7 @@ public class AnalysisInfo implements Writable {
boolean isExternalTableLevelTask, boolean partitionOnly, boolean samplingPartition,
boolean isAllPartition, long partitionCount, CronExpression cronExpression, boolean forceFull,
boolean usingSqlForPartitionColumn, long tblUpdateTime, long rowCount, boolean userInject,
- long updateRows) {
+ long updateRows, JobPriority priority) {
this.jobId = jobId;
this.taskId = taskId;
this.taskIds = taskIds;
@@ -249,6 +253,7 @@ public class AnalysisInfo implements Writable {
this.rowCount = rowCount;
this.userInject = userInject;
this.updateRows = updateRows;
+ this.priority = priority;
}
@Override
@@ -293,6 +298,7 @@ public class AnalysisInfo implements Writable {
sj.add("rowCount: " + rowCount);
sj.add("userInject: " + userInject);
sj.add("updateRows: " + updateRows);
+ sj.add("priority: " + priority.name());
return sj.toString();
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfoBuilder.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfoBuilder.java
index 527d503fd52..2f60b258598 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfoBuilder.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisInfoBuilder.java
@@ -65,6 +65,7 @@ public class AnalysisInfoBuilder {
private long rowCount;
private boolean userInject;
private long updateRows;
+ private JobPriority priority;
public AnalysisInfoBuilder() {
}
@@ -105,6 +106,7 @@ public class AnalysisInfoBuilder {
rowCount = info.rowCount;
userInject = info.userInject;
updateRows = info.updateRows;
+ priority = info.priority;
}
public AnalysisInfoBuilder setJobId(long jobId) {
@@ -282,12 +284,18 @@ public class AnalysisInfoBuilder {
return this;
}
+ public AnalysisInfoBuilder setPriority(JobPriority priority) {
+ this.priority = priority;
+ return this;
+ }
+
public AnalysisInfo build() {
return new AnalysisInfo(jobId, taskId, taskIds, catalogId, dbId, tblId, colToPartitions, partitionNames,
colName, indexId, jobType, analysisMode, analysisMethod, analysisType, samplePercent,
sampleRows, maxBucketNum, periodTimeInMs, message, lastExecTimeInMs, timeCostInMs, state, scheduleType,
externalTableLevelTask, partitionOnly, samplingPartition, isAllPartition, partitionCount,
- cronExpression, forceFull, usingSqlForPartitionColumn, tblUpdateTime, rowCount, userInject, updateRows);
+ cronExpression, forceFull, usingSqlForPartitionColumn, tblUpdateTime, rowCount, userInject, updateRows,
+ priority);
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
index c6713b3a7c1..ddcdf459e6e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
@@ -219,7 +219,8 @@ public class AnalysisManager implements Writable {
public void createAnalysisJob(AnalyzeTblStmt stmt, boolean proxy) throws DdlException {
// Using auto analyzer if user specifies.
if (stmt.getAnalyzeProperties().getProperties().containsKey("use.auto.analyzer")) {
- Env.getCurrentEnv().getStatisticsAutoCollector().processOneJob(stmt.getTable(), stmt.getColumnNames());
+ Env.getCurrentEnv().getStatisticsAutoCollector()
+ .processOneJob(stmt.getTable(), stmt.getColumnNames(), JobPriority.HIGH);
return;
}
AnalysisInfo jobInfo = buildAndAssignJob(stmt);
@@ -422,6 +423,7 @@ public class AnalysisManager implements Writable {
infoBuilder.setRowCount(rowCount);
TableStatsMeta tableStatsStatus = findTableStatsStatus(table.getId());
infoBuilder.setUpdateRows(tableStatsStatus == null ? 0 : tableStatsStatus.updatedRows.get());
+ infoBuilder.setPriority(JobPriority.MANUAL);
return infoBuilder.build();
}
@@ -1230,12 +1232,14 @@ public class AnalysisManager implements Writable {
public void mergeFollowerQueryColumns(Collection<TQueryColumn> highColumns,
Collection<TQueryColumn> midColumns) {
for (TQueryColumn c : highColumns) {
- if (!highPriorityColumns.offer(new HighPriorityColumn(c.catalogId, c.dbId, c.tblId, c.colName))) {
+ if (!highPriorityColumns.offer(new HighPriorityColumn(Long.parseLong(c.catalogId), Long.parseLong(c.dbId),
+ Long.parseLong(c.tblId), c.colName))) {
break;
}
}
for (TQueryColumn c : midColumns) {
- if (!midPriorityColumns.offer(new HighPriorityColumn(c.catalogId, c.dbId, c.tblId, c.colName))) {
+ if (!midPriorityColumns.offer(new HighPriorityColumn(Long.parseLong(c.catalogId), Long.parseLong(c.dbId),
+ Long.parseLong(c.tblId), c.colName))) {
break;
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/FollowerColumnSender.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/FollowerColumnSender.java
index 181000c1ef2..0a804152694 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/FollowerColumnSender.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/FollowerColumnSender.java
@@ -32,14 +32,16 @@ import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import java.net.InetSocketAddress;
+import java.util.ArrayList;
import java.util.List;
+import java.util.Set;
import java.util.stream.Collectors;
public class FollowerColumnSender extends MasterDaemon {
private static final Logger LOG = LogManager.getLogger(FollowerColumnSender.class);
- public static final long INTERVAL = 5000;
+ public static final long INTERVAL = 60000;
public FollowerColumnSender() {
super("Follower Column Sender", INTERVAL);
@@ -68,21 +70,28 @@ public class FollowerColumnSender extends MasterDaemon {
if (analysisManager.highPriorityColumns.isEmpty() && analysisManager.midPriorityColumns.isEmpty()) {
return;
}
- List<TQueryColumn> highPriorityColumns
+ Set<TQueryColumn> highPriorityColumns
= analysisManager.highPriorityColumns
.stream()
+ .filter(c -> StatisticsUtil.needAnalyzeColumn(c))
.map(HighPriorityColumn::toThrift)
- .collect(Collectors.toList());
- List<TQueryColumn> midPriorityColumns
+ .collect(Collectors.toSet());
+ Set<TQueryColumn> midPriorityColumns
= analysisManager.midPriorityColumns
.stream()
+ .filter(c -> StatisticsUtil.needAnalyzeColumn(c))
+ .filter(c -> !highPriorityColumns.contains(c))
.map(HighPriorityColumn::toThrift)
- .collect(Collectors.toList());
+ .collect(Collectors.toSet());
analysisManager.highPriorityColumns.clear();
analysisManager.midPriorityColumns.clear();
TSyncQueryColumns queryColumns = new TSyncQueryColumns();
- queryColumns.highPriorityColumns = highPriorityColumns;
- queryColumns.midPriorityColumns = midPriorityColumns;
+ List<TQueryColumn> highs = new ArrayList<>();
+ highs.addAll(highPriorityColumns);
+ queryColumns.highPriorityColumns = highs;
+ List<TQueryColumn> mids = new ArrayList<>();
+ mids.addAll(midPriorityColumns);
+ queryColumns.midPriorityColumns = mids;
Frontend master = null;
try {
InetSocketAddress masterAddress = currentEnv.getHaProtocol().getLeader();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java
index b2292ef725d..d619ef82c08 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java
@@ -57,9 +57,9 @@ public class HighPriorityColumn {
public TQueryColumn toThrift() {
TQueryColumn tQueryColumn = new TQueryColumn();
- tQueryColumn.catalogId = catalogId;
- tQueryColumn.dbId = dbId;
- tQueryColumn.tblId = tblId;
+ tQueryColumn.catalogId = String.valueOf(catalogId);
+ tQueryColumn.dbId = String.valueOf(dbId);
+ tQueryColumn.tblId = String.valueOf(tblId);
tQueryColumn.colName = colName;
return tQueryColumn;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/JobPriority.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/JobPriority.java
index 2786b063563..c3656b92927 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/JobPriority.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/JobPriority.java
@@ -20,5 +20,6 @@ package org.apache.doris.statistics;
public enum JobPriority {
HIGH,
MID,
- LOW;
+ LOW,
+ MANUAL;
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
index 227074dbb5c..c26e7b05efd 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsAutoCollector.java
@@ -23,9 +23,9 @@ import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.TableIf;
import org.apache.doris.common.Config;
import org.apache.doris.common.DdlException;
+import org.apache.doris.common.Pair;
import org.apache.doris.common.util.TimeUtils;
import org.apache.doris.datasource.hive.HMSExternalTable;
-import org.apache.doris.datasource.hive.HMSExternalTable.DLAType;
import org.apache.doris.statistics.AnalysisInfo.AnalysisMethod;
import org.apache.doris.statistics.AnalysisInfo.JobType;
import org.apache.doris.statistics.AnalysisInfo.ScheduleType;
@@ -39,6 +39,7 @@ import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
+import java.util.Map.Entry;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.TimeUnit;
@@ -58,29 +59,22 @@ public class StatisticsAutoCollector extends StatisticsCollector {
@Override
protected void collect() {
while (canCollect()) {
- Map.Entry<TableName, Set<String>> job = getJob();
+ Pair<Entry<TableName, Set<String>>, JobPriority> job = getJob();
if (job == null) {
// No more job to process, break and sleep.
break;
}
try {
- TableName tblName = job.getKey();
+ TableName tblName = job.first.getKey();
TableIf table = StatisticsUtil.findTable(tblName.getCtl(), tblName.getDb(), tblName.getTbl());
if (!supportAutoAnalyze(table)) {
continue;
}
- Set<String> columns = job.getValue()
- .stream()
- .filter(c -> {
- boolean needAnalyzeColumn = needAnalyzeColumn(table, c);
- LOG.info("Need analyze column " + c + " ? " + needAnalyzeColumn);
- return needAnalyzeColumn;
- })
- .collect(Collectors.toSet());
- processOneJob(table, columns);
+ Set<String> columns = job.first.getValue().stream().collect(Collectors.toSet());
+ processOneJob(table, columns, job.second);
} catch (Exception e) {
- LOG.warn("Failed to analyze table {} with columns [{}]",
- job.getKey().getTbl(), job.getValue().stream().collect(Collectors.joining(",")), e);
+ LOG.warn("Failed to analyze table {} with columns [{}]", job.first.getKey().getTbl(),
+ job.first.getValue().stream().collect(Collectors.joining(",")), e);
}
}
}
@@ -90,18 +84,18 @@ public class StatisticsAutoCollector extends StatisticsCollector {
&& StatisticsUtil.inAnalyzeTime(LocalTime.now(TimeUtils.getTimeZone().toZoneId()));
}
- protected Map.Entry<TableName, Set<String>> getJob() {
+ protected Pair<Entry<TableName, Set<String>>, JobPriority> getJob() {
AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
- Optional<Map.Entry<TableName, Set<String>>> job = fetchJobFromMap(manager.highPriorityJobs);
+ Optional<Entry<TableName, Set<String>>> job = fetchJobFromMap(manager.highPriorityJobs);
if (job.isPresent()) {
- return job.get();
+ return Pair.of(job.get(), JobPriority.HIGH);
}
job = fetchJobFromMap(manager.midPriorityJobs);
if (job.isPresent()) {
- return job.get();
+ return Pair.of(job.get(), JobPriority.MID);
}
job = fetchJobFromMap(manager.lowPriorityJobs);
- return job.isPresent() ? job.get() : null;
+ return job.isPresent() ? Pair.of(job.get(), JobPriority.LOW) : null;
}
protected Optional<Map.Entry<TableName, Set<String>>> fetchJobFromMap(Map<TableName, Set<String>> jobMap) {
@@ -112,12 +106,12 @@ public class StatisticsAutoCollector extends StatisticsCollector {
}
}
- protected void processOneJob(TableIf table, Set<String> columns) throws DdlException {
+ protected void processOneJob(TableIf table, Set<String> columns, JobPriority priority) throws DdlException {
appendPartitionColumns(table, columns);
if (columns.isEmpty()) {
return;
}
- AnalysisInfo analyzeJob = createAnalyzeJobForTbl(table, columns);
+ AnalysisInfo analyzeJob = createAnalyzeJobForTbl(table, columns, priority);
LOG.info("Analyze job : {}", analyzeJob.toString());
createSystemAnalysisJob(analyzeJob);
}
@@ -134,69 +128,6 @@ public class StatisticsAutoCollector extends StatisticsCollector {
}
}
- // TODO: Need refactor, hard to understand now.
- protected boolean needAnalyzeColumn(TableIf table, String column) {
- AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
- TableStatsMeta tableStatsStatus = manager.findTableStatsStatus(table.getId());
- if (tableStatsStatus == null) {
- return true;
- }
- if (tableStatsStatus.userInjected) {
- return false;
- }
- ColStatsMeta columnStatsMeta = tableStatsStatus.findColumnStatsMeta(column);
- if (columnStatsMeta == null) {
- return true;
- }
- if (table instanceof OlapTable) {
- long currentUpdatedRows = tableStatsStatus.updatedRows.get();
- long lastAnalyzeUpdateRows = columnStatsMeta.updatedRows;
- if (lastAnalyzeUpdateRows == 0 && currentUpdatedRows > 0) {
- return true;
- }
- if (lastAnalyzeUpdateRows > currentUpdatedRows) {
- // Shouldn't happen. Just in case.
- return true;
- }
- OlapTable olapTable = (OlapTable) table;
- long currentRowCount = olapTable.getRowCount();
- long lastAnalyzeRowCount = columnStatsMeta.rowCount;
- if (tableStatsStatus.newPartitionLoaded.get() && olapTable.isPartitionColumn(column)) {
- return true;
- }
- if (lastAnalyzeRowCount == 0 && currentRowCount > 0) {
- return true;
- }
- if (currentUpdatedRows == lastAnalyzeUpdateRows) {
- return false;
- }
- double healthValue = ((double) (currentUpdatedRows - lastAnalyzeUpdateRows)
- / (double) currentUpdatedRows) * 100.0;
- LOG.info("Column " + column + " update rows health value is " + healthValue);
- if (healthValue < StatisticsUtil.getTableStatsHealthThreshold()) {
- return true;
- }
- if (currentRowCount == 0 && lastAnalyzeRowCount != 0) {
- return true;
- }
- if (currentRowCount == 0 && lastAnalyzeRowCount == 0) {
- return false;
- }
- healthValue = ((double) (currentRowCount - lastAnalyzeRowCount) / (double) currentRowCount) * 100.0;
- return healthValue < StatisticsUtil.getTableStatsHealthThreshold();
- } else {
- if (!(table instanceof HMSExternalTable)) {
- return false;
- }
- HMSExternalTable hmsTable = (HMSExternalTable) table;
- if (!hmsTable.getDlaType().equals(DLAType.HIVE)) {
- return false;
- }
- return System.currentTimeMillis()
- - tableStatsStatus.updatedTime > StatisticsUtil.getExternalTableAutoAnalyzeIntervalInMillis();
- }
- }
-
protected boolean supportAutoAnalyze(TableIf tableIf) {
if (tableIf == null) {
return false;
@@ -206,7 +137,7 @@ public class StatisticsAutoCollector extends StatisticsCollector {
&& ((HMSExternalTable) tableIf).getDlaType().equals(HMSExternalTable.DLAType.HIVE);
}
- protected AnalysisInfo createAnalyzeJobForTbl(TableIf table, Set<String> columns) {
+ protected AnalysisInfo createAnalyzeJobForTbl(TableIf table, Set<String> columns, JobPriority priority) {
AnalysisMethod analysisMethod = table.getDataSize(true) >= StatisticsUtil.getHugeTableLowerBoundSizeInBytes()
? AnalysisMethod.SAMPLE : AnalysisMethod.FULL;
AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
@@ -236,6 +167,7 @@ public class StatisticsAutoCollector extends StatisticsCollector {
.setRowCount(rowCount)
.setUpdateRows(tableStatsStatus == null ? 0 : tableStatsStatus.updatedRows.get())
.setColToPartitions(colToPartitions)
+ .setPriority(priority)
.build();
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
index 93d03a3fdb8..9e07c65e2fe 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/StatisticsJobAppender.java
@@ -86,6 +86,9 @@ public class StatisticsJobAppender extends MasterDaemon {
int size = columnQueue.size();
for (int i = 0; i < size; i++) {
HighPriorityColumn column = columnQueue.poll();
+ if (!StatisticsUtil.needAnalyzeColumn(column)) {
+ continue;
+ }
LOG.info("Process column " + column.tblId + "." + column.colName);
TableIf table = StatisticsUtil.findTable(column.catalogId, column.dbId, column.tblId);
TableName tableName = new TableName(table.getDatabase().getCatalog().getName(),
@@ -132,8 +135,9 @@ public class StatisticsJobAppender extends MasterDaemon {
if (!jobsMap.containsKey(tableName) && jobsMap.size() >= JOB_MAP_SIZE) {
return;
}
- Set<String> columns
- = t.getColumns().stream().filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
+ Set<String> columns = t.getColumns().stream()
+ .filter(c -> !StatisticsUtil.isUnsupportedType(c.getType()))
+ .filter(c -> StatisticsUtil.needAnalyzeColumn(t, c.getName()))
.map(c -> c.getName()).collect(Collectors.toSet());
if (jobsMap.containsKey(tableName)) {
jobsMap.get(tableName).addAll(columns);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
index 65c2ee9e6da..27aa83ba716 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/util/StatisticsUtil.java
@@ -58,6 +58,7 @@ import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.InternalCatalog;
import org.apache.doris.datasource.hive.HMSExternalCatalog;
import org.apache.doris.datasource.hive.HMSExternalTable;
+import org.apache.doris.datasource.hive.HMSExternalTable.DLAType;
import org.apache.doris.datasource.hive.HiveMetaStoreCache;
import org.apache.doris.datasource.hive.HivePartition;
import org.apache.doris.nereids.trees.expressions.literal.DateTimeLiteral;
@@ -68,11 +69,15 @@ import org.apache.doris.qe.QueryState;
import org.apache.doris.qe.SessionVariable;
import org.apache.doris.qe.StmtExecutor;
import org.apache.doris.qe.VariableMgr;
+import org.apache.doris.statistics.AnalysisManager;
+import org.apache.doris.statistics.ColStatsMeta;
import org.apache.doris.statistics.ColumnStatistic;
import org.apache.doris.statistics.ColumnStatisticBuilder;
+import org.apache.doris.statistics.HighPriorityColumn;
import org.apache.doris.statistics.Histogram;
import org.apache.doris.statistics.ResultRow;
import org.apache.doris.statistics.StatisticConstants;
+import org.apache.doris.statistics.TableStatsMeta;
import org.apache.doris.system.Frontend;
import com.google.common.base.Preconditions;
@@ -874,7 +879,7 @@ public class StatisticsUtil {
} catch (Exception e) {
LOG.warn("Fail to get value of enable auto analyze internal catalog, return false by default", e);
}
- return false;
+ return true;
}
public static int getInsertMergeCount() {
@@ -984,4 +989,85 @@ public class StatisticsUtil {
|| columnName.startsWith(CreateMaterializedViewStmt.MATERIALIZED_VIEW_AGGREGATE_NAME_PREFIX);
}
+ // TODO: Need refactor, hard to understand now.
+ public static boolean needAnalyzeColumn(TableIf table, String column) {
+ AnalysisManager manager = Env.getServingEnv().getAnalysisManager();
+ TableStatsMeta tableStatsStatus = manager.findTableStatsStatus(table.getId());
+ if (tableStatsStatus == null) {
+ return true;
+ }
+ if (tableStatsStatus.userInjected) {
+ return false;
+ }
+ ColStatsMeta columnStatsMeta = tableStatsStatus.findColumnStatsMeta(column);
+ if (columnStatsMeta == null) {
+ return true;
+ }
+ if (table instanceof OlapTable) {
+ long currentUpdatedRows = tableStatsStatus.updatedRows.get();
+ long lastAnalyzeUpdateRows = columnStatsMeta.updatedRows;
+ if (lastAnalyzeUpdateRows == 0 && currentUpdatedRows > 0) {
+ return true;
+ }
+ if (lastAnalyzeUpdateRows > currentUpdatedRows) {
+ // Shouldn't happen. Just in case.
+ return true;
+ }
+ OlapTable olapTable = (OlapTable) table;
+ long currentRowCount = olapTable.getRowCount();
+ long lastAnalyzeRowCount = columnStatsMeta.rowCount;
+ if (tableStatsStatus.newPartitionLoaded.get() && olapTable.isPartitionColumn(column)) {
+ return true;
+ }
+ if (lastAnalyzeRowCount == 0 && currentRowCount > 0) {
+ return true;
+ }
+ if (currentUpdatedRows == lastAnalyzeUpdateRows) {
+ return false;
+ }
+ double healthValue = ((double) (currentUpdatedRows - lastAnalyzeUpdateRows)
+ / (double) currentUpdatedRows) * 100.0;
+ LOG.info("Column " + column + " update rows health value is " + healthValue);
+ if (healthValue < StatisticsUtil.getTableStatsHealthThreshold()) {
+ return true;
+ }
+ if (currentRowCount == 0 && lastAnalyzeRowCount != 0) {
+ return true;
+ }
+ if (currentRowCount == 0 && lastAnalyzeRowCount == 0) {
+ return false;
+ }
+ healthValue = ((double) (currentRowCount - lastAnalyzeRowCount) / (double) currentRowCount) * 100.0;
+ return healthValue < StatisticsUtil.getTableStatsHealthThreshold();
+ } else {
+ if (!(table instanceof HMSExternalTable)) {
+ return false;
+ }
+ HMSExternalTable hmsTable = (HMSExternalTable) table;
+ if (!hmsTable.getDlaType().equals(DLAType.HIVE)) {
+ return false;
+ }
+ return System.currentTimeMillis()
+ - tableStatsStatus.updatedTime > StatisticsUtil.getExternalTableAutoAnalyzeIntervalInMillis();
+ }
+ }
+
+ public static boolean needAnalyzeColumn(HighPriorityColumn column) {
+ if (column == null) {
+ return false;
+ }
+ TableIf table;
+ Column col;
+ try {
+ table = StatisticsUtil.findTable(column.catalogId, column.dbId, column.tblId);
+ col = table.getColumn(column.colName);
+ } catch (Exception e) {
+ LOG.warn("Failed to find table for column {}", column.colName, e);
+ return false;
+ }
+ return col != null
+ && !StatisticsUtil.isUnsupportedType(col.getType())
+ && StatisticsUtil.needAnalyzeColumn(table, column.colName);
+ }
+
}
diff --git a/gensrc/thrift/FrontendService.thrift b/gensrc/thrift/FrontendService.thrift
index d2cdecbbe56..6be02613d5e 100644
--- a/gensrc/thrift/FrontendService.thrift
+++ b/gensrc/thrift/FrontendService.thrift
@@ -1411,9 +1411,9 @@ struct TReportCommitTxnResultRequest {
}
struct TQueryColumn {
- 1: optional i64 catalogId
- 2: optional i64 dbId
- 3: optional i64 tblId
+ 1: optional string catalogId
+ 2: optional string dbId
+ 3: optional string tblId
4: optional string colName
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 01/07: Collect high priority columns. (#31235)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lijibing pushed a commit to branch high-priority-column
in repository https://gitbox.apache.org/repos/asf/doris.git
commit a816e7c646cfd2eff256ab7f43158a3219d9615b
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Wed Feb 28 11:54:40 2024 +0800
Collect high priority columns. (#31235)
---
.../doris/nereids/jobs/executor/Rewriter.java | 4 +-
.../org/apache/doris/nereids/rules/RuleType.java | 1 +
.../expression/HighPriorityColumnCollector.java | 202 +++++++++++++++++++++
.../apache/doris/statistics/AnalysisManager.java | 44 +++++
.../doris/statistics/HighPriorityColumn.java | 55 ++++++
5 files changed, 305 insertions(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java
index db463354120..8976e274893 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/jobs/executor/Rewriter.java
@@ -32,6 +32,7 @@ import org.apache.doris.nereids.rules.expression.CheckLegalityAfterRewrite;
import org.apache.doris.nereids.rules.expression.ExpressionNormalization;
import org.apache.doris.nereids.rules.expression.ExpressionOptimization;
import org.apache.doris.nereids.rules.expression.ExpressionRewrite;
+import org.apache.doris.nereids.rules.expression.HighPriorityColumnCollector;
import org.apache.doris.nereids.rules.rewrite.AddDefaultLimit;
import org.apache.doris.nereids.rules.rewrite.AdjustConjunctsReturnType;
import org.apache.doris.nereids.rules.rewrite.AdjustNullable;
@@ -453,7 +454,8 @@ public class Rewriter extends AbstractBatchJobExecutor {
new CollectFilterAboveConsumer(),
new CollectProjectAboveConsumer()
)
- )
+ ),
+ topic("Collect used column", custom(RuleType.COLLECT_COLUMNS, HighPriorityColumnCollector::new))
);
private static final List<RewriteJob> WHOLE_TREE_REWRITE_JOBS
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java
index 90f2c222091..e54dd7486ea 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/RuleType.java
@@ -300,6 +300,7 @@ public enum RuleType {
LEADING_JOIN(RuleTypeClass.REWRITE),
REWRITE_SENTINEL(RuleTypeClass.REWRITE),
+ COLLECT_COLUMNS(RuleTypeClass.REWRITE),
// topn opts
DEFER_MATERIALIZE_TOP_N_RESULT(RuleTypeClass.REWRITE),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/HighPriorityColumnCollector.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/HighPriorityColumnCollector.java
new file mode 100644
index 00000000000..ed67ad97005
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/HighPriorityColumnCollector.java
@@ -0,0 +1,202 @@
+// 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.rules.expression;
+
+import org.apache.doris.catalog.Column;
+import org.apache.doris.catalog.Env;
+import org.apache.doris.catalog.TableIf;
+import org.apache.doris.nereids.jobs.JobContext;
+import org.apache.doris.nereids.rules.expression.HighPriorityColumnCollector.CollectorContext;
+import org.apache.doris.nereids.trees.expressions.NamedExpression;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
+import org.apache.doris.nereids.trees.plans.Plan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalAggregate;
+import org.apache.doris.nereids.trees.plans.logical.LogicalCatalogRelation;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFileScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalFilter;
+import org.apache.doris.nereids.trees.plans.logical.LogicalHaving;
+import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
+import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
+import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
+import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
+import org.apache.doris.qe.ConnectContext;
+import org.apache.doris.statistics.AnalysisManager;
+import org.apache.doris.statistics.util.StatisticsUtil;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+/**
+ * Used to collect High priority column.
+ */
+public class HighPriorityColumnCollector extends DefaultPlanRewriter<CollectorContext> implements CustomRewriter {
+
+ @Override
+ public Plan rewriteRoot(Plan plan, JobContext jobContext) {
+ ConnectContext connectContext = ConnectContext.get();
+ if (connectContext != null && connectContext.getSessionVariable().internalSession) {
+ return plan;
+ }
+ CollectorContext context = new CollectorContext();
+ plan.accept(this, context);
+ if (StatisticsUtil.enableAutoAnalyze()) {
+ context.queried.removeAll(context.usedInPredicate);
+ AnalysisManager analysisManager = Env.getCurrentEnv().getAnalysisManager();
+ analysisManager.updateColumnUsedInPredicate(context.usedInPredicate);
+ analysisManager.updateQueriedColumn(context.queried);
+ }
+ return plan;
+ }
+
+ /**
+ * Context.
+ */
+ public static class CollectorContext {
+ public Map<Slot/*project output column*/, NamedExpression/*Actual project expr*/> projects = new HashMap<>();
+
+ public Set<Slot> usedInPredicate = new HashSet<>();
+
+ public Set<Slot> queried = new HashSet<>();
+ }
+
+ @Override
+ public Plan visitLogicalProject(LogicalProject<? extends Plan> project, CollectorContext context) {
+ project.child().accept(this, context);
+ List<NamedExpression> projects = project.getOutputs();
+ List<Slot> slots = project.computeOutput();
+ for (int i = 0; i < slots.size(); i++) {
+ context.projects.put(slots.get(i), projects.get(i));
+ }
+ if (project.child() instanceof LogicalCatalogRelation
+ || project.child() instanceof LogicalFilter
+ && ((LogicalFilter) project.child()).child() instanceof LogicalCatalogRelation) {
+ Set<Slot> allUsed = project.getExpressions()
+ .stream().flatMap(e -> e.<Set<SlotReference>>collect(n -> n instanceof SlotReference).stream())
+ .collect(Collectors.toSet());
+ LogicalCatalogRelation scan = project.child() instanceof LogicalCatalogRelation
+ ? (LogicalCatalogRelation) project.child()
+ : (LogicalCatalogRelation) project.child().child(0);
+ List<Slot> outputOfScan = scan.getOutput();
+ for (Slot slot : outputOfScan) {
+ if (!allUsed.contains(slot)) {
+ context.queried.remove(slot);
+ }
+ }
+ }
+ return project;
+ }
+
+ @Override
+ public Plan visitLogicalJoin(LogicalJoin<? extends Plan, ? extends Plan> join, CollectorContext context) {
+ join.child(0).accept(this, context);
+ join.child(1).accept(this, context);
+ context.usedInPredicate.addAll(
+ (join.isMarkJoin() ? join.getLeftConditionSlot() : join.getConditionSlot())
+ .stream().flatMap(s -> backtrace(s, context).stream())
+ .collect(Collectors.toSet())
+ );
+ return join;
+ }
+
+ @Override
+ public Plan visitLogicalAggregate(LogicalAggregate<? extends Plan> aggregate, CollectorContext context) {
+ aggregate.child(0).accept(this, context);
+ context.usedInPredicate.addAll(aggregate.getGroupByExpressions()
+ .stream()
+ .flatMap(e -> e.<Set<SlotReference>>collect(n -> n instanceof SlotReference).stream())
+ .flatMap(s -> backtrace(s, context).stream())
+ .collect(Collectors.toSet()));
+ return aggregate;
+ }
+
+ @Override
+ public Plan visitLogicalHaving(LogicalHaving<? extends Plan> having, CollectorContext context) {
+ having.child(0).accept(this, context);
+ context.usedInPredicate.addAll(
+ having.getExpressions().stream()
+ .flatMap(e -> e.<Set<SlotReference>>collect(n -> n instanceof SlotReference).stream())
+ .flatMap(s -> backtrace(s, context).stream())
+ .collect(Collectors.toSet()));
+ return having;
+ }
+
+ @Override
+ public Plan visitLogicalOlapScan(LogicalOlapScan olapScan, CollectorContext context) {
+ List<Slot> slots = olapScan.getOutput();
+ context.queried.addAll(slots);
+ return olapScan;
+ }
+
+ @Override
+ public Plan visitLogicalFileScan(LogicalFileScan fileScan, CollectorContext context) {
+ List<Slot> slots = fileScan.getOutput();
+ context.queried.addAll(slots);
+ return fileScan;
+ }
+
+ @Override
+ public Plan visitLogicalFilter(LogicalFilter<? extends Plan> filter, CollectorContext context) {
+ filter.child(0).accept(this, context);
+ context.usedInPredicate.addAll(filter
+ .getExpressions()
+ .stream()
+ .flatMap(e -> e.<Set<SlotReference>>collect(n -> n instanceof SlotReference).stream())
+ .flatMap(s -> backtrace(s, context).stream())
+ .collect(Collectors.toSet()));
+ return filter;
+ }
+
+ private Set<Slot> backtrace(Slot slot, CollectorContext context) {
+ return backtrace(slot, new HashSet<>(), context);
+ }
+
+ private Set<Slot> backtrace(Slot slot, Set<Slot> path, CollectorContext context) {
+ if (path.contains(slot)) {
+ return Collections.emptySet();
+ }
+ path.add(slot);
+ if (slot instanceof SlotReference) {
+ SlotReference slotReference = (SlotReference) slot;
+ Optional<Column> col = slotReference.getColumn();
+ Optional<TableIf> table = slotReference.getTable();
+ if (col.isPresent() && table.isPresent()) {
+ return Collections.singleton(slot);
+ }
+ }
+ NamedExpression namedExpression = context.projects.get(slot);
+ if (namedExpression == null) {
+ return Collections.emptySet();
+ }
+ Set<SlotReference> slotReferences
+ = namedExpression.<Set<SlotReference>>collect(n -> n instanceof SlotReference);
+ Set<Slot> refCol = new HashSet<>();
+ for (SlotReference slotReference : slotReferences) {
+ refCol.addAll(backtrace(slotReference, path, context));
+ }
+ return refCol;
+ }
+
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
index f90ce83a088..cf3764719ba 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/AnalysisManager.java
@@ -47,6 +47,8 @@ import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.datasource.ExternalTable;
import org.apache.doris.datasource.hive.HMSExternalTable;
import org.apache.doris.mysql.privilege.PrivPredicate;
+import org.apache.doris.nereids.trees.expressions.Slot;
+import org.apache.doris.nereids.trees.expressions.SlotReference;
import org.apache.doris.persist.AnalyzeDeletionLog;
import org.apache.doris.persist.gson.GsonUtils;
import org.apache.doris.qe.ConnectContext;
@@ -89,9 +91,11 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.NavigableMap;
import java.util.Optional;
+import java.util.Queue;
import java.util.Set;
import java.util.StringJoiner;
import java.util.TreeMap;
+import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.CountDownLatch;
@@ -104,6 +108,16 @@ public class AnalysisManager implements Writable {
private static final Logger LOG = LogManager.getLogger(AnalysisManager.class);
+ /**
+ * Mem only.
+ */
+ public final Queue<HighPriorityColumn> predicateColumns = new ArrayBlockingQueue<>(100);
+
+ /**
+ * Mem only.
+ */
+ public final Queue<HighPriorityColumn> queryColumns = new ArrayBlockingQueue<>(100);
+
// Tracking running manually submitted async tasks, keep in mem only
protected final ConcurrentMap<Long, Map<Long, BaseAnalysisTask>> analysisJobIdToTaskMap = new ConcurrentHashMap<>();
@@ -1141,4 +1155,34 @@ public class AnalysisManager implements Writable {
}
return false;
}
+
+
+ public void updateColumnUsedInPredicate(Set<Slot> slotReferences) {
+ updateColumn(slotReferences, predicateColumns);
+ }
+
+ public void updateQueriedColumn(Collection<Slot> slotReferences) {
+ updateColumn(slotReferences, queryColumns);
+ }
+
+ protected void updateColumn(Collection<Slot> slotReferences, Queue<HighPriorityColumn> queue) {
+ for (Slot s : slotReferences) {
+ if (!(s instanceof SlotReference)) {
+ return;
+ }
+ Optional<Column> optionalColumn = ((SlotReference) s).getColumn();
+ Optional<TableIf> optionalTable = ((SlotReference) s).getTable();
+ if (optionalColumn.isPresent() && optionalTable.isPresent()) {
+ TableIf table = optionalTable.get();
+ DatabaseIf database = table.getDatabase();
+ if (database != null) {
+ CatalogIf catalog = database.getCatalog();
+ if (catalog != null) {
+ queue.offer(new HighPriorityColumn(catalog.getId(), database.getId(),
+ table.getId(), optionalColumn.get().getName()));
+ }
+ }
+ }
+ }
+ }
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java b/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java
new file mode 100644
index 00000000000..c4bc20c399a
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/statistics/HighPriorityColumn.java
@@ -0,0 +1,55 @@
+// 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.statistics;
+
+import java.util.Objects;
+
+public class HighPriorityColumn {
+
+ public final long catalogId;
+ public final long dbId;
+ public final long tblId;
+ public final String colName;
+
+ public HighPriorityColumn(long catalogId, long dbId, long tblId, String colName) {
+ this.catalogId = catalogId;
+ this.dbId = dbId;
+ this.tblId = tblId;
+ this.colName = colName;
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(catalogId, dbId, tblId, colName);
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (!(other instanceof HighPriorityColumn)) {
+ return false;
+ }
+ HighPriorityColumn otherCriticalColumn = (HighPriorityColumn) other;
+ return this.catalogId == otherCriticalColumn.catalogId
+ && this.dbId == otherCriticalColumn.dbId
+ && this.tblId == otherCriticalColumn.tblId
+ && this.colName.equals(otherCriticalColumn.colName);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
(doris) 07/07: support window (#32094)
Posted by li...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
lijibing pushed a commit to branch high-priority-column
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 5944b418dfa512b83a95b6fdc9d14205e23c5d86
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Tue Mar 12 13:25:23 2024 +0800
support window (#32094)
---
.../rules/expression/HighPriorityColumnCollector.java | 13 +++++++++++++
1 file changed, 13 insertions(+)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/HighPriorityColumnCollector.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/HighPriorityColumnCollector.java
index ed67ad97005..e20363b1202 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/HighPriorityColumnCollector.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/expression/HighPriorityColumnCollector.java
@@ -34,6 +34,7 @@ import org.apache.doris.nereids.trees.plans.logical.LogicalHaving;
import org.apache.doris.nereids.trees.plans.logical.LogicalJoin;
import org.apache.doris.nereids.trees.plans.logical.LogicalOlapScan;
import org.apache.doris.nereids.trees.plans.logical.LogicalProject;
+import org.apache.doris.nereids.trees.plans.logical.LogicalWindow;
import org.apache.doris.nereids.trees.plans.visitor.CustomRewriter;
import org.apache.doris.nereids.trees.plans.visitor.DefaultPlanRewriter;
import org.apache.doris.qe.ConnectContext;
@@ -169,6 +170,18 @@ public class HighPriorityColumnCollector extends DefaultPlanRewriter<CollectorCo
return filter;
}
+ @Override
+ public Plan visitLogicalWindow(LogicalWindow<? extends Plan> window, CollectorContext context) {
+ window.child(0).accept(this, context);
+ context.usedInPredicate.addAll(window
+ .getWindowExpressions()
+ .stream()
+ .flatMap(e -> e.<Set<SlotReference>>collect(n -> n instanceof SlotReference).stream())
+ .flatMap(s -> backtrace(s, context).stream())
+ .collect(Collectors.toSet()));
+ return window;
+ }
+
private Set<Slot> backtrace(Slot slot, CollectorContext context) {
return backtrace(slot, new HashSet<>(), context);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org