You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by kx...@apache.org on 2023/07/17 15:48:33 UTC
[doris] branch branch-2.0 updated (edceca404c -> 9bb22d480b)
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a change to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
from edceca404c [Bug](CSVReader) fix null pointer coredump in CSVReader in p2 (#20811)
new 1d9571a909 [Bug](materialized-view) fix some problems of mv and make ssb mv work on nereids (#21559)
new ec36d27fc0 [Feature](table function) support explode_json_array_json (#21795)
new c865daee08 [chore](topn opt) disable two phase read when light schema change is disabled (#21809)
new 350095e569 [improvement](catalog) reduce the size thrift params for external table query (#21771)
new 9afdc02186 Fix incorrect priority queue refactor (#21857)
new 0f74793c1d [refactor](profile)add node id info in pipeline profile (#21823)
new 68b54dde2e [Fix](statistics) Fix analyze db always use internal catalog bug (#21850)
new ec5277d5bd [enhancement](jdbc catalog) Add mysql jdbc url param `rewriteBatchedStatements=true` (#21864)
new 373e586c08 [Feature](Compaction)Support full compaction (#21177)
new 9bb22d480b [fix](clone) fix cannot further repair clone replica which miss version data (#21382)
The 10 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
be/src/common/status.h | 3 +
be/src/http/action/compaction_action.cpp | 17 +-
be/src/http/action/compaction_action.h | 1 +
be/src/io/io_common.h | 3 +-
be/src/olap/compaction.cpp | 4 +
be/src/olap/full_compaction.cpp | 215 ++++++++++++++++++++
be/src/olap/full_compaction.h | 62 ++++++
be/src/olap/olap_common.h | 2 +-
be/src/olap/push_handler.cpp | 7 +-
be/src/olap/reader.cpp | 2 +
be/src/olap/rowset/rowset_meta.h | 2 +-
be/src/olap/single_replica_compaction.cpp | 2 +
be/src/olap/tablet.cpp | 18 ++
be/src/olap/tablet.h | 17 ++
be/src/pipeline/pipeline_fragment_context.cpp | 90 ++++----
be/src/pipeline/pipeline_task.cpp | 11 +-
be/src/pipeline/task_queue.cpp | 2 +-
be/src/runtime/fragment_mgr.cpp | 6 +
be/src/runtime/query_context.h | 4 +
be/src/vec/exec/format/csv/csv_reader.cpp | 21 +-
be/src/vec/exec/format/json/new_json_reader.cpp | 7 +-
be/src/vec/exec/format/orc/vorc_reader.cpp | 30 ++-
be/src/vec/exec/format/orc/vorc_reader.h | 8 +-
be/src/vec/exec/format/parquet/vparquet_reader.cpp | 7 +-
be/src/vec/exec/scan/avro_jni_reader.cpp | 8 +-
be/src/vec/exec/scan/new_file_scan_node.cpp | 11 +-
be/src/vec/exec/scan/vfile_scanner.cpp | 67 +++---
be/src/vec/exec/scan/vfile_scanner.h | 2 +-
.../table_function/table_function_factory.cpp | 3 +
.../exprs/table_function/vexplode_json_array.cpp | 25 +++
.../vec/exprs/table_function/vexplode_json_array.h | 7 +-
be/src/vec/functions/function_fake.cpp | 3 +
.../table-functions/explode-json-array.md | 16 ++
.../table-functions/explode-json-array.md | 150 +++++++++++---
.../org/apache/doris/analysis/AnalyzeDBStmt.java | 2 +-
.../java/org/apache/doris/analysis/SelectStmt.java | 3 +
.../catalog/BuiltinTableGeneratingFunctions.java | 6 +-
.../java/org/apache/doris/catalog/FunctionSet.java | 6 +
.../org/apache/doris/catalog/JdbcResource.java | 5 +-
.../main/java/org/apache/doris/catalog/Tablet.java | 6 +-
.../mv/AbstractSelectMaterializedIndexRule.java | 7 +-
.../functions/generator/ExplodeJsonArrayJson.java | 66 ++++++
.../generator/ExplodeJsonArrayJsonOuter.java | 66 ++++++
.../visitor/TableGeneratingFunctionVisitor.java | 11 +
.../doris/nereids/util/TypeCoercionUtils.java | 26 ++-
.../doris/planner/external/FileQueryScanNode.java | 82 ++++----
.../main/java/org/apache/doris/qe/Coordinator.java | 19 +-
.../ExternalFileTableValuedFunction.java | 5 +-
.../doris/planner/TableFunctionPlanTest.java | 6 +
gensrc/thrift/PaloInternalService.thrift | 5 +
gensrc/thrift/PlanNodes.thrift | 8 +
.../data/compaction/test_full_compaction.out | 41 ++++
.../hive/test_hive_analyze_db.out | 20 ++
regression-test/data/mv_p0/k1s2m3/k1s2m3.out | 64 ++++++
.../ssb/multiple_no_where/multiple_no_where.out | 24 +++
.../table_function/explode_json_array.out | 14 ++
.../plugins/plugin_curl_requester.groovy | 6 +
.../suites/compaction/test_full_compaction.groovy | 177 ++++++++++++++++
.../hive/test_external_brown.groovy | 2 +-
.../hive/test_hive_analyze_db.groovy | 40 ++++
regression-test/suites/mv_p0/k1s2m3/k1s2m3.groovy | 101 +++++++++
.../ssb/multiple_no_where/multiple_no_where.groovy | 226 +++++++++++++++++++++
.../sql_functions/test_arith_functions.groovy | 2 +-
.../table_function/explode_json_array.groovy | 3 +
64 files changed, 1700 insertions(+), 182 deletions(-)
create mode 100644 be/src/olap/full_compaction.cpp
create mode 100644 be/src/olap/full_compaction.h
create mode 100644 fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/generator/ExplodeJsonArrayJson.java
create mode 100644 fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/generator/ExplodeJsonArrayJsonOuter.java
create mode 100644 regression-test/data/compaction/test_full_compaction.out
create mode 100644 regression-test/data/external_table_emr_p2/hive/test_hive_analyze_db.out
create mode 100644 regression-test/data/mv_p0/k1s2m3/k1s2m3.out
create mode 100644 regression-test/data/mv_p0/ssb/multiple_no_where/multiple_no_where.out
create mode 100644 regression-test/suites/compaction/test_full_compaction.groovy
create mode 100644 regression-test/suites/external_table_emr_p2/hive/test_hive_analyze_db.groovy
create mode 100644 regression-test/suites/mv_p0/k1s2m3/k1s2m3.groovy
create mode 100644 regression-test/suites/mv_p0/ssb/multiple_no_where/multiple_no_where.groovy
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 01/10: [Bug](materialized-view) fix some problems of mv and make ssb mv work on nereids (#21559)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 1d9571a9091c3d4ea3e678c29cbdfa5e8c77db98
Author: Pxl <px...@qq.com>
AuthorDate: Mon Jul 17 10:08:25 2023 +0800
[Bug](materialized-view) fix some problems of mv and make ssb mv work on nereids (#21559)
fix some problems of mv and make ssb mv work on nereids
---
.../mv/AbstractSelectMaterializedIndexRule.java | 7 +-
.../doris/nereids/util/TypeCoercionUtils.java | 26 ++-
regression-test/data/mv_p0/k1s2m3/k1s2m3.out | 64 ++++++
.../ssb/multiple_no_where/multiple_no_where.out | 24 +++
regression-test/suites/mv_p0/k1s2m3/k1s2m3.groovy | 101 +++++++++
.../ssb/multiple_no_where/multiple_no_where.groovy | 226 +++++++++++++++++++++
.../sql_functions/test_arith_functions.groovy | 2 +-
7 files changed, 444 insertions(+), 6 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/mv/AbstractSelectMaterializedIndexRule.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/mv/AbstractSelectMaterializedIndexRule.java
index 1960dbabdc..bc77b8d058 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/mv/AbstractSelectMaterializedIndexRule.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/rules/rewrite/mv/AbstractSelectMaterializedIndexRule.java
@@ -71,6 +71,7 @@ import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.TreeMap;
+import java.util.TreeSet;
import java.util.function.Function;
import java.util.stream.Collectors;
@@ -98,12 +99,12 @@ public abstract class AbstractSelectMaterializedIndexRule {
OlapTable table = scan.getTable();
Set<String> requiredMvColumnNames = requiredScanOutput.stream()
- .map(s -> normalizeName(Column.getNameWithoutMvPrefix(s.getName())))
- .collect(Collectors.toSet());
+ .map(s -> normalizeName(Column.getNameWithoutMvPrefix(s.getName())))
+ .collect(Collectors.toCollection(() -> new TreeSet<String>(String.CASE_INSENSITIVE_ORDER)));
Set<String> mvColNames = table.getSchemaByIndexId(index.getId(), true).stream()
.map(c -> normalizeName(parseMvColumnToSql(c.getNameWithoutMvPrefix())))
- .collect(Collectors.toSet());
+ .collect(Collectors.toCollection(() -> new TreeSet<String>(String.CASE_INSENSITIVE_ORDER)));
return mvColNames.containsAll(requiredMvColumnNames)
|| requiredExpr.stream()
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java
index 05de35a388..afdc19b2fe 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/util/TypeCoercionUtils.java
@@ -114,6 +114,18 @@ import java.util.stream.Stream;
*/
public class TypeCoercionUtils {
+ /**
+ * integer type precedence for type promotion.
+ * bigger numeric has smaller ordinal
+ */
+ public static final List<DataType> INTEGER_PRECEDENCE = ImmutableList.of(
+ LargeIntType.INSTANCE,
+ BigIntType.INSTANCE,
+ IntegerType.INSTANCE,
+ SmallIntType.INSTANCE,
+ TinyIntType.INSTANCE
+ );
+
/**
* numeric type precedence for type promotion.
* bigger numeric has smaller ordinal
@@ -486,8 +498,18 @@ public class TypeCoercionUtils {
left = castIfNotSameType(left, t1);
right = castIfNotSameType(right, t2);
- Expression newLeft = TypeCoercionUtils.castIfNotSameType(left, BigIntType.INSTANCE);
- Expression newRight = TypeCoercionUtils.castIfNotSameType(right, BigIntType.INSTANCE);
+ DataType commonType = BigIntType.INSTANCE;
+ if (t1.isIntegralType() && t2.isIntegralType()) {
+ for (DataType dataType : TypeCoercionUtils.INTEGER_PRECEDENCE) {
+ if (t1.equals(dataType) || t2.equals(dataType)) {
+ commonType = dataType;
+ break;
+ }
+ }
+ }
+
+ Expression newLeft = TypeCoercionUtils.castIfNotSameType(left, commonType);
+ Expression newRight = TypeCoercionUtils.castIfNotSameType(right, commonType);
return divide.withChildren(newLeft, newRight);
}
diff --git a/regression-test/data/mv_p0/k1s2m3/k1s2m3.out b/regression-test/data/mv_p0/k1s2m3/k1s2m3.out
new file mode 100644
index 0000000000..1da9198a32
--- /dev/null
+++ b/regression-test/data/mv_p0/k1s2m3/k1s2m3.out
@@ -0,0 +1,64 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_star --
+\N 4 \N d
+-4 -4 -4 d
+1 1 1 a
+2 2 2 b
+3 -3 \N c
+
+-- !select_mv --
+\N \N
+-4 16
+1 1
+2 4
+3 \N
+
+-- !select_mv --
+\N \N
+-4 16
+1 1
+2 4
+3 \N
+
+-- !select_mv --
+\N \N
+-4 16
+1 1
+2 4
+3 \N
+
+-- !select_mv --
+\N \N
+-4 16
+1 1
+2 4
+3 \N
+
+-- !select_mv --
+\N \N
+-4 16
+1 1
+2 4
+3 \N
+
+-- !select_mv --
+\N \N
+-4 16
+1 1
+2 4
+3 \N
+
+-- !select_mv --
+\N \N
+-4 16
+1 1
+2 4
+3 \N
+
+-- !select_mv --
+\N \N
+-4 16
+1 1
+2 4
+3 \N
+
diff --git a/regression-test/data/mv_p0/ssb/multiple_no_where/multiple_no_where.out b/regression-test/data/mv_p0/ssb/multiple_no_where/multiple_no_where.out
new file mode 100644
index 0000000000..dd404cf8ed
--- /dev/null
+++ b/regression-test/data/mv_p0/ssb/multiple_no_where/multiple_no_where.out
@@ -0,0 +1,24 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_star --
+1 1 1 1 1 1 1 1 1 1 1 1 1 1 1 2023-06-09 shipmode name address city nation AMERICA phone mktsegment name address city nation AMERICA phone name MFGR#1 category brand color type 4 container
+2 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2023-06-09 shipmode name address city nation region phone mktsegment name address city nation region phone name mfgr category brand color type 4 container
+19920101 1 1 1 1 1 1 1 1 1 1 100 1 1 1 2023-06-09 ASIA ASIA ASIA ASIA ASIA ASIA ASIA ASIA ASIA ASIA ASIA ASIA ASIA ASIA ASIA MFGR#12 MFGR#12 brand color type 4 container
+19930101 1 1 1 1 1 1 1 1 1 1 100 1 1 1 2023-06-09 shipmode name address city nation AMERICA phone mktsegment name address city nation AMERICA phone name MFGR#1 category brand color type 4 container
+19930101 1 1 1 1 1 1 1 1 1 1 100 1 1 1 2023-06-09 shipmode name address city nation AMERICA phone mktsegment name address city nation AMERICA phone name MFGR#12 MFGR#12 brand color type 4 container
+19930101 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2023-06-09 shipmode name address city nation region phone mktsegment name address city nation region phone name mfgr category brand color type 4 container
+19930101 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2023-06-09 shipmode name address city nation region phone mktsegment name address city nation region phone name mfgr category brand color type 4 container
+19930101 2 2 2 2 2 2 2 2 2 2 2 2 2 2 2023-06-09 shipmode name address city nation region phone mktsegment name address city nation region phone name mfgr category brand color type 4 container
+
+-- !select_q_1_1 --
+12
+
+-- !select_q_2_1 --
+1 1993 brand
+
+-- !select_mv --
+ASIA ASIA 1992 1
+
+-- !select_mv --
+0 nation 0
+1993 nation 0
+
diff --git a/regression-test/suites/mv_p0/k1s2m3/k1s2m3.groovy b/regression-test/suites/mv_p0/k1s2m3/k1s2m3.groovy
new file mode 100644
index 0000000000..9c732f5699
--- /dev/null
+++ b/regression-test/suites/mv_p0/k1s2m3/k1s2m3.groovy
@@ -0,0 +1,101 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.codehaus.groovy.runtime.IOGroovyMethods
+
+suite ("k1s2m3") {
+
+ sql """ DROP TABLE IF EXISTS d_table; """
+
+ sql """
+ create table d_table(
+ k1 int null,
+ k2 int not null,
+ k3 bigint null,
+ k4 varchar(100) null
+ )
+ duplicate key (k1,k2,k3)
+ distributed BY hash(k1) buckets 3
+ properties("replication_num" = "1");
+ """
+
+ sql "insert into d_table select 1,1,1,'a';"
+ sql "insert into d_table select 2,2,2,'b';"
+ sql "insert into d_table select 3,-3,null,'c';"
+
+ createMV("create materialized view k1s2m3 as select k1,sum(k2*k3) from d_table group by k1;")
+
+ sql "insert into d_table select -4,-4,-4,'d';"
+ sql "insert into d_table(k4,k2) values('d',4);"
+
+ qt_select_star "select * from d_table order by k1;"
+
+ sql "set enable_nereids_planner=false"
+ explain {
+ sql("select k1,sum(k2*k3) from d_table group by k1 order by k1;")
+ contains "(k1s2m3)"
+ }
+ qt_select_mv "select k1,sum(k2*k3) from d_table group by k1 order by k1;"
+
+ explain {
+ sql("select K1,sum(K2*K3) from d_table group by K1 order by K1;")
+ contains "(k1s2m3)"
+ }
+ qt_select_mv "select K1,sum(K2*K3) from d_table group by K1 order by K1;"
+
+ sql "set enable_nereids_planner=true"
+ explain {
+ sql("select k1,sum(k2*k3) from d_table group by k1 order by k1;")
+ contains "(k1s2m3)"
+ }
+ qt_select_mv "select k1,sum(k2*k3) from d_table group by k1 order by k1;"
+
+ explain {
+ sql("select K1,sum(K2*K3) from d_table group by K1 order by K1;")
+ contains "(k1s2m3)"
+ }
+ qt_select_mv "select K1,sum(K2*K3) from d_table group by K1 order by K1;"
+
+ sql""" drop materialized view k1s2m3 on d_table; """
+ createMV("create materialized view k1s2m3 as select K1,sum(K2*K3) from d_table group by K1;")
+
+ sql "set enable_nereids_planner=false"
+ explain {
+ sql("select k1,sum(k2*k3) from d_table group by k1 order by k1;")
+ contains "(k1s2m3)"
+ }
+ qt_select_mv "select k1,sum(k2*k3) from d_table group by k1 order by k1;"
+
+ explain {
+ sql("select K1,sum(K2*K3) from d_table group by K1 order by K1;")
+ contains "(k1s2m3)"
+ }
+ qt_select_mv "select K1,sum(K2*K3) from d_table group by K1 order by K1;"
+
+ sql "set enable_nereids_planner=true"
+ explain {
+ sql("select k1,sum(k2*k3) from d_table group by k1 order by k1;")
+ contains "(k1s2m3)"
+ }
+ qt_select_mv "select k1,sum(k2*k3) from d_table group by k1 order by k1;"
+
+ explain {
+ sql("select K1,sum(K2*K3) from d_table group by K1 order by K1;")
+ contains "(k1s2m3)"
+ }
+ qt_select_mv "select K1,sum(K2*K3) from d_table group by K1 order by K1;"
+}
diff --git a/regression-test/suites/mv_p0/ssb/multiple_no_where/multiple_no_where.groovy b/regression-test/suites/mv_p0/ssb/multiple_no_where/multiple_no_where.groovy
new file mode 100644
index 0000000000..36f6fc634a
--- /dev/null
+++ b/regression-test/suites/mv_p0/ssb/multiple_no_where/multiple_no_where.groovy
@@ -0,0 +1,226 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.codehaus.groovy.runtime.IOGroovyMethods
+
+suite ("multiple_no_where") {
+ sql """ DROP TABLE IF EXISTS lineorder_flat; """
+ sql """set enable_nereids_planner=true"""
+ sql """SET enable_fallback_to_original_planner=false"""
+
+ sql """
+ CREATE TABLE IF NOT EXISTS `lineorder_flat` (
+ `LO_ORDERDATE` int(11) NOT NULL COMMENT "",
+ `LO_ORDERKEY` int(11) NOT NULL COMMENT "",
+ `LO_LINENUMBER` tinyint(4) NOT NULL COMMENT "",
+ `LO_CUSTKEY` int(11) NOT NULL COMMENT "",
+ `LO_PARTKEY` int(11) NOT NULL COMMENT "",
+ `LO_SUPPKEY` int(11) NOT NULL COMMENT "",
+ `LO_ORDERPRIORITY` varchar(100) NOT NULL COMMENT "",
+ `LO_SHIPPRIORITY` tinyint(4) NOT NULL COMMENT "",
+ `LO_QUANTITY` tinyint(4) NOT NULL COMMENT "",
+ `LO_EXTENDEDPRICE` int(11) NOT NULL COMMENT "",
+ `LO_ORDTOTALPRICE` int(11) NOT NULL COMMENT "",
+ `LO_DISCOUNT` tinyint(4) NOT NULL COMMENT "",
+ `LO_REVENUE` int(11) NOT NULL COMMENT "",
+ `LO_SUPPLYCOST` int(11) NOT NULL COMMENT "",
+ `LO_TAX` tinyint(4) NOT NULL COMMENT "",
+ `LO_COMMITDATE` date NOT NULL COMMENT "",
+ `LO_SHIPMODE` varchar(100) NOT NULL COMMENT "",
+ `C_NAME` varchar(100) NOT NULL COMMENT "",
+ `C_ADDRESS` varchar(100) NOT NULL COMMENT "",
+ `C_CITY` varchar(100) NOT NULL COMMENT "",
+ `C_NATION` varchar(100) NOT NULL COMMENT "",
+ `C_REGION` varchar(100) NOT NULL COMMENT "",
+ `C_PHONE` varchar(100) NOT NULL COMMENT "",
+ `C_MKTSEGMENT` varchar(100) NOT NULL COMMENT "",
+ `S_NAME` varchar(100) NOT NULL COMMENT "",
+ `S_ADDRESS` varchar(100) NOT NULL COMMENT "",
+ `S_CITY` varchar(100) NOT NULL COMMENT "",
+ `S_NATION` varchar(100) NOT NULL COMMENT "",
+ `S_REGION` varchar(100) NOT NULL COMMENT "",
+ `S_PHONE` varchar(100) NOT NULL COMMENT "",
+ `P_NAME` varchar(100) NOT NULL COMMENT "",
+ `P_MFGR` varchar(100) NOT NULL COMMENT "",
+ `P_CATEGORY` varchar(100) NOT NULL COMMENT "",
+ `P_BRAND` varchar(100) NOT NULL COMMENT "",
+ `P_COLOR` varchar(100) NOT NULL COMMENT "",
+ `P_TYPE` varchar(100) NOT NULL COMMENT "",
+ `P_SIZE` tinyint(4) NOT NULL COMMENT "",
+ `P_CONTAINER` varchar(100) NOT NULL COMMENT ""
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`LO_ORDERDATE`, `LO_ORDERKEY`)
+ COMMENT "OLAP"
+ PARTITION BY RANGE(`LO_ORDERDATE`)
+ (PARTITION p1992 VALUES [("-2147483648"), ("19930101")),
+ PARTITION p1993 VALUES [("19930101"), ("19940101")),
+ PARTITION p1994 VALUES [("19940101"), ("19950101")),
+ PARTITION p1995 VALUES [("19950101"), ("19960101")),
+ PARTITION p1996 VALUES [("19960101"), ("19970101")),
+ PARTITION p1997 VALUES [("19970101"), ("19980101")),
+ PARTITION p1998 VALUES [("19980101"), ("19990101")))
+ DISTRIBUTED BY HASH(`LO_ORDERKEY`) BUCKETS 48
+ PROPERTIES (
+ "replication_num" = "1",
+ "colocate_with" = "groupxx1",
+ "in_memory" = "false",
+ "storage_format" = "DEFAULT"
+ );
+ """
+
+ sql """INSERT INTO lineorder_flat (LO_ORDERDATE, LO_ORDERKEY, LO_LINENUMBER, LO_CUSTKEY, LO_PARTKEY, LO_SUPPKEY, LO_ORDERPRIORITY, LO_SHIPPRIORITY, LO_QUANTITY, LO_EXTENDEDPRICE, LO_ORDTOTALPRICE, LO_DISCOUNT, LO_REVENUE, LO_SUPPLYCOST, LO_TAX, LO_COMMITDATE, LO_SHIPMODE, C_NAME, C_ADDRESS, C_CITY, C_NATION, C_REGION, C_PHONE, C_MKTSEGMENT, S_NAME, S_ADDRESS, S_CITY, S_NATION, S_REGION, S_PHONE, P_NAME, P_MFGR, P_CATEGORY, P_BRAND, P_COLOR,P_TYPE,P_SIZE,P_CONTAINER) VALUES (19930101 [...]
+
+ test {
+ sql """create materialized view lineorder_q_1_1 as
+ SELECT LO_ORDERKEY, SUM(LO_EXTENDEDPRICE * LO_DISCOUNT)
+ FROM lineorder_flat GROUP BY
+ LO_ORDERKEY, LO_ORDERDATE, LO_DISCOUNT, LO_QUANTITY;"""
+ exception "not in select list"
+ }
+
+ createMV ("""create materialized view lineorder_q_1_1 as
+ SELECT LO_ORDERKEY, LO_ORDERDATE, LO_DISCOUNT, LO_QUANTITY, SUM(LO_EXTENDEDPRICE * LO_DISCOUNT)
+ FROM lineorder_flat GROUP BY
+ LO_ORDERKEY, LO_ORDERDATE, LO_DISCOUNT, LO_QUANTITY;""")
+
+ createMV ("""create materialized view lineorder_q_2_1 as
+ SELECT
+ (LO_ORDERDATE DIV 10000) AS YEAR,
+ P_BRAND, P_CATEGORY, S_REGION,
+ SUM(LO_REVENUE)
+ FROM lineorder_flat
+ GROUP BY YEAR, P_BRAND, P_CATEGORY,S_REGION;""")
+
+ createMV ("""create materialized view lineorder_q_3_1 as
+ SELECT
+ C_NATION,
+ S_NATION, (LO_ORDERDATE DIV 10000) AS YEAR, C_REGION, S_REGION, LO_ORDERDATE,
+ SUM(LO_REVENUE) AS revenue
+ FROM lineorder_flat
+ GROUP BY C_NATION, S_NATION, YEAR, C_REGION, S_REGION, LO_ORDERDATE;""")
+
+ createMV ("""create materialized view lineorder_q_4_1 as
+ SELECT (LO_ORDERDATE DIV 10000) AS YEAR,
+ C_NATION,C_REGION,S_REGION,P_MFGR,
+ SUM(LO_REVENUE - LO_SUPPLYCOST) AS profit
+ FROM lineorder_flat
+ GROUP BY YEAR, C_NATION,C_REGION,S_REGION,P_MFGR;""")
+
+ sql """INSERT INTO lineorder_flat (LO_ORDERDATE, LO_ORDERKEY, LO_LINENUMBER, LO_CUSTKEY, LO_PARTKEY, LO_SUPPKEY, LO_ORDERPRIORITY, LO_SHIPPRIORITY, LO_QUANTITY, LO_EXTENDEDPRICE, LO_ORDTOTALPRICE, LO_DISCOUNT, LO_REVENUE, LO_SUPPLYCOST, LO_TAX, LO_COMMITDATE, LO_SHIPMODE,C_NAME,C_ADDRESS,C_CITY,C_NATION,C_REGION,C_PHONE,C_MKTSEGMENT,S_NAME,S_ADDRESS,S_CITY,S_NATION,S_REGION,S_PHONE,P_NAME,P_MFGR,P_CATEGORY,P_BRAND,P_COLOR,P_TYPE,P_SIZE,P_CONTAINER) VALUES (19930101 , 2 , 2 , 2 , 2 , [...]
+
+ sql """INSERT INTO lineorder_flat (LO_ORDERDATE, LO_ORDERKEY, LO_LINENUMBER, LO_CUSTKEY, LO_PARTKEY, LO_SUPPKEY, LO_ORDERPRIORITY, LO_SHIPPRIORITY, LO_QUANTITY, LO_EXTENDEDPRICE, LO_ORDTOTALPRICE, LO_DISCOUNT, LO_REVENUE, LO_SUPPLYCOST, LO_TAX, LO_COMMITDATE, LO_SHIPMODE, C_NAME, C_ADDRESS, C_CITY, C_NATION, C_REGION, C_PHONE, C_MKTSEGMENT, S_NAME, S_ADDRESS, S_CITY, S_NATION, S_REGION, S_PHONE, P_NAME, P_MFGR, P_CATEGORY, P_BRAND, P_COLOR,P_TYPE,P_SIZE,P_CONTAINER) VALUES (19930101 [...]
+
+ sql """INSERT INTO lineorder_flat (LO_ORDERDATE, LO_ORDERKEY, LO_LINENUMBER, LO_CUSTKEY, LO_PARTKEY, LO_SUPPKEY, LO_ORDERPRIORITY, LO_SHIPPRIORITY, LO_QUANTITY, LO_EXTENDEDPRICE, LO_ORDTOTALPRICE, LO_DISCOUNT, LO_REVENUE, LO_SUPPLYCOST, LO_TAX, LO_COMMITDATE, LO_SHIPMODE,C_NAME,C_ADDRESS,C_CITY,C_NATION,C_REGION,C_PHONE,C_MKTSEGMENT,S_NAME,S_ADDRESS,S_CITY,S_NATION,S_REGION,S_PHONE,P_NAME,P_MFGR,P_CATEGORY,P_BRAND,P_COLOR,P_TYPE,P_SIZE,P_CONTAINER) VALUES (19930101 , 2 , 2 , 2 , 2 , [...]
+
+ sql """INSERT INTO lineorder_flat (LO_ORDERDATE, LO_ORDERKEY, LO_LINENUMBER, LO_CUSTKEY, LO_PARTKEY, LO_SUPPKEY, LO_ORDERPRIORITY, LO_SHIPPRIORITY, LO_QUANTITY, LO_EXTENDEDPRICE, LO_ORDTOTALPRICE, LO_DISCOUNT, LO_REVENUE, LO_SUPPLYCOST, LO_TAX, LO_COMMITDATE, LO_SHIPMODE, C_NAME, C_ADDRESS, C_CITY, C_NATION, C_REGION, C_PHONE, C_MKTSEGMENT, S_NAME, S_ADDRESS, S_CITY, S_NATION, S_REGION, S_PHONE, P_NAME, P_MFGR, P_CATEGORY, P_BRAND, P_COLOR,P_TYPE,P_SIZE,P_CONTAINER) VALUES (19920101 [...]
+
+ sql """INSERT INTO lineorder_flat (LO_ORDERDATE, LO_ORDERKEY, LO_LINENUMBER, LO_CUSTKEY, LO_PARTKEY, LO_SUPPKEY, LO_ORDERPRIORITY, LO_SHIPPRIORITY, LO_QUANTITY, LO_EXTENDEDPRICE, LO_ORDTOTALPRICE, LO_DISCOUNT, LO_REVENUE, LO_SUPPLYCOST, LO_TAX, LO_COMMITDATE, LO_SHIPMODE,C_NAME,C_ADDRESS,C_CITY,C_NATION,C_REGION,C_PHONE,C_MKTSEGMENT,S_NAME,S_ADDRESS,S_CITY,S_NATION,S_REGION,S_PHONE,P_NAME,P_MFGR,P_CATEGORY,P_BRAND,P_COLOR,P_TYPE,P_SIZE,P_CONTAINER) VALUES (19930101 , 2 , 2 , 2 , 2 , [...]
+
+ sql """INSERT INTO lineorder_flat (LO_ORDERDATE, LO_ORDERKEY, LO_LINENUMBER, LO_CUSTKEY, LO_PARTKEY, LO_SUPPKEY, LO_ORDERPRIORITY, LO_SHIPPRIORITY, LO_QUANTITY, LO_EXTENDEDPRICE, LO_ORDTOTALPRICE, LO_DISCOUNT, LO_REVENUE, LO_SUPPLYCOST, LO_TAX, LO_COMMITDATE, LO_SHIPMODE,C_NAME,C_ADDRESS,C_CITY,C_NATION,C_REGION,C_PHONE,C_MKTSEGMENT,S_NAME,S_ADDRESS,S_CITY,S_NATION,S_REGION,S_PHONE,P_NAME,P_MFGR,P_CATEGORY,P_BRAND,P_COLOR,P_TYPE,P_SIZE,P_CONTAINER) VALUES (2 , 2 , 2 , 2 , 2 , 2 ,'2', [...]
+
+ sql """INSERT INTO lineorder_flat (LO_ORDERDATE, LO_ORDERKEY, LO_LINENUMBER, LO_CUSTKEY, LO_PARTKEY, LO_SUPPKEY, LO_ORDERPRIORITY, LO_SHIPPRIORITY, LO_QUANTITY, LO_EXTENDEDPRICE, LO_ORDTOTALPRICE, LO_DISCOUNT, LO_REVENUE, LO_SUPPLYCOST, LO_TAX, LO_COMMITDATE, LO_SHIPMODE, C_NAME, C_ADDRESS, C_CITY, C_NATION, C_REGION, C_PHONE, C_MKTSEGMENT, S_NAME, S_ADDRESS, S_CITY, S_NATION, S_REGION, S_PHONE, P_NAME, P_MFGR, P_CATEGORY, P_BRAND, P_COLOR,P_TYPE,P_SIZE,P_CONTAINER) VALUES (1 , 1 , 1 [...]
+
+ qt_select_star "select * from lineorder_flat order by 1,2;"
+
+ explain {
+ sql("""SELECT SUM(LO_EXTENDEDPRICE * LO_DISCOUNT) AS revenue
+ FROM lineorder_flat
+ WHERE
+ LO_ORDERDATE >= 19930101
+ AND LO_ORDERDATE <= 19931231
+ AND LO_DISCOUNT >= 1 AND LO_DISCOUNT <= 3
+ AND LO_QUANTITY < 25;""")
+ contains "(lineorder_q_1_1)"
+ }
+ qt_select_q_1_1 """SELECT SUM(LO_EXTENDEDPRICE * LO_DISCOUNT) AS revenue
+ FROM lineorder_flat
+ WHERE
+ LO_ORDERDATE >= 19930101
+ AND LO_ORDERDATE <= 19931231
+ AND LO_DISCOUNT >= 1 AND LO_DISCOUNT <= 3
+ AND LO_QUANTITY < 25;"""
+
+ explain {
+ sql("""SELECT
+ SUM(LO_REVENUE), (LO_ORDERDATE DIV 10000) AS YEAR,
+ P_BRAND
+ FROM lineorder_flat
+ WHERE P_CATEGORY = 'MFGR#12' AND S_REGION = 'AMERICA'
+ GROUP BY (LO_ORDERDATE DIV 10000), P_BRAND
+ ORDER BY YEAR, P_BRAND;""")
+ contains "(lineorder_q_2_1)"
+ }
+ qt_select_q_2_1 """SELECT
+ SUM(LO_REVENUE), (LO_ORDERDATE DIV 10000) AS YEAR,
+ P_BRAND
+ FROM lineorder_flat
+ WHERE P_CATEGORY = 'MFGR#12' AND S_REGION = 'AMERICA'
+ GROUP BY YEAR, P_BRAND
+ ORDER BY YEAR, P_BRAND;"""
+
+ explain {
+ sql("""SELECT
+ C_NATION,
+ S_NATION, (LO_ORDERDATE DIV 10000) AS YEAR,
+ SUM(LO_REVENUE) AS revenue
+ FROM lineorder_flat
+ WHERE
+ C_REGION = 'ASIA'
+ AND S_REGION = 'ASIA'
+ AND LO_ORDERDATE >= 19920101
+ AND LO_ORDERDATE <= 19971231
+ GROUP BY C_NATION, S_NATION, YEAR
+ ORDER BY YEAR ASC, revenue DESC;""")
+ contains "(lineorder_q_3_1)"
+ }
+ qt_select_mv """SELECT
+ C_NATION,
+ S_NATION, (LO_ORDERDATE DIV 10000) AS YEAR,
+ SUM(LO_REVENUE) AS revenue
+ FROM lineorder_flat
+ WHERE
+ C_REGION = 'ASIA'
+ AND S_REGION = 'ASIA'
+ AND LO_ORDERDATE >= 19920101
+ AND LO_ORDERDATE <= 19971231
+ GROUP BY C_NATION, S_NATION, YEAR
+ ORDER BY YEAR ASC, revenue DESC;"""
+
+ explain {
+ sql("""SELECT (LO_ORDERDATE DIV 10000) AS YEAR,
+ C_NATION,
+ SUM(LO_REVENUE - LO_SUPPLYCOST) AS profit
+ FROM lineorder_flat
+ WHERE
+ C_REGION = 'AMERICA'
+ AND S_REGION = 'AMERICA'
+ AND P_MFGR IN ('MFGR#1', 'MFGR#2')
+ GROUP BY YEAR, C_NATION
+ ORDER BY YEAR ASC, C_NATION ASC;""")
+ contains "(lineorder_q_4_1)"
+ }
+ qt_select_mv """SELECT (LO_ORDERDATE DIV 10000) AS YEAR,
+ C_NATION,
+ SUM(LO_REVENUE - LO_SUPPLYCOST) AS profit
+ FROM lineorder_flat
+ WHERE
+ C_REGION = 'AMERICA'
+ AND S_REGION = 'AMERICA'
+ AND P_MFGR IN ('MFGR#1', 'MFGR#2')
+ GROUP BY YEAR, C_NATION
+ ORDER BY YEAR ASC, C_NATION ASC;"""
+}
diff --git a/regression-test/suites/nereids_p0/sql_functions/test_arith_functions.groovy b/regression-test/suites/nereids_p0/sql_functions/test_arith_functions.groovy
index 5294a0e58e..2778d0f359 100644
--- a/regression-test/suites/nereids_p0/sql_functions/test_arith_functions.groovy
+++ b/regression-test/suites/nereids_p0/sql_functions/test_arith_functions.groovy
@@ -26,7 +26,7 @@ suite("test_arith_functions") {
}
test {
sql 'select int_divide(1, 1), bitand(1, 1), bitor(2, 2), bitxor(3.0, 2.0), bitnot(3.0)'
- result([[1L, 1, 2, 1L, -4L]])
+ result([[1, 1, 2, 1L, -4L]])
}
test {
sql 'select add(k1, k2) + subtract(k2, k3) + multiply(k3, k4), cast(divide(k4, k3) + mod(k4, k3) as bigint) from test order by k1 limit 1'
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 06/10: [refactor](profile)add node id info in pipeline profile (#21823)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 0f74793c1dfd701188ddafc4c6220c10a37467c2
Author: zhangstar333 <87...@users.noreply.github.com>
AuthorDate: Mon Jul 17 15:24:02 2023 +0800
[refactor](profile)add node id info in pipeline profile (#21823)
---
be/src/pipeline/pipeline_fragment_context.cpp | 90 +++++++++++++--------------
be/src/pipeline/pipeline_task.cpp | 11 ++--
2 files changed, 49 insertions(+), 52 deletions(-)
diff --git a/be/src/pipeline/pipeline_fragment_context.cpp b/be/src/pipeline/pipeline_fragment_context.cpp
index ea0a6c9240..69848be82a 100644
--- a/be/src/pipeline/pipeline_fragment_context.cpp
+++ b/be/src/pipeline/pipeline_fragment_context.cpp
@@ -438,15 +438,14 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
case TPlanNodeType::FILE_SCAN_NODE:
case TPlanNodeType::META_SCAN_NODE:
case TPlanNodeType::ES_SCAN_NODE: {
- OperatorBuilderPtr operator_t =
- std::make_shared<ScanOperatorBuilder>(next_operator_builder_id(), node);
+ OperatorBuilderPtr operator_t = std::make_shared<ScanOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(operator_t));
break;
}
case TPlanNodeType::MYSQL_SCAN_NODE: {
#ifdef DORIS_WITH_MYSQL
OperatorBuilderPtr operator_t =
- std::make_shared<MysqlScanOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<MysqlScanOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(operator_t));
break;
#else
@@ -456,25 +455,24 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
}
case TPlanNodeType::SCHEMA_SCAN_NODE: {
OperatorBuilderPtr operator_t =
- std::make_shared<SchemaScanOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<SchemaScanOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(operator_t));
break;
}
case TPlanNodeType::EXCHANGE_NODE: {
OperatorBuilderPtr operator_t =
- std::make_shared<ExchangeSourceOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<ExchangeSourceOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(operator_t));
break;
}
case TPlanNodeType::EMPTY_SET_NODE: {
OperatorBuilderPtr operator_t =
- std::make_shared<EmptySetSourceOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<EmptySetSourceOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(operator_t));
break;
}
case TPlanNodeType::DATA_GEN_SCAN_NODE: {
- OperatorBuilderPtr operator_t =
- std::make_shared<DataGenOperatorBuilder>(next_operator_builder_id(), node);
+ OperatorBuilderPtr operator_t = std::make_shared<DataGenOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(operator_t));
break;
}
@@ -483,7 +481,7 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
if (union_node->children_count() == 0 &&
union_node->get_first_materialized_child_idx() == 0) { // only have const expr
OperatorBuilderPtr builder =
- std::make_shared<ConstValueOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<ConstValueOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(builder));
} else {
int child_count = union_node->children_count();
@@ -492,11 +490,11 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
auto new_child_pipeline = add_pipeline();
RETURN_IF_ERROR(_build_pipelines(union_node->child(child_id), new_child_pipeline));
OperatorBuilderPtr child_sink_builder = std::make_shared<UnionSinkOperatorBuilder>(
- next_operator_builder_id(), child_id, union_node, data_queue);
+ union_node->id(), child_id, union_node, data_queue);
RETURN_IF_ERROR(new_child_pipeline->set_sink(child_sink_builder));
}
OperatorBuilderPtr source_builder = std::make_shared<UnionSourceOperatorBuilder>(
- next_operator_builder_id(), union_node, data_queue);
+ node->id(), union_node, data_queue);
RETURN_IF_ERROR(cur_pipe->add_operator(source_builder));
}
break;
@@ -508,19 +506,19 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
if (agg_node->is_streaming_preagg()) {
auto data_queue = std::make_shared<DataQueue>(1);
OperatorBuilderPtr pre_agg_sink = std::make_shared<StreamingAggSinkOperatorBuilder>(
- next_operator_builder_id(), agg_node, data_queue);
+ node->id(), agg_node, data_queue);
RETURN_IF_ERROR(new_pipe->set_sink(pre_agg_sink));
OperatorBuilderPtr pre_agg_source = std::make_shared<StreamingAggSourceOperatorBuilder>(
- next_operator_builder_id(), agg_node, data_queue);
+ node->id(), agg_node, data_queue);
RETURN_IF_ERROR(cur_pipe->add_operator(pre_agg_source));
} else {
OperatorBuilderPtr agg_sink =
- std::make_shared<AggSinkOperatorBuilder>(next_operator_builder_id(), agg_node);
+ std::make_shared<AggSinkOperatorBuilder>(node->id(), agg_node);
RETURN_IF_ERROR(new_pipe->set_sink(agg_sink));
- OperatorBuilderPtr agg_source = std::make_shared<AggSourceOperatorBuilder>(
- next_operator_builder_id(), agg_node);
+ OperatorBuilderPtr agg_source =
+ std::make_shared<AggSourceOperatorBuilder>(node->id(), agg_node);
RETURN_IF_ERROR(cur_pipe->add_operator(agg_source));
}
break;
@@ -529,12 +527,11 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
auto new_pipeline = add_pipeline();
RETURN_IF_ERROR(_build_pipelines(node->child(0), new_pipeline));
- OperatorBuilderPtr sort_sink =
- std::make_shared<SortSinkOperatorBuilder>(next_operator_builder_id(), node);
+ OperatorBuilderPtr sort_sink = std::make_shared<SortSinkOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(new_pipeline->set_sink(sort_sink));
OperatorBuilderPtr sort_source =
- std::make_shared<SortSourceOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<SortSourceOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(sort_source));
break;
}
@@ -542,13 +539,12 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
auto new_pipeline = add_pipeline();
RETURN_IF_ERROR(_build_pipelines(node->child(0), new_pipeline));
- OperatorBuilderPtr partition_sort_sink = std::make_shared<PartitionSortSinkOperatorBuilder>(
- next_operator_builder_id(), node);
+ OperatorBuilderPtr partition_sort_sink =
+ std::make_shared<PartitionSortSinkOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(new_pipeline->set_sink(partition_sort_sink));
OperatorBuilderPtr partition_sort_source =
- std::make_shared<PartitionSortSourceOperatorBuilder>(next_operator_builder_id(),
- node);
+ std::make_shared<PartitionSortSourceOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(partition_sort_source));
break;
}
@@ -557,32 +553,31 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
RETURN_IF_ERROR(_build_pipelines(node->child(0), new_pipeline));
OperatorBuilderPtr analytic_sink =
- std::make_shared<AnalyticSinkOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<AnalyticSinkOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(new_pipeline->set_sink(analytic_sink));
OperatorBuilderPtr analytic_source =
- std::make_shared<AnalyticSourceOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<AnalyticSourceOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(analytic_source));
break;
}
case TPlanNodeType::REPEAT_NODE: {
RETURN_IF_ERROR(_build_pipelines(node->child(0), cur_pipe));
- OperatorBuilderPtr builder =
- std::make_shared<RepeatOperatorBuilder>(next_operator_builder_id(), node);
+ OperatorBuilderPtr builder = std::make_shared<RepeatOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(builder));
break;
}
case TPlanNodeType::ASSERT_NUM_ROWS_NODE: {
RETURN_IF_ERROR(_build_pipelines(node->child(0), cur_pipe));
OperatorBuilderPtr builder =
- std::make_shared<AssertNumRowsOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<AssertNumRowsOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(builder));
break;
}
case TPlanNodeType::TABLE_FUNCTION_NODE: {
RETURN_IF_ERROR(_build_pipelines(node->child(0), cur_pipe));
OperatorBuilderPtr builder =
- std::make_shared<TableFunctionOperatorBuilder>(next_operator_builder_id(), node);
+ std::make_shared<TableFunctionOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(builder));
break;
}
@@ -593,16 +588,16 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
RETURN_IF_ERROR(_build_pipelines(node->child(1), new_pipe));
} else {
OperatorBuilderPtr builder = std::make_shared<EmptySourceOperatorBuilder>(
- next_operator_builder_id(), node->child(1)->row_desc(), node->child(1));
+ node->child(1)->id(), node->child(1)->row_desc(), node->child(1));
new_pipe->add_operator(builder);
}
OperatorBuilderPtr join_sink =
- std::make_shared<HashJoinBuildSinkBuilder>(next_operator_builder_id(), join_node);
+ std::make_shared<HashJoinBuildSinkBuilder>(node->id(), join_node);
RETURN_IF_ERROR(new_pipe->set_sink(join_sink));
RETURN_IF_ERROR(_build_pipelines(node->child(0), cur_pipe));
- OperatorBuilderPtr join_source = std::make_shared<HashJoinProbeOperatorBuilder>(
- next_operator_builder_id(), join_node);
+ OperatorBuilderPtr join_source =
+ std::make_shared<HashJoinProbeOperatorBuilder>(node->id(), join_node);
RETURN_IF_ERROR(cur_pipe->add_operator(join_source));
cur_pipe->add_dependency(new_pipe);
@@ -611,13 +606,13 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
case TPlanNodeType::CROSS_JOIN_NODE: {
auto new_pipe = add_pipeline();
RETURN_IF_ERROR(_build_pipelines(node->child(1), new_pipe));
- OperatorBuilderPtr join_sink = std::make_shared<NestLoopJoinBuildOperatorBuilder>(
- next_operator_builder_id(), node);
+ OperatorBuilderPtr join_sink =
+ std::make_shared<NestLoopJoinBuildOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(new_pipe->set_sink(join_sink));
RETURN_IF_ERROR(_build_pipelines(node->child(0), cur_pipe));
- OperatorBuilderPtr join_source = std::make_shared<NestLoopJoinProbeOperatorBuilder>(
- next_operator_builder_id(), node);
+ OperatorBuilderPtr join_source =
+ std::make_shared<NestLoopJoinProbeOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(join_source));
cur_pipe->add_dependency(new_pipe);
@@ -633,8 +628,7 @@ Status PipelineFragmentContext::_build_pipelines(ExecNode* node, PipelinePtr cur
}
case TPlanNodeType::SELECT_NODE: {
RETURN_IF_ERROR(_build_pipelines(node->child(0), cur_pipe));
- OperatorBuilderPtr builder =
- std::make_shared<SelectOperatorBuilder>(next_operator_builder_id(), node);
+ OperatorBuilderPtr builder = std::make_shared<SelectOperatorBuilder>(node->id(), node);
RETURN_IF_ERROR(cur_pipe->add_operator(builder));
break;
}
@@ -650,21 +644,21 @@ Status PipelineFragmentContext::_build_operators_for_set_operation_node(ExecNode
PipelinePtr cur_pipe) {
auto build_pipeline = add_pipeline();
RETURN_IF_ERROR(_build_pipelines(node->child(0), build_pipeline));
- OperatorBuilderPtr sink_builder = std::make_shared<SetSinkOperatorBuilder<is_intersect>>(
- next_operator_builder_id(), node);
+ OperatorBuilderPtr sink_builder =
+ std::make_shared<SetSinkOperatorBuilder<is_intersect>>(node->id(), node);
RETURN_IF_ERROR(build_pipeline->set_sink(sink_builder));
for (int child_id = 1; child_id < node->children_count(); ++child_id) {
auto probe_pipeline = add_pipeline();
RETURN_IF_ERROR(_build_pipelines(node->child(child_id), probe_pipeline));
OperatorBuilderPtr probe_sink_builder =
- std::make_shared<SetProbeSinkOperatorBuilder<is_intersect>>(
- next_operator_builder_id(), child_id, node);
+ std::make_shared<SetProbeSinkOperatorBuilder<is_intersect>>(node->id(), child_id,
+ node);
RETURN_IF_ERROR(probe_pipeline->set_sink(probe_sink_builder));
}
- OperatorBuilderPtr source_builder = std::make_shared<SetSourceOperatorBuilder<is_intersect>>(
- next_operator_builder_id(), node);
+ OperatorBuilderPtr source_builder =
+ std::make_shared<SetSourceOperatorBuilder<is_intersect>>(node->id(), node);
return cur_pipe->add_operator(source_builder);
}
@@ -734,7 +728,7 @@ Status PipelineFragmentContext::_create_sink(int sender_id, const TDataSink& thr
OperatorBuilderPtr sink_;
switch (thrift_sink.type) {
case TDataSinkType::DATA_STREAM_SINK: {
- sink_ = std::make_shared<ExchangeSinkOperatorBuilder>(next_operator_builder_id(),
+ sink_ = std::make_shared<ExchangeSinkOperatorBuilder>(thrift_sink.stream_sink.dest_node_id,
_sink.get(), this);
break;
}
@@ -755,8 +749,8 @@ Status PipelineFragmentContext::_create_sink(int sender_id, const TDataSink& thr
break;
}
case TDataSinkType::RESULT_FILE_SINK: {
- sink_ = std::make_shared<ResultFileSinkOperatorBuilder>(next_operator_builder_id(),
- _sink.get());
+ sink_ = std::make_shared<ResultFileSinkOperatorBuilder>(
+ thrift_sink.result_file_sink.dest_node_id, _sink.get());
break;
}
case TDataSinkType::MULTI_CAST_DATA_STREAM_SINK: {
diff --git a/be/src/pipeline/pipeline_task.cpp b/be/src/pipeline/pipeline_task.cpp
index 2d9e45004b..e48d729f46 100644
--- a/be/src/pipeline/pipeline_task.cpp
+++ b/be/src/pipeline/pipeline_task.cpp
@@ -94,15 +94,18 @@ Status PipelineTask::prepare(RuntimeState* state) {
RETURN_IF_ERROR(o->prepare(state));
}
- _task_profile->add_info_string("Sink", fmt::format("{}({})", _sink->get_name(), _sink->id()));
+ _task_profile->add_info_string("Sink",
+ fmt::format("{}(dst_id={})", _sink->get_name(), _sink->id()));
fmt::memory_buffer operator_ids_str;
for (size_t i = 0; i < _operators.size(); i++) {
if (i == 0) {
- fmt::format_to(operator_ids_str,
- fmt::format("[{}({})", _operators[i]->get_name(), _operators[i]->id()));
+ fmt::format_to(
+ operator_ids_str,
+ fmt::format("[{}(node_id={})", _operators[i]->get_name(), _operators[i]->id()));
} else {
fmt::format_to(operator_ids_str,
- fmt::format(", {}({})", _operators[i]->get_name(), _operators[i]->id()));
+ fmt::format(", {}(node_id={})", _operators[i]->get_name(),
+ _operators[i]->id()));
}
}
fmt::format_to(operator_ids_str, "]");
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 08/10: [enhancement](jdbc catalog) Add mysql jdbc url param `rewriteBatchedStatements=true` (#21864)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit ec5277d5bd0b73b0d7ef3cde4417cf56e1102916
Author: zy-kkk <zh...@gmail.com>
AuthorDate: Mon Jul 17 17:39:26 2023 +0800
[enhancement](jdbc catalog) Add mysql jdbc url param `rewriteBatchedStatements=true` (#21864)
When `rewriteBatchedStatements=false`, the JDBC driver will not merge multiple insert statements into one larger insert statement. Therefore, during the batch insertion process, each insert statement needs to be sent to the MySQL server individually, leading to a higher number of network roundtrips. Network latency could potentially be a significant factor contributing to the performance degradation. For this reason, we propose to set this parameter to true by default, to enhance the [...]
---
fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java | 5 +++--
1 file changed, 3 insertions(+), 2 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java
index 2e8a71f92a..8a76b2bc34 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/JdbcResource.java
@@ -303,10 +303,11 @@ public class JdbcResource extends Resource {
// However when tinyInt1isBit=false, GetColumnClassName of MySQL returns java.lang.Boolean,
// while that of Doris returns java.lang.Integer. In order to be compatible with both MySQL and Doris,
// Jdbc params should set tinyInt1isBit=true&transformedBitIsBoolean=true
- newJdbcUrl = checkAndSetJdbcBoolParam(newJdbcUrl, "tinyInt1isBit", "true", "true");
- newJdbcUrl = checkAndSetJdbcBoolParam(newJdbcUrl, "transformedBitIsBoolean", "true", "true");
+ newJdbcUrl = checkAndSetJdbcBoolParam(newJdbcUrl, "tinyInt1isBit", "false", "true");
+ newJdbcUrl = checkAndSetJdbcBoolParam(newJdbcUrl, "transformedBitIsBoolean", "false", "true");
// set useUnicode and characterEncoding to false and utf-8
newJdbcUrl = checkAndSetJdbcBoolParam(newJdbcUrl, "useUnicode", "false", "true");
+ newJdbcUrl = checkAndSetJdbcBoolParam(newJdbcUrl, "rewriteBatchedStatements", "false", "true");
newJdbcUrl = checkAndSetJdbcParam(newJdbcUrl, "characterEncoding", "utf-8");
if (dbType.equals(OCEANBASE)) {
// set useCursorFetch to true
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 03/10: [chore](topn opt) disable two phase read when light schema change is disabled (#21809)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit c865daee0866a2d8222f687fc82e4e29db0e7c9f
Author: lihangyu <15...@163.com>
AuthorDate: Mon Jul 17 12:46:28 2023 +0800
[chore](topn opt) disable two phase read when light schema change is disabled (#21809)
---
fe/fe-core/src/main/java/org/apache/doris/analysis/SelectStmt.java | 3 +++
1 file changed, 3 insertions(+)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/SelectStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/SelectStmt.java
index 468fd07284..e2ce149c4d 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/SelectStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/SelectStmt.java
@@ -758,6 +758,9 @@ public class SelectStmt extends QueryStmt {
LOG.debug("only support duplicate key or MOW model");
return false;
}
+ if (!olapTable.getEnableLightSchemaChange()) {
+ return false;
+ }
if (getOrderByElements() != null) {
if (!evaluateOrderBy) {
// Need evaluate orderby, if sort node was eliminated then this optmization
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 09/10: [Feature](Compaction)Support full compaction (#21177)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 373e586c088f96f363838da5ebd109179ba31ffd
Author: abmdocrt <Yu...@gmail.com>
AuthorDate: Sun Jul 16 13:21:15 2023 +0800
[Feature](Compaction)Support full compaction (#21177)
---
be/src/common/status.h | 3 +
be/src/http/action/compaction_action.cpp | 17 +-
be/src/http/action/compaction_action.h | 1 +
be/src/io/io_common.h | 3 +-
be/src/olap/compaction.cpp | 4 +
be/src/olap/full_compaction.cpp | 215 +++++++++++++++++++++
be/src/olap/full_compaction.h | 62 ++++++
be/src/olap/olap_common.h | 2 +-
be/src/olap/reader.cpp | 2 +
be/src/olap/rowset/rowset_meta.h | 2 +-
be/src/olap/single_replica_compaction.cpp | 2 +
be/src/olap/tablet.cpp | 18 ++
be/src/olap/tablet.h | 17 ++
.../data/compaction/test_full_compaction.out | 41 ++++
.../plugins/plugin_curl_requester.groovy | 6 +
.../suites/compaction/test_full_compaction.groovy | 177 +++++++++++++++++
16 files changed, 568 insertions(+), 4 deletions(-)
diff --git a/be/src/common/status.h b/be/src/common/status.h
index eda6a34d55..d64b70615c 100644
--- a/be/src/common/status.h
+++ b/be/src/common/status.h
@@ -229,6 +229,8 @@ E(CUMULATIVE_INVALID_NEED_MERGED_VERSIONS, -2004);
E(CUMULATIVE_ERROR_DELETE_ACTION, -2005);
E(CUMULATIVE_MISS_VERSION, -2006);
E(CUMULATIVE_CLONE_OCCURRED, -2007);
+E(FULL_NO_SUITABLE_VERSION, -2008);
+E(FULL_MISS_VERSION, -2009);
E(META_INVALID_ARGUMENT, -3000);
E(META_OPEN_DB_ERROR, -3001);
E(META_KEY_NOT_FOUND, -3002);
@@ -285,6 +287,7 @@ inline bool capture_stacktrace(int code) {
&& code != ErrorCode::PUSH_TRANSACTION_ALREADY_EXIST
&& code != ErrorCode::BE_NO_SUITABLE_VERSION
&& code != ErrorCode::CUMULATIVE_NO_SUITABLE_VERSION
+ && code != ErrorCode::FULL_NO_SUITABLE_VERSION
&& code != ErrorCode::PUBLISH_VERSION_NOT_CONTINUOUS
&& code != ErrorCode::ROWSET_RENAME_FILE_FAILED
&& code != ErrorCode::SEGCOMPACTION_INIT_READER
diff --git a/be/src/http/action/compaction_action.cpp b/be/src/http/action/compaction_action.cpp
index 48dbe78ab4..81f6ad4d30 100644
--- a/be/src/http/action/compaction_action.cpp
+++ b/be/src/http/action/compaction_action.cpp
@@ -37,6 +37,7 @@
#include "olap/base_compaction.h"
#include "olap/cumulative_compaction.h"
#include "olap/cumulative_compaction_policy.h"
+#include "olap/full_compaction.h"
#include "olap/olap_define.h"
#include "olap/storage_engine.h"
#include "olap/tablet_manager.h"
@@ -89,7 +90,8 @@ Status CompactionAction::_handle_run_compaction(HttpRequest* req, std::string* j
// check compaction_type equals 'base' or 'cumulative'
std::string compaction_type = req->param(PARAM_COMPACTION_TYPE);
if (compaction_type != PARAM_COMPACTION_BASE &&
- compaction_type != PARAM_COMPACTION_CUMULATIVE) {
+ compaction_type != PARAM_COMPACTION_CUMULATIVE &&
+ compaction_type != PARAM_COMPACTION_FULL) {
return Status::NotSupported("The compaction type '{}' is not supported", compaction_type);
}
@@ -229,6 +231,19 @@ Status CompactionAction::_execute_compaction_callback(TabletSharedPtr tablet,
<< ", table=" << tablet->full_name();
}
}
+ } else if (compaction_type == PARAM_COMPACTION_FULL) {
+ FullCompaction full_compaction(tablet);
+ res = full_compaction.compact();
+ if (!res) {
+ if (res.is<FULL_NO_SUITABLE_VERSION>()) {
+ // Ignore this error code.
+ VLOG_NOTICE << "failed to init full compaction due to no suitable version,"
+ << "tablet=" << tablet->full_name();
+ } else {
+ LOG(WARNING) << "failed to do full compaction. res=" << res
+ << ", table=" << tablet->full_name();
+ }
+ }
}
timer.stop();
diff --git a/be/src/http/action/compaction_action.h b/be/src/http/action/compaction_action.h
index 1feb7989e7..9f52c625f1 100644
--- a/be/src/http/action/compaction_action.h
+++ b/be/src/http/action/compaction_action.h
@@ -39,6 +39,7 @@ enum class CompactionActionType {
const std::string PARAM_COMPACTION_TYPE = "compact_type";
const std::string PARAM_COMPACTION_BASE = "base";
const std::string PARAM_COMPACTION_CUMULATIVE = "cumulative";
+const std::string PARAM_COMPACTION_FULL = "full";
/// This action is used for viewing the compaction status.
/// See compaction-action.md for details.
diff --git a/be/src/io/io_common.h b/be/src/io/io_common.h
index 18391f2a32..8849330746 100644
--- a/be/src/io/io_common.h
+++ b/be/src/io/io_common.h
@@ -29,7 +29,8 @@ enum class ReaderType {
READER_CHECKSUM = 4,
READER_COLD_DATA_COMPACTION = 5,
READER_SEGMENT_COMPACTION = 6,
- UNKNOWN = 7
+ READER_FULL_COMPACTION = 7,
+ UNKNOWN = 8
};
namespace io {
diff --git a/be/src/olap/compaction.cpp b/be/src/olap/compaction.cpp
index e4c0b59d8c..d67248919f 100644
--- a/be/src/olap/compaction.cpp
+++ b/be/src/olap/compaction.cpp
@@ -301,6 +301,8 @@ Status Compaction::do_compaction_impl(int64_t permits) {
_tablet->set_last_cumu_compaction_success_time(now);
} else if (compaction_type() == ReaderType::READER_BASE_COMPACTION) {
_tablet->set_last_base_compaction_success_time(now);
+ } else if (compaction_type() == ReaderType::READER_FULL_COMPACTION) {
+ _tablet->set_last_full_compaction_success_time(now);
}
auto cumu_policy = _tablet->cumulative_compaction_policy();
LOG(INFO) << "succeed to do ordered data " << compaction_name()
@@ -451,6 +453,8 @@ Status Compaction::do_compaction_impl(int64_t permits) {
_tablet->set_last_cumu_compaction_success_time(now);
} else if (compaction_type() == ReaderType::READER_BASE_COMPACTION) {
_tablet->set_last_base_compaction_success_time(now);
+ } else if (compaction_type() == ReaderType::READER_FULL_COMPACTION) {
+ _tablet->set_last_full_compaction_success_time(now);
}
int64_t current_max_version;
diff --git a/be/src/olap/full_compaction.cpp b/be/src/olap/full_compaction.cpp
new file mode 100644
index 0000000000..e2acd1cdb0
--- /dev/null
+++ b/be/src/olap/full_compaction.cpp
@@ -0,0 +1,215 @@
+// 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.
+
+#include "olap/full_compaction.h"
+
+#include <glog/logging.h>
+#include <time.h>
+
+#include <memory>
+#include <mutex>
+#include <ostream>
+#include <shared_mutex>
+
+#include "common/config.h"
+#include "common/status.h"
+#include "olap/cumulative_compaction_policy.h"
+#include "olap/olap_common.h"
+#include "olap/olap_define.h"
+#include "olap/rowset/beta_rowset.h"
+#include "olap/rowset/rowset.h"
+#include "olap/schema_change.h"
+#include "olap/tablet_meta.h"
+#include "runtime/thread_context.h"
+#include "util/thread.h"
+#include "util/trace.h"
+
+namespace doris {
+using namespace ErrorCode;
+
+FullCompaction::FullCompaction(const TabletSharedPtr& tablet)
+ : Compaction(tablet, "FullCompaction:" + std::to_string(tablet->tablet_id())) {}
+
+FullCompaction::~FullCompaction() {}
+
+Status FullCompaction::prepare_compact() {
+ if (!_tablet->init_succeeded()) {
+ return Status::Error<INVALID_ARGUMENT>("Full compaction init failed");
+ }
+
+ std::unique_lock full_lock(_tablet->get_full_compaction_lock());
+ std::unique_lock base_lock(_tablet->get_base_compaction_lock());
+ std::unique_lock cumu_lock(_tablet->get_cumulative_compaction_lock());
+
+ // 1. pick rowsets to compact
+ RETURN_IF_ERROR(pick_rowsets_to_compact());
+ _tablet->set_clone_occurred(false);
+
+ return Status::OK();
+}
+
+Status FullCompaction::execute_compact_impl() {
+ std::unique_lock full_lock(_tablet->get_full_compaction_lock());
+ std::unique_lock base_lock(_tablet->get_base_compaction_lock());
+ std::unique_lock cumu_lock(_tablet->get_cumulative_compaction_lock());
+
+ // Clone task may happen after compaction task is submitted to thread pool, and rowsets picked
+ // for compaction may change. In this case, current compaction task should not be executed.
+ if (_tablet->get_clone_occurred()) {
+ _tablet->set_clone_occurred(false);
+ return Status::Error<BE_CLONE_OCCURRED>("get_clone_occurred failed");
+ }
+
+ SCOPED_ATTACH_TASK(_mem_tracker);
+
+ // 2. do full compaction, merge rowsets
+ int64_t permits = get_compaction_permits();
+ RETURN_IF_ERROR(do_compaction(permits));
+
+ // 3. set state to success
+ _state = CompactionState::SUCCESS;
+
+ // 4. set cumulative point
+ Version last_version = _input_rowsets.back()->version();
+ _tablet->cumulative_compaction_policy()->update_cumulative_point(_tablet.get(), _input_rowsets,
+ _output_rowset, last_version);
+ VLOG_CRITICAL << "after cumulative compaction, current cumulative point is "
+ << _tablet->cumulative_layer_point() << ", tablet=" << _tablet->full_name();
+
+ return Status::OK();
+}
+
+Status FullCompaction::pick_rowsets_to_compact() {
+ _input_rowsets = _tablet->pick_candidate_rowsets_to_full_compaction();
+ RETURN_IF_ERROR(check_version_continuity(_input_rowsets));
+ RETURN_IF_ERROR(_check_all_version(_input_rowsets));
+ if (_input_rowsets.size() <= 1) {
+ return Status::Error<FULL_NO_SUITABLE_VERSION>("There is no suitable version");
+ }
+
+ if (_input_rowsets.size() == 2 && _input_rowsets[0]->end_version() == 1) {
+ // the tablet is with rowset: [0-1], [2-y]
+ // and [0-1] has no data. in this situation, no need to do full compaction.
+ return Status::Error<FULL_NO_SUITABLE_VERSION>("There is no suitable version");
+ }
+
+ return Status::OK();
+}
+
+Status FullCompaction::modify_rowsets(const Merger::Statistics* stats) {
+ if (_tablet->keys_type() == KeysType::UNIQUE_KEYS &&
+ _tablet->enable_unique_key_merge_on_write()) {
+ RETURN_IF_ERROR(
+ _full_compaction_update_delete_bitmap(_output_rowset, _output_rs_writer.get()));
+ }
+ std::vector<RowsetSharedPtr> output_rowsets(1, _output_rowset);
+ RETURN_IF_ERROR(_tablet->modify_rowsets(output_rowsets, _input_rowsets, true));
+ _tablet->save_meta();
+ return Status::OK();
+}
+
+Status FullCompaction::_check_all_version(const std::vector<RowsetSharedPtr>& rowsets) {
+ if (rowsets.empty()) {
+ return Status::Error<FULL_MISS_VERSION>("There is no input rowset when do full compaction");
+ }
+ const RowsetSharedPtr& last_rowset = rowsets.back();
+ const RowsetSharedPtr& first_rowset = rowsets.front();
+ if (last_rowset->version() != _tablet->max_version() || first_rowset->version().first != 0) {
+ return Status::Error<FULL_MISS_VERSION>(
+ "Full compaction rowsets' versions not equal to all exist rowsets' versions. "
+ "full compaction rowsets max version={}-{}"
+ ", current rowsets max version={}-{}"
+ "full compaction rowsets min version={}-{}, current rowsets min version=0-1",
+ last_rowset->start_version(), last_rowset->end_version(),
+ _tablet->max_version().first, _tablet->max_version().second,
+ first_rowset->start_version(), first_rowset->end_version());
+ }
+ return Status::OK();
+}
+
+Status FullCompaction::_full_compaction_update_delete_bitmap(const RowsetSharedPtr& rowset,
+ RowsetWriter* rowset_writer) {
+ std::vector<RowsetSharedPtr> tmp_rowsets {};
+
+ // tablet is under alter process. The delete bitmap will be calculated after conversion.
+ if (_tablet->tablet_state() == TABLET_NOTREADY &&
+ SchemaChangeHandler::tablet_in_converting(_tablet->tablet_id())) {
+ LOG(INFO) << "tablet is under alter process, update delete bitmap later, tablet_id="
+ << _tablet->tablet_id();
+ return Status::OK();
+ }
+
+ int64_t max_version = _tablet->max_version().second;
+ DCHECK(max_version >= rowset->version().second);
+ if (max_version > rowset->version().second) {
+ _tablet->capture_consistent_rowsets({rowset->version().second + 1, max_version},
+ &tmp_rowsets);
+ }
+
+ for (const auto& it : tmp_rowsets) {
+ const int64_t& cur_version = it->rowset_meta()->start_version();
+ RETURN_IF_ERROR(
+ _full_compaction_calc_delete_bitmap(it, rowset, cur_version, rowset_writer));
+ }
+
+ std::lock_guard rowset_update_lock(_tablet->get_rowset_update_lock());
+ std::lock_guard header_lock(_tablet->get_header_lock());
+ for (const auto& it : _tablet->rowset_map()) {
+ const int64_t& cur_version = it.first.first;
+ const RowsetSharedPtr& published_rowset = it.second;
+ if (cur_version > max_version) {
+ RETURN_IF_ERROR(_full_compaction_calc_delete_bitmap(published_rowset, rowset,
+ cur_version, rowset_writer));
+ }
+ }
+
+ return Status::OK();
+}
+
+Status FullCompaction::_full_compaction_calc_delete_bitmap(const RowsetSharedPtr& published_rowset,
+ const RowsetSharedPtr& rowset,
+ const int64_t& cur_version,
+ RowsetWriter* rowset_writer) {
+ std::vector<segment_v2::SegmentSharedPtr> segments;
+ auto beta_rowset = reinterpret_cast<BetaRowset*>(published_rowset.get());
+ RETURN_IF_ERROR(beta_rowset->load_segments(&segments));
+ DeleteBitmapPtr delete_bitmap =
+ std::make_shared<DeleteBitmap>(_tablet->tablet_meta()->tablet_id());
+ std::vector<RowsetSharedPtr> specified_rowsets(1, rowset);
+
+ OlapStopWatch watch;
+ RETURN_IF_ERROR(_tablet->calc_delete_bitmap(published_rowset, segments, specified_rowsets,
+ delete_bitmap, cur_version, rowset_writer));
+ size_t total_rows = std::accumulate(
+ segments.begin(), segments.end(), 0,
+ [](size_t sum, const segment_v2::SegmentSharedPtr& s) { return sum += s->num_rows(); });
+ VLOG_DEBUG << "[Full compaction] construct delete bitmap tablet: " << _tablet->tablet_id()
+ << ", published rowset version: [" << published_rowset->version().first << "-"
+ << published_rowset->version().second << "]"
+ << ", full compaction rowset version: [" << rowset->version().first << "-"
+ << rowset->version().second << "]"
+ << ", cost: " << watch.get_elapse_time_us() << "(us), total rows: " << total_rows;
+
+ for (const auto& [k, v] : delete_bitmap->delete_bitmap) {
+ _tablet->tablet_meta()->delete_bitmap().merge({std::get<0>(k), std::get<1>(k), cur_version},
+ v);
+ }
+
+ return Status::OK();
+}
+
+} // namespace doris
diff --git a/be/src/olap/full_compaction.h b/be/src/olap/full_compaction.h
new file mode 100644
index 0000000000..bce9ac745b
--- /dev/null
+++ b/be/src/olap/full_compaction.h
@@ -0,0 +1,62 @@
+// 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.
+
+#pragma once
+
+#include <butil/macros.h>
+
+#include <string>
+#include <vector>
+
+#include "common/status.h"
+#include "io/io_common.h"
+#include "olap/compaction.h"
+#include "olap/rowset/rowset.h"
+#include "olap/tablet.h"
+
+namespace doris {
+
+class FullCompaction : public Compaction {
+public:
+ FullCompaction(const TabletSharedPtr& tablet);
+ ~FullCompaction() override;
+
+ Status prepare_compact() override;
+ Status execute_compact_impl() override;
+ Status modify_rowsets(const Merger::Statistics* stats = nullptr) override;
+
+ std::vector<RowsetSharedPtr> get_input_rowsets() { return _input_rowsets; }
+
+protected:
+ Status pick_rowsets_to_compact() override;
+ std::string compaction_name() const override { return "full compaction"; }
+
+ ReaderType compaction_type() const override { return ReaderType::READER_FULL_COMPACTION; }
+
+private:
+ Status _check_all_version(const std::vector<RowsetSharedPtr>& rowsets);
+ Status _full_compaction_update_delete_bitmap(const RowsetSharedPtr& rowset,
+ RowsetWriter* rowset_writer);
+ Status _full_compaction_calc_delete_bitmap(const RowsetSharedPtr& published_rowset,
+ const RowsetSharedPtr& rowset,
+ const int64_t& cur_version,
+ RowsetWriter* rowset_writer);
+
+ DISALLOW_COPY_AND_ASSIGN(FullCompaction);
+};
+
+} // namespace doris
diff --git a/be/src/olap/olap_common.h b/be/src/olap/olap_common.h
index 86f59af14c..c9030d1317 100644
--- a/be/src/olap/olap_common.h
+++ b/be/src/olap/olap_common.h
@@ -48,7 +48,7 @@ using uint128_t = unsigned __int128;
using TabletUid = UniqueId;
-enum CompactionType { BASE_COMPACTION = 1, CUMULATIVE_COMPACTION = 2 };
+enum CompactionType { BASE_COMPACTION = 1, CUMULATIVE_COMPACTION = 2, FULL_COMPACTION = 3 };
struct DataDirInfo {
std::string path;
diff --git a/be/src/olap/reader.cpp b/be/src/olap/reader.cpp
index 730fdf5145..468c6289d5 100644
--- a/be/src/olap/reader.cpp
+++ b/be/src/olap/reader.cpp
@@ -324,6 +324,7 @@ Status TabletReader::_init_return_columns(const ReaderParams& read_params) {
} else if ((read_params.reader_type == ReaderType::READER_CUMULATIVE_COMPACTION ||
read_params.reader_type == ReaderType::READER_SEGMENT_COMPACTION ||
read_params.reader_type == ReaderType::READER_BASE_COMPACTION ||
+ read_params.reader_type == ReaderType::READER_FULL_COMPACTION ||
read_params.reader_type == ReaderType::READER_COLD_DATA_COMPACTION ||
read_params.reader_type == ReaderType::READER_ALTER_TABLE) &&
!read_params.return_columns.empty()) {
@@ -613,6 +614,7 @@ Status TabletReader::_init_delete_condition(const ReaderParams& read_params) {
// QUERY will filter the row in query layer to keep right result use where clause.
// CUMULATIVE_COMPACTION will lost the filter_delete info of base rowset
if (read_params.reader_type == ReaderType::READER_BASE_COMPACTION ||
+ read_params.reader_type == ReaderType::READER_FULL_COMPACTION ||
read_params.reader_type == ReaderType::READER_COLD_DATA_COMPACTION ||
read_params.reader_type == ReaderType::READER_CHECKSUM) {
_filter_delete = true;
diff --git a/be/src/olap/rowset/rowset_meta.h b/be/src/olap/rowset/rowset_meta.h
index 68bb1e7075..22c15d790d 100644
--- a/be/src/olap/rowset/rowset_meta.h
+++ b/be/src/olap/rowset/rowset_meta.h
@@ -293,7 +293,7 @@ public:
// `segments_overlap()` only return the value of "segments_overlap" field in rowset meta,
// but "segments_overlap" may be UNKNOWN.
//
- // Returns true iff all of the following conditions are met
+ // Returns true if all of the following conditions are met
// 1. the rowset contains more than one segment
// 2. the rowset's start version == end version (non-singleton rowset was generated by compaction process
// which always produces non-overlapped segments)
diff --git a/be/src/olap/single_replica_compaction.cpp b/be/src/olap/single_replica_compaction.cpp
index 8f927022f1..c7671aee36 100644
--- a/be/src/olap/single_replica_compaction.cpp
+++ b/be/src/olap/single_replica_compaction.cpp
@@ -164,6 +164,8 @@ Status SingleReplicaCompaction::_do_single_replica_compaction_impl() {
_tablet->set_last_cumu_compaction_success_time(UnixMillis());
} else if (compaction_type() == ReaderType::READER_BASE_COMPACTION) {
_tablet->set_last_base_compaction_success_time(UnixMillis());
+ } else if (compaction_type() == ReaderType::READER_FULL_COMPACTION) {
+ _tablet->set_last_full_compaction_success_time(UnixMillis());
}
int64_t current_max_version;
diff --git a/be/src/olap/tablet.cpp b/be/src/olap/tablet.cpp
index 417fdcee5c..9a16a7c636 100644
--- a/be/src/olap/tablet.cpp
+++ b/be/src/olap/tablet.cpp
@@ -244,8 +244,10 @@ Tablet::Tablet(TabletMetaSharedPtr tablet_meta, DataDir* data_dir,
_is_bad(false),
_last_cumu_compaction_failure_millis(0),
_last_base_compaction_failure_millis(0),
+ _last_full_compaction_failure_millis(0),
_last_cumu_compaction_success_millis(0),
_last_base_compaction_success_millis(0),
+ _last_full_compaction_success_millis(0),
_cumulative_point(K_INVALID_CUMULATIVE_POINT),
_newly_created_rowset_num(0),
_last_checkpoint_time(0),
@@ -1313,6 +1315,10 @@ std::vector<RowsetSharedPtr> Tablet::pick_candidate_rowsets_to_base_compaction()
return candidate_rowsets;
}
+std::vector<RowsetSharedPtr> Tablet::pick_candidate_rowsets_to_full_compaction() {
+ return pick_candidate_rowsets_to_single_replica_compaction();
+}
+
std::vector<RowsetSharedPtr> Tablet::pick_candidate_rowsets_to_build_inverted_index(
const std::set<int32_t>& alter_index_uids, bool is_drop_op) {
std::vector<RowsetSharedPtr> candidate_rowsets;
@@ -1393,6 +1399,10 @@ void Tablet::get_compaction_status(std::string* json_result) {
format_str = ToStringFromUnixMillis(_last_base_compaction_failure_millis.load());
base_value.SetString(format_str.c_str(), format_str.length(), root.GetAllocator());
root.AddMember("last base failure time", base_value, root.GetAllocator());
+ rapidjson::Value full_value;
+ format_str = ToStringFromUnixMillis(_last_full_compaction_failure_millis.load());
+ base_value.SetString(format_str.c_str(), format_str.length(), root.GetAllocator());
+ root.AddMember("last full failure time", full_value, root.GetAllocator());
rapidjson::Value cumu_success_value;
format_str = ToStringFromUnixMillis(_last_cumu_compaction_success_millis.load());
cumu_success_value.SetString(format_str.c_str(), format_str.length(), root.GetAllocator());
@@ -1401,6 +1411,10 @@ void Tablet::get_compaction_status(std::string* json_result) {
format_str = ToStringFromUnixMillis(_last_base_compaction_success_millis.load());
base_success_value.SetString(format_str.c_str(), format_str.length(), root.GetAllocator());
root.AddMember("last base success time", base_success_value, root.GetAllocator());
+ rapidjson::Value full_success_value;
+ format_str = ToStringFromUnixMillis(_last_full_compaction_success_millis.load());
+ full_success_value.SetString(format_str.c_str(), format_str.length(), root.GetAllocator());
+ root.AddMember("last full success time", full_success_value, root.GetAllocator());
// print all rowsets' version as an array
rapidjson::Document versions_arr;
@@ -1761,6 +1775,8 @@ void Tablet::execute_single_replica_compaction(CompactionType compaction_type) {
set_last_cumu_compaction_failure_time(UnixMillis());
} else if (compaction_type == CompactionType::BASE_COMPACTION) {
set_last_base_compaction_failure_time(UnixMillis());
+ } else if (compaction_type == CompactionType::FULL_COMPACTION) {
+ set_last_full_compaction_failure_time(UnixMillis());
}
LOG(WARNING) << "failed to do single replica compaction. res=" << res
<< ", tablet=" << full_name();
@@ -1770,6 +1786,8 @@ void Tablet::execute_single_replica_compaction(CompactionType compaction_type) {
set_last_cumu_compaction_failure_time(0);
} else if (compaction_type == CompactionType::BASE_COMPACTION) {
set_last_base_compaction_failure_time(0);
+ } else if (compaction_type == CompactionType::FULL_COMPACTION) {
+ set_last_full_compaction_failure_time(0);
}
}
diff --git a/be/src/olap/tablet.h b/be/src/olap/tablet.h
index 8b764326ed..9852c38a02 100644
--- a/be/src/olap/tablet.h
+++ b/be/src/olap/tablet.h
@@ -199,6 +199,7 @@ public:
std::mutex& get_push_lock() { return _ingest_lock; }
std::mutex& get_base_compaction_lock() { return _base_compaction_lock; }
std::mutex& get_cumulative_compaction_lock() { return _cumulative_compaction_lock; }
+ std::mutex& get_full_compaction_lock() { return _full_compaction_lock; }
std::shared_mutex& get_migration_lock() { return _migration_lock; }
@@ -234,6 +235,11 @@ public:
_last_base_compaction_failure_millis = millis;
}
+ int64_t last_full_compaction_failure_time() { return _last_full_compaction_failure_millis; }
+ void set_last_full_compaction_failure_time(int64_t millis) {
+ _last_full_compaction_failure_millis = millis;
+ }
+
int64_t last_cumu_compaction_success_time() { return _last_cumu_compaction_success_millis; }
void set_last_cumu_compaction_success_time(int64_t millis) {
_last_cumu_compaction_success_millis = millis;
@@ -244,6 +250,11 @@ public:
_last_base_compaction_success_millis = millis;
}
+ int64_t last_full_compaction_success_time() { return _last_full_compaction_success_millis; }
+ void set_last_full_compaction_success_time(int64_t millis) {
+ _last_full_compaction_success_millis = millis;
+ }
+
void delete_all_files();
void check_tablet_path_exists();
@@ -255,6 +266,7 @@ public:
std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_cumulative_compaction();
std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_base_compaction();
+ std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_full_compaction();
std::vector<RowsetSharedPtr> pick_candidate_rowsets_to_build_inverted_index(
const std::set<int32_t>& alter_index_uids, bool is_drop_op);
@@ -591,6 +603,7 @@ private:
std::mutex _ingest_lock;
std::mutex _base_compaction_lock;
std::mutex _cumulative_compaction_lock;
+ std::mutex _full_compaction_lock;
std::mutex _schema_change_lock;
std::shared_mutex _migration_lock;
std::mutex _build_inverted_index_lock;
@@ -618,10 +631,14 @@ private:
std::atomic<int64_t> _last_cumu_compaction_failure_millis;
// timestamp of last base compaction failure
std::atomic<int64_t> _last_base_compaction_failure_millis;
+ // timestamp of last full compaction failure
+ std::atomic<int64_t> _last_full_compaction_failure_millis;
// timestamp of last cumu compaction success
std::atomic<int64_t> _last_cumu_compaction_success_millis;
// timestamp of last base compaction success
std::atomic<int64_t> _last_base_compaction_success_millis;
+ // timestamp of last full compaction success
+ std::atomic<int64_t> _last_full_compaction_success_millis;
std::atomic<int64_t> _cumulative_point;
std::atomic<int64_t> _cumulative_promotion_size;
std::atomic<int32_t> _newly_created_rowset_num;
diff --git a/regression-test/data/compaction/test_full_compaction.out b/regression-test/data/compaction/test_full_compaction.out
new file mode 100644
index 0000000000..b5ed2dffc1
--- /dev/null
+++ b/regression-test/data/compaction/test_full_compaction.out
@@ -0,0 +1,41 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !1 --
+1 1
+2 2
+
+-- !2 --
+1 10
+2 20
+
+-- !3 --
+1 100
+2 200
+
+-- !4 --
+1 100
+2 200
+3 300
+
+-- !5 --
+1 100
+2 200
+3 100
+
+-- !6 --
+1 100
+2 200
+
+-- !skip_delete --
+1 1
+1 10
+1 100
+2 2
+2 20
+2 200
+3 300
+3 100
+
+-- !select_final --
+1 100
+2 200
+
diff --git a/regression-test/plugins/plugin_curl_requester.groovy b/regression-test/plugins/plugin_curl_requester.groovy
index 2e777cb8d7..a268339e63 100644
--- a/regression-test/plugins/plugin_curl_requester.groovy
+++ b/regression-test/plugins/plugin_curl_requester.groovy
@@ -61,3 +61,9 @@ Suite.metaClass.be_run_cumulative_compaction = { String ip, String port, String
}
logger.info("Added 'be_run_cumulative_compaction' function to Suite")
+
+Suite.metaClass.be_run_full_compaction = { String ip, String port, String tablet_id /* param */->
+ return curl("POST", String.format("http://%s:%s/api/compaction/run?tablet_id=%s&compact_type=full", ip, port, tablet_id))
+}
+
+logger.info("Added 'be_run_full_compaction' function to Suite")
diff --git a/regression-test/suites/compaction/test_full_compaction.groovy b/regression-test/suites/compaction/test_full_compaction.groovy
new file mode 100644
index 0000000000..41813fde3b
--- /dev/null
+++ b/regression-test/suites/compaction/test_full_compaction.groovy
@@ -0,0 +1,177 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+import org.codehaus.groovy.runtime.IOGroovyMethods
+
+suite("test_full_compaction") {
+ def tableName = "test_full_compaction"
+
+ try {
+ String backend_id;
+
+ def backendId_to_backendIP = [:]
+ def backendId_to_backendHttpPort = [:]
+ getBackendIpHttpPort(backendId_to_backendIP, backendId_to_backendHttpPort);
+
+ backend_id = backendId_to_backendIP.keySet()[0]
+ def (code, out, err) = show_be_config(backendId_to_backendIP.get(backend_id), backendId_to_backendHttpPort.get(backend_id))
+ logger.info("Show config: code=" + code + ", out=" + out + ", err=" + err)
+ assertEquals(code, 0)
+ def configList = parseJson(out.trim())
+ assert configList instanceof List
+
+ boolean disableAutoCompaction = true
+ for (Object ele in (List) configList) {
+ assert ele instanceof List<String>
+ if (((List<String>) ele)[0] == "disable_auto_compaction") {
+ disableAutoCompaction = Boolean.parseBoolean(((List<String>) ele)[2])
+ }
+ }
+
+ sql """ DROP TABLE IF EXISTS ${tableName} """
+ sql """
+ CREATE TABLE ${tableName} (
+ `user_id` INT NOT NULL, `value` INT NOT NULL)
+ UNIQUE KEY(`user_id`)
+ DISTRIBUTED BY HASH(`user_id`)
+ BUCKETS 1
+ PROPERTIES ("replication_allocation" = "tag.location.default: 1",
+ "disable_auto_compaction" = "true",
+ "enable_unique_key_merge_on_write" = "true");"""
+
+ // version1 (1,1)(2,2)
+ sql """ INSERT INTO ${tableName} VALUES
+ (1,1),(2,2)
+ """
+ qt_1 """select * from ${tableName} order by user_id"""
+
+
+ // version2 (1,10)(2,20)
+ sql """ INSERT INTO ${tableName} VALUES
+ (1,10),(2,20)
+ """
+ qt_2 """select * from ${tableName} order by user_id"""
+
+
+ // version3 (1,100)(2,200)
+ sql """ INSERT INTO ${tableName} VALUES
+ (1,100),(2,200)
+ """
+ qt_3 """select * from ${tableName} order by user_id"""
+
+
+ // version4 (1,100)(2,200)(3,300)
+ sql """ INSERT INTO ${tableName} VALUES
+ (3,300)
+ """
+ qt_4 """select * from ${tableName} order by user_id"""
+
+
+ // version5 (1,100)(2,200)(3,100)
+ sql """update ${tableName} set value = 100 where user_id = 3"""
+ qt_5 """select * from ${tableName} order by user_id"""
+
+
+ // version6 (1,100)(2,200)
+ sql """delete from ${tableName} where user_id = 3"""
+ qt_6 """select * from ${tableName} order by user_id"""
+
+ sql "SET skip_delete_predicate = true"
+ sql "SET skip_delete_sign = true"
+ sql "SET skip_delete_bitmap = true"
+ // show all hidden data
+ // (1,10)(1,100)(2,2)(2,20)(2,200)(3,300)(3,100)
+ qt_skip_delete """select * from ${tableName} order by user_id"""
+
+ //TabletId,ReplicaId,BackendId,SchemaHash,Version,LstSuccessVersion,LstFailedVersion,LstFailedTime,LocalDataSize,RemoteDataSize,RowCount,State,LstConsistencyCheckTime,CheckVersion,VersionCount,PathHash,MetaUrl,CompactionStatus
+ String[][] tablets = sql """ show tablets from ${tableName}; """
+
+ // before full compaction, there are 7 rowsets.
+ int rowsetCount = 0
+ for (String[] tablet in tablets) {
+ String tablet_id = tablet[0]
+ def compactionStatusUrlIndex = 18
+ (code, out, err) = curl("GET", tablet[compactionStatusUrlIndex])
+ logger.info("Show tablets status: code=" + code + ", out=" + out + ", err=" + err)
+ assertEquals(code, 0)
+ def tabletJson = parseJson(out.trim())
+ assert tabletJson.rowsets instanceof List
+ rowsetCount +=((List<String>) tabletJson.rowsets).size()
+ }
+ assert (rowsetCount == 7)
+
+ // trigger full compactions for all tablets in ${tableName}
+ for (String[] tablet in tablets) {
+ String tablet_id = tablet[0]
+ backend_id = tablet[2]
+ times = 1
+
+ do{
+ (code, out, err) = be_run_full_compaction(backendId_to_backendIP.get(backend_id), backendId_to_backendHttpPort.get(backend_id), tablet_id)
+ logger.info("Run compaction: code=" + code + ", out=" + out + ", err=" + err)
+ ++times
+ sleep(2000)
+ } while (parseJson(out.trim()).status.toLowerCase()!="success" && times<=10)
+
+ def compactJson = parseJson(out.trim())
+ if (compactJson.status.toLowerCase() == "fail") {
+ assertEquals(disableAutoCompaction, false)
+ logger.info("Compaction was done automatically!")
+ }
+ if (disableAutoCompaction) {
+ assertEquals("success", compactJson.status.toLowerCase())
+ }
+ }
+
+ // wait for full compaction done
+ for (String[] tablet in tablets) {
+ boolean running = true
+ do {
+ Thread.sleep(1000)
+ String tablet_id = tablet[0]
+ backend_id = tablet[2]
+ (code, out, err) = be_get_compaction_status(backendId_to_backendIP.get(backend_id), backendId_to_backendHttpPort.get(backend_id), tablet_id)
+ logger.info("Get compaction status: code=" + code + ", out=" + out + ", err=" + err)
+ assertEquals(code, 0)
+ def compactionStatus = parseJson(out.trim())
+ assertEquals("success", compactionStatus.status.toLowerCase())
+ running = compactionStatus.run_status
+ } while (running)
+ }
+
+ // after full compaction, there is only 1 rowset.
+
+ rowsetCount = 0
+ for (String[] tablet in tablets) {
+ String tablet_id = tablet[0]
+ def compactionStatusUrlIndex = 18
+ (code, out, err) = curl("GET", tablet[compactionStatusUrlIndex])
+ logger.info("Show tablets status: code=" + code + ", out=" + out + ", err=" + err)
+ assertEquals(code, 0)
+ def tabletJson = parseJson(out.trim())
+ assert tabletJson.rowsets instanceof List
+ rowsetCount +=((List<String>) tabletJson.rowsets).size()
+ }
+ assert (rowsetCount == 1)
+
+ // make sure all hidden data has been deleted
+ // (1,100)(2,200)
+ qt_select_final """select * from ${tableName} order by user_id"""
+ } finally {
+ try_sql("DROP TABLE IF EXISTS ${tableName}")
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 02/10: [Feature](table function) support explode_json_array_json (#21795)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit ec36d27fc0ce00dac962c41cbcd17d7541a1e5d7
Author: zy-kkk <zh...@gmail.com>
AuthorDate: Mon Jul 17 11:40:02 2023 +0800
[Feature](table function) support explode_json_array_json (#21795)
---
.../table_function/table_function_factory.cpp | 3 +
.../exprs/table_function/vexplode_json_array.cpp | 25 ++++
.../vec/exprs/table_function/vexplode_json_array.h | 7 +-
be/src/vec/functions/function_fake.cpp | 3 +
.../table-functions/explode-json-array.md | 16 +++
.../table-functions/explode-json-array.md | 150 +++++++++++++++++----
.../catalog/BuiltinTableGeneratingFunctions.java | 6 +-
.../java/org/apache/doris/catalog/FunctionSet.java | 6 +
.../functions/generator/ExplodeJsonArrayJson.java | 66 +++++++++
.../generator/ExplodeJsonArrayJsonOuter.java | 66 +++++++++
.../visitor/TableGeneratingFunctionVisitor.java | 11 ++
.../doris/planner/TableFunctionPlanTest.java | 6 +
.../table_function/explode_json_array.out | 14 ++
.../table_function/explode_json_array.groovy | 3 +
14 files changed, 355 insertions(+), 27 deletions(-)
diff --git a/be/src/vec/exprs/table_function/table_function_factory.cpp b/be/src/vec/exprs/table_function/table_function_factory.cpp
index 06b69a27c7..bafe47b69e 100644
--- a/be/src/vec/exprs/table_function/table_function_factory.cpp
+++ b/be/src/vec/exprs/table_function/table_function_factory.cpp
@@ -49,6 +49,8 @@ inline auto VExplodeJsonArrayDoubleCreator =
TableFunctionCreator<VExplodeJsonArrayTableFunction> {ExplodeJsonArrayType::DOUBLE};
inline auto VExplodeJsonArrayStringCreator =
TableFunctionCreator<VExplodeJsonArrayTableFunction> {ExplodeJsonArrayType::STRING};
+inline auto VExplodeJsonArrayJsonCreator =
+ TableFunctionCreator<VExplodeJsonArrayTableFunction> {ExplodeJsonArrayType::JSON};
const std::unordered_map<std::string, std::function<std::unique_ptr<TableFunction>()>>
TableFunctionFactory::_function_map {
@@ -57,6 +59,7 @@ const std::unordered_map<std::string, std::function<std::unique_ptr<TableFunctio
{"explode_json_array_int", VExplodeJsonArrayIntCreator},
{"explode_json_array_double", VExplodeJsonArrayDoubleCreator},
{"explode_json_array_string", VExplodeJsonArrayStringCreator},
+ {"explode_json_array_json", VExplodeJsonArrayJsonCreator},
{"explode_bitmap", TableFunctionCreator<VExplodeBitmapTableFunction>()},
{"explode", TableFunctionCreator<VExplodeTableFunction> {}}};
diff --git a/be/src/vec/exprs/table_function/vexplode_json_array.cpp b/be/src/vec/exprs/table_function/vexplode_json_array.cpp
index 16f6c39bf7..840923d018 100644
--- a/be/src/vec/exprs/table_function/vexplode_json_array.cpp
+++ b/be/src/vec/exprs/table_function/vexplode_json_array.cpp
@@ -24,6 +24,8 @@
#include <algorithm>
#include "common/status.h"
+#include "rapidjson/stringbuffer.h"
+#include "rapidjson/writer.h"
#include "vec/columns/column.h"
#include "vec/core/block.h"
#include "vec/core/column_with_type_and_name.h"
@@ -125,6 +127,29 @@ int ParsedData::set_output(ExplodeJsonArrayType type, rapidjson::Document& docum
}
break;
}
+ case ExplodeJsonArrayType::JSON: {
+ _data_string.clear();
+ _backup_string.clear();
+ _string_nulls.clear();
+ for (auto& v : document.GetArray()) {
+ if (v.IsObject()) {
+ rapidjson::StringBuffer buffer;
+ rapidjson::Writer<rapidjson::StringBuffer> writer(buffer);
+ v.Accept(writer);
+ _backup_string.emplace_back(buffer.GetString(), buffer.GetSize());
+ _string_nulls.push_back(false);
+ } else {
+ _data_string.push_back({});
+ _string_nulls.push_back(true);
+ }
+ }
+ // Must set _data_string at the end, so that we can
+ // save the real addr of string in `_backup_string` to `_data_string`.
+ for (auto& str : _backup_string) {
+ _data_string.emplace_back(str);
+ }
+ break;
+ }
default:
CHECK(false) << type;
break;
diff --git a/be/src/vec/exprs/table_function/vexplode_json_array.h b/be/src/vec/exprs/table_function/vexplode_json_array.h
index 2bdae9ed8e..db1784e5ad 100644
--- a/be/src/vec/exprs/table_function/vexplode_json_array.h
+++ b/be/src/vec/exprs/table_function/vexplode_json_array.h
@@ -40,7 +40,7 @@ class Block;
namespace doris::vectorized {
-enum ExplodeJsonArrayType { INT = 0, DOUBLE, STRING };
+enum ExplodeJsonArrayType { INT = 0, DOUBLE, STRING, JSON };
struct ParsedData {
static std::string true_value;
@@ -66,6 +66,7 @@ struct ParsedData {
_data.clear();
_backup_double.clear();
break;
+ case ExplodeJsonArrayType::JSON:
case ExplodeJsonArrayType::STRING:
_data_string.clear();
_backup_string.clear();
@@ -82,6 +83,7 @@ struct ParsedData {
case ExplodeJsonArrayType::INT:
case ExplodeJsonArrayType::DOUBLE:
return _data[offset];
+ case ExplodeJsonArrayType::JSON:
case ExplodeJsonArrayType::STRING:
return _string_nulls[offset] ? nullptr
: real ? reinterpret_cast<void*>(_backup_string[offset].data())
@@ -92,7 +94,8 @@ struct ParsedData {
}
int64 get_value_length(ExplodeJsonArrayType type, int64_t offset) {
- if (type == ExplodeJsonArrayType::STRING && !_string_nulls[offset]) {
+ if ((type == ExplodeJsonArrayType::STRING || type == ExplodeJsonArrayType::JSON) &&
+ !_string_nulls[offset]) {
return _backup_string[offset].size();
}
return 0;
diff --git a/be/src/vec/functions/function_fake.cpp b/be/src/vec/functions/function_fake.cpp
index 4fd626d17b..26896c028c 100644
--- a/be/src/vec/functions/function_fake.cpp
+++ b/be/src/vec/functions/function_fake.cpp
@@ -26,6 +26,7 @@
#include "vec/aggregate_functions/aggregate_function.h"
#include "vec/data_types/data_type_array.h"
+#include "vec/data_types/data_type_jsonb.h"
#include "vec/data_types/data_type_nullable.h"
#include "vec/data_types/data_type_number.h"
#include "vec/data_types/data_type_string.h"
@@ -102,6 +103,8 @@ void register_function_fake(SimpleFunctionFactory& factory) {
register_table_function_expand_outer_default<DataTypeInt64>(factory, "explode_json_array_int");
register_table_function_expand_outer_default<DataTypeString>(factory,
"explode_json_array_string");
+ register_table_function_expand_outer_default<DataTypeString>(factory,
+ "explode_json_array_json");
register_table_function_expand_outer_default<DataTypeFloat64>(factory,
"explode_json_array_double");
register_table_function_expand_outer_default<DataTypeInt64>(factory, "explode_bitmap");
diff --git a/docs/en/docs/sql-manual/sql-functions/table-functions/explode-json-array.md b/docs/en/docs/sql-manual/sql-functions/table-functions/explode-json-array.md
index ee4cfe1d8f..daff67381f 100644
--- a/docs/en/docs/sql-manual/sql-functions/table-functions/explode-json-array.md
+++ b/docs/en/docs/sql-manual/sql-functions/table-functions/explode-json-array.md
@@ -38,6 +38,7 @@ Expand a json array. According to the array element type, there are three functi
explode_json_array_int(json_str)
explode_json_array_double(json_str)
explode_json_array_string(json_str)
+explode_json_array_json(json_str)
```
### example
@@ -276,6 +277,21 @@ mysql> select k1, e1 from example1 lateral view explode_json_array_string('{"a":
| 2 | NULL |
| 3 | NULL |
+------+------+
+
+mysql> select k1, e1 from example1 lateral view explode_json_array_json('[{"id":1,"name":"John"},{"id":2,"name":"Mary"},{"id":3,"name":"Bob"}]') tmp1 as e1 order by k1, e1;
++------+------------------------+
+| k1 | e1 |
++------+------------------------+
+| 1 | {"id":1,"name":"John"} |
+| 1 | {"id":2,"name":"Mary"} |
+| 1 | {"id":3,"name":"Bob"} |
+| 2 | {"id":1,"name":"John"} |
+| 2 | {"id":2,"name":"Mary"} |
+| 2 | {"id":3,"name":"Bob"} |
+| 3 | {"id":1,"name":"John"} |
+| 3 | {"id":2,"name":"Mary"} |
+| 3 | {"id":3,"name":"Bob"} |
++------+------------------------+
```
### keywords
diff --git a/docs/zh-CN/docs/sql-manual/sql-functions/table-functions/explode-json-array.md b/docs/zh-CN/docs/sql-manual/sql-functions/table-functions/explode-json-array.md
index 15591a595c..c7f09139e1 100644
--- a/docs/zh-CN/docs/sql-manual/sql-functions/table-functions/explode-json-array.md
+++ b/docs/zh-CN/docs/sql-manual/sql-functions/table-functions/explode-json-array.md
@@ -37,28 +37,22 @@ under the License.
explode_json_array_int(json_str)
explode_json_array_double(json_str)
explode_json_array_string(json_str)
+explode_json_array_json(json_str)
```
### example
原表数据：
-```
-mysql> select k1 from example1 order by k1;
-+------+
-| k1 |
-+------+
-| 1 |
-| 2 |
-| 3 |
-+------+
-```
-
-Lateral View:
-
```
mysql> select k1, e1 from example1 lateral view explode_json_array_int('[]') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 2 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_int('[1,2,3]') tmp1 as e1 order by k1, e1;
+------+------+
@@ -79,28 +73,79 @@ mysql> select k1, e1 from example1 lateral view explode_json_array_int('[1,"b",3
+------+------+
| k1 | e1 |
+------+------+
+| 1 | NULL |
| 1 | 1 |
| 1 | 3 |
+| 2 | NULL |
| 2 | 1 |
| 2 | 3 |
+| 3 | NULL |
| 3 | 1 |
| 3 | 3 |
+------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_int('["a","b","c"]') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 1 | NULL |
+| 1 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 3 | NULL |
+| 3 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_int('{"a": 3}') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 2 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_double('[]') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 2 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_double('[1,2,3]') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 1 | NULL |
+| 1 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 3 | NULL |
+| 3 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_double('[1,"b",3]') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 1 | NULL |
+| 1 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 3 | NULL |
+| 3 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_double('[1.0,2.0,3.0]') tmp1 as e1 order by k1, e1;
+------+------+
@@ -118,16 +163,52 @@ mysql> select k1, e1 from example1 lateral view explode_json_array_double('[1.0,
+------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_double('[1,"b",3]') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 1 | NULL |
+| 1 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 3 | NULL |
+| 3 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_double('["a","b","c"]') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 1 | NULL |
+| 1 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 2 | NULL |
+| 3 | NULL |
+| 3 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_double('{"a": 3}') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 2 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_string('[]') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 2 | NULL |
+| 3 | NULL |
++------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_string('[1.0,2.0,3.0]') tmp1 as e1 order by k1, e1;
+------+----------+
@@ -175,7 +256,28 @@ mysql> select k1, e1 from example1 lateral view explode_json_array_string('["a",
+------+------+
mysql> select k1, e1 from example1 lateral view explode_json_array_string('{"a": 3}') tmp1 as e1 order by k1, e1;
-Empty set
++------+------+
+| k1 | e1 |
++------+------+
+| 1 | NULL |
+| 2 | NULL |
+| 3 | NULL |
++------+------+
+
+mysql> select k1, e1 from example1 lateral view explode_json_array_json('[{"id":1,"name":"John"},{"id":2,"name":"Mary"},{"id":3,"name":"Bob"}]') tmp1 as e1 order by k1, e1;
++------+------------------------+
+| k1 | e1 |
++------+------------------------+
+| 1 | {"id":1,"name":"John"} |
+| 1 | {"id":2,"name":"Mary"} |
+| 1 | {"id":3,"name":"Bob"} |
+| 2 | {"id":1,"name":"John"} |
+| 2 | {"id":2,"name":"Mary"} |
+| 2 | {"id":3,"name":"Bob"} |
+| 3 | {"id":1,"name":"John"} |
+| 3 | {"id":2,"name":"Mary"} |
+| 3 | {"id":3,"name":"Bob"} |
++------+------------------------+
```
### keywords
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableGeneratingFunctions.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableGeneratingFunctions.java
index fc82eafcbb..b1dd9c8306 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableGeneratingFunctions.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/BuiltinTableGeneratingFunctions.java
@@ -23,6 +23,8 @@ import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJso
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayDoubleOuter;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayInt;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayIntOuter;
+import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayJson;
+import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayJsonOuter;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayString;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayStringOuter;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeNumbers;
@@ -53,7 +55,9 @@ public class BuiltinTableGeneratingFunctions implements FunctionHelper {
tableGenerating(ExplodeJsonArrayDouble.class, "explode_json_array_double"),
tableGenerating(ExplodeJsonArrayDoubleOuter.class, "explode_json_array_double_outer"),
tableGenerating(ExplodeJsonArrayString.class, "explode_json_array_string"),
- tableGenerating(ExplodeJsonArrayStringOuter.class, "explode_json_array_string_outer")
+ tableGenerating(ExplodeJsonArrayStringOuter.class, "explode_json_array_string_outer"),
+ tableGenerating(ExplodeJsonArrayJson.class, "explode_json_array_json"),
+ tableGenerating(ExplodeJsonArrayJsonOuter.class, "explode_json_array_json_outer")
);
public static final BuiltinTableGeneratingFunctions INSTANCE = new BuiltinTableGeneratingFunctions();
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java
index f86bba5006..558d328de2 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/FunctionSet.java
@@ -1588,6 +1588,7 @@ public class FunctionSet<T> {
public static final String EXPLODE_JSON_ARRAY_INT = "explode_json_array_int";
public static final String EXPLODE_JSON_ARRAY_DOUBLE = "explode_json_array_double";
public static final String EXPLODE_JSON_ARRAY_STRING = "explode_json_array_string";
+ public static final String EXPLODE_JSON_ARRAY_JSON = "explode_json_array_json";
public static final String EXPLODE_NUMBERS = "explode_numbers";
public static final String EXPLODE = "explode";
@@ -1635,6 +1636,11 @@ public class FunctionSet<T> {
Function.NullableMode.DEPEND_ON_ARGUMENT, Lists.newArrayList(Type.VARCHAR), false,
"_ZN5doris19DummyTableFunctions25explode_json_array_stringEPN9doris_udf15FunctionContextERKNS1_9StringValE");
+ initTableFunctionListWithCombinator(EXPLODE_JSON_ARRAY_JSON);
+ addTableFunctionWithCombinator(EXPLODE_JSON_ARRAY_JSON, Type.VARCHAR,
+ Function.NullableMode.DEPEND_ON_ARGUMENT, Lists.newArrayList(Type.VARCHAR), false,
+ "_ZN5doris19DummyTableFunctions25explode_json_array_jsonEPN9doris_udf15FunctionContextERKNS1_9StringValE");
+
initTableFunctionListWithCombinator(EXPLODE_NUMBERS);
addTableFunctionWithCombinator(EXPLODE_NUMBERS, Type.INT, Function.NullableMode.DEPEND_ON_ARGUMENT,
Lists.newArrayList(Type.INT), false,
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/generator/ExplodeJsonArrayJson.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/generator/ExplodeJsonArrayJson.java
new file mode 100644
index 0000000000..a07e0e5d8f
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/generator/ExplodeJsonArrayJson.java
@@ -0,0 +1,66 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.expressions.functions.generator;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.PropagateNullable;
+import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.VarcharType;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+/**
+ * explode_json_array_json("[{"id":1,"name":"John"},{"id":2,"name":"Mary"},{"id":3,"name":"Bob"}]"),
+ * generate 3 lines include '{"id":1,"name":"John"}', '{"id":2,"name":"Mary"}' and '{"id":3,"name":"Bob"}'.
+ */
+public class ExplodeJsonArrayJson extends TableGeneratingFunction implements UnaryExpression, PropagateNullable {
+ public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
+ FunctionSignature.ret(VarcharType.SYSTEM_DEFAULT).args(VarcharType.SYSTEM_DEFAULT)
+ );
+
+ /**
+ * constructor with 1 argument.
+ */
+ public ExplodeJsonArrayJson(Expression arg) {
+ super("explode_json_array_json", arg);
+ }
+
+ /**
+ * withChildren.
+ */
+ @Override
+ public ExplodeJsonArrayJson withChildren(List<Expression> children) {
+ Preconditions.checkArgument(children.size() == 1);
+ return new ExplodeJsonArrayJson(children.get(0));
+ }
+
+ @Override
+ public List<FunctionSignature> getSignatures() {
+ return SIGNATURES;
+ }
+
+ @Override
+ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+ return visitor.visitExplodeJsonArrayJson(this, context);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/generator/ExplodeJsonArrayJsonOuter.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/generator/ExplodeJsonArrayJsonOuter.java
new file mode 100644
index 0000000000..bb4a34905a
--- /dev/null
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/functions/generator/ExplodeJsonArrayJsonOuter.java
@@ -0,0 +1,66 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.nereids.trees.expressions.functions.generator;
+
+import org.apache.doris.catalog.FunctionSignature;
+import org.apache.doris.nereids.trees.expressions.Expression;
+import org.apache.doris.nereids.trees.expressions.functions.PropagateNullable;
+import org.apache.doris.nereids.trees.expressions.shape.UnaryExpression;
+import org.apache.doris.nereids.trees.expressions.visitor.ExpressionVisitor;
+import org.apache.doris.nereids.types.VarcharType;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ImmutableList;
+
+import java.util.List;
+
+/**
+ * explode_json_array_json_outer("[{"id":1,"name":"John"},{"id":2,"name":"Mary"},{"id":3,"name":"Bob"}]"),
+ * generate 3 lines include '{"id":1,"name":"John"}', '{"id":2,"name":"Mary"}' and '{"id":3,"name":"Bob"}'.
+ */
+public class ExplodeJsonArrayJsonOuter extends TableGeneratingFunction implements UnaryExpression, PropagateNullable {
+ public static final List<FunctionSignature> SIGNATURES = ImmutableList.of(
+ FunctionSignature.ret(VarcharType.SYSTEM_DEFAULT).args(VarcharType.SYSTEM_DEFAULT)
+ );
+
+ /**
+ * constructor with 1 argument.
+ */
+ public ExplodeJsonArrayJsonOuter(Expression arg) {
+ super("explode_json_array_json_outer", arg);
+ }
+
+ /**
+ * withChildren.
+ */
+ @Override
+ public ExplodeJsonArrayJsonOuter withChildren(List<Expression> children) {
+ Preconditions.checkArgument(children.size() == 1);
+ return new ExplodeJsonArrayJsonOuter(children.get(0));
+ }
+
+ @Override
+ public List<FunctionSignature> getSignatures() {
+ return SIGNATURES;
+ }
+
+ @Override
+ public <R, C> R accept(ExpressionVisitor<R, C> visitor, C context) {
+ return visitor.visitExplodeJsonArrayJsonOuter(this, context);
+ }
+}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableGeneratingFunctionVisitor.java b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableGeneratingFunctionVisitor.java
index a869ab0d7e..040b279df8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableGeneratingFunctionVisitor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/nereids/trees/expressions/visitor/TableGeneratingFunctionVisitor.java
@@ -23,6 +23,8 @@ import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJso
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayDoubleOuter;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayInt;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayIntOuter;
+import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayJson;
+import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayJsonOuter;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayString;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeJsonArrayStringOuter;
import org.apache.doris.nereids.trees.expressions.functions.generator.ExplodeNumbers;
@@ -84,4 +86,13 @@ public interface TableGeneratingFunctionVisitor<R, C> {
default R visitExplodeJsonArrayStringOuter(ExplodeJsonArrayStringOuter explodeJsonArrayStringOuter, C context) {
return visitTableGeneratingFunction(explodeJsonArrayStringOuter, context);
}
+
+ default R visitExplodeJsonArrayJson(ExplodeJsonArrayJson explodeJsonArrayJson, C context) {
+ return visitTableGeneratingFunction(explodeJsonArrayJson, context);
+ }
+
+ default R visitExplodeJsonArrayJsonOuter(ExplodeJsonArrayJsonOuter explodeJsonArrayJsonOuter, C context) {
+ return visitTableGeneratingFunction(explodeJsonArrayJsonOuter, context);
+ }
+
}
diff --git a/fe/fe-core/src/test/java/org/apache/doris/planner/TableFunctionPlanTest.java b/fe/fe-core/src/test/java/org/apache/doris/planner/TableFunctionPlanTest.java
index 0b1308c9b1..f1e187ccc0 100644
--- a/fe/fe-core/src/test/java/org/apache/doris/planner/TableFunctionPlanTest.java
+++ b/fe/fe-core/src/test/java/org/apache/doris/planner/TableFunctionPlanTest.java
@@ -447,6 +447,12 @@ public class TableFunctionPlanTest {
System.out.println(explainString);
Assert.assertTrue(explainString.contains("table function: explode_json_array_double('[1.1, 2.2, 3.3]')"));
Assert.assertTrue(explainString.contains("output slot id: 0 1"));
+
+ sql = "desc select /*+ SET_VAR(enable_nereids_planner=false) */ k1, e1 from db1.tbl2 lateral view explode_json_array_json('[{\"id\":1,\"name\":\"John\"},{\"id\":2,\"name\":\"Mary\"},{\"id\":3,\"name\":\"Bob\"}]') tmp1 as e1 ";
+ explainString = UtFrameUtils.getSQLPlanOrErrorMsg(ctx, sql, true);
+ System.out.println(explainString);
+ Assert.assertTrue(explainString.contains("table function: explode_json_array_json('[{\"id\":1,\"name\":\"John\"},{\"id\":2,\"name\":\"Mary\"},{\"id\":3,\"name\":\"Bob\"}]')"));
+ Assert.assertTrue(explainString.contains("output slot id: 0 1"));
}
/*
diff --git a/regression-test/data/query_p0/sql_functions/table_function/explode_json_array.out b/regression-test/data/query_p0/sql_functions/table_function/explode_json_array.out
index a9f68ee30c..0b1994a22c 100644
--- a/regression-test/data/query_p0/sql_functions/table_function/explode_json_array.out
+++ b/regression-test/data/query_p0/sql_functions/table_function/explode_json_array.out
@@ -75,3 +75,17 @@
\N 80 3
\N 80 b
+-- !outer_join_explode_json_array11 --
+\N \N {"id":1,"name":"John"}
+\N \N {"id":2,"name":"Mary"}
+\N \N {"id":3,"name":"Bob"}
+\N 30 {"id":1,"name":"John"}
+\N 30 {"id":2,"name":"Mary"}
+\N 30 {"id":3,"name":"Bob"}
+\N 50 {"id":1,"name":"John"}
+\N 50 {"id":2,"name":"Mary"}
+\N 50 {"id":3,"name":"Bob"}
+\N 80 {"id":1,"name":"John"}
+\N 80 {"id":2,"name":"Mary"}
+\N 80 {"id":3,"name":"Bob"}
+
diff --git a/regression-test/suites/query_p0/sql_functions/table_function/explode_json_array.groovy b/regression-test/suites/query_p0/sql_functions/table_function/explode_json_array.groovy
index fc94aa5655..7ee801702b 100644
--- a/regression-test/suites/query_p0/sql_functions/table_function/explode_json_array.groovy
+++ b/regression-test/suites/query_p0/sql_functions/table_function/explode_json_array.groovy
@@ -57,4 +57,7 @@ suite("explode_json_array") {
qt_outer_join_explode_json_array11 """SELECT id, age, e1 FROM (SELECT id, age, e1 FROM (SELECT b.id, a.age FROM
${tableName} a LEFT JOIN ${tableName} b ON a.id=b.age)T LATERAL VIEW EXPLODE_JSON_ARRAY_STRING('[1, "b", 3]')
TMP AS e1) AS T ORDER BY age, e1"""
+ qt_outer_join_explode_json_array11 """SELECT id, age, e1 FROM (SELECT id, age, e1 FROM (SELECT b.id, a.age FROM
+ ${tableName} a LEFT JOIN ${tableName} b ON a.id=b.age)T LATERAL VIEW EXPLODE_JSON_ARRAY_JSON('[{"id":1,"name":"John"},{"id":2,"name":"Mary"},{"id":3,"name":"Bob"}]')
+ TMP AS e1) AS T ORDER BY age, e1"""
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 10/10: [fix](clone) fix cannot further repair clone replica which miss version data (#21382)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 9bb22d480badec9220883b0df919de9481247745
Author: yujun <yu...@gmail.com>
AuthorDate: Mon Jul 17 20:00:50 2023 +0800
[fix](clone) fix cannot further repair clone replica which miss version data (#21382)
---
fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java | 6 +++++-
1 file changed, 5 insertions(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
index 078179b823..74f1c31cbf 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/Tablet.java
@@ -505,10 +505,14 @@ public class Tablet extends MetaObject implements Writable {
}
alive++;
+ // this replica is alive but version incomplete
if (replica.getLastFailedVersion() > 0 || replica.getVersion() < visibleVersion) {
- // this replica is alive but version incomplete
+ if (replica.needFurtherRepair() && backend.isScheduleAvailable()) {
+ needFurtherRepairReplica = replica;
+ }
continue;
}
+
aliveAndVersionComplete++;
if (!backend.isScheduleAvailable()) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 04/10: [improvement](catalog) reduce the size thrift params for external table query (#21771)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 350095e569a15cfb8c12b253b3230d17f6125417
Author: Mingyu Chen <mo...@163.com>
AuthorDate: Mon Jul 17 13:37:02 2023 +0800
[improvement](catalog) reduce the size thrift params for external table query (#21771)
### 1
In previous implementation, for each FileSplit, there will be a `TFileScanRange`, and each `TFileScanRange`
contains a list of `TFileRangeDesc` and a `TFileScanRangeParams`.
So if there are thousands of FileSplit, there will be thousands of `TFileScanRange`, which cause the thrift
data send to BE too large, resulting in:
1. the rpc of sending fragment may fail due to timeout
2. FE will OOM
For a certain query request, the `TFileScanRangeParams` is the common part and is same of all `TFileScanRange`.
So I move this to the `TExecPlanFragmentParams`.
After that, for each FileSplit, there is only a list of `TFileRangeDesc`.
In my test, to query a hive table with 100000 partitions, the size of thrift data reduced from 151MB to 15MB,
and the above 2 issues are gone.
### 2
Support when setting `max_external_file_meta_cache_num` <=0, the file meta cache for parquet footer will
not be used.
Because I found that for some wide table, the footer is too large(1MB after compact, and much more after
deserialized to thrift), it will consuming too much memory of BE when there are many files.
This will be optimized later, here I just support to disable this cache.
---
be/src/olap/push_handler.cpp | 7 +-
be/src/runtime/fragment_mgr.cpp | 6 ++
be/src/runtime/query_context.h | 4 ++
be/src/vec/exec/format/csv/csv_reader.cpp | 21 +++++-
be/src/vec/exec/format/json/new_json_reader.cpp | 7 +-
be/src/vec/exec/format/orc/vorc_reader.cpp | 30 ++++++--
be/src/vec/exec/format/orc/vorc_reader.h | 8 ++-
be/src/vec/exec/format/parquet/vparquet_reader.cpp | 7 +-
be/src/vec/exec/scan/avro_jni_reader.cpp | 8 ++-
be/src/vec/exec/scan/new_file_scan_node.cpp | 11 ++-
be/src/vec/exec/scan/vfile_scanner.cpp | 67 ++++++++++--------
be/src/vec/exec/scan/vfile_scanner.h | 2 +-
.../doris/planner/external/FileQueryScanNode.java | 82 ++++++++++++----------
.../main/java/org/apache/doris/qe/Coordinator.java | 19 ++++-
.../ExternalFileTableValuedFunction.java | 5 +-
gensrc/thrift/PaloInternalService.thrift | 5 ++
gensrc/thrift/PlanNodes.thrift | 8 +++
.../hive/test_external_brown.groovy | 2 +-
18 files changed, 211 insertions(+), 88 deletions(-)
diff --git a/be/src/olap/push_handler.cpp b/be/src/olap/push_handler.cpp
index 380576fdc1..d852b03814 100644
--- a/be/src/olap/push_handler.cpp
+++ b/be/src/olap/push_handler.cpp
@@ -322,7 +322,6 @@ PushBrokerReader::PushBrokerReader(const Schema* schema, const TBrokerScanRange&
if (0 == _ranges.size()) {
return;
}
- _file_params.file_type = _ranges[0].file_type;
_file_params.format_type = _ranges[0].format_type;
_file_params.src_tuple_id = _params.src_tuple_id;
_file_params.dest_tuple_id = _params.dest_tuple_id;
@@ -336,6 +335,12 @@ PushBrokerReader::PushBrokerReader(const Schema* schema, const TBrokerScanRange&
for (int i = 0; i < _ranges.size(); ++i) {
TFileRangeDesc file_range;
+ // TODO(cmy): in previous implementation, the file_type is set in _file_params
+ // and it use _ranges[0].file_type.
+ // Later, this field is moved to TFileRangeDesc, but here we still only use _ranges[0]'s
+ // file_type.
+ // Because I don't know if other range has this field, so just keep it same as before.
+ file_range.file_type = _ranges[0].file_type;
file_range.load_id = _ranges[i].load_id;
file_range.path = _ranges[i].path;
file_range.start_offset = _ranges[i].start_offset;
diff --git a/be/src/runtime/fragment_mgr.cpp b/be/src/runtime/fragment_mgr.cpp
index b5fcb933cb..f8e7750151 100644
--- a/be/src/runtime/fragment_mgr.cpp
+++ b/be/src/runtime/fragment_mgr.cpp
@@ -676,6 +676,12 @@ Status FragmentMgr::_get_query_ctx(const Params& params, TUniqueId query_id, boo
query_ctx->query_id = query_id;
RETURN_IF_ERROR(DescriptorTbl::create(&(query_ctx->obj_pool), params.desc_tbl,
&(query_ctx->desc_tbl)));
+
+ // set file scan range params
+ if (params.__isset.file_scan_params) {
+ query_ctx->file_scan_range_params_map = params.file_scan_params;
+ }
+
query_ctx->coord_addr = params.coord;
LOG(INFO) << "query_id: " << UniqueId(query_ctx->query_id.hi, query_ctx->query_id.lo)
<< " coord_addr " << query_ctx->coord_addr
diff --git a/be/src/runtime/query_context.h b/be/src/runtime/query_context.h
index 8ed0150db5..c7651a687d 100644
--- a/be/src/runtime/query_context.h
+++ b/be/src/runtime/query_context.h
@@ -197,6 +197,10 @@ public:
std::vector<TUniqueId> fragment_ids;
+ // plan node id -> TFileScanRangeParams
+ // only for file scan node
+ std::map<int, TFileScanRangeParams> file_scan_range_params_map;
+
private:
ExecEnv* _exec_env;
vectorized::VecDateTimeValue _start_time;
diff --git a/be/src/vec/exec/format/csv/csv_reader.cpp b/be/src/vec/exec/format/csv/csv_reader.cpp
index 09531a2c2b..55a43ed4c5 100644
--- a/be/src/vec/exec/format/csv/csv_reader.cpp
+++ b/be/src/vec/exec/format/csv/csv_reader.cpp
@@ -88,7 +88,12 @@ CsvReader::CsvReader(RuntimeState* state, RuntimeProfile* profile, ScannerCounte
_io_ctx(io_ctx) {
_file_format_type = _params.format_type;
_is_proto_format = _file_format_type == TFileFormatType::FORMAT_PROTO;
- _file_compress_type = _params.compress_type;
+ if (_range.__isset.compress_type) {
+ // for compatibility
+ _file_compress_type = _range.compress_type;
+ } else {
+ _file_compress_type = _params.compress_type;
+ }
_size = _range.size;
_split_values.reserve(sizeof(Slice) * _file_slot_descs.size());
@@ -110,7 +115,12 @@ CsvReader::CsvReader(RuntimeProfile* profile, const TFileScanRangeParams& params
_decompressor(nullptr),
_io_ctx(io_ctx) {
_file_format_type = _params.format_type;
- _file_compress_type = _params.compress_type;
+ if (_range.__isset.compress_type) {
+ // for compatibility
+ _file_compress_type = _range.compress_type;
+ } else {
+ _file_compress_type = _params.compress_type;
+ }
_size = _range.size;
_init_system_properties();
_init_file_description();
@@ -119,7 +129,12 @@ CsvReader::CsvReader(RuntimeProfile* profile, const TFileScanRangeParams& params
CsvReader::~CsvReader() = default;
void CsvReader::_init_system_properties() {
- _system_properties.system_type = _params.file_type;
+ if (_range.__isset.file_type) {
+ // for compatibility
+ _system_properties.system_type = _range.file_type;
+ } else {
+ _system_properties.system_type = _params.file_type;
+ }
_system_properties.properties = _params.properties;
_system_properties.hdfs_params = _params.hdfs_params;
if (_params.__isset.broker_addresses) {
diff --git a/be/src/vec/exec/format/json/new_json_reader.cpp b/be/src/vec/exec/format/json/new_json_reader.cpp
index 7d17f650c6..94c376ae69 100644
--- a/be/src/vec/exec/format/json/new_json_reader.cpp
+++ b/be/src/vec/exec/format/json/new_json_reader.cpp
@@ -135,7 +135,12 @@ NewJsonReader::NewJsonReader(RuntimeProfile* profile, const TFileScanRangeParams
}
void NewJsonReader::_init_system_properties() {
- _system_properties.system_type = _params.file_type;
+ if (_range.__isset.file_type) {
+ // for compatibility
+ _system_properties.system_type = _range.file_type;
+ } else {
+ _system_properties.system_type = _params.file_type;
+ }
_system_properties.properties = _params.properties;
_system_properties.hdfs_params = _params.hdfs_params;
if (_params.__isset.broker_addresses) {
diff --git a/be/src/vec/exec/format/orc/vorc_reader.cpp b/be/src/vec/exec/format/orc/vorc_reader.cpp
index 7f87f77590..001d32b821 100644
--- a/be/src/vec/exec/format/orc/vorc_reader.cpp
+++ b/be/src/vec/exec/format/orc/vorc_reader.cpp
@@ -181,7 +181,9 @@ void OrcReader::_collect_profile_on_close() {
COUNTER_UPDATE(_orc_profile.read_bytes, _statistics.fs_read_bytes);
COUNTER_UPDATE(_orc_profile.column_read_time, _statistics.column_read_time);
COUNTER_UPDATE(_orc_profile.get_batch_time, _statistics.get_batch_time);
- COUNTER_UPDATE(_orc_profile.parse_meta_time, _statistics.parse_meta_time);
+ COUNTER_UPDATE(_orc_profile.create_reader_time, _statistics.create_reader_time);
+ COUNTER_UPDATE(_orc_profile.init_column_time, _statistics.init_column_time);
+ COUNTER_UPDATE(_orc_profile.set_fill_column_time, _statistics.set_fill_column_time);
COUNTER_UPDATE(_orc_profile.decode_value_time, _statistics.decode_value_time);
COUNTER_UPDATE(_orc_profile.decode_null_map_time, _statistics.decode_null_map_time);
}
@@ -200,7 +202,11 @@ void OrcReader::_init_profile() {
_orc_profile.read_bytes = ADD_COUNTER(_profile, "FileReadBytes", TUnit::BYTES);
_orc_profile.column_read_time = ADD_CHILD_TIMER(_profile, "ColumnReadTime", orc_profile);
_orc_profile.get_batch_time = ADD_CHILD_TIMER(_profile, "GetBatchTime", orc_profile);
- _orc_profile.parse_meta_time = ADD_CHILD_TIMER(_profile, "ParseMetaTime", orc_profile);
+ _orc_profile.create_reader_time =
+ ADD_CHILD_TIMER(_profile, "CreateReaderTime", orc_profile);
+ _orc_profile.init_column_time = ADD_CHILD_TIMER(_profile, "InitColumnTime", orc_profile);
+ _orc_profile.set_fill_column_time =
+ ADD_CHILD_TIMER(_profile, "SetFillColumnTime", orc_profile);
_orc_profile.decode_value_time = ADD_CHILD_TIMER(_profile, "DecodeValueTime", orc_profile);
_orc_profile.decode_null_map_time =
ADD_CHILD_TIMER(_profile, "DecodeNullMapTime", orc_profile);
@@ -254,9 +260,14 @@ Status OrcReader::init_reader(
not_single_slot_filter_conjuncts->end());
}
_obj_pool = std::make_shared<ObjectPool>();
- SCOPED_RAW_TIMER(&_statistics.parse_meta_time);
- RETURN_IF_ERROR(_create_file_reader());
- RETURN_IF_ERROR(_init_read_columns());
+ {
+ SCOPED_RAW_TIMER(&_statistics.create_reader_time);
+ RETURN_IF_ERROR(_create_file_reader());
+ }
+ {
+ SCOPED_RAW_TIMER(&_statistics.init_column_time);
+ RETURN_IF_ERROR(_init_read_columns());
+ }
return Status::OK();
}
@@ -646,7 +657,7 @@ Status OrcReader::set_fill_columns(
const std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>&
partition_columns,
const std::unordered_map<std::string, VExprContextSPtr>& missing_columns) {
- SCOPED_RAW_TIMER(&_statistics.parse_meta_time);
+ SCOPED_RAW_TIMER(&_statistics.set_fill_column_time);
// std::unordered_map<column_name, std::pair<col_id, slot_id>>
std::unordered_map<std::string, std::pair<uint32_t, int>> predicate_columns;
@@ -864,7 +875,12 @@ void OrcReader::_init_bloom_filter(
}
void OrcReader::_init_system_properties() {
- _system_properties.system_type = _scan_params.file_type;
+ if (_scan_range.__isset.file_type) {
+ // for compatibility
+ _system_properties.system_type = _scan_range.file_type;
+ } else {
+ _system_properties.system_type = _scan_params.file_type;
+ }
_system_properties.properties = _scan_params.properties;
_system_properties.hdfs_params = _scan_params.hdfs_params;
if (_scan_params.__isset.broker_addresses) {
diff --git a/be/src/vec/exec/format/orc/vorc_reader.h b/be/src/vec/exec/format/orc/vorc_reader.h
index 85a73e0e21..d05fb4c478 100644
--- a/be/src/vec/exec/format/orc/vorc_reader.h
+++ b/be/src/vec/exec/format/orc/vorc_reader.h
@@ -126,7 +126,9 @@ public:
int64_t fs_read_bytes = 0;
int64_t column_read_time = 0;
int64_t get_batch_time = 0;
- int64_t parse_meta_time = 0;
+ int64_t create_reader_time = 0;
+ int64_t init_column_time = 0;
+ int64_t set_fill_column_time = 0;
int64_t decode_value_time = 0;
int64_t decode_null_map_time = 0;
};
@@ -200,7 +202,9 @@ private:
RuntimeProfile::Counter* read_bytes;
RuntimeProfile::Counter* column_read_time;
RuntimeProfile::Counter* get_batch_time;
- RuntimeProfile::Counter* parse_meta_time;
+ RuntimeProfile::Counter* create_reader_time;
+ RuntimeProfile::Counter* init_column_time;
+ RuntimeProfile::Counter* set_fill_column_time;
RuntimeProfile::Counter* decode_value_time;
RuntimeProfile::Counter* decode_null_map_time;
};
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
index fed33b6d28..ed4aa6e6ae 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
@@ -269,7 +269,12 @@ Status ParquetReader::open() {
}
void ParquetReader::_init_system_properties() {
- _system_properties.system_type = _scan_params.file_type;
+ if (_scan_range.__isset.file_type) {
+ // for compatibility
+ _system_properties.system_type = _scan_range.file_type;
+ } else {
+ _system_properties.system_type = _scan_params.file_type;
+ }
_system_properties.properties = _scan_params.properties;
_system_properties.hdfs_params = _scan_params.hdfs_params;
if (_scan_params.__isset.broker_addresses) {
diff --git a/be/src/vec/exec/scan/avro_jni_reader.cpp b/be/src/vec/exec/scan/avro_jni_reader.cpp
index 5d1ef40cbc..ffca8682c4 100644
--- a/be/src/vec/exec/scan/avro_jni_reader.cpp
+++ b/be/src/vec/exec/scan/avro_jni_reader.cpp
@@ -74,7 +74,13 @@ Status AvroJNIReader::init_fetch_table_reader(
index++;
}
- TFileType::type type = _params.file_type;
+ TFileType::type type;
+ if (_range.__isset.file_type) {
+ // for compatibility
+ type = _range.file_type;
+ } else {
+ type = _params.file_type;
+ }
std::map<String, String> required_param = {
{"required_fields", required_fields.str()},
{"columns_types", columns_types.str()},
diff --git a/be/src/vec/exec/scan/new_file_scan_node.cpp b/be/src/vec/exec/scan/new_file_scan_node.cpp
index 127539d26e..0b8d318551 100644
--- a/be/src/vec/exec/scan/new_file_scan_node.cpp
+++ b/be/src/vec/exec/scan/new_file_scan_node.cpp
@@ -49,6 +49,12 @@ Status NewFileScanNode::init(const TPlanNode& tnode, RuntimeState* state) {
Status NewFileScanNode::prepare(RuntimeState* state) {
RETURN_IF_ERROR(VScanNode::prepare(state));
+ if (state->get_query_ctx() != nullptr &&
+ state->get_query_ctx()->file_scan_range_params_map.count(id()) > 0) {
+ TFileScanRangeParams& params = state->get_query_ctx()->file_scan_range_params_map[id()];
+ _input_tuple_id = params.src_tuple_id;
+ _output_tuple_id = params.dest_tuple_id;
+ }
return Status::OK();
}
@@ -74,7 +80,10 @@ void NewFileScanNode::set_scan_ranges(const std::vector<TScanRangeParams>& scan_
_scan_ranges.shrink_to_fit();
LOG(INFO) << "Merge " << scan_ranges.size() << " scan ranges to " << _scan_ranges.size();
}
- if (scan_ranges.size() > 0) {
+ if (scan_ranges.size() > 0 &&
+ scan_ranges[0].scan_range.ext_scan_range.file_scan_range.__isset.params) {
+ // for compatibility.
+ // in new implement, the tuple id is set in prepare phase
_input_tuple_id =
scan_ranges[0].scan_range.ext_scan_range.file_scan_range.params.src_tuple_id;
_output_tuple_id =
diff --git a/be/src/vec/exec/scan/vfile_scanner.cpp b/be/src/vec/exec/scan/vfile_scanner.cpp
index b95d2aa773..38574262de 100644
--- a/be/src/vec/exec/scan/vfile_scanner.cpp
+++ b/be/src/vec/exec/scan/vfile_scanner.cpp
@@ -89,7 +89,6 @@ VFileScanner::VFileScanner(RuntimeState* state, NewFileScanNode* parent, int64_t
const TFileScanRange& scan_range, RuntimeProfile* profile,
ShardedKVCache* kv_cache)
: VScanner(state, static_cast<VScanNode*>(parent), limit, profile),
- _params(scan_range.params),
_ranges(scan_range.ranges),
_next_range(0),
_cur_reader(nullptr),
@@ -99,6 +98,14 @@ VFileScanner::VFileScanner(RuntimeState* state, NewFileScanNode* parent, int64_t
if (scan_range.params.__isset.strict_mode) {
_strict_mode = scan_range.params.strict_mode;
}
+
+ if (state->get_query_ctx() != nullptr &&
+ state->get_query_ctx()->file_scan_range_params_map.count(parent->id()) > 0) {
+ _params = &(state->get_query_ctx()->file_scan_range_params_map[parent->id()]);
+ } else {
+ CHECK(scan_range.__isset.params);
+ _params = &(scan_range.params);
+ }
}
Status VFileScanner::prepare(
@@ -133,13 +140,13 @@ Status VFileScanner::prepare(
std::vector<TupleId>({_input_tuple_desc->id()}),
std::vector<bool>({false})));
// prepare pre filters
- if (_params.__isset.pre_filter_exprs_list) {
+ if (_params->__isset.pre_filter_exprs_list) {
RETURN_IF_ERROR(doris::vectorized::VExpr::create_expr_trees(
- _params.pre_filter_exprs_list, _pre_conjunct_ctxs));
- } else if (_params.__isset.pre_filter_exprs) {
+ _params->pre_filter_exprs_list, _pre_conjunct_ctxs));
+ } else if (_params->__isset.pre_filter_exprs) {
VExprContextSPtr context;
RETURN_IF_ERROR(
- doris::vectorized::VExpr::create_expr_tree(_params.pre_filter_exprs, context));
+ doris::vectorized::VExpr::create_expr_tree(_params->pre_filter_exprs, context));
_pre_conjunct_ctxs.emplace_back(context);
}
@@ -569,9 +576,10 @@ Status VFileScanner::_get_next_reader() {
// create reader for specific format
Status init_status;
- TFileFormatType::type format_type = _params.format_type;
+ TFileFormatType::type format_type = _params->format_type;
// JNI reader can only push down column value range
- bool push_down_predicates = !_is_load && _params.format_type != TFileFormatType::FORMAT_JNI;
+ bool push_down_predicates =
+ !_is_load && _params->format_type != TFileFormatType::FORMAT_JNI;
if (format_type == TFileFormatType::FORMAT_JNI && range.__isset.table_format_params &&
range.table_format_params.table_format_type == "hudi") {
if (range.table_format_params.hudi_params.delta_logs.empty()) {
@@ -598,9 +606,9 @@ Status VFileScanner::_get_next_reader() {
->init_reader(_colname_to_value_range);
} else if (range.__isset.table_format_params &&
range.table_format_params.table_format_type == "hudi") {
- _cur_reader =
- HudiJniReader::create_unique(_params, range.table_format_params.hudi_params,
- _file_slot_descs, _state, _profile);
+ _cur_reader = HudiJniReader::create_unique(*_params,
+ range.table_format_params.hudi_params,
+ _file_slot_descs, _state, _profile);
init_status =
((HudiJniReader*)_cur_reader.get())->init_reader(_colname_to_value_range);
}
@@ -608,9 +616,11 @@ Status VFileScanner::_get_next_reader() {
}
case TFileFormatType::FORMAT_PARQUET: {
std::unique_ptr<ParquetReader> parquet_reader = ParquetReader::create_unique(
- _profile, _params, range, _state->query_options().batch_size,
+ _profile, *_params, range, _state->query_options().batch_size,
const_cast<cctz::time_zone*>(&_state->timezone_obj()), _io_ctx.get(), _state,
- ExecEnv::GetInstance()->file_meta_cache(),
+ config::max_external_file_meta_cache_num <= 0
+ ? nullptr
+ : ExecEnv::GetInstance()->file_meta_cache(),
_state->query_options().enable_parquet_lazy_mat);
{
SCOPED_TIMER(_open_reader_timer);
@@ -627,7 +637,7 @@ Status VFileScanner::_get_next_reader() {
range.table_format_params.table_format_type == "iceberg") {
std::unique_ptr<IcebergTableReader> iceberg_reader =
IcebergTableReader::create_unique(std::move(parquet_reader), _profile,
- _state, _params, range, _kv_cache,
+ _state, *_params, range, _kv_cache,
_io_ctx.get());
init_status = iceberg_reader->init_reader(
_file_col_names, _col_id_name_map, _colname_to_value_range,
@@ -649,7 +659,7 @@ Status VFileScanner::_get_next_reader() {
}
case TFileFormatType::FORMAT_ORC: {
std::unique_ptr<OrcReader> orc_reader = OrcReader::create_unique(
- _profile, _state, _params, range, _state->query_options().batch_size,
+ _profile, _state, *_params, range, _state->query_options().batch_size,
_state->timezone(), _io_ctx.get(), _state->query_options().enable_orc_lazy_mat);
if (push_down_predicates && _push_down_conjuncts.empty() && !_conjuncts.empty()) {
_push_down_conjuncts.resize(_conjuncts.size());
@@ -662,7 +672,7 @@ Status VFileScanner::_get_next_reader() {
range.table_format_params.table_format_type == "transactional_hive") {
std::unique_ptr<TransactionalHiveReader> tran_orc_reader =
TransactionalHiveReader::create_unique(std::move(orc_reader), _profile,
- _state, _params, range,
+ _state, *_params, range,
_io_ctx.get());
init_status = tran_orc_reader->init_reader(
_file_col_names, _colname_to_value_range, _push_down_conjuncts,
@@ -686,13 +696,13 @@ Status VFileScanner::_get_next_reader() {
case TFileFormatType::FORMAT_CSV_LZOP:
case TFileFormatType::FORMAT_CSV_DEFLATE:
case TFileFormatType::FORMAT_PROTO: {
- _cur_reader = CsvReader::create_unique(_state, _profile, &_counter, _params, range,
+ _cur_reader = CsvReader::create_unique(_state, _profile, &_counter, *_params, range,
_file_slot_descs, _io_ctx.get());
init_status = ((CsvReader*)(_cur_reader.get()))->init_reader(_is_load);
break;
}
case TFileFormatType::FORMAT_JSON: {
- _cur_reader = NewJsonReader::create_unique(_state, _profile, &_counter, _params, range,
+ _cur_reader = NewJsonReader::create_unique(_state, _profile, &_counter, *_params, range,
_file_slot_descs, &_scanner_eof,
_io_ctx.get(), _is_dynamic_schema);
init_status =
@@ -700,13 +710,14 @@ Status VFileScanner::_get_next_reader() {
break;
}
case TFileFormatType::FORMAT_AVRO: {
- _cur_reader = AvroJNIReader::create_unique(_state, _profile, _params, _file_slot_descs);
+ _cur_reader =
+ AvroJNIReader::create_unique(_state, _profile, *_params, _file_slot_descs);
init_status = ((AvroJNIReader*)(_cur_reader.get()))
->init_fetch_table_reader(_colname_to_value_range);
break;
}
default:
- return Status::InternalError("Not supported file format: {}", _params.format_type);
+ return Status::InternalError("Not supported file format: {}", _params->format_type);
}
if (init_status.is<END_OF_FILE>()) {
@@ -810,8 +821,8 @@ Status VFileScanner::_init_expr_ctxes() {
}
}
- _num_of_columns_from_file = _params.num_of_columns_from_file;
- for (const auto& slot_info : _params.required_slots) {
+ _num_of_columns_from_file = _params->num_of_columns_from_file;
+ for (const auto& slot_info : _params->required_slots) {
auto slot_id = slot_info.slot_id;
auto it = full_src_slot_map.find(slot_id);
if (it == std::end(full_src_slot_map)) {
@@ -843,8 +854,8 @@ Status VFileScanner::_init_expr_ctxes() {
continue;
}
vectorized::VExprContextSPtr ctx;
- auto it = _params.default_value_of_src_slot.find(slot_desc->id());
- if (it != std::end(_params.default_value_of_src_slot)) {
+ auto it = _params->default_value_of_src_slot.find(slot_desc->id());
+ if (it != std::end(_params->default_value_of_src_slot)) {
if (!it->second.nodes.empty()) {
RETURN_IF_ERROR(vectorized::VExpr::create_expr_tree(it->second, ctx));
RETURN_IF_ERROR(ctx->prepare(_state, *_default_val_row_desc));
@@ -857,14 +868,14 @@ Status VFileScanner::_init_expr_ctxes() {
if (_is_load) {
// follow desc expr map is only for load task.
- bool has_slot_id_map = _params.__isset.dest_sid_to_src_sid_without_trans;
+ bool has_slot_id_map = _params->__isset.dest_sid_to_src_sid_without_trans;
int idx = 0;
for (auto slot_desc : _output_tuple_desc->slots()) {
if (!slot_desc->is_materialized()) {
continue;
}
- auto it = _params.expr_of_dest_slot.find(slot_desc->id());
- if (it == std::end(_params.expr_of_dest_slot)) {
+ auto it = _params->expr_of_dest_slot.find(slot_desc->id());
+ if (it == std::end(_params->expr_of_dest_slot)) {
return Status::InternalError("No expr for dest slot, id={}, name={}",
slot_desc->id(), slot_desc->col_name());
}
@@ -879,8 +890,8 @@ Status VFileScanner::_init_expr_ctxes() {
_dest_slot_name_to_idx[slot_desc->col_name()] = idx++;
if (has_slot_id_map) {
- auto it1 = _params.dest_sid_to_src_sid_without_trans.find(slot_desc->id());
- if (it1 == std::end(_params.dest_sid_to_src_sid_without_trans)) {
+ auto it1 = _params->dest_sid_to_src_sid_without_trans.find(slot_desc->id());
+ if (it1 == std::end(_params->dest_sid_to_src_sid_without_trans)) {
_src_slot_descs_order_by_dest.emplace_back(nullptr);
} else {
auto _src_slot_it = full_src_slot_map.find(it1->second);
diff --git a/be/src/vec/exec/scan/vfile_scanner.h b/be/src/vec/exec/scan/vfile_scanner.h
index 0b1d7a558a..9a16d056ed 100644
--- a/be/src/vec/exec/scan/vfile_scanner.h
+++ b/be/src/vec/exec/scan/vfile_scanner.h
@@ -89,7 +89,7 @@ protected:
protected:
std::unique_ptr<TextConverter> _text_converter;
- const TFileScanRangeParams& _params;
+ const TFileScanRangeParams* _params;
const std::vector<TFileRangeDesc>& _ranges;
int _next_range;
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
index 09ed6361e5..5b6238bf14 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/FileQueryScanNode.java
@@ -180,11 +180,6 @@ public abstract class FileQueryScanNode extends FileScanNode {
}
// Update required slots and column_idxs in scanRangeLocations.
setColumnPositionMapping();
- for (TScanRangeLocations location : scanRangeLocations) {
- TFileScanRangeParams rangeParams = location.getScanRange().getExtScanRange().getFileScanRange().getParams();
- rangeParams.setRequiredSlots(params.getRequiredSlots());
- rangeParams.setColumnIdxs(params.getColumnIdxs());
- }
}
@Override
@@ -227,6 +222,10 @@ public abstract class FileQueryScanNode extends FileScanNode {
params.setColumnIdxs(columnIdxs);
}
+ public TFileScanRangeParams getFileScanRangeParams() {
+ return params;
+ }
+
@Override
public void createScanRangeLocations() throws UserException {
long start = System.currentTimeMillis();
@@ -237,41 +236,24 @@ public abstract class FileQueryScanNode extends FileScanNode {
}
TFileFormatType fileFormatType = getFileFormatType();
params.setFormatType(fileFormatType);
+ boolean isCsvOrJson = Util.isCsvFormat(fileFormatType) || fileFormatType == TFileFormatType.FORMAT_JSON;
+ if (isCsvOrJson) {
+ params.setFileAttributes(getFileAttributes());
+ }
+
+ Map<String, String> locationProperties = getLocationProperties();
+ // for JNI, only need to set properties
+ if (fileFormatType == TFileFormatType.FORMAT_JNI) {
+ params.setProperties(locationProperties);
+ }
+
List<String> pathPartitionKeys = getPathPartitionKeys();
for (Split split : inputSplits) {
- TFileScanRangeParams scanRangeParams = new TFileScanRangeParams(params);
FileSplit fileSplit = (FileSplit) split;
TFileType locationType = getLocationType(fileSplit.getPath().toString());
- scanRangeParams.setFileType(locationType);
- TFileCompressType fileCompressType = getFileCompressType(fileSplit);
- scanRangeParams.setCompressType(fileCompressType);
- boolean isCsvOrJson = Util.isCsvFormat(fileFormatType) || fileFormatType == TFileFormatType.FORMAT_JSON;
- if (isCsvOrJson) {
- scanRangeParams.setFileAttributes(getFileAttributes());
- }
-
- // set hdfs params for hdfs file type.
- Map<String, String> locationProperties = getLocationProperties();
- if (fileFormatType == TFileFormatType.FORMAT_JNI || locationType == TFileType.FILE_S3) {
- scanRangeParams.setProperties(locationProperties);
- }
- if (locationType == TFileType.FILE_HDFS || locationType == TFileType.FILE_BROKER) {
- String fsName = getFsName(fileSplit);
- THdfsParams tHdfsParams = HdfsResource.generateHdfsParam(locationProperties);
- tHdfsParams.setFsName(fsName);
- scanRangeParams.setHdfsParams(tHdfsParams);
-
- if (locationType == TFileType.FILE_BROKER) {
- FsBroker broker = Env.getCurrentEnv().getBrokerMgr().getAnyAliveBroker();
- if (broker == null) {
- throw new UserException("No alive broker.");
- }
- scanRangeParams.addToBrokerAddresses(new TNetworkAddress(broker.host, broker.port));
- }
- }
-
- TScanRangeLocations curLocations = newLocations(scanRangeParams);
+ setLocationPropertiesIfNecessary(locationType, fileSplit, locationProperties);
+ TScanRangeLocations curLocations = newLocations();
// If fileSplit has partition values, use the values collected from hive partitions.
// Otherwise, use the values in file path.
boolean isACID = false;
@@ -285,6 +267,8 @@ public abstract class FileQueryScanNode extends FileScanNode {
TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit, partitionValuesFromPath, pathPartitionKeys,
locationType);
+ TFileCompressType fileCompressType = getFileCompressType(fileSplit);
+ rangeDesc.setCompressType(fileCompressType);
if (isACID) {
HiveSplit hiveSplit = (HiveSplit) split;
hiveSplit.setTableFormatType(TableFormatType.TRANSACTIONAL_HIVE);
@@ -332,11 +316,31 @@ public abstract class FileQueryScanNode extends FileScanNode {
scanRangeLocations.size(), (System.currentTimeMillis() - start));
}
- private TScanRangeLocations newLocations(TFileScanRangeParams params) {
+ private void setLocationPropertiesIfNecessary(TFileType locationType, FileSplit fileSplit,
+ Map<String, String> locationProperties) throws UserException {
+ if (locationType == TFileType.FILE_HDFS || locationType == TFileType.FILE_BROKER) {
+ if (!params.isSetHdfsParams()) {
+ String fsName = getFsName(fileSplit);
+ THdfsParams tHdfsParams = HdfsResource.generateHdfsParam(locationProperties);
+ tHdfsParams.setFsName(fsName);
+ params.setHdfsParams(tHdfsParams);
+ }
+
+ if (locationType == TFileType.FILE_BROKER && !params.isSetBrokerAddresses()) {
+ FsBroker broker = Env.getCurrentEnv().getBrokerMgr().getAnyAliveBroker();
+ if (broker == null) {
+ throw new UserException("No alive broker.");
+ }
+ params.addToBrokerAddresses(new TNetworkAddress(broker.host, broker.port));
+ }
+ } else if (locationType == TFileType.FILE_S3 && !params.isSetProperties()) {
+ params.setProperties(locationProperties);
+ }
+ }
+
+ private TScanRangeLocations newLocations() {
// Generate on file scan range
TFileScanRange fileScanRange = new TFileScanRange();
- fileScanRange.setParams(params);
-
// Scan range
TExternalScanRange externalScanRange = new TExternalScanRange();
externalScanRange.setFileScanRange(fileScanRange);
@@ -361,6 +365,7 @@ public abstract class FileQueryScanNode extends FileScanNode {
rangeDesc.setColumnsFromPath(columnsFromPath);
rangeDesc.setColumnsFromPathKeys(columnsFromPathKeys);
+ rangeDesc.setFileType(locationType);
if (locationType == TFileType.FILE_HDFS) {
rangeDesc.setPath(fileSplit.getPath().toUri().getPath());
} else if (locationType == TFileType.FILE_S3
@@ -424,3 +429,4 @@ public abstract class FileQueryScanNode extends FileScanNode {
}
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
index 180e3e1c75..a81bd1984b 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/Coordinator.java
@@ -61,6 +61,7 @@ import org.apache.doris.planner.ScanNode;
import org.apache.doris.planner.SetOperationNode;
import org.apache.doris.planner.UnionNode;
import org.apache.doris.planner.external.ExternalScanNode;
+import org.apache.doris.planner.external.FileQueryScanNode;
import org.apache.doris.planner.external.FileScanNode;
import org.apache.doris.proto.InternalService;
import org.apache.doris.proto.InternalService.PExecPlanFragmentResult;
@@ -83,6 +84,7 @@ import org.apache.doris.thrift.TExecPlanFragmentParams;
import org.apache.doris.thrift.TExecPlanFragmentParamsList;
import org.apache.doris.thrift.TExternalScanRange;
import org.apache.doris.thrift.TFileScanRange;
+import org.apache.doris.thrift.TFileScanRangeParams;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TPaloScanRange;
import org.apache.doris.thrift.TPipelineFragmentParams;
@@ -165,6 +167,9 @@ public class Coordinator {
// copied from TQueryExecRequest; constant across all fragments
private final TDescriptorTable descTable;
+ // scan node id -> TFileScanRangeParams
+ private Map<Integer, TFileScanRangeParams> fileScanRangeParamsMap = Maps.newHashMap();
+
// Why do we use query global?
// When `NOW()` function is in sql, we need only one now(),
// but, we execute `NOW()` distributed.
@@ -1937,6 +1942,11 @@ public class Coordinator {
k -> Sets.newHashSet());
scanNodeIds.add(scanNode.getId().asInt());
+ if (scanNode instanceof FileQueryScanNode) {
+ fileScanRangeParamsMap.put(
+ scanNode.getId().asInt(), ((FileQueryScanNode) scanNode).getFileScanRangeParams());
+ }
+
FragmentScanRangeAssignment assignment
= fragmentExecParamsMap.get(scanNode.getFragmentId()).scanRangeAssignment;
boolean fragmentContainsColocateJoin = isColocateFragment(scanNode.getFragment(),
@@ -2294,7 +2304,7 @@ public class Coordinator {
return executionProfile.isAllInstancesDone();
}
- // map from an impalad host address to the per-node assigned scan ranges;
+ // map from a BE host address to the per-node assigned scan ranges;
// records scan range assignment for a single fragment
class FragmentScanRangeAssignment
extends HashMap<TNetworkAddress, Map<Integer, List<TScanRangeParams>>> {
@@ -2580,6 +2590,7 @@ public class Coordinator {
*/
public void unsetFields() {
this.rpcParams.unsetDescTbl();
+ this.rpcParams.unsetFileScanParams();
this.rpcParams.unsetCoord();
this.rpcParams.unsetQueryGlobals();
this.rpcParams.unsetResourceInfo();
@@ -2731,6 +2742,7 @@ public class Coordinator {
*/
public void unsetFields() {
this.rpcParams.unsetDescTbl();
+ this.rpcParams.unsetFileScanParams();
this.rpcParams.unsetCoord();
this.rpcParams.unsetQueryGlobals();
this.rpcParams.unsetResourceInfo();
@@ -3148,6 +3160,8 @@ public class Coordinator {
rf.getFilterId().asInt(), rf.toThrift());
}
}
+
+ params.setFileScanParams(fileScanRangeParamsMap);
paramsList.add(params);
}
return paramsList;
@@ -3186,6 +3200,8 @@ public class Coordinator {
if (tWorkloadGroups != null) {
params.setWorkloadGroups(tWorkloadGroups);
}
+
+ params.setFileScanParams(fileScanRangeParamsMap);
res.put(instanceExecParam.host, params);
}
TPipelineFragmentParams params = res.get(instanceExecParam.host);
@@ -3420,3 +3436,4 @@ public class Coordinator {
}
+
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
index 8a28c82ec9..7e1a4b6698 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/ExternalFileTableValuedFunction.java
@@ -442,7 +442,6 @@ public abstract class ExternalFileTableValuedFunction extends TableValuedFunctio
private PFetchTableSchemaRequest getFetchTableStructureRequest() throws AnalysisException, TException {
// set TFileScanRangeParams
TFileScanRangeParams fileScanRangeParams = new TFileScanRangeParams();
- fileScanRangeParams.setFileType(getTFileType());
fileScanRangeParams.setFormatType(fileFormatType);
fileScanRangeParams.setProperties(locationProperties);
fileScanRangeParams.setFileAttributes(getFileAttributes());
@@ -466,9 +465,10 @@ public abstract class ExternalFileTableValuedFunction extends TableValuedFunctio
throw new AnalysisException("Can not get first file, please check uri.");
}
- fileScanRangeParams.setCompressType(Util.getOrInferCompressType(compressionType, firstFile.getPath()));
// set TFileRangeDesc
TFileRangeDesc fileRangeDesc = new TFileRangeDesc();
+ fileRangeDesc.setFileType(getTFileType());
+ fileRangeDesc.setCompressType(Util.getOrInferCompressType(compressionType, firstFile.getPath()));
fileRangeDesc.setPath(firstFile.getPath());
fileRangeDesc.setStartOffset(0);
fileRangeDesc.setSize(firstFile.getSize());
@@ -483,3 +483,4 @@ public abstract class ExternalFileTableValuedFunction extends TableValuedFunctio
}
}
+
diff --git a/gensrc/thrift/PaloInternalService.thrift b/gensrc/thrift/PaloInternalService.thrift
index 40457481b7..95dc716b7b 100644
--- a/gensrc/thrift/PaloInternalService.thrift
+++ b/gensrc/thrift/PaloInternalService.thrift
@@ -420,6 +420,9 @@ struct TExecPlanFragmentParams {
22: optional list<Types.TUniqueId> instances_sharing_hash_table;
23: optional string table_name;
+
+ // scan node id -> scan range params, only for external file scan
+ 24: optional map<Types.TPlanNodeId, PlanNodes.TFileScanRangeParams> file_scan_params
}
struct TExecPlanFragmentParamsList {
@@ -632,6 +635,8 @@ struct TPipelineFragmentParams {
26: optional list<TPipelineWorkloadGroup> workload_groups
27: optional TTxnParams txn_conf
28: optional string table_name
+ // scan node id -> scan range params, only for external file scan
+ 29: optional map<Types.TPlanNodeId, PlanNodes.TFileScanRangeParams> file_scan_params
}
struct TPipelineFragmentParamsList {
diff --git a/gensrc/thrift/PlanNodes.thrift b/gensrc/thrift/PlanNodes.thrift
index 1b74b0cc70..9dacaac889 100644
--- a/gensrc/thrift/PlanNodes.thrift
+++ b/gensrc/thrift/PlanNodes.thrift
@@ -335,8 +335,10 @@ struct TTableFormatFileDesc {
}
struct TFileScanRangeParams {
+ // deprecated, move to TFileScanRange
1: optional Types.TFileType file_type;
2: optional TFileFormatType format_type;
+ // deprecated, move to TFileScanRange
3: optional TFileCompressType compress_type;
// If this is for load job, src point to the source table and dest point to the doris table.
// If this is for query, only dest_tuple_id is set, including both file slot and partition slot.
@@ -395,6 +397,8 @@ struct TFileRangeDesc {
8: optional TTableFormatFileDesc table_format_params
// Use modification time to determine whether the file is changed
9: optional i64 modification_time
+ 10: optional Types.TFileType file_type;
+ 11: optional TFileCompressType compress_type;
}
// TFileScanRange represents a set of descriptions of a file and the rules for reading and converting it.
@@ -402,6 +406,10 @@ struct TFileRangeDesc {
// list<TFileRangeDesc>: file location and range
struct TFileScanRange {
1: optional list<TFileRangeDesc> ranges
+ // If file_scan_params in TExecPlanFragmentParams is set in TExecPlanFragmentParams
+ // will use that field, otherwise, use this field.
+ // file_scan_params in TExecPlanFragmentParams will always be set in query request,
+ // and TFileScanRangeParams here is used for some other request such as fetch table schema for tvf.
2: optional TFileScanRangeParams params
}
diff --git a/regression-test/suites/external_table_emr_p2/hive/test_external_brown.groovy b/regression-test/suites/external_table_emr_p2/hive/test_external_brown.groovy
index 455d0333f4..cd8e1ee501 100644
--- a/regression-test/suites/external_table_emr_p2/hive/test_external_brown.groovy
+++ b/regression-test/suites/external_table_emr_p2/hive/test_external_brown.groovy
@@ -285,7 +285,7 @@ suite("test_external_brown", "p2") {
if (enabled != null && enabled.equalsIgnoreCase("true")) {
String extHiveHmsHost = context.config.otherConfigs.get("extHiveHmsHost")
String extHiveHmsPort = context.config.otherConfigs.get("extHiveHmsPort")
- String catalog_name = "external_yandex"
+ String catalog_name = "external_brown"
sql """drop catalog if exists ${catalog_name};"""
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 07/10: [Fix](statistics) Fix analyze db always use internal catalog bug (#21850)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 68b54dde2e2f73e5e93d60a7341ae56adcfd3f6f
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Mon Jul 17 15:28:54 2023 +0800
[Fix](statistics) Fix analyze db always use internal catalog bug (#21850)
`Analyze database db_name ` command couldn't use current catalog, it is always using the internal catalog. This will cause the command failed to find the db. This pr is to fix this bug.
---
.../org/apache/doris/analysis/AnalyzeDBStmt.java | 2 +-
.../hive/test_hive_analyze_db.out | 20 +++++++++++
.../hive/test_hive_analyze_db.groovy | 40 ++++++++++++++++++++++
3 files changed, 61 insertions(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/AnalyzeDBStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/AnalyzeDBStmt.java
index db155f07ae..4957bd2cb8 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/AnalyzeDBStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/AnalyzeDBStmt.java
@@ -44,7 +44,7 @@ public class AnalyzeDBStmt extends AnalyzeStmt {
@Override
public void analyze(Analyzer analyzer) throws AnalysisException, UserException {
if (ctlName == null) {
- ctlIf = Env.getCurrentInternalCatalog();
+ ctlIf = Env.getCurrentEnv().getCurrentCatalog();
} else {
ctlIf = Env.getCurrentEnv().getCatalogMgr().getCatalogOrAnalysisException(ctlName);
}
diff --git a/regression-test/data/external_table_emr_p2/hive/test_hive_analyze_db.out b/regression-test/data/external_table_emr_p2/hive/test_hive_analyze_db.out
new file mode 100644
index 0000000000..0b75316476
--- /dev/null
+++ b/regression-test/data/external_table_emr_p2/hive/test_hive_analyze_db.out
@@ -0,0 +1,20 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !1 --
+lo_orderpriority 100.0 5.0 0.0 888.8000000000001 8.8 '1-URGENT' '5-LOW'
+lo_custkey 100.0 26.0 0.0 404.0 4.0 67423 2735521
+lo_partkey 100.0 100.0 0.0 404.0 4.0 2250 989601
+lo_revenue 100.0 100.0 0.0 404.0 4.0 101171 8703450
+lo_commitdate 100.0 95.0 0.0 404.0 4.0 19920515 19981016
+lo_quantity 100.0 46.0 0.0 404.0 4.0 1 50
+lo_orderkey 100.0 26.0 0.0 404.0 4.0 1 98
+lo_suppkey 100.0 100.0 0.0 404.0 4.0 4167 195845
+lo_supplycost 100.0 100.0 0.0 404.0 4.0 58023 121374
+lo_shipmode 100.0 7.0 0.0 425.21 4.21 'AIR' 'TRUCK'
+lo_orderdate 100.0 26.0 0.0 404.0 4.0 19920221 19980721
+lo_linenumber 100.0 7.0 0.0 404.0 4.0 1 7
+lo_shippriority 100.0 1.0 0.0 404.0 4.0 0 0
+lo_ordtotalprice 100.0 26.0 0.0 404.0 4.0 3428256 36771805
+lo_extendedprice 100.0 100.0 0.0 404.0 4.0 104300 9066094
+lo_tax 100.0 9.0 0.0 404.0 4.0 0 8
+lo_discount 100.0 11.0 0.0 404.0 4.0 0 10
+
diff --git a/regression-test/suites/external_table_emr_p2/hive/test_hive_analyze_db.groovy b/regression-test/suites/external_table_emr_p2/hive/test_hive_analyze_db.groovy
new file mode 100644
index 0000000000..a30c87275b
--- /dev/null
+++ b/regression-test/suites/external_table_emr_p2/hive/test_hive_analyze_db.groovy
@@ -0,0 +1,40 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+suite("test_hive_analyze_db", "p2") {
+ String enabled = context.config.otherConfigs.get("enableExternalHiveTest")
+ if (enabled != null && enabled.equalsIgnoreCase("true")) {
+ String extHiveHmsHost = context.config.otherConfigs.get("extHiveHmsHost")
+ String extHiveHmsPort = context.config.otherConfigs.get("extHiveHmsPort")
+ String catalog_name = "test_hive_analyze_db"
+ sql """drop catalog if exists ${catalog_name};"""
+ sql """
+ create catalog if not exists ${catalog_name} properties (
+ 'type'='hms',
+ 'hadoop.username' = 'hadoop',
+ 'hive.metastore.uris' = 'thrift://${extHiveHmsHost}:${extHiveHmsPort}'
+ );
+ """
+ logger.info("catalog " + catalog_name + " created")
+ sql """switch ${catalog_name};"""
+ logger.info("switched to catalog " + catalog_name)
+ sql """use statistics;"""
+ sql """analyze database statistics with sync"""
+ qt_1 "show column stats statistics"
+ }
+}
+
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 05/10: Fix incorrect priority queue refactor (#21857)
Posted by kx...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
kxiao pushed a commit to branch branch-2.0
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 9afdc0218616eb49dda053300dc7cf75c5e9d163
Author: wangbo <wa...@apache.org>
AuthorDate: Mon Jul 17 14:56:24 2023 +0800
Fix incorrect priority queue refactor (#21857)
---
be/src/pipeline/task_queue.cpp | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/be/src/pipeline/task_queue.cpp b/be/src/pipeline/task_queue.cpp
index 2c09ac8edd..715c9601cd 100644
--- a/be/src/pipeline/task_queue.cpp
+++ b/be/src/pipeline/task_queue.cpp
@@ -43,7 +43,7 @@ PipelineTask* SubTaskQueue::try_take(bool is_steal) {
PriorityTaskQueue::PriorityTaskQueue() : _closed(false) {
double factor = 1;
- for (int i = 0; i < SUB_QUEUE_LEVEL; ++i) {
+ for (int i = SUB_QUEUE_LEVEL - 1; i >= 0; i--) {
_sub_queues[i].set_level_factor(factor);
factor *= LEVEL_QUEUE_TIME_FACTOR;
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org