You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@doris.apache.org by mo...@apache.org on 2022/12/24 09:09:52 UTC
[doris] branch branch-1.2-lts updated (fd51474649 -> 9d637531e9)
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a change to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
from fd51474649 [feature](planner) mark join to support subquery in disjunction (#14579) (#15291)
new b89c88bccb [enhancement] (streamload) allow table in url when do two-phase commit (#15246) (#15248)
new 609f8e9882 [Bug](decimalv3) Fix wrong precision of DECIMALV3 (#15302)
new e8bb4a3bcc [Bug](Agg) fix crash when encountering not supported agg function like last_value(bitmap) (#15257)
new 5c4a99dff9 [fix](s3 outfile) Add the`use_path_style` parameter for s3 outfile (#15288)
new 8e5472502e [fix](Inbitmap) fix in bitmap result error when left expr is constant (#15271)
new 024a848a66 [fix](bitmapfilter) fix core dump caused by bitmap filter (#15296)
new 2e24b3c08a [fix](union)the union node should not pass through children in some case (#15286)
new 8c67998f80 [fix](fe)fix bug of the bucket shuffle join is not recognized (#15255)
new 4036ac38cf [Fix](multi catalog)Fix VFileScanner file not found status bug. #15226
new ddee3b4b85 [fix](jdbc catalog) fix bugs of jdbc catalog and table valued function (#15216)
new 3c6f78e945 [fix](iceberg-v2) fix fe iceberg split, add regression case (#15299)
new 049c1f0980 [feature](multi-catalog) support use catalog.db when client connect to the doris server (#15293)
new d2b6746586 [fix](scan) fix that be may core dump when the predicates are all false (#15332)
new f251343fa4 [fix](multi-catalog) fix and optimize iceberg v2 reader (#15274)
new 9d637531e9 [fix](compile) fix compile error caused by `mysql_scan_node.cpp` not being found when enabling `WITH_MYSQL` (#15277)
The 15 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/exec/CMakeLists.txt | 7 -
be/src/exprs/runtime_filter.cpp | 8 +-
be/src/runtime/types.h | 5 +
be/src/service/http_service.cpp | 2 +
be/src/service/internal_service.cpp | 2 +-
be/src/vec/CMakeLists.txt | 7 +
.../aggregate_function_window.cpp | 4 +-
be/src/vec/exec/format/csv/csv_reader.cpp | 4 +-
be/src/vec/exec/format/csv/csv_reader.h | 4 +-
be/src/vec/exec/format/generic_reader.h | 6 +-
be/src/vec/exec/format/json/new_json_reader.cpp | 4 +-
be/src/vec/exec/format/json/new_json_reader.h | 4 +-
be/src/vec/exec/format/orc/vorc_reader.cpp | 4 +-
be/src/vec/exec/format/orc/vorc_reader.h | 4 +-
be/src/vec/exec/format/parquet/parquet_common.h | 2 +-
.../exec/format/parquet/vparquet_column_reader.cpp | 30 +--
.../exec/format/parquet/vparquet_column_reader.h | 19 +-
.../exec/format/parquet/vparquet_group_reader.cpp | 199 +++++++++-----
.../exec/format/parquet/vparquet_group_reader.h | 83 ++++--
be/src/vec/exec/format/parquet/vparquet_reader.cpp | 243 +++++++----------
be/src/vec/exec/format/parquet/vparquet_reader.h | 30 ++-
be/src/vec/exec/format/table/iceberg_reader.cpp | 293 ++++++++++-----------
be/src/vec/exec/format/table/iceberg_reader.h | 30 +--
be/src/vec/exec/format/table/table_format_reader.h | 5 +-
be/src/vec/exec/scan/scanner_scheduler.cpp | 5 +-
be/src/vec/exec/scan/vscan_node.cpp | 5 +-
be/src/vec/exec/vanalytic_eval_node.cpp | 4 +
be/src/vec/exprs/vectorized_agg_fn.cpp | 2 +-
be/test/vec/exec/parquet/parquet_thrift_test.cpp | 12 +-
.../docker-compose/mysql/init/03-create-table.sql | 9 +
.../docker-compose/mysql/init/04-insert.sql | 5 +
.../import/import-way/stream-load-manual.md | 38 +--
.../Data-Manipulation-Statements/OUTFILE.md | 1 +
.../import/import-way/stream-load-manual.md | 24 +-
.../Data-Manipulation-Statements/OUTFILE.md | 1 +
.../org/apache/doris/analysis/DescribeStmt.java | 3 +-
.../org/apache/doris/analysis/OutFileClause.java | 4 +
.../org/apache/doris/analysis/StmtRewriter.java | 45 +++-
.../java/org/apache/doris/catalog/FunctionSet.java | 2 +-
.../java/org/apache/doris/catalog/ScalarType.java | 10 +-
.../org/apache/doris/external/jdbc/JdbcClient.java | 25 +-
.../org/apache/doris/httpv2/rest/LoadAction.java | 9 +
.../java/org/apache/doris/mysql/MysqlProto.java | 31 ++-
.../apache/doris/planner/DistributedPlanner.java | 2 +-
.../java/org/apache/doris/planner/PlanNode.java | 17 ++
.../org/apache/doris/planner/SetOperationNode.java | 3 +
.../doris/planner/external/QueryScanProvider.java | 2 +-
.../java/org/apache/doris/qe/ConnectProcessor.java | 2 +
.../ExternalFileTableValuedFunction.java | 2 +-
.../doris/tablefunction/S3TableValuedFunction.java | 2 +-
.../table_valued_function/test_hdfs_tvf.out | 9 +
...jection.out => test_union_subquery_groupby.out} | 0
..._constant.out => test_with_false_predicate.out} | 5 +-
.../decimalv3/test_arithmetic_expressions.out | 19 ++
.../jdbc_catalog_p0/test_mysql_jdbc_catalog.out | 4 +
.../data/query_p0/join/test_bitmap_filter.out | 19 ++
.../table_valued_function/test_hdfs_tvf.groovy | 9 +
...join.groovy => test_bucket_shuffle_join.groovy} | 13 +-
...t.groovy => test_union_subquery_groupby.groovy} | 42 ++-
...dow.groovy => test_with_false_predicate.groovy} | 41 ++-
.../decimalv3/test_arithmetic_expressions.groovy} | 38 ++-
.../jdbc_catalog_p0/test_mysql_jdbc_catalog.groovy | 7 +-
.../suites/query_p0/join/test_bitmap_filter.groovy | 12 +-
.../function/test_not_supported_agg_fn.groovy | 43 +++
64 files changed, 938 insertions(+), 587 deletions(-)
copy regression-test/data/correctness_p0/{test_join_with_projection.out => test_union_subquery_groupby.out} (100%)
copy regression-test/data/correctness_p0/{test_group_by_constant.out => test_with_false_predicate.out} (83%)
create mode 100644 regression-test/data/datatype_p0/decimalv3/test_arithmetic_expressions.out
copy regression-test/suites/correctness_p0/{test_colocate_join.groovy => test_bucket_shuffle_join.groovy} (87%)
copy regression-test/suites/correctness_p0/{test_implict_cast.groovy => test_union_subquery_groupby.groovy} (58%)
copy regression-test/suites/correctness_p0/{test_no_grouping_window.groovy => test_with_false_predicate.groovy} (55%)
copy regression-test/suites/{bloom_filter_p0/test_bloom_filter_is_not_null.groovy => datatype_p0/decimalv3/test_arithmetic_expressions.groovy} (54%)
create mode 100644 regression-test/suites/tpch_sf1_p1/function/test_not_supported_agg_fn.groovy
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 07/15: [fix](union)the union node should not pass through children in some case (#15286)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 2e24b3c08a5e9b3986b671e32d4d5c31c9de8cfc
Author: starocean999 <40...@users.noreply.github.com>
AuthorDate: Fri Dec 23 10:27:49 2022 +0800
[fix](union)the union node should not pass through children in some case (#15286)
the union node will make children pass through in wrong condition. If the children's materialized slots are different from union node, children can't be passed through.
---
.../org/apache/doris/planner/SetOperationNode.java | 3 ++
.../correctness_p0/test_union_subquery_groupby.out | 4 ++
.../test_union_subquery_groupby.groovy | 57 ++++++++++++++++++++++
3 files changed, 64 insertions(+)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java
index be4782188b..e79bfe7cf4 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/SetOperationNode.java
@@ -285,6 +285,9 @@ public abstract class SetOperationNode extends PlanNode {
for (int i = 0; i < setOpResultExprs.size(); ++i) {
if (!setOpTupleDescriptor.getSlots().get(i).isMaterialized()) {
+ if (VectorizedUtil.isVectorized() && childTupleDescriptor.getSlots().get(i).isMaterialized()) {
+ return false;
+ }
continue;
}
SlotRef setOpSlotRef = setOpResultExprs.get(i).unwrapSlotRef(false);
diff --git a/regression-test/data/correctness_p0/test_union_subquery_groupby.out b/regression-test/data/correctness_p0/test_union_subquery_groupby.out
new file mode 100644
index 0000000000..72d126351a
--- /dev/null
+++ b/regression-test/data/correctness_p0/test_union_subquery_groupby.out
@@ -0,0 +1,4 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select --
+1
+
diff --git a/regression-test/suites/correctness_p0/test_union_subquery_groupby.groovy b/regression-test/suites/correctness_p0/test_union_subquery_groupby.groovy
new file mode 100644
index 0000000000..fe400781d0
--- /dev/null
+++ b/regression-test/suites/correctness_p0/test_union_subquery_groupby.groovy
@@ -0,0 +1,57 @@
+// 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_union_subquery_groupby") {
+ sql """
+ drop table if exists t_union_subquery_group_by;
+ """
+
+ sql """
+ CREATE TABLE `t_union_subquery_group_by` (
+ `org_id` bigint(20) NULL,
+ `org_name` text NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`org_id`)
+ COMMENT "OLAP"
+ DISTRIBUTED BY HASH(`org_id`) BUCKETS 3
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "in_memory" = "false",
+ "storage_format" = "V2"
+ );
+ """
+
+ sql """
+ insert into t_union_subquery_group_by values(1,'1'),(2,'1');
+ """
+
+ qt_select """
+ SELECT `org_name`
+ FROM
+ ( SELECT org_id, org_name
+ FROM t_union_subquery_group_by
+ GROUP BY org_id, org_name
+ UNION ALL
+ SELECT org_id, org_name
+ FROM t_union_subquery_group_by
+ GROUP BY org_id, org_name) T GROUP BY `org_name`;
+ """
+
+ sql """
+ drop table if exists t_union_subquery_group_by;
+ """
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 05/15: [fix](Inbitmap) fix in bitmap result error when left expr is constant (#15271)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 8e5472502eea2d9a9f409ca767dc456805ebe5f9
Author: luozenglin <37...@users.noreply.github.com>
AuthorDate: Thu Dec 22 19:25:09 2022 +0800
[fix](Inbitmap) fix in bitmap result error when left expr is constant (#15271)
* [fix](Inbitmap) fix in bitmap result error when left expr is constant
1. When left expr of the in predicate is a constant, instead of generating a bitmap filter, rewrite sql to use `bitmap_contains`.
For example,"select k1, k2 from (select 2 k1, 11 k2) t where k1 in (select bitmap_col from bitmap_tbl)"
=> "select k1, k2 from (select 2 k1, 11 k2) t left semi join bitmap_tbl b on bitmap_contains(b.bitmap_col, t.k1)"
* add regression test
---
.../org/apache/doris/analysis/StmtRewriter.java | 45 +++++++++++++++++-----
.../data/query_p0/join/test_bitmap_filter.out | 10 +++++
.../suites/query_p0/join/test_bitmap_filter.groovy | 8 ++++
3 files changed, 53 insertions(+), 10 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/StmtRewriter.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/StmtRewriter.java
index f837bf6af5..a93c21692c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/StmtRewriter.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/StmtRewriter.java
@@ -50,6 +50,8 @@ import java.util.List;
public class StmtRewriter {
private static final Logger LOG = LoggerFactory.getLogger(StmtRewriter.class);
+ private static final String BITMAP_CONTAINS = "bitmap_contains";
+
/**
* Rewrite the statement of an analysis result. The unanalyzed rewritten
* statement is returned.
@@ -799,8 +801,12 @@ public class StmtRewriter {
if (!hasEqJoinPred && !inlineView.isCorrelated()) {
// Join with InPredicate is actually an equal join, so we choose HashJoin.
if (expr instanceof ExistsPredicate) {
- joinOp = ((ExistsPredicate) expr).isNotExists() ? JoinOperator.LEFT_ANTI_JOIN
- : JoinOperator.LEFT_SEMI_JOIN;
+ joinOp = ((ExistsPredicate) expr).isNotExists() ? JoinOperator.LEFT_ANTI_JOIN :
+ JoinOperator.LEFT_SEMI_JOIN;
+ } else if (expr instanceof InPredicate && joinConjunct instanceof FunctionCallExpr
+ && (((FunctionCallExpr) joinConjunct).getFnName().getFunction()
+ .equalsIgnoreCase(BITMAP_CONTAINS))) {
+ joinOp = ((InPredicate) expr).isNotIn() ? JoinOperator.LEFT_ANTI_JOIN : JoinOperator.LEFT_SEMI_JOIN;
} else {
joinOp = JoinOperator.CROSS_JOIN;
// We can equal the aggregate subquery using a cross join. All conjuncts
@@ -1166,6 +1172,32 @@ public class StmtRewriter {
}
}
+ // If left expr of in predicate is a constant and a bitmap filter cannot be used, then the normal nested loop join
+ // process is used, with the join Conjunct being `bitmap_contains`,
+ // e.g. 'select k1, k2 from (select 2 k1, 11 k2) t where k1 in (select bitmap_col from bitmap_tbl)'.
+ private static Expr createInBitmapConjunct(Expr exprWithSubquery, SlotRef bitmapSlotRef, Analyzer analyzer,
+ boolean isCorrelated) throws AnalysisException {
+ if (isCorrelated) {
+ throw new AnalysisException("In bitmap does not support correlated subquery: " + exprWithSubquery.toSql());
+ }
+
+ boolean useBitmapFilter = false;
+ List<SlotRef> slotRefs = Lists.newArrayList();
+ exprWithSubquery.getChild(0).collect(SlotRef.class, slotRefs);
+ for (SlotRef slotRef : slotRefs) {
+ List<Expr> sourceExprs = slotRef.getDesc().getSourceExprs();
+ if (sourceExprs.isEmpty() || sourceExprs.stream().anyMatch(expr -> !expr.isConstant())) {
+ useBitmapFilter = true;
+ }
+ }
+
+ Expr pred = useBitmapFilter ? new BitmapFilterPredicate(exprWithSubquery.getChild(0), bitmapSlotRef,
+ ((InPredicate) exprWithSubquery).isNotIn()) : new FunctionCallExpr(new FunctionName(BITMAP_CONTAINS),
+ Lists.newArrayList(bitmapSlotRef, exprWithSubquery.getChild(0)));
+ pred.analyze(analyzer);
+ return pred;
+ }
+
/**
* Converts an expr containing a subquery into an analyzed conjunct to be
* used in a join. The conversion is performed in place by replacing the
@@ -1189,14 +1221,7 @@ public class StmtRewriter {
Expr subquerySubstitute = slotRef;
if (exprWithSubquery instanceof InPredicate) {
if (slotRef.getType().isBitmapType()) {
- if (isCorrelated) {
- throw new AnalysisException(
- "In bitmap does not support correlated subquery: " + exprWithSubquery.toSql());
- }
- Expr pred = new BitmapFilterPredicate(exprWithSubquery.getChild(0), slotRef,
- ((InPredicate) exprWithSubquery).isNotIn());
- pred.analyze(analyzer);
- return pred;
+ return createInBitmapConjunct(exprWithSubquery, slotRef, analyzer, isCorrelated);
}
BinaryPredicate pred = new BinaryPredicate(BinaryPredicate.Operator.EQ,
exprWithSubquery.getChild(0), slotRef);
diff --git a/regression-test/data/query_p0/join/test_bitmap_filter.out b/regression-test/data/query_p0/join/test_bitmap_filter.out
index 24d7773720..83bcb3aed9 100644
--- a/regression-test/data/query_p0/join/test_bitmap_filter.out
+++ b/regression-test/data/query_p0/join/test_bitmap_filter.out
@@ -56,3 +56,13 @@
1991 2
32767 1
+-- !sql7 --
+
+-- !sql8 --
+11 11
+
+-- !sql9 --
+2 11
+
+-- !sql10 --
+
diff --git a/regression-test/suites/query_p0/join/test_bitmap_filter.groovy b/regression-test/suites/query_p0/join/test_bitmap_filter.groovy
index 0292c5c93a..90c0fe5f74 100644
--- a/regression-test/suites/query_p0/join/test_bitmap_filter.groovy
+++ b/regression-test/suites/query_p0/join/test_bitmap_filter.groovy
@@ -51,6 +51,14 @@ suite("test_bitmap_filter", "query_p0") {
qt_sql6 "select k2, count(k2) from ${tbl1} where k1 in (select k2 from ${tbl2}) group by k2 order by k2;"
+ qt_sql7 "select k1, k2 from (select 2 k1, 2 k2) t where k1 in (select k2 from ${tbl2}) order by 1, 2;"
+
+ qt_sql8 "select k1, k2 from (select 11 k1, 11 k2) t where k1 in (select k2 from ${tbl2}) order by 1, 2;"
+
+ qt_sql9 "select k1, k2 from (select 2 k1, 11 k2) t where k1 not in (select k2 from ${tbl2}) order by 1, 2;"
+
+ qt_sql10 "select k1, k2 from (select 1 k1, 11 k2) t where k1 not in (select k2 from ${tbl2}) order by 1, 2;"
+
test {
sql "select k1, k2 from ${tbl1} b1 where k1 in (select k2 from ${tbl2} b2 where b1.k2 = b2.k1) order by k1;"
exception "In bitmap does not support correlated subquery"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 08/15: [fix](fe)fix bug of the bucket shuffle join is not recognized (#15255)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 8c67998f80764480dfebea7ab306805fd8b02311
Author: starocean999 <40...@users.noreply.github.com>
AuthorDate: Fri Dec 23 16:44:44 2022 +0800
[fix](fe)fix bug of the bucket shuffle join is not recognized (#15255)
* [fix](fe)fix bug of the bucket shuffle join is not recognized
* use broadcast join for empty table
---
.../apache/doris/planner/DistributedPlanner.java | 2 +-
.../java/org/apache/doris/planner/PlanNode.java | 17 +++++
.../correctness_p0/test_bucket_shuffle_join.groovy | 81 ++++++++++++++++++++++
3 files changed, 99 insertions(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java b/fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java
index 6382787d1b..6c35b62006 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/DistributedPlanner.java
@@ -656,7 +656,7 @@ public class DistributedPlanner {
continue;
}
- SlotRef leftSlot = lhsJoinExpr.unwrapSlotRef();
+ SlotRef leftSlot = node.getChild(0).findSrcSlotRef(lhsJoinExpr.getSrcSlotRef());
if (leftSlot.getTable() instanceof OlapTable
&& leftScanNode.desc.getSlots().contains(leftSlot.getDesc())) {
// table name in SlotRef is not the really name. `select * from test as t`
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
index 2c3baf6ec8..36263f5795 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/PlanNode.java
@@ -32,6 +32,7 @@ import org.apache.doris.analysis.SlotRef;
import org.apache.doris.analysis.TupleDescriptor;
import org.apache.doris.analysis.TupleId;
import org.apache.doris.catalog.Function;
+import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Type;
import org.apache.doris.common.AnalysisException;
import org.apache.doris.common.NotImplementedException;
@@ -926,6 +927,22 @@ public abstract class PlanNode extends TreeNode<PlanNode> implements PlanStats {
return null;
}
+ public SlotRef findSrcSlotRef(SlotRef slotRef) {
+ if (slotRef.getTable() instanceof OlapTable) {
+ return slotRef;
+ }
+ if (this instanceof HashJoinNode) {
+ HashJoinNode hashJoinNode = (HashJoinNode) this;
+ SlotRef inputSlotRef = hashJoinNode.getMappedInputSlotRef(slotRef);
+ if (inputSlotRef != null) {
+ return hashJoinNode.getChild(0).findSrcSlotRef(inputSlotRef);
+ } else {
+ return slotRef;
+ }
+ }
+ return slotRef;
+ }
+
protected void addRuntimeFilter(RuntimeFilter filter) {
runtimeFilters.add(filter);
}
diff --git a/regression-test/suites/correctness_p0/test_bucket_shuffle_join.groovy b/regression-test/suites/correctness_p0/test_bucket_shuffle_join.groovy
new file mode 100644
index 0000000000..febd57a353
--- /dev/null
+++ b/regression-test/suites/correctness_p0/test_bucket_shuffle_join.groovy
@@ -0,0 +1,81 @@
+// 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_bucket_shuffle_join") {
+ sql """ DROP TABLE IF EXISTS `test_colo1` """
+ sql """ DROP TABLE IF EXISTS `test_colo2` """
+ sql """ DROP TABLE IF EXISTS `test_colo3` """
+ sql """
+ CREATE TABLE `test_colo1` (
+ `id` varchar(64) NULL,
+ `name` varchar(64) NULL,
+ `age` int NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`,`name`)
+ COMMENT 'OLAP'
+ DISTRIBUTED BY HASH(`id`,`name`) BUCKETS 4
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "in_memory" = "false",
+ "storage_format" = "V2",
+ "disable_auto_compaction" = "false"
+ );
+ """
+ sql """
+ CREATE TABLE `test_colo2` (
+ `id` varchar(64) NULL,
+ `name` varchar(64) NULL,
+ `age` int NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`,`name`)
+ COMMENT 'OLAP'
+ DISTRIBUTED BY HASH(`id`,`name`) BUCKETS 5
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "in_memory" = "false",
+ "storage_format" = "V2",
+ "disable_auto_compaction" = "false"
+ );
+ """
+
+ sql """
+ CREATE TABLE `test_colo3` (
+ `id` varchar(64) NULL,
+ `name` varchar(64) NULL,
+ `age` int NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`id`,`name`)
+ COMMENT 'OLAP'
+ DISTRIBUTED BY HASH(`id`,`name`) BUCKETS 6
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "in_memory" = "false",
+ "storage_format" = "V2",
+ "disable_auto_compaction" = "false"
+ );
+ """
+
+ sql """insert into test_colo1 values('1','a',12);"""
+ sql """insert into test_colo2 values('1','a',12);"""
+ sql """insert into test_colo3 values('1','a',12);"""
+
+ explain {
+ sql("select a.id,a.name,b.id,b.name,c.id,c.name from test_colo1 a inner join test_colo2 b on a.id = b.id and a.name = b.name inner join test_colo3 c on a.id=c.id and a.name= c.name")
+ contains "4:VHASH JOIN\n | join op: INNER JOIN(BUCKET_SHUFFLE)"
+ contains "2:VHASH JOIN\n | join op: INNER JOIN(BUCKET_SHUFFLE)"
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 11/15: [fix](iceberg-v2) fix fe iceberg split, add regression case (#15299)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 3c6f78e94543b1a8df7e50a2429ab4653ce767cf
Author: slothever <18...@users.noreply.github.com>
AuthorDate: Fri Dec 23 19:33:00 2022 +0800
[fix](iceberg-v2) fix fe iceberg split, add regression case (#15299)
---
.../main/java/org/apache/doris/planner/external/QueryScanProvider.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/planner/external/QueryScanProvider.java b/fe/fe-core/src/main/java/org/apache/doris/planner/external/QueryScanProvider.java
index a2ce044bb7..a7453394f6 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/planner/external/QueryScanProvider.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/planner/external/QueryScanProvider.java
@@ -106,7 +106,7 @@ public abstract class QueryScanProvider implements FileScanProviderIf {
TFileRangeDesc rangeDesc = createFileRangeDesc(fileSplit, partitionValuesFromPath, pathPartitionKeys);
// external data lake table
if (split instanceof IcebergSplit) {
- IcebergScanProvider.setIcebergParams(rangeDesc, (IcebergSplit) inputSplit);
+ IcebergScanProvider.setIcebergParams(rangeDesc, (IcebergSplit) split);
}
curLocations.getScanRange().getExtScanRange().getFileScanRange().addToRanges(rangeDesc);
LOG.debug("assign to backend {} with table split: {} ({}, {}), location: {}",
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 10/15: [fix](jdbc catalog) fix bugs of jdbc catalog and table valued function (#15216)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit ddee3b4b855f5910a3797994050a97ff952339c8
Author: Tiewei Fang <43...@users.noreply.github.com>
AuthorDate: Fri Dec 23 16:46:39 2022 +0800
[fix](jdbc catalog) fix bugs of jdbc catalog and table valued function (#15216)
* fix bugs
* add `desc function` test
* add test
* fix
---
.../docker-compose/mysql/init/03-create-table.sql | 9 ++++++++
.../docker-compose/mysql/init/04-insert.sql | 5 +++++
.../org/apache/doris/analysis/DescribeStmt.java | 3 ++-
.../org/apache/doris/external/jdbc/JdbcClient.java | 25 +++++++++++++++++++++-
.../java/org/apache/doris/qe/ConnectProcessor.java | 2 ++
.../ExternalFileTableValuedFunction.java | 2 +-
.../doris/tablefunction/S3TableValuedFunction.java | 2 +-
.../table_valued_function/test_hdfs_tvf.out | 9 ++++++++
.../jdbc_catalog_p0/test_mysql_jdbc_catalog.out | 4 ++++
.../table_valued_function/test_hdfs_tvf.groovy | 9 ++++++++
.../jdbc_catalog_p0/test_mysql_jdbc_catalog.groovy | 7 +++++-
11 files changed, 72 insertions(+), 5 deletions(-)
diff --git a/docker/thirdparties/docker-compose/mysql/init/03-create-table.sql b/docker/thirdparties/docker-compose/mysql/init/03-create-table.sql
index b0e4890013..6c8371e7c7 100644
--- a/docker/thirdparties/docker-compose/mysql/init/03-create-table.sql
+++ b/docker/thirdparties/docker-compose/mysql/init/03-create-table.sql
@@ -214,4 +214,13 @@ create table ex_tb19 (
timestamp_value timestamp
) engine=innodb charset=utf8;
+create table doris_test.ex_tb20 (
+ decimal_normal decimal(38, 5),
+ decimal_unsigned decimal(37, 5) unsigned,
+ decimal_out1 decimal(39, 5),
+ decimal_unsigned_out1 decimal(38, 5) unsigned,
+ decimal_long decimal(65, 5),
+ decimal_unsigned_long decimal(65, 5) unsigned
+) engine=innodb charset=utf8;
+
diff --git a/docker/thirdparties/docker-compose/mysql/init/04-insert.sql b/docker/thirdparties/docker-compose/mysql/init/04-insert.sql
index 684a4137e2..d978a2298d 100644
--- a/docker/thirdparties/docker-compose/mysql/init/04-insert.sql
+++ b/docker/thirdparties/docker-compose/mysql/init/04-insert.sql
@@ -1119,3 +1119,8 @@ INSERT INTO ex_tb18 VALUES
INSERT INTO ex_tb19 VALUES
('2022-11-27', '07:09:51', '2022', '2022-11-27 07:09:51', '2022-11-27 07:09:51');
+
+INSERT INTO doris_test.ex_tb20 VALUES
+(1.12345, 1.12345, 1.12345, 1.12345, 1.12345, 1.12345),
+(123456789012345678901234567890123.12345, 12345678901234567890123456789012.12345, 1234567890123456789012345678901234.12345, 123456789012345678901234567890123.12345,
+123456789012345678901234567890123456789012345678901234567890.12345, 123456789012345678901234567890123456789012345678901234567890.12345);
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/DescribeStmt.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/DescribeStmt.java
index 37370394b9..8a71f8b4be 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/DescribeStmt.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/DescribeStmt.java
@@ -110,7 +110,8 @@ public class DescribeStmt extends ShowStmt {
@Override
public void analyze(Analyzer analyzer) throws UserException {
if (!isAllTables && isTableValuedFunction) {
- List<Column> columns = tableValuedFunctionRef.getTableFunction().getTable().getBaseSchema();
+ tableValuedFunctionRef.analyze(analyzer);
+ List<Column> columns = tableValuedFunctionRef.getTable().getBaseSchema();
for (Column column : columns) {
List<String> row = Arrays.asList(
column.getDisplayName(),
diff --git a/fe/fe-core/src/main/java/org/apache/doris/external/jdbc/JdbcClient.java b/fe/fe-core/src/main/java/org/apache/doris/external/jdbc/JdbcClient.java
index 9148629d46..0c3b7702eb 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/external/jdbc/JdbcClient.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/external/jdbc/JdbcClient.java
@@ -21,6 +21,7 @@ import org.apache.doris.catalog.Column;
import org.apache.doris.catalog.PrimitiveType;
import org.apache.doris.catalog.ScalarType;
import org.apache.doris.catalog.Type;
+import org.apache.doris.common.Config;
import org.apache.doris.common.FeConstants;
import org.apache.doris.common.util.Util;
@@ -78,6 +79,7 @@ public class JdbcClient {
this.dbType = parseDbType(jdbcUrl);
this.checkSum = computeObjectChecksum();
+ ClassLoader oldClassLoader = Thread.currentThread().getContextClassLoader();
try {
// TODO(ftw): The problem here is that the jar package is handled by FE
// and URLClassLoader may load the jar package directly into memory
@@ -94,6 +96,8 @@ public class JdbcClient {
dataSource = new HikariDataSource(config);
} catch (MalformedURLException e) {
throw new JdbcClientException("MalformedURLException to load class about " + driverUrl, e);
+ } finally {
+ Thread.currentThread().setContextClassLoader(oldClassLoader);
}
}
@@ -338,6 +342,17 @@ public class JdbcClient {
return Type.BIGINT;
case "BIGINT":
return ScalarType.createStringType();
+ case "DECIMAL":
+ int precision = fieldSchema.getColumnSize() + 1;
+ int scale = fieldSchema.getDecimalDigits();
+ if (precision <= ScalarType.MAX_DECIMAL128_PRECISION) {
+ if (!Config.enable_decimal_conversion && precision > ScalarType.MAX_DECIMALV2_PRECISION) {
+ return ScalarType.createStringType();
+ }
+ return ScalarType.createDecimalType(precision, scale);
+ } else {
+ return ScalarType.createStringType();
+ }
default:
throw new JdbcClientException("Unknown UNSIGNED type of mysql, type: [" + mysqlType + "]");
}
@@ -368,7 +383,14 @@ public class JdbcClient {
case "DECIMAL":
int precision = fieldSchema.getColumnSize();
int scale = fieldSchema.getDecimalDigits();
- return ScalarType.createDecimalType(precision, scale);
+ if (precision <= ScalarType.MAX_DECIMAL128_PRECISION) {
+ if (!Config.enable_decimal_conversion && precision > ScalarType.MAX_DECIMALV2_PRECISION) {
+ return ScalarType.createStringType();
+ }
+ return ScalarType.createDecimalType(precision, scale);
+ } else {
+ return ScalarType.createStringType();
+ }
case "CHAR":
ScalarType charType = ScalarType.createType(PrimitiveType.CHAR);
charType.setLength(fieldSchema.columnSize);
@@ -392,6 +414,7 @@ public class JdbcClient {
case "BIT":
case "BINARY":
case "VARBINARY":
+ case "ENUM":
return ScalarType.createStringType();
default:
throw new JdbcClientException("Can not convert mysql data type to doris data type for type ["
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
index 2715aa3a6c..54de1449d0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/ConnectProcessor.java
@@ -429,6 +429,8 @@ public class ConnectProcessor {
channel.sendOnePacket(serializer.toByteBuffer());
}
+ } catch (Throwable throwable) {
+ handleQueryException(throwable, "", null, null);
} finally {
table.readUnlock();
}
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 4abed5362f..7cadc26db2 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
@@ -141,7 +141,7 @@ public abstract class ExternalFileTableValuedFunction extends TableValuedFunctio
try {
BrokerUtil.parseFile(path, brokerDesc, fileStatuses);
} catch (UserException e) {
- throw new AnalysisException("parse file failed, path = " + path);
+ throw new AnalysisException("parse file failed, path = " + path, e);
}
}
diff --git a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
index e0b4e310f5..ccf5c3eac5 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/tablefunction/S3TableValuedFunction.java
@@ -85,7 +85,7 @@ public class S3TableValuedFunction extends ExternalFileTableValuedFunction {
try {
s3uri = S3URI.create(validParams.get(S3_URI), forceVirtualHosted);
} catch (UserException e) {
- throw new AnalysisException("parse s3 uri failed, uri = " + validParams.get(S3_URI));
+ throw new AnalysisException("parse s3 uri failed, uri = " + validParams.get(S3_URI), e);
}
if (forceVirtualHosted) {
// s3uri.getVirtualBucket() is: virtualBucket.endpoint, Eg:
diff --git a/regression-test/data/correctness_p0/table_valued_function/test_hdfs_tvf.out b/regression-test/data/correctness_p0/table_valued_function/test_hdfs_tvf.out
index 6ed17594b9..e18f42d8ff 100644
--- a/regression-test/data/correctness_p0/table_valued_function/test_hdfs_tvf.out
+++ b/regression-test/data/correctness_p0/table_valued_function/test_hdfs_tvf.out
@@ -287,3 +287,12 @@
4 shenzhen 2345674
5 guangzhou 2345675
+-- !desc --
+s_suppkey INT Yes false \N NONE
+s_name TEXT Yes false \N NONE
+s_address TEXT Yes false \N NONE
+s_nationkey INT Yes false \N NONE
+s_phone TEXT Yes false \N NONE
+s_acctbal DECIMAL(9,0) Yes false \N NONE
+s_comment TEXT Yes false \N NONE
+
diff --git a/regression-test/data/jdbc_catalog_p0/test_mysql_jdbc_catalog.out b/regression-test/data/jdbc_catalog_p0/test_mysql_jdbc_catalog.out
index 9c4a43e959..00360cb185 100644
--- a/regression-test/data/jdbc_catalog_p0/test_mysql_jdbc_catalog.out
+++ b/regression-test/data/jdbc_catalog_p0/test_mysql_jdbc_catalog.out
@@ -152,3 +152,7 @@ bca 2022-11-02 2022-11-02 8012 vivo
-- !ex_tb19 --
2022-11-27 07:09:51 2022 2022-11-27T07:09:51 2022-11-27T07:09:51
+-- !ex_tb20 --
+1.12345 1.12345 1.12345 1.12345 1.12345 1.12345
+123456789012345678901234567890123.12345 12345678901234567890123456789012.12345 1234567890123456789012345678901234.12345 123456789012345678901234567890123.12345 123456789012345678901234567890123456789012345678901234567890.12345 123456789012345678901234567890123456789012345678901234567890.12345
+
diff --git a/regression-test/suites/correctness_p0/table_valued_function/test_hdfs_tvf.groovy b/regression-test/suites/correctness_p0/table_valued_function/test_hdfs_tvf.groovy
index 07b8363b0b..4c4f940026 100644
--- a/regression-test/suites/correctness_p0/table_valued_function/test_hdfs_tvf.groovy
+++ b/regression-test/suites/correctness_p0/table_valued_function/test_hdfs_tvf.groovy
@@ -191,6 +191,15 @@ suite("test_hdfs_tvf") {
sql "sync"
assertTrue(result2[0][0] == 5, "Insert should update 12 rows")
qt_insert """ select * from test_hdfs_tvf order by id; """
+
+ // test desc function
+ uri = "${defaultFS}" + "/user/doris/preinstalled_data/hdfs_tvf/test_parquet.snappy.parquet"
+ format = "parquet"
+ qt_desc """ desc function HDFS(
+ "uri" = "${uri}",
+ "fs.defaultFS"= "${defaultFS}",
+ "hadoop.username" = "${hdfsUserName}",
+ "format" = "${format}"); """
} finally {
}
}
diff --git a/regression-test/suites/jdbc_catalog_p0/test_mysql_jdbc_catalog.groovy b/regression-test/suites/jdbc_catalog_p0/test_mysql_jdbc_catalog.groovy
index 97922eee57..77f7108275 100644
--- a/regression-test/suites/jdbc_catalog_p0/test_mysql_jdbc_catalog.groovy
+++ b/regression-test/suites/jdbc_catalog_p0/test_mysql_jdbc_catalog.groovy
@@ -43,8 +43,9 @@ suite("test_mysql_jdbc_catalog", "p0") {
String ex_tb17 = "ex_tb17";
String ex_tb18 = "ex_tb18";
String ex_tb19 = "ex_tb19";
+ String ex_tb20 = "ex_tb20";
-
+ sql """ADMIN SET FRONTEND CONFIG ("enable_decimal_conversion" = "true");"""
sql """drop catalog if exists ${catalog_name} """
// if use 'com.mysql.cj.jdbc.Driver' here, it will report: ClassNotFound
@@ -94,5 +95,9 @@ suite("test_mysql_jdbc_catalog", "p0") {
order_qt_ex_tb17 """ select * from ${ex_tb17} order by id; """
order_qt_ex_tb18 """ select * from ${ex_tb18} order by num_tinyint; """
order_qt_ex_tb19 """ select * from ${ex_tb19} order by date_value; """
+ order_qt_ex_tb20 """ select * from ${ex_tb20} order by decimal_normal; """
+
+ sql """drop catalog if exists ${catalog_name} """
+ sql """drop resource if exists jdbc_resource_catalog_mysql"""
}
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 09/15: [Fix](multi catalog)Fix VFileScanner file not found status bug. #15226
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 4036ac38cfe46f6233ecae1128565b7bea1dea2f
Author: Jibing-Li <64...@users.noreply.github.com>
AuthorDate: Fri Dec 23 16:45:54 2022 +0800
[Fix](multi catalog)Fix VFileScanner file not found status bug. #15226
The if condition to check NOT FOUND status for VFileScanner is incorrect, fix it.
---
be/src/vec/exec/scan/scanner_scheduler.cpp | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/be/src/vec/exec/scan/scanner_scheduler.cpp b/be/src/vec/exec/scan/scanner_scheduler.cpp
index 0a8dd47932..e67ff364be 100644
--- a/be/src/vec/exec/scan/scanner_scheduler.cpp
+++ b/be/src/vec/exec/scan/scanner_scheduler.cpp
@@ -235,8 +235,11 @@ void ScannerScheduler::_scanner_scan(ScannerScheduler* scheduler, ScannerContext
VLOG_ROW << "VOlapScanNode input rows: " << block->rows() << ", eos: " << eos;
// The VFileScanner for external table may try to open not exist files,
// Because FE file cache for external table may out of date.
+ // So, NOT_FOUND for VFileScanner is not a fail case.
+ // Will remove this after file reader refactor.
if (!status.ok() && (typeid(*scanner) != typeid(doris::vectorized::VFileScanner) ||
- !status.is_not_found())) {
+ (typeid(*scanner) == typeid(doris::vectorized::VFileScanner) &&
+ !status.is_not_found()))) {
LOG(WARNING) << "Scan thread read VOlapScanner failed: " << status.to_string();
// Add block ptr in blocks, prevent mem leak in read failed
blocks.push_back(block);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 13/15: [fix](scan) fix that be may core dump when the predicates are all false (#15332)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit d2b67465865cff517a53bc68610832e5ef17b2f3
Author: Xin Liao <li...@126.com>
AuthorDate: Sat Dec 24 15:27:43 2022 +0800
[fix](scan) fix that be may core dump when the predicates are all false (#15332)
---
be/src/vec/exec/scan/vscan_node.cpp | 3 +
.../correctness_p0/test_with_false_predicate.out | 8 +++
.../test_with_false_predicate.groovy | 64 ++++++++++++++++++++++
3 files changed, 75 insertions(+)
diff --git a/be/src/vec/exec/scan/vscan_node.cpp b/be/src/vec/exec/scan/vscan_node.cpp
index e7953c3703..56f9aec91a 100644
--- a/be/src/vec/exec/scan/vscan_node.cpp
+++ b/be/src/vec/exec/scan/vscan_node.cpp
@@ -93,6 +93,9 @@ Status VScanNode::open(RuntimeState* state) {
RETURN_IF_ERROR(_acquire_runtime_filter());
RETURN_IF_ERROR(_process_conjuncts());
+ if (_eos) {
+ return Status::OK();
+ }
std::list<VScanner*> scanners;
RETURN_IF_ERROR(_init_scanners(&scanners));
diff --git a/regression-test/data/correctness_p0/test_with_false_predicate.out b/regression-test/data/correctness_p0/test_with_false_predicate.out
new file mode 100644
index 0000000000..5884eb2e3a
--- /dev/null
+++ b/regression-test/data/correctness_p0/test_with_false_predicate.out
@@ -0,0 +1,8 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !sql --
+1 1 1 1
+2 2 2 2
+3 3 3 3
+
+-- !sql --
+
diff --git a/regression-test/suites/correctness_p0/test_with_false_predicate.groovy b/regression-test/suites/correctness_p0/test_with_false_predicate.groovy
new file mode 100644
index 0000000000..ffa64f99fa
--- /dev/null
+++ b/regression-test/suites/correctness_p0/test_with_false_predicate.groovy
@@ -0,0 +1,64 @@
+// 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_with_false_predicate") {
+ def tableName = "false_predicate_table"
+
+ sql """
+ DROP TABLE IF EXISTS $tableName
+ """
+
+ sql """
+ CREATE TABLE IF NOT EXISTS `$tableName` (
+ `k1` int NOT NULL,
+ `k2` int NOT NULL,
+ `k3` int NOT NULL,
+ `v1` int NOT NULL
+ ) ENGINE=OLAP
+ DUPLICATE KEY(`k1`, `k2`, `k3`)
+ COMMENT "OLAP"
+ DISTRIBUTED BY HASH(`k1`) BUCKETS 1
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "in_memory" = "false",
+ "storage_format" = "V2"
+ )
+ """
+
+ sql """
+ INSERT INTO $tableName VALUES (1, 1, 1, 1);
+ """
+ sql """
+ INSERT INTO $tableName VALUES (2, 2, 2, 2);
+ """
+ sql """
+ INSERT INTO $tableName VALUES (3, 3, 3, 3);
+ """
+
+ sql " sync "
+
+ qt_sql """
+ select * from $tableName order by k1;
+ """
+
+ qt_sql """
+ select k1, k2 from $tableName where BITMAP_EMPTY() is NULL order by k1;
+ """
+
+ sql " DROP TABLE $tableName "
+
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 04/15: [fix](s3 outfile) Add the`use_path_style` parameter for s3 outfile (#15288)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 5c4a99dff9ebda02434f45c54db440ff7f9d6ed2
Author: Tiewei Fang <43...@users.noreply.github.com>
AuthorDate: Fri Dec 23 16:22:06 2022 +0800
[fix](s3 outfile) Add the`use_path_style` parameter for s3 outfile (#15288)
Currently, `outfile` did not support `use_path_style` parameter and use `virtual-host style` by default,
however some Object-storage may only support `use_path_style` access mode.
This pr add the`use_path_style` parameter for s3 outfile, so that different object-storage can use different access mode.
---
.../sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md | 1 +
.../sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md | 1 +
fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java | 4 ++++
3 files changed, 6 insertions(+)
diff --git a/docs/en/docs/sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md b/docs/en/docs/sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md
index b094fcdd3f..d46b2164d7 100644
--- a/docs/en/docs/sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md
+++ b/docs/en/docs/sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md
@@ -101,6 +101,7 @@ illustrate:
AWS_ACCESS_KEY
AWS_SECRET_KEY
AWS_REGION
+ use_path_stype: (optional) default false . The S3 SDK uses the virtual-hosted style by default. However, some object storage systems may not be enabled or support virtual-hosted style access. At this time, we can add the use_path_style parameter to force the use of path style access method.
```
### example
diff --git a/docs/zh-CN/docs/sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md b/docs/zh-CN/docs/sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md
index 82f8ecec33..25b35f5894 100644
--- a/docs/zh-CN/docs/sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md
+++ b/docs/zh-CN/docs/sql-manual/sql-reference/Data-Manipulation-Statements/OUTFILE.md
@@ -104,6 +104,7 @@ INTO OUTFILE "file_path"
AWS_ACCESS_KEY
AWS_SECRET_KEY
AWS_REGION
+ use_path_stype: (选填) 默认为false 。S3 SDK 默认使用 virtual-hosted style 方式。但某些对象存储系统可能没开启或不支持virtual-hosted style 方式的访问，此时可以添加 use_path_style 参数来强制使用 path style 访问方式。
```
### example
diff --git a/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java b/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
index 3d3f82b74b..78f57cff9c 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/analysis/OutFileClause.java
@@ -648,6 +648,10 @@ public class OutFileClause {
}
}
if (storageType == StorageBackend.StorageType.S3) {
+ if (properties.containsKey(S3Resource.USE_PATH_STYLE)) {
+ brokerProps.put(S3Resource.USE_PATH_STYLE, properties.get(S3Resource.USE_PATH_STYLE));
+ processedPropKeys.add(S3Resource.USE_PATH_STYLE);
+ }
S3Storage.checkS3(brokerProps);
} else if (storageType == StorageBackend.StorageType.HDFS) {
if (!brokerProps.containsKey(HdfsResource.HADOOP_FS_NAME)) {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 01/15: [enhancement] (streamload) allow table in url when do two-phase commit (#15246) (#15248)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit b89c88bccb3d8860fa83d41ea235acfa7105991b
Author: zhengyu <fr...@gmail.com>
AuthorDate: Thu Dec 22 17:00:51 2022 +0800
[enhancement] (streamload) allow table in url when do two-phase commit (#15246) (#15248)
Make it works even if user provide us with (unnecessary) table info in url.
i.e. `curl -X PUT --location-trusted -u user:passwd -H "txn_id:18036" -H \
"txn_operation:commit" http://fe_host:http_port/api/{db}/{table}/_stream_load_2pc`
can still works!
Signed-off-by: freemandealer <fr...@gmail.com>
---
be/src/service/http_service.cpp | 2 ++
.../import/import-way/stream-load-manual.md | 38 ++++++++++++----------
.../import/import-way/stream-load-manual.md | 24 ++++++++------
.../org/apache/doris/httpv2/rest/LoadAction.java | 9 +++++
4 files changed, 46 insertions(+), 27 deletions(-)
diff --git a/be/src/service/http_service.cpp b/be/src/service/http_service.cpp
index 24045a10df..b62e54e6b1 100644
--- a/be/src/service/http_service.cpp
+++ b/be/src/service/http_service.cpp
@@ -66,6 +66,8 @@ Status HttpService::start() {
StreamLoad2PCAction* streamload_2pc_action = _pool.add(new StreamLoad2PCAction(_env));
_ev_http_server->register_handler(HttpMethod::PUT, "/api/{db}/_stream_load_2pc",
streamload_2pc_action);
+ _ev_http_server->register_handler(HttpMethod::PUT, "/api/{db}/{table}/_stream_load_2pc",
+ streamload_2pc_action);
// register download action
std::vector<std::string> allow_paths;
diff --git a/docs/en/docs/data-operate/import/import-way/stream-load-manual.md b/docs/en/docs/data-operate/import/import-way/stream-load-manual.md
index e9917e160f..51103df3b9 100644
--- a/docs/en/docs/data-operate/import/import-way/stream-load-manual.md
+++ b/docs/en/docs/data-operate/import/import-way/stream-load-manual.md
@@ -5,7 +5,7 @@
}
---
-<!--
+<!--
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
@@ -108,7 +108,7 @@ Stream load uses HTTP protocol, so all parameters related to import tasks are se
+ column_separator
-
+
Used to specify the column separator in the load file. The default is `\t`. If it is an invisible character, you need to add `\x` as a prefix and hexadecimal to indicate the separator.
For example, the separator `\x01` of the hive file needs to be specified as `-H "column_separator:\x01"`.
@@ -116,7 +116,7 @@ Stream load uses HTTP protocol, so all parameters related to import tasks are se
You can use a combination of multiple characters as the column separator.
+ line_delimiter
-
+
Used to specify the line delimiter in the load file. The default is `\n`.
You can use a combination of multiple characters as the column separator.
@@ -150,13 +150,13 @@ The number of rows in the original file = `dpp.abnorm.ALL + dpp.norm.ALL`
The function transformation configuration of data to be imported includes the sequence change of columns and the expression transformation, in which the expression transformation method is consistent with the query statement.
```
- Examples of column order transformation: There are three columns of original data (src_c1,src_c2,src_c3), and there are also three columns （dst_c1,dst_c2,dst_c3) in the doris table at present.
+ Examples of column order transformation: There are three columns of original data (src_c1,src_c2,src_c3), and there are also three columns （dst_c1,dst_c2,dst_c3) in the doris table at present.
when the first column src_c1 of the original file corresponds to the dst_c1 column of the target table, while the second column src_c2 of the original file corresponds to the dst_c2 column of the target table and the third column src_c3 of the original file corresponds to the dst_c3 column of the target table,which is written as follows:
columns: dst_c1, dst_c2, dst_c3
-
+
when the first column src_c1 of the original file corresponds to the dst_c2 column of the target table, while the second column src_c2 of the original file corresponds to the dst_c3 column of the target table and the third column src_c3 of the original file corresponds to the dst_c1 column of the target table,which is written as follows:
columns: dst_c2, dst_c3, dst_c1
-
+
Example of expression transformation: There are two columns in the original file and two columns in the target table (c1, c2). However, both columns in the original file need to be transformed by functions to correspond to the two columns in the target table.
columns: tmp_c1, tmp_c2, c1 = year(tmp_c1), c2 = mouth(tmp_c2)
Tmp_* is a placeholder, representing two original columns in the original file.
@@ -201,17 +201,21 @@ The number of rows in the original file = `dpp.abnorm.ALL + dpp.norm.ALL`
"CommitAndPublishTimeMs": 0
}
```
- 2. Trigger the commit operation on the transaction
+ 2. Trigger the commit operation on the transaction.
+ Note 1) requesting to fe and be both works
+ Note 2) `{table}` in url can be omit when commit
```shell
- curl -X PUT --location-trusted -u user:passwd -H "txn_id:18036" -H "txn_operation:commit" http://fe_host:http_port/api/{db}/_stream_load_2pc
+ curl -X PUT --location-trusted -u user:passwd -H "txn_id:18036" -H "txn_operation:commit" http://fe_host:http_port/api/{db}/{table}/_stream_load_2pc
{
"status": "Success",
"msg": "transaction [18036] commit successfully."
}
```
3. Trigger an abort operation on a transaction
+ Note 1) requesting to fe and be both works
+ Note 2) `{table}` in url can be omit when abort
```shell
- curl -X PUT --location-trusted -u user:passwd -H "txn_id:18037" -H "txn_operation:abort" http://fe_host:http_port/api/{db}/_stream_load_2pc
+ curl -X PUT --location-trusted -u user:passwd -H "txn_id:18037" -H "txn_operation:abort" http://fe_host:http_port/api/{db}/{table}/_stream_load_2pc
{
"status": "Success",
"msg": "transaction [18037] abort successfully."
@@ -261,7 +265,7 @@ The following main explanations are given for the Stream load import result para
"Label Already Exists": Label duplicate, need to be replaced Label.
"Fail": Import failed.
-
+
+ ExistingJobStatus: The state of the load job corresponding to the existing Label.
This field is displayed only when the status is "Label Already Exists". The user can know the status of the load job corresponding to Label through this state. "RUNNING" means that the job is still executing, and "FINISHED" means that the job is successful.
@@ -283,7 +287,7 @@ The following main explanations are given for the Stream load import result para
+ BeginTxnTimeMs: The time cost for RPC to Fe to begin a transaction, Unit milliseconds.
+ StreamLoadPutTimeMs: The time cost for RPC to Fe to get a stream load plan, Unit milliseconds.
-
+
+ ReadDataTimeMs: Read data time, Unit milliseconds.
+ WriteDataTimeMs: Write data time, Unit milliseconds.
@@ -387,21 +391,21 @@ Cluster situation: The concurrency of Stream load is not affected by cluster siz
To sort out the possible methods mentioned above: Search FE Master's log with Label to see if there are two ``redirect load action to destination = ``redirect load action to destination cases in the same Label. If so, the request is submitted repeatedly by the Client side.
It is recommended that the user calculate the approximate import time based on the amount of data currently requested, and change the request overtime on the client side to a value greater than the import timeout time according to the import timeout time to avoid multiple submissions of the request by the client side.
-
+
3. Connection reset abnormal
-
+
In the community version 0.14.0 and earlier versions, the connection reset exception occurred after Http V2 was enabled, because the built-in web container is tomcat, and Tomcat has pits in 307 (Temporary Redirect). There is a problem with the implementation of this protocol. All In the case of using Stream load to import a large amount of data, a connect reset exception will occur. This is because tomcat started data transmission before the 307 jump, which resulted in the lack of aut [...]
-
+
After the upgrade, also upgrade the http client version of your program to `4.5.13`，Introduce the following dependencies in your pom.xml file
-
+
```xml
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<version>4.5.13</version>
- </dependency>
+ </dependency>
```
-
+
## More Help
diff --git a/docs/zh-CN/docs/data-operate/import/import-way/stream-load-manual.md b/docs/zh-CN/docs/data-operate/import/import-way/stream-load-manual.md
index 816ab3b46b..0d4526ce20 100644
--- a/docs/zh-CN/docs/data-operate/import/import-way/stream-load-manual.md
+++ b/docs/zh-CN/docs/data-operate/import/import-way/stream-load-manual.md
@@ -5,7 +5,7 @@
}
---
-<!--
+<!--
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
@@ -78,7 +78,7 @@ Stream Load 通过 HTTP 协议提交和传输数据。这里通过 `curl` 命令
```shell
curl --location-trusted -u user:passwd [-H ""...] -T data.file -XPUT http://fe_host:http_port/api/{db}/{table}/_stream_load
-# Header 中支持属性见下面的 ‘导入任务参数’ 说明
+# Header 中支持属性见下面的 ‘导入任务参数’ 说明
# 格式为: -H "key1:value1"
```
@@ -152,13 +152,13 @@ Stream Load 由于使用的是 HTTP 协议，所以所有导入任务有关的
```text
列顺序变换例子：原始数据有三列(src_c1,src_c2,src_c3), 目前doris表也有三列（dst_c1,dst_c2,dst_c3）
-
+
如果原始表的src_c1列对应目标表dst_c1列，原始表的src_c2列对应目标表dst_c2列，原始表的src_c3列对应目标表dst_c3列，则写法如下：
columns: dst_c1, dst_c2, dst_c3
-
+
如果原始表的src_c1列对应目标表dst_c2列，原始表的src_c2列对应目标表dst_c3列，原始表的src_c3列对应目标表dst_c1列，则写法如下：
columns: dst_c2, dst_c3, dst_c1
-
+
表达式变换例子：原始文件有两列，目标表也有两列（c1,c2）但是原始文件的两列均需要经过函数变换才能对应目标表的两列，则写法如下：
columns: tmp_c1, tmp_c2, c1 = year(tmp_c1), c2 = month(tmp_c2)
其中 tmp_*是一个占位符，代表的是原始文件中的两个原始列。
@@ -186,10 +186,10 @@ Stream Load 由于使用的是 HTTP 协议，所以所有导入任务有关的
默认的两阶段批量事务提交为关闭。
- > **开启方式：** 在be.conf中配置`disable_stream_load_2pc=false` 并且 在 HEADER 中声明 `two_phase_commit=true` 。
-
+ > **开启方式：** 在be.conf中配置`disable_stream_load_2pc=false` 并且 在 HEADER 中声明 `two_phase_commit=true` 。
+
示例：
-
+
1. 发起stream load预提交操作
```shell
curl --location-trusted -u user:passwd -H "two_phase_commit:true" -T test.txt http://fe_host:http_port/api/{db}/{table}/_stream_load
@@ -213,16 +213,20 @@ Stream Load 由于使用的是 HTTP 协议，所以所有导入任务有关的
}
```
2. 对事务触发commit操作
+ 注意1) 请求发往fe或be均可
+ 注意2) commit 的时候可以省略 url 中的 `{table}`
```shell
- curl -X PUT --location-trusted -u user:passwd -H "txn_id:18036" -H "txn_operation:commit" http://fe_host:http_port/api/{db}/_stream_load_2pc
+ curl -X PUT --location-trusted -u user:passwd -H "txn_id:18036" -H "txn_operation:commit" http://fe_host:http_port/api/{db}/{table}/_stream_load_2pc
{
"status": "Success",
"msg": "transaction [18036] commit successfully."
}
```
3. 对事务触发abort操作
+ 注意1) 请求发往fe或be均可
+ 注意2) abort 的时候可以省略 url 中的 `{table}`
```shell
- curl -X PUT --location-trusted -u user:passwd -H "txn_id:18037" -H "txn_operation:abort" http://fe_host:http_port/api/{db}/_stream_load_2pc
+ curl -X PUT --location-trusted -u user:passwd -H "txn_id:18037" -H "txn_operation:abort" http://fe_host:http_port/api/{db}/{table}/_stream_load_2pc
{
"status": "Success",
"msg": "transaction [18037] abort successfully."
diff --git a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java
index 0a9313e925..b5ca33b058 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/httpv2/rest/LoadAction.java
@@ -85,6 +85,15 @@ public class LoadAction extends RestBaseController {
return executeStreamLoad2PC(request, db);
}
+ @RequestMapping(path = "/api/{" + DB_KEY + "}/{" + TABLE_KEY + "}/_stream_load_2pc", method = RequestMethod.PUT)
+ public Object streamLoad2PC_table(HttpServletRequest request,
+ HttpServletResponse response,
+ @PathVariable(value = DB_KEY) String db,
+ @PathVariable(value = TABLE_KEY) String table) {
+ executeCheckPassword(request, response);
+ return executeStreamLoad2PC(request, db);
+ }
+
// Same as Multi load, to be compatible with http v1's response body,
// we return error by using RestBaseResult.
private Object executeWithoutPassword(HttpServletRequest request,
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 06/15: [fix](bitmapfilter) fix core dump caused by bitmap filter (#15296)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 024a848a66f0ec4dd1763a03a47ad92977e4f263
Author: luozenglin <37...@users.noreply.github.com>
AuthorDate: Fri Dec 23 16:42:45 2022 +0800
[fix](bitmapfilter) fix core dump caused by bitmap filter (#15296)
Do not push down the bitmap filter to a non-integer column
---
be/src/exprs/runtime_filter.cpp | 8 ++++++--
be/src/runtime/types.h | 5 +++++
be/src/vec/exec/scan/vscan_node.cpp | 2 +-
regression-test/data/query_p0/join/test_bitmap_filter.out | 9 +++++++++
regression-test/suites/query_p0/join/test_bitmap_filter.groovy | 4 +++-
5 files changed, 24 insertions(+), 4 deletions(-)
diff --git a/be/src/exprs/runtime_filter.cpp b/be/src/exprs/runtime_filter.cpp
index d584f06d17..0444d3cc42 100644
--- a/be/src/exprs/runtime_filter.cpp
+++ b/be/src/exprs/runtime_filter.cpp
@@ -1325,8 +1325,12 @@ Status IRuntimeFilter::init_with_desc(const TRuntimeFilterDesc* desc, const TQue
doris::vectorized::VExprContext* bitmap_target_ctx = nullptr;
RETURN_IF_ERROR(doris::vectorized::VExpr::create_expr_tree(_pool, desc->bitmap_target_expr,
&bitmap_target_ctx));
- auto* target_expr = doris::vectorized::VExpr::expr_without_cast(bitmap_target_ctx->root());
- params.column_return_type = const_cast<doris::vectorized::VExpr*>(target_expr)->type().type;
+ auto type = const_cast<vectorized::VExpr*>(
+ vectorized::VExpr::expr_without_cast(bitmap_target_ctx->root()))
+ ->type();
+ // The bitmap filter evaluates only integers.
+ params.column_return_type =
+ type.is_integer_type() ? type.type : bitmap_target_ctx->root()->type().type;
if (desc->__isset.bitmap_filter_not_in) {
params.bitmap_filter_not_in = desc->bitmap_filter_not_in;
diff --git a/be/src/runtime/types.h b/be/src/runtime/types.h
index d4a30b3207..4f4c2ef0fe 100644
--- a/be/src/runtime/types.h
+++ b/be/src/runtime/types.h
@@ -160,6 +160,11 @@ struct TypeDescriptor {
void to_protobuf(PTypeDesc* ptype) const;
+ bool is_integer_type() const {
+ return type == TYPE_TINYINT || type == TYPE_SMALLINT || type == TYPE_INT ||
+ type == TYPE_BIGINT;
+ }
+
bool is_string_type() const {
return type == TYPE_VARCHAR || type == TYPE_CHAR || type == TYPE_HLL ||
type == TYPE_OBJECT || type == TYPE_QUANTILE_STATE || type == TYPE_STRING;
diff --git a/be/src/vec/exec/scan/vscan_node.cpp b/be/src/vec/exec/scan/vscan_node.cpp
index cbd3a9e592..e7953c3703 100644
--- a/be/src/vec/exec/scan/vscan_node.cpp
+++ b/be/src/vec/exec/scan/vscan_node.cpp
@@ -501,7 +501,7 @@ Status VScanNode::_normalize_bloom_filter(VExpr* expr, VExprContext* expr_ctx, S
Status VScanNode::_normalize_bitmap_filter(VExpr* expr, VExprContext* expr_ctx,
SlotDescriptor* slot, PushDownType* pdt) {
- if (TExprNodeType::BITMAP_PRED == expr->node_type()) {
+ if (TExprNodeType::BITMAP_PRED == expr->node_type() && expr->type().is_integer_type()) {
DCHECK(expr->children().size() == 1);
PushDownType temp_pdt = _should_push_down_bitmap_filter();
if (temp_pdt != PushDownType::UNACCEPTABLE) {
diff --git a/regression-test/data/query_p0/join/test_bitmap_filter.out b/regression-test/data/query_p0/join/test_bitmap_filter.out
index 83bcb3aed9..bf8be1dca2 100644
--- a/regression-test/data/query_p0/join/test_bitmap_filter.out
+++ b/regression-test/data/query_p0/join/test_bitmap_filter.out
@@ -66,3 +66,12 @@
-- !sql10 --
+-- !sql11 --
+1991-08-11
+1991-08-11
+2012-03-14
+2015-04-02
+2015-04-02
+2015-04-02
+2015-04-02
+
diff --git a/regression-test/suites/query_p0/join/test_bitmap_filter.groovy b/regression-test/suites/query_p0/join/test_bitmap_filter.groovy
index 90c0fe5f74..d73de718ac 100644
--- a/regression-test/suites/query_p0/join/test_bitmap_filter.groovy
+++ b/regression-test/suites/query_p0/join/test_bitmap_filter.groovy
@@ -36,7 +36,7 @@ suite("test_bitmap_filter", "query_p0") {
"replication_allocation" = "tag.location.default: 1"
);
"""
- sql """insert into bitmap_table values (1, bitmap_from_string('1, 3, 5, 7, 9, 11, 13, 99'),
+ sql """insert into bitmap_table values (1, bitmap_from_string('1, 3, 5, 7, 9, 11, 13, 99, 19910811, 20150402'),
bitmap_from_string('32767, 1985, 255, 789, 1991')),(2, bitmap_from_string('10, 11, 12, 13, 14'), bitmap_empty());"""
qt_sql1 "select k1, k2 from ${tbl1} where k1 in (select k2 from ${tbl2}) order by k1;"
@@ -59,6 +59,8 @@ suite("test_bitmap_filter", "query_p0") {
qt_sql10 "select k1, k2 from (select 1 k1, 11 k2) t where k1 not in (select k2 from ${tbl2}) order by 1, 2;"
+ qt_sql11 "select k10 from ${tbl1} where cast(k10 as bigint) in (select bitmap_or(k2, to_bitmap(20120314)) from ${tbl2} b) order by 1;"
+
test {
sql "select k1, k2 from ${tbl1} b1 where k1 in (select k2 from ${tbl2} b2 where b1.k2 = b2.k1) order by k1;"
exception "In bitmap does not support correlated subquery"
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 02/15: [Bug](decimalv3) Fix wrong precision of DECIMALV3 (#15302)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 609f8e9882e7b17e9b0e129ae7b4a0fd9224aeef
Author: Gabriel <ga...@gmail.com>
AuthorDate: Fri Dec 23 14:11:08 2022 +0800
[Bug](decimalv3) Fix wrong precision of DECIMALV3 (#15302)
* [Bug](decimalv3) Fix wrong precision of DECIMALV3
* update
---
.../java/org/apache/doris/catalog/ScalarType.java | 10 +++--
.../decimalv3/test_arithmetic_expressions.out | 19 +++++++++
.../decimalv3/test_arithmetic_expressions.groovy | 49 ++++++++++++++++++++++
3 files changed, 75 insertions(+), 3 deletions(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/catalog/ScalarType.java b/fe/fe-core/src/main/java/org/apache/doris/catalog/ScalarType.java
index e97471f73d..3cd713b507 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/catalog/ScalarType.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/catalog/ScalarType.java
@@ -1022,9 +1022,13 @@ public class ScalarType extends Type {
}
if (t1.isDecimalV3() && t2.isDecimalV3()) {
- return ScalarType.createDecimalV3Type(Math.max(t1.decimalPrecision() - t1.decimalScale(),
- t2.decimalPrecision() - t2.decimalScale()) + Math.max(t1.decimalScale(),
- t2.decimalScale()), Math.max(t1.decimalScale(), t2.decimalScale()));
+ ScalarType finalType = ScalarType.createDecimalV3Type(Math.max(t1.decimalPrecision() - t1.decimalScale(),
+ t2.decimalPrecision() - t2.decimalScale()) + Math.max(t1.decimalScale(),
+ t2.decimalScale()), Math.max(t1.decimalScale(), t2.decimalScale()));
+ if (finalType.getPrecision() > MAX_PRECISION) {
+ finalType = ScalarType.createDecimalV3Type(MAX_PRECISION, finalType.getScalarScale());
+ }
+ return finalType;
}
PrimitiveType smallerType =
diff --git a/regression-test/data/datatype_p0/decimalv3/test_arithmetic_expressions.out b/regression-test/data/datatype_p0/decimalv3/test_arithmetic_expressions.out
new file mode 100644
index 0000000000..a8d9df8d8d
--- /dev/null
+++ b/regression-test/data/datatype_p0/decimalv3/test_arithmetic_expressions.out
@@ -0,0 +1,19 @@
+-- This file is automatically generated. You should know what you did if you want to edit this
+-- !select_all --
+1.100000000000000000 1.200000000000000000 1.300000000000000000
+1.200000000000000000 1.200000000000000000 1.300000000000000000
+1.500000000000000000 1.200000000000000000 1.300000000000000000
+
+-- !select --
+1.320000000000000000000000000000000000
+1.440000000000000000000000000000000000
+1.800000000000000000000000000000000000
+
+-- !select --
+1.300000000000000000000000000000000000
+1.300000000000000000000000000000000000
+1.300000000000000000000000000000000000
+1.320000000000000000000000000000000000
+1.440000000000000000000000000000000000
+1.800000000000000000000000000000000000
+
diff --git a/regression-test/suites/datatype_p0/decimalv3/test_arithmetic_expressions.groovy b/regression-test/suites/datatype_p0/decimalv3/test_arithmetic_expressions.groovy
new file mode 100644
index 0000000000..44cd313fa0
--- /dev/null
+++ b/regression-test/suites/datatype_p0/decimalv3/test_arithmetic_expressions.groovy
@@ -0,0 +1,49 @@
+// 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_arithmetic_expressions") {
+
+ def table1 = "test_arithmetic_expressions"
+
+ sql "drop table if exists ${table1}"
+
+ sql """
+ CREATE TABLE IF NOT EXISTS `${table1}` (
+ `k1` decimalv3(38, 18) NULL COMMENT "",
+ `k2` decimalv3(38, 18) NULL COMMENT "",
+ `k3` decimalv3(38, 18) NULL COMMENT ""
+ ) ENGINE=OLAP
+ COMMENT "OLAP"
+ DISTRIBUTED BY HASH(`k1`, `k2`, `k3`) BUCKETS 8
+ PROPERTIES (
+ "replication_allocation" = "tag.location.default: 1",
+ "in_memory" = "false",
+ "storage_format" = "V2"
+ )
+ """
+
+ sql """insert into ${table1} values(1.1,1.2,1.3),
+ (1.2,1.2,1.3),
+ (1.5,1.2,1.3)
+ """
+ qt_select_all "select * from ${table1} order by k1"
+
+ qt_select "select k1 * k2 from ${table1} order by k1"
+ qt_select "select * from (select k1 * k2 from ${table1} union all select k3 from ${table1}) a order by 1"
+
+ sql "drop table if exists ${table1}"
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 12/15: [feature](multi-catalog) support use catalog.db when client connect to the doris server (#15293)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 049c1f09803906ac4cd67d1d8bb463d48f63dadc
Author: xueweizhang <zx...@163.com>
AuthorDate: Sat Dec 24 15:20:00 2022 +0800
[feature](multi-catalog) support use catalog.db when client connect to the doris server (#15293)
---
.../java/org/apache/doris/mysql/MysqlProto.java | 31 +++++++++++++++++++++-
1 file changed, 30 insertions(+), 1 deletion(-)
diff --git a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlProto.java b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlProto.java
index ff2b757892..9dd346ce6e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlProto.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/mysql/MysqlProto.java
@@ -26,6 +26,7 @@ import org.apache.doris.common.DdlException;
import org.apache.doris.common.ErrorCode;
import org.apache.doris.common.ErrorReport;
import org.apache.doris.common.LdapConfig;
+import org.apache.doris.datasource.CatalogIf;
import org.apache.doris.ldap.LdapAuthenticate;
import org.apache.doris.mysql.privilege.PaloAuth;
import org.apache.doris.mysql.privilege.UserResource;
@@ -276,8 +277,36 @@ public class MysqlProto {
// set database
String db = authPacket.getDb();
if (!Strings.isNullOrEmpty(db)) {
+ String catalogName = null;
+ String dbName = null;
+ String[] dbNames = db.split("\\.");
+ if (dbNames.length == 1) {
+ dbName = db;
+ } else if (dbNames.length == 2) {
+ catalogName = dbNames[0];
+ dbName = dbNames[1];
+ } else if (dbNames.length > 2) {
+ context.getState().setError(ErrorCode.ERR_BAD_DB_ERROR, "Only one dot can be in the name: " + db);
+ return false;
+ }
+ String dbFullName = ClusterNamespace.getFullName(context.getClusterName(), dbName);
+
+ // check catalog and db exists
+ if (catalogName != null) {
+ CatalogIf catalogIf = context.getEnv().getCatalogMgr().getCatalogNullable(catalogName);
+ if (catalogIf == null) {
+ context.getState().setError(ErrorCode.ERR_BAD_DB_ERROR, "No match catalog in doris: " + db);
+ return false;
+ }
+ if (catalogIf.getDbNullable(dbName) == null) {
+ context.getState().setError(ErrorCode.ERR_BAD_DB_ERROR, "No match database in doris: " + db);
+ return false;
+ }
+ }
try {
- String dbFullName = ClusterNamespace.getFullName(context.getClusterName(), db);
+ if (catalogName != null) {
+ context.getEnv().changeCatalog(context, catalogName);
+ }
Env.getCurrentEnv().changeDb(context, dbFullName);
} catch (DdlException e) {
context.getState().setError(e.getMysqlErrorCode(), e.getMessage());
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 14/15: [fix](multi-catalog) fix and optimize iceberg v2 reader (#15274)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit f251343fa42878e4e4454723c8d92924eca381ea
Author: Ashin Gau <As...@users.noreply.github.com>
AuthorDate: Sat Dec 24 16:02:07 2022 +0800
[fix](multi-catalog) fix and optimize iceberg v2 reader (#15274)
Fix three bugs when read iceberg v2 tables:
1. The `delete position` in `delete file` represents the position of delete row in the entire file, but the `read range` in
`RowGroupReader` represents the position in current row group. Therefore, we need to subtract the position of first
row of current row group from `delete position`.
2. When only reading the partition columns, `RowGroupReader` skips processing the `delete position`.
3. If the `delete position` has delete all rows in a row group, the `read range` is empty, but we read the whole row
group in such case.
Optimize four performance issues:
1. We change `delete position` to `delete range`, and then merge `delete range` and `read range` into the final read
ranges. This process is too tedious and time-consuming. . we can merge `delete position` and `read range` directly.
2. `delete position` is ordered in a `delete file`, so we can use merge-sort, instead of ordered-set.
3. Initialize `RowGroupReader` when reading, instead of initialize all row groups when opening a `ParquetReader`, to
save memory usage, and the same as `IcebergReader`.
4. Change the recursive call of `_do_lazy_read` to loop logic.
---
be/src/service/internal_service.cpp | 2 +-
be/src/vec/exec/format/csv/csv_reader.cpp | 4 +-
be/src/vec/exec/format/csv/csv_reader.h | 4 +-
be/src/vec/exec/format/generic_reader.h | 6 +-
be/src/vec/exec/format/json/new_json_reader.cpp | 4 +-
be/src/vec/exec/format/json/new_json_reader.h | 4 +-
be/src/vec/exec/format/orc/vorc_reader.cpp | 4 +-
be/src/vec/exec/format/orc/vorc_reader.h | 4 +-
be/src/vec/exec/format/parquet/parquet_common.h | 2 +-
.../exec/format/parquet/vparquet_column_reader.cpp | 30 +--
.../exec/format/parquet/vparquet_column_reader.h | 19 +-
.../exec/format/parquet/vparquet_group_reader.cpp | 199 +++++++++-----
.../exec/format/parquet/vparquet_group_reader.h | 83 ++++--
be/src/vec/exec/format/parquet/vparquet_reader.cpp | 243 +++++++----------
be/src/vec/exec/format/parquet/vparquet_reader.h | 30 ++-
be/src/vec/exec/format/table/iceberg_reader.cpp | 293 ++++++++++-----------
be/src/vec/exec/format/table/iceberg_reader.h | 30 +--
be/src/vec/exec/format/table/table_format_reader.h | 5 +-
be/test/vec/exec/parquet/parquet_thrift_test.cpp | 12 +-
19 files changed, 512 insertions(+), 466 deletions(-)
diff --git a/be/src/service/internal_service.cpp b/be/src/service/internal_service.cpp
index 28b9b8f4ca..381782af85 100644
--- a/be/src/service/internal_service.cpp
+++ b/be/src/service/internal_service.cpp
@@ -478,7 +478,7 @@ void PInternalServiceImpl::fetch_table_schema(google::protobuf::RpcController* c
}
std::vector<std::string> col_names;
std::vector<TypeDescriptor> col_types;
- st = reader->get_parsered_schema(&col_names, &col_types);
+ st = reader->get_parsed_schema(&col_names, &col_types);
if (!st.ok()) {
LOG(WARNING) << "fetch table schema failed, errmsg=" << st.get_error_msg();
st.to_protobuf(result->mutable_status());
diff --git a/be/src/vec/exec/format/csv/csv_reader.cpp b/be/src/vec/exec/format/csv/csv_reader.cpp
index 2e710ca44f..48f8d84dc7 100644
--- a/be/src/vec/exec/format/csv/csv_reader.cpp
+++ b/be/src/vec/exec/format/csv/csv_reader.cpp
@@ -202,8 +202,8 @@ Status CsvReader::get_columns(std::unordered_map<std::string, TypeDescriptor>* n
return Status::OK();
}
-Status CsvReader::get_parsered_schema(std::vector<std::string>* col_names,
- std::vector<TypeDescriptor>* col_types) {
+Status CsvReader::get_parsed_schema(std::vector<std::string>* col_names,
+ std::vector<TypeDescriptor>* col_types) {
size_t read_line = 0;
bool is_parse_name = false;
RETURN_IF_ERROR(_prepare_parse(&read_line, &is_parse_name));
diff --git a/be/src/vec/exec/format/csv/csv_reader.h b/be/src/vec/exec/format/csv/csv_reader.h
index 5083c00d2d..5bb14523e3 100644
--- a/be/src/vec/exec/format/csv/csv_reader.h
+++ b/be/src/vec/exec/format/csv/csv_reader.h
@@ -49,8 +49,8 @@ public:
// 1. header_type is empty, get schema from first line.
// 2. header_type is CSV_WITH_NAMES, get schema from first line.
// 3. header_type is CSV_WITH_NAMES_AND_TYPES, get schema from first two line.
- Status get_parsered_schema(std::vector<std::string>* col_names,
- std::vector<TypeDescriptor>* col_types) override;
+ Status get_parsed_schema(std::vector<std::string>* col_names,
+ std::vector<TypeDescriptor>* col_types) override;
private:
// used for stream/broker load of csv file.
diff --git a/be/src/vec/exec/format/generic_reader.h b/be/src/vec/exec/format/generic_reader.h
index dd2bdd249c..30e93aacd8 100644
--- a/be/src/vec/exec/format/generic_reader.h
+++ b/be/src/vec/exec/format/generic_reader.h
@@ -39,9 +39,9 @@ public:
return Status::NotSupported("get_columns is not implemented");
}
- virtual Status get_parsered_schema(std::vector<std::string>* col_names,
- std::vector<TypeDescriptor>* col_types) {
- return Status::NotSupported("get_parser_schema is not implemented for this reader.");
+ virtual Status get_parsed_schema(std::vector<std::string>* col_names,
+ std::vector<TypeDescriptor>* col_types) {
+ return Status::NotSupported("get_parsed_schema is not implemented for this reader.");
}
virtual ~GenericReader() = default;
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 2228f9701d..68a3f089e5 100644
--- a/be/src/vec/exec/format/json/new_json_reader.cpp
+++ b/be/src/vec/exec/format/json/new_json_reader.cpp
@@ -139,8 +139,8 @@ Status NewJsonReader::get_columns(std::unordered_map<std::string, TypeDescriptor
return Status::OK();
}
-Status NewJsonReader::get_parsered_schema(std::vector<std::string>* col_names,
- std::vector<TypeDescriptor>* col_types) {
+Status NewJsonReader::get_parsed_schema(std::vector<std::string>* col_names,
+ std::vector<TypeDescriptor>* col_types) {
RETURN_IF_ERROR(_get_range_params());
RETURN_IF_ERROR(_open_file_reader());
diff --git a/be/src/vec/exec/format/json/new_json_reader.h b/be/src/vec/exec/format/json/new_json_reader.h
index 6b003c30fe..5a057d32fe 100644
--- a/be/src/vec/exec/format/json/new_json_reader.h
+++ b/be/src/vec/exec/format/json/new_json_reader.h
@@ -48,8 +48,8 @@ public:
Status get_next_block(Block* block, size_t* read_rows, bool* eof) override;
Status get_columns(std::unordered_map<std::string, TypeDescriptor>* name_to_type,
std::unordered_set<std::string>* missing_cols) override;
- Status get_parsered_schema(std::vector<std::string>* col_names,
- std::vector<TypeDescriptor>* col_types) override;
+ Status get_parsed_schema(std::vector<std::string>* col_names,
+ std::vector<TypeDescriptor>* col_types) override;
private:
Status _get_range_params();
diff --git a/be/src/vec/exec/format/orc/vorc_reader.cpp b/be/src/vec/exec/format/orc/vorc_reader.cpp
index 4998bc2486..2748160e97 100644
--- a/be/src/vec/exec/format/orc/vorc_reader.cpp
+++ b/be/src/vec/exec/format/orc/vorc_reader.cpp
@@ -176,8 +176,8 @@ Status OrcReader::init_reader(
return Status::OK();
}
-Status OrcReader::get_parsered_schema(std::vector<std::string>* col_names,
- std::vector<TypeDescriptor>* col_types) {
+Status OrcReader::get_parsed_schema(std::vector<std::string>* col_names,
+ std::vector<TypeDescriptor>* col_types) {
if (_file_reader == nullptr) {
std::unique_ptr<FileReader> inner_reader;
RETURN_IF_ERROR(FileFactory::create_file_reader(_profile, _scan_params, _scan_range.path,
diff --git a/be/src/vec/exec/format/orc/vorc_reader.h b/be/src/vec/exec/format/orc/vorc_reader.h
index 9b0e4b6a44..98c2fc7c02 100644
--- a/be/src/vec/exec/format/orc/vorc_reader.h
+++ b/be/src/vec/exec/format/orc/vorc_reader.h
@@ -99,8 +99,8 @@ public:
Status get_columns(std::unordered_map<std::string, TypeDescriptor>* name_to_type,
std::unordered_set<std::string>* missing_cols) override;
- Status get_parsered_schema(std::vector<std::string>* col_names,
- std::vector<TypeDescriptor>* col_types) override;
+ Status get_parsed_schema(std::vector<std::string>* col_names,
+ std::vector<TypeDescriptor>* col_types) override;
private:
struct OrcProfile {
diff --git a/be/src/vec/exec/format/parquet/parquet_common.h b/be/src/vec/exec/format/parquet/parquet_common.h
index 611ce969f1..112be6a223 100644
--- a/be/src/vec/exec/format/parquet/parquet_common.h
+++ b/be/src/vec/exec/format/parquet/parquet_common.h
@@ -41,7 +41,7 @@ namespace doris::vectorized {
using level_t = int16_t;
struct RowRange {
- RowRange() {}
+ RowRange() = default;
RowRange(int64_t first, int64_t last) : first_row(first), last_row(last) {}
int64_t first_row;
diff --git a/be/src/vec/exec/format/parquet/vparquet_column_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_column_reader.cpp
index c0557c482a..75f43b4730 100644
--- a/be/src/vec/exec/format/parquet/vparquet_column_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_column_reader.cpp
@@ -19,7 +19,6 @@
#include <common/status.h>
#include <gen_cpp/parquet_types.h>
-#include <vec/columns/columns_number.h>
#include "schema_desc.h"
#include "vec/data_types/data_type_array.h"
@@ -28,7 +27,9 @@
namespace doris::vectorized {
Status ParquetColumnReader::create(FileReader* file, FieldSchema* field,
- const tparquet::RowGroup& row_group, cctz::time_zone* ctz,
+ const ParquetReadColumn& column,
+ const tparquet::RowGroup& row_group,
+ const std::vector<RowRange>& row_ranges, cctz::time_zone* ctz,
std::unique_ptr<ParquetColumnReader>& reader,
size_t max_buf_size) {
if (field->type.type == TYPE_MAP || field->type.type == TYPE_STRUCT) {
@@ -36,13 +37,13 @@ Status ParquetColumnReader::create(FileReader* file, FieldSchema* field,
}
if (field->type.type == TYPE_ARRAY) {
tparquet::ColumnChunk chunk = row_group.columns[field->children[0].physical_column_index];
- ArrayColumnReader* array_reader = new ArrayColumnReader(ctz);
+ ArrayColumnReader* array_reader = new ArrayColumnReader(row_ranges, ctz);
array_reader->init_column_metadata(chunk);
RETURN_IF_ERROR(array_reader->init(file, field, &chunk, max_buf_size));
reader.reset(array_reader);
} else {
tparquet::ColumnChunk chunk = row_group.columns[field->physical_column_index];
- ScalarColumnReader* scalar_reader = new ScalarColumnReader(ctz);
+ ScalarColumnReader* scalar_reader = new ScalarColumnReader(row_ranges, ctz);
scalar_reader->init_column_metadata(chunk);
RETURN_IF_ERROR(scalar_reader->init(file, field, &chunk, max_buf_size));
reader.reset(scalar_reader);
@@ -61,23 +62,19 @@ void ParquetColumnReader::init_column_metadata(const tparquet::ColumnChunk& chun
void ParquetColumnReader::_generate_read_ranges(int64_t start_index, int64_t end_index,
std::list<RowRange>& read_ranges) {
- if (_row_ranges->empty()) {
- read_ranges.emplace_back(start_index, end_index);
- return;
- }
int index = _row_range_index;
- while (index < _row_ranges->size()) {
- const RowRange& row_range = (*_row_ranges)[index];
- if (row_range.last_row <= start_index) {
+ while (index < _row_ranges.size()) {
+ const RowRange& read_range = _row_ranges[index];
+ if (read_range.last_row <= start_index) {
index++;
_row_range_index++;
continue;
}
- if (row_range.first_row >= end_index) {
+ if (read_range.first_row >= end_index) {
break;
}
- int64_t start = row_range.first_row < start_index ? start_index : row_range.first_row;
- int64_t end = row_range.last_row < end_index ? row_range.last_row : end_index;
+ int64_t start = read_range.first_row < start_index ? start_index : read_range.first_row;
+ int64_t end = read_range.last_row < end_index ? read_range.last_row : end_index;
read_ranges.emplace_back(start, end);
index++;
}
@@ -216,7 +213,7 @@ Status ScalarColumnReader::read_column_data(ColumnPtr& doris_column, DataTypePtr
// lazy read
size_t remaining_num_values = 0;
for (auto& range : read_ranges) {
- remaining_num_values = range.last_row - range.first_row;
+ remaining_num_values += range.last_row - range.first_row;
}
if (batch_size >= remaining_num_values &&
select_vector.can_filter_all(remaining_num_values)) {
@@ -330,6 +327,7 @@ Status ArrayColumnReader::read_column_data(ColumnPtr& doris_column, DataTypePtr&
data_column = doris_column->assume_mutable();
}
+ // generate array offset
size_t real_batch_size = 0;
size_t num_values = 0;
std::vector<size_t> element_offsets;
@@ -339,7 +337,7 @@ Status ArrayColumnReader::read_column_data(ColumnPtr& doris_column, DataTypePtr&
level_t* definitions = _def_levels_buf.get();
_chunk_reader->get_def_levels(definitions, num_values);
_def_offset = 0;
- // read_range delete_row_range
+
// generate the row ranges that should be read
std::list<RowRange> read_ranges;
_generate_read_ranges(_current_row_index, _current_row_index + real_batch_size, read_ranges);
diff --git a/be/src/vec/exec/format/parquet/vparquet_column_reader.h b/be/src/vec/exec/format/parquet/vparquet_column_reader.h
index b7d062e7c6..de0ec185b9 100644
--- a/be/src/vec/exec/format/parquet/vparquet_column_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_column_reader.h
@@ -96,7 +96,8 @@ public:
}
};
- ParquetColumnReader(cctz::time_zone* ctz) : _ctz(ctz) {};
+ ParquetColumnReader(const std::vector<RowRange>& row_ranges, cctz::time_zone* ctz)
+ : _row_ranges(row_ranges), _ctz(ctz) {};
virtual ~ParquetColumnReader() {
if (_stream_reader != nullptr) {
delete _stream_reader;
@@ -106,12 +107,12 @@ public:
virtual Status read_column_data(ColumnPtr& doris_column, DataTypePtr& type,
ColumnSelectVector& select_vector, size_t batch_size,
size_t* read_rows, bool* eof) = 0;
- static Status create(FileReader* file, FieldSchema* field, const tparquet::RowGroup& row_group,
- cctz::time_zone* ctz, std::unique_ptr<ParquetColumnReader>& reader,
- size_t max_buf_size);
+ static Status create(FileReader* file, FieldSchema* field, const ParquetReadColumn& column,
+ const tparquet::RowGroup& row_group,
+ const std::vector<RowRange>& row_ranges, cctz::time_zone* ctz,
+ std::unique_ptr<ParquetColumnReader>& reader, size_t max_buf_size);
void init_column_metadata(const tparquet::ColumnChunk& chunk);
void add_offset_index(tparquet::OffsetIndex* offset_index) { _offset_index = offset_index; }
- void set_row_ranges(const std::vector<RowRange>* row_ranges) { _row_ranges = row_ranges; };
Statistics statistics() {
return Statistics(_stream_reader->statistics(), _chunk_reader->statistics(),
_decode_null_map_time);
@@ -124,7 +125,7 @@ protected:
BufferedFileStreamReader* _stream_reader;
std::unique_ptr<ParquetColumnMetadata> _metadata;
- const std::vector<RowRange>* _row_ranges;
+ const std::vector<RowRange>& _row_ranges;
cctz::time_zone* _ctz;
std::unique_ptr<ColumnChunkReader> _chunk_reader;
tparquet::OffsetIndex* _offset_index;
@@ -135,7 +136,8 @@ protected:
class ScalarColumnReader : public ParquetColumnReader {
public:
- ScalarColumnReader(cctz::time_zone* ctz) : ParquetColumnReader(ctz) {};
+ ScalarColumnReader(const std::vector<RowRange>& row_ranges, cctz::time_zone* ctz)
+ : ParquetColumnReader(row_ranges, ctz) {};
~ScalarColumnReader() override { close(); };
Status init(FileReader* file, FieldSchema* field, tparquet::ColumnChunk* chunk,
size_t max_buf_size);
@@ -150,7 +152,8 @@ public:
class ArrayColumnReader : public ParquetColumnReader {
public:
- ArrayColumnReader(cctz::time_zone* ctz) : ParquetColumnReader(ctz) {};
+ ArrayColumnReader(const std::vector<RowRange>& row_ranges, cctz::time_zone* ctz)
+ : ParquetColumnReader(row_ranges, ctz) {};
~ArrayColumnReader() override { close(); };
Status init(FileReader* file, FieldSchema* field, tparquet::ColumnChunk* chunk,
size_t max_buf_size);
diff --git a/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp
index 42baf3a283..24441896a9 100644
--- a/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_group_reader.cpp
@@ -23,26 +23,31 @@
namespace doris::vectorized {
+const std::vector<int64_t> RowGroupReader::NO_DELETE = {};
+
RowGroupReader::RowGroupReader(doris::FileReader* file_reader,
const std::vector<ParquetReadColumn>& read_columns,
- const RowGroupIndex& row_group_idx,
- const tparquet::RowGroup& row_group, cctz::time_zone* ctz,
+ const int32_t row_group_id, const tparquet::RowGroup& row_group,
+ cctz::time_zone* ctz,
+ const PositionDeleteContext& position_delete_ctx,
const LazyReadContext& lazy_read_ctx)
: _file_reader(file_reader),
_read_columns(read_columns),
- _row_group_idx(row_group_idx),
+ _row_group_id(row_group_id),
_row_group_meta(row_group),
_remaining_rows(row_group.num_rows),
_ctz(ctz),
+ _position_delete_ctx(position_delete_ctx),
_lazy_read_ctx(lazy_read_ctx) {}
RowGroupReader::~RowGroupReader() {
_column_readers.clear();
}
-Status RowGroupReader::init(const FieldDescriptor& schema,
+Status RowGroupReader::init(const FieldDescriptor& schema, std::vector<RowRange>& row_ranges,
std::unordered_map<int, tparquet::OffsetIndex>& col_offsets) {
- if (_read_columns.size() == 0) {
+ _merge_read_ranges(row_ranges);
+ if (_read_columns.empty()) {
// Query task that only select columns in path.
return Status::OK();
}
@@ -52,15 +57,15 @@ Status RowGroupReader::init(const FieldDescriptor& schema,
for (auto& read_col : _read_columns) {
auto field = const_cast<FieldSchema*>(schema.get_column(read_col._file_slot_name));
std::unique_ptr<ParquetColumnReader> reader;
- RETURN_IF_ERROR(ParquetColumnReader::create(_file_reader, field, _row_group_meta, _ctz,
- reader, max_buf_size));
+ RETURN_IF_ERROR(ParquetColumnReader::create(_file_reader, field, read_col, _row_group_meta,
+ _read_ranges, _ctz, reader, max_buf_size));
auto col_iter = col_offsets.find(read_col._parquet_col_id);
if (col_iter != col_offsets.end()) {
tparquet::OffsetIndex oi = col_iter->second;
reader->add_offset_index(&oi);
}
if (reader == nullptr) {
- VLOG_DEBUG << "Init row group(" << _row_group_idx.row_group_id << ") reader failed";
+ VLOG_DEBUG << "Init row group(" << _row_group_id << ") reader failed";
return Status::Corruption("Init row group reader failed");
}
_column_readers[read_col._file_slot_name] = std::move(reader);
@@ -68,17 +73,11 @@ Status RowGroupReader::init(const FieldDescriptor& schema,
return Status::OK();
}
-void RowGroupReader::set_row_ranges(const std::vector<doris::vectorized::RowRange>& row_ranges) {
- for (auto& read_col : _read_columns) {
- _column_readers[read_col._file_slot_name]->set_row_ranges(&row_ranges);
- }
-}
-
Status RowGroupReader::next_batch(Block* block, size_t batch_size, size_t* read_rows,
- bool* _batch_eof) {
+ bool* batch_eof) {
// Process external table query task that select columns are all from path.
if (_read_columns.empty()) {
- RETURN_IF_ERROR(_read_empty_batch(batch_size, read_rows, _batch_eof));
+ RETURN_IF_ERROR(_read_empty_batch(batch_size, read_rows, batch_eof));
RETURN_IF_ERROR(
_fill_partition_columns(block, *read_rows, _lazy_read_ctx.partition_columns));
RETURN_IF_ERROR(_fill_missing_columns(block, *read_rows, _lazy_read_ctx.missing_columns));
@@ -90,11 +89,11 @@ Status RowGroupReader::next_batch(Block* block, size_t batch_size, size_t* read_
}
if (_lazy_read_ctx.can_lazy_read) {
// call _do_lazy_read recursively when current batch is skipped
- return _do_lazy_read(block, batch_size, read_rows, _batch_eof);
+ return _do_lazy_read(block, batch_size, read_rows, batch_eof);
} else {
ColumnSelectVector run_length_vector;
RETURN_IF_ERROR(_read_column_data(block, _lazy_read_ctx.all_read_columns, batch_size,
- read_rows, _batch_eof, run_length_vector));
+ read_rows, batch_eof, run_length_vector));
RETURN_IF_ERROR(
_fill_partition_columns(block, *read_rows, _lazy_read_ctx.partition_columns));
RETURN_IF_ERROR(_fill_missing_columns(block, *read_rows, _lazy_read_ctx.missing_columns));
@@ -106,8 +105,39 @@ Status RowGroupReader::next_batch(Block* block, size_t batch_size, size_t* read_
}
}
+void RowGroupReader::_merge_read_ranges(std::vector<RowRange>& row_ranges) {
+ // row_ranges is generated from page index, and the row index begins with 0 in each row group.
+ // _position_delete_ctx is generated from delete file, and the row index begins with 0 in parquet file
+ for (auto& range : row_ranges) {
+ int64_t start_row_id = range.first_row;
+ while (_position_delete_ctx.index < _position_delete_ctx.end_index) {
+ const int64_t& delete_row_id =
+ _position_delete_ctx.delete_rows[_position_delete_ctx.index] -
+ _position_delete_ctx.first_row_id;
+ if (delete_row_id < range.first_row) {
+ _position_delete_ctx.index++;
+ } else if (delete_row_id < range.last_row) {
+ if (start_row_id < delete_row_id) {
+ _read_ranges.emplace_back(start_row_id, delete_row_id);
+ }
+ start_row_id = delete_row_id + 1;
+ _position_delete_ctx.index++;
+ } else { // delete_row_id >= range.last_row
+ if (start_row_id < range.last_row) {
+ _read_ranges.emplace_back(start_row_id, range.last_row);
+ start_row_id = range.last_row + 1;
+ }
+ break;
+ }
+ }
+ if (start_row_id < range.last_row) {
+ _read_ranges.emplace_back(start_row_id, range.last_row);
+ }
+ }
+}
+
Status RowGroupReader::_read_column_data(Block* block, const std::vector<std::string>& columns,
- size_t batch_size, size_t* read_rows, bool* _batch_eof,
+ size_t batch_size, size_t* read_rows, bool* batch_eof,
ColumnSelectVector& select_vector) {
size_t batch_read_rows = 0;
bool has_eof = false;
@@ -138,57 +168,74 @@ Status RowGroupReader::_read_column_data(Block* block, const std::vector<std::st
col_idx++;
}
*read_rows = batch_read_rows;
- *_batch_eof = has_eof;
+ *batch_eof = has_eof;
return Status::OK();
}
Status RowGroupReader::_do_lazy_read(Block* block, size_t batch_size, size_t* read_rows,
bool* batch_eof) {
- // read predicate columns
+ std::unique_ptr<ColumnSelectVector> select_vector_ptr = nullptr;
size_t pre_read_rows;
bool pre_eof;
- ColumnSelectVector run_length_vector;
- RETURN_IF_ERROR(_read_column_data(block, _lazy_read_ctx.predicate_columns, batch_size,
- &pre_read_rows, &pre_eof, run_length_vector));
- RETURN_IF_ERROR(_fill_partition_columns(block, pre_read_rows,
- _lazy_read_ctx.predicate_partition_columns));
- RETURN_IF_ERROR(
- _fill_missing_columns(block, pre_read_rows, _lazy_read_ctx.predicate_missing_columns));
- // generate filter vector
- if (_lazy_read_ctx.resize_first_column) {
- // VExprContext.execute has an optimization, the filtering is executed when block->rows() > 0
- // The following process may be tricky and time-consuming, but we have no other way.
- block->get_by_position(0).column->assume_mutable()->resize(pre_read_rows);
- }
size_t origin_column_num = block->columns();
int filter_column_id = -1;
- RETURN_IF_ERROR(_lazy_read_ctx.vconjunct_ctx->execute(block, &filter_column_id));
- ColumnPtr& sv = block->get_by_position(filter_column_id).column;
- if (_lazy_read_ctx.resize_first_column) {
- // We have to clean the first column to insert right data.
- block->get_by_position(0).column->assume_mutable()->clear();
- }
-
- // build filter map
- bool can_filter_all = false;
- const uint8_t* filter_map = _build_filter_map(sv, pre_read_rows, &can_filter_all);
- ColumnSelectVector select_vector(filter_map, pre_read_rows, can_filter_all);
- if (select_vector.filter_all() && !pre_eof) {
- // If continuous batches are skipped, we can cache them to skip a whole page
- _cached_filtered_rows += pre_read_rows;
- for (auto& col : _lazy_read_ctx.predicate_columns) {
- // clean block to read predicate columns
- block->get_by_name(col).column->assume_mutable()->clear();
+ while (true) {
+ // read predicate columns
+ pre_read_rows = 0;
+ pre_eof = false;
+ ColumnSelectVector run_length_vector;
+ RETURN_IF_ERROR(_read_column_data(block, _lazy_read_ctx.predicate_columns, batch_size,
+ &pre_read_rows, &pre_eof, run_length_vector));
+ if (pre_read_rows == 0) {
+ DCHECK_EQ(pre_eof, true);
+ break;
}
- for (auto& col : _lazy_read_ctx.predicate_partition_columns) {
- block->get_by_name(col.first).column->assume_mutable()->clear();
+ RETURN_IF_ERROR(_fill_partition_columns(block, pre_read_rows,
+ _lazy_read_ctx.predicate_partition_columns));
+ RETURN_IF_ERROR(_fill_missing_columns(block, pre_read_rows,
+ _lazy_read_ctx.predicate_missing_columns));
+ // generate filter vector
+ if (_lazy_read_ctx.resize_first_column) {
+ // VExprContext.execute has an optimization, the filtering is executed when block->rows() > 0
+ // The following process may be tricky and time-consuming, but we have no other way.
+ block->get_by_position(0).column->assume_mutable()->resize(pre_read_rows);
}
- for (auto& col : _lazy_read_ctx.predicate_missing_columns) {
- block->get_by_name(col.first).column->assume_mutable()->clear();
+ RETURN_IF_ERROR(_lazy_read_ctx.vconjunct_ctx->execute(block, &filter_column_id));
+ ColumnPtr& sv = block->get_by_position(filter_column_id).column;
+ if (_lazy_read_ctx.resize_first_column) {
+ // We have to clean the first column to insert right data.
+ block->get_by_position(0).column->assume_mutable()->clear();
}
- Block::erase_useless_column(block, origin_column_num);
- return _do_lazy_read(block, batch_size, read_rows, batch_eof);
+
+ // build filter map
+ bool can_filter_all = false;
+ const uint8_t* filter_map = _build_filter_map(sv, pre_read_rows, &can_filter_all);
+ select_vector_ptr.reset(new ColumnSelectVector(filter_map, pre_read_rows, can_filter_all));
+ if (select_vector_ptr->filter_all() && !pre_eof) {
+ // If continuous batches are skipped, we can cache them to skip a whole page
+ _cached_filtered_rows += pre_read_rows;
+ for (auto& col : _lazy_read_ctx.predicate_columns) {
+ // clean block to read predicate columns
+ block->get_by_name(col).column->assume_mutable()->clear();
+ }
+ for (auto& col : _lazy_read_ctx.predicate_partition_columns) {
+ block->get_by_name(col.first).column->assume_mutable()->clear();
+ }
+ for (auto& col : _lazy_read_ctx.predicate_missing_columns) {
+ block->get_by_name(col.first).column->assume_mutable()->clear();
+ }
+ Block::erase_useless_column(block, origin_column_num);
+ } else {
+ break;
+ }
+ }
+ if (select_vector_ptr == nullptr) {
+ DCHECK_EQ(pre_read_rows + _cached_filtered_rows, 0);
+ *read_rows = 0;
+ *batch_eof = true;
}
+
+ ColumnSelectVector& select_vector = *select_vector_ptr;
std::unique_ptr<uint8_t[]> rebuild_filter_map = nullptr;
if (_cached_filtered_rows != 0) {
_rebuild_select_vector(select_vector, rebuild_filter_map, pre_read_rows);
@@ -345,15 +392,37 @@ Status RowGroupReader::_fill_missing_columns(
return Status::OK();
}
-Status RowGroupReader::_read_empty_batch(size_t batch_size, size_t* read_rows, bool* _batch_eof) {
- if (batch_size < _remaining_rows) {
- *read_rows = batch_size;
- _remaining_rows -= batch_size;
- *_batch_eof = false;
+Status RowGroupReader::_read_empty_batch(size_t batch_size, size_t* read_rows, bool* batch_eof) {
+ if (_position_delete_ctx.has_filter) {
+ int64_t start_row_id = _position_delete_ctx.current_row_id;
+ int64_t end_row_id = std::min(_position_delete_ctx.current_row_id + (int64_t)batch_size,
+ _position_delete_ctx.last_row_id);
+ int64_t num_delete_rows = 0;
+ while (_position_delete_ctx.index < _position_delete_ctx.end_index) {
+ const int64_t& delete_row_id =
+ _position_delete_ctx.delete_rows[_position_delete_ctx.index];
+ if (delete_row_id < start_row_id) {
+ _position_delete_ctx.index++;
+ } else if (delete_row_id < end_row_id) {
+ num_delete_rows++;
+ _position_delete_ctx.index++;
+ } else { // delete_row_id >= end_row_id
+ break;
+ }
+ }
+ *read_rows = end_row_id - start_row_id - num_delete_rows;
+ _position_delete_ctx.current_row_id = end_row_id;
+ *batch_eof = _position_delete_ctx.current_row_id == _position_delete_ctx.last_row_id;
} else {
- *read_rows = _remaining_rows;
- _remaining_rows = 0;
- *_batch_eof = true;
+ if (batch_size < _remaining_rows) {
+ *read_rows = batch_size;
+ _remaining_rows -= batch_size;
+ *batch_eof = false;
+ } else {
+ *read_rows = _remaining_rows;
+ _remaining_rows = 0;
+ *batch_eof = true;
+ }
}
return Status::OK();
}
diff --git a/be/src/vec/exec/format/parquet/vparquet_group_reader.h b/be/src/vec/exec/format/parquet/vparquet_group_reader.h
index ef73eba21c..53e8a052c0 100644
--- a/be/src/vec/exec/format/parquet/vparquet_group_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_group_reader.h
@@ -25,16 +25,18 @@
namespace doris::vectorized {
-struct RowGroupIndex {
- int32_t row_group_id;
- int32_t first_row;
- int32_t last_row;
- RowGroupIndex(int32_t id, int32_t first, int32_t last)
- : row_group_id(id), first_row(first), last_row(last) {}
-};
-
class RowGroupReader {
public:
+ static const std::vector<int64_t> NO_DELETE;
+
+ struct RowGroupIndex {
+ int32_t row_group_id;
+ int64_t first_row;
+ int64_t last_row;
+ RowGroupIndex(int32_t id, int64_t first, int64_t last)
+ : row_group_id(id), first_row(first), last_row(last) {}
+ };
+
struct LazyReadContext {
VExprContext* vconjunct_ctx = nullptr;
bool can_lazy_read = false;
@@ -56,26 +58,66 @@ public:
std::unordered_map<std::string, VExprContext*> missing_columns;
};
+ /**
+ * Support row-level delete in iceberg:
+ * https://iceberg.apache.org/spec/#position-delete-files
+ */
+ struct PositionDeleteContext {
+ // the filtered rows in current row group
+ const std::vector<int64_t>& delete_rows;
+ // the first row id of current row group in parquet file
+ const int64_t first_row_id;
+ // the number of rows in current row group
+ const int64_t num_rows;
+ const int64_t last_row_id;
+ // current row id to read in the row group
+ int64_t current_row_id;
+ // start index in delete_rows
+ const int64_t start_index;
+ // end index in delete_rows
+ const int64_t end_index;
+ // current index in delete_rows
+ int64_t index;
+ const bool has_filter;
+
+ PositionDeleteContext(const std::vector<int64_t>& delete_rows, const int64_t num_rows,
+ const int64_t first_row_id, const int64_t start_index,
+ const int64_t end_index)
+ : delete_rows(delete_rows),
+ first_row_id(first_row_id),
+ num_rows(num_rows),
+ last_row_id(first_row_id + num_rows),
+ current_row_id(first_row_id),
+ start_index(start_index),
+ end_index(end_index),
+ index(start_index),
+ has_filter(end_index > start_index) {}
+
+ PositionDeleteContext(const int64_t num_rows, const int64_t first_row)
+ : PositionDeleteContext(NO_DELETE, num_rows, first_row, 0, 0) {}
+
+ PositionDeleteContext(const PositionDeleteContext& filter) = default;
+ };
+
RowGroupReader(doris::FileReader* file_reader,
- const std::vector<ParquetReadColumn>& read_columns,
- const RowGroupIndex& _row_group_idx, const tparquet::RowGroup& row_group,
- cctz::time_zone* ctz, const LazyReadContext& lazy_read_ctx);
+ const std::vector<ParquetReadColumn>& read_columns, const int32_t row_group_id,
+ const tparquet::RowGroup& row_group, cctz::time_zone* ctz,
+ const PositionDeleteContext& position_delete_ctx,
+ const LazyReadContext& lazy_read_ctx);
~RowGroupReader();
- Status init(const FieldDescriptor& schema,
+ Status init(const FieldDescriptor& schema, std::vector<RowRange>& row_ranges,
std::unordered_map<int, tparquet::OffsetIndex>& col_offsets);
- Status next_batch(Block* block, size_t batch_size, size_t* read_rows, bool* _batch_eof);
- int64_t lazy_read_filtered_rows() { return _lazy_read_filtered_rows; }
- const RowGroupIndex& index() { return _row_group_idx; }
- void set_row_ranges(const std::vector<doris::vectorized::RowRange>& row_ranges);
+ Status next_batch(Block* block, size_t batch_size, size_t* read_rows, bool* batch_eof);
int64_t lazy_read_filtered_rows() const { return _lazy_read_filtered_rows; }
ParquetColumnReader::Statistics statistics();
private:
- Status _read_empty_batch(size_t batch_size, size_t* read_rows, bool* _batch_eof);
+ void _merge_read_ranges(std::vector<RowRange>& row_ranges);
+ Status _read_empty_batch(size_t batch_size, size_t* read_rows, bool* batch_eof);
Status _read_column_data(Block* block, const std::vector<std::string>& columns,
- size_t batch_size, size_t* read_rows, bool* _batch_eof,
+ size_t batch_size, size_t* read_rows, bool* batch_eof,
ColumnSelectVector& select_vector);
Status _do_lazy_read(Block* block, size_t batch_size, size_t* read_rows, bool* batch_eof);
const uint8_t* _build_filter_map(ColumnPtr& sv, size_t num_rows, bool* can_filter_all);
@@ -92,10 +134,13 @@ private:
doris::FileReader* _file_reader;
std::unordered_map<std::string, std::unique_ptr<ParquetColumnReader>> _column_readers;
const std::vector<ParquetReadColumn>& _read_columns;
- const RowGroupIndex& _row_group_idx;
+ const int32_t _row_group_id;
const tparquet::RowGroup& _row_group_meta;
int64_t _remaining_rows;
cctz::time_zone* _ctz;
+ PositionDeleteContext _position_delete_ctx;
+ // merge the row ranges generated from page index and position delete.
+ std::vector<RowRange> _read_ranges;
const LazyReadContext& _lazy_read_ctx;
int64_t _lazy_read_filtered_rows = 0;
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.cpp b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
index c8d4f6b3c4..6613b1f987 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.cpp
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.cpp
@@ -19,6 +19,7 @@
#include <algorithm>
+#include "common/status.h"
#include "io/file_factory.h"
#include "parquet_pred_cmp.h"
#include "parquet_thrift_util.h"
@@ -150,24 +151,12 @@ Status ParquetReader::_open_file() {
return Status::OK();
}
-Status ParquetReader::file_metadata(FileMetaData** metadata) {
- Status open_status = _open_file();
- if (!open_status.ok()) {
- return open_status;
- }
- *metadata = _file_metadata.get();
- return Status::OK();
-}
-
Status ParquetReader::init_reader(
const std::vector<std::string>& column_names,
std::unordered_map<std::string, ColumnValueRangeType>* colname_to_value_range,
VExprContext* vconjunct_ctx, bool filter_groups) {
SCOPED_RAW_TIMER(&_statistics.parse_meta_time);
- Status open_status = _open_file();
- if (!open_status.ok()) {
- return open_status;
- }
+ RETURN_IF_ERROR(_open_file());
_column_names = &column_names;
_t_metadata = &_file_metadata->to_thrift();
_total_groups = _t_metadata->row_groups.size();
@@ -182,7 +171,7 @@ Status ParquetReader::init_reader(
RETURN_IF_ERROR(_init_read_columns());
// build column predicates for column lazy read
_lazy_read_ctx.vconjunct_ctx = vconjunct_ctx;
- RETURN_IF_ERROR(_init_row_group_readers(filter_groups));
+ RETURN_IF_ERROR(_init_row_groups(filter_groups));
return Status::OK();
}
@@ -317,20 +306,9 @@ std::unordered_map<std::string, TypeDescriptor> ParquetReader::get_name_to_type(
return map;
}
-Status ParquetReader::get_parsered_schema(std::vector<std::string>* col_names,
- std::vector<TypeDescriptor>* col_types) {
- if (_file_reader == nullptr) {
- RETURN_IF_ERROR(FileFactory::create_file_reader(_profile, _scan_params, _scan_range.path,
- _scan_range.start_offset,
- _scan_range.file_size, 0, _file_reader));
- }
- if (_file_metadata == nullptr) {
- RETURN_IF_ERROR(_file_reader->open());
- if (_file_reader->size() == 0) {
- return Status::EndOfFile("Empty Parquet File");
- }
- RETURN_IF_ERROR(parse_thrift_footer(_file_reader.get(), _file_metadata));
- }
+Status ParquetReader::get_parsed_schema(std::vector<std::string>* col_names,
+ std::vector<TypeDescriptor>* col_types) {
+ RETURN_IF_ERROR(_open_file());
_t_metadata = &_file_metadata->to_thrift();
_total_groups = _t_metadata->row_groups.size();
@@ -362,143 +340,90 @@ Status ParquetReader::get_columns(std::unordered_map<std::string, TypeDescriptor
return Status::OK();
}
-void ParquetReader::merge_delete_row_ranges(const std::set<RowRange>& delete_row_ranges) {
- if (_row_ranges.empty()) {
- _current_group_reader->set_row_ranges(_row_ranges);
- return;
- }
- if (!delete_row_ranges.empty()) {
- std::vector<RowRange> candidate_ranges;
- auto start_range = _row_ranges.begin();
- auto delete_range = delete_row_ranges.begin();
- int64_t processed_range_start_idx = start_range->first_row;
- while (start_range != _row_ranges.end() && delete_range != delete_row_ranges.end()) {
- int64_t delete_start = delete_range->first_row;
- int64_t delete_end = delete_range->last_row;
- int64_t range_start = start_range->first_row;
- int64_t range_end = start_range->last_row;
- if (delete_end > range_end) {
- if (range_start < processed_range_start_idx) {
- // rows before processed_range_start_idx have been processed
- range_start = processed_range_start_idx;
- }
- if (range_end < delete_start) {
- /**
- * start_range
- * || --------- || - |--------- |
- * delete_range
- */
- candidate_ranges.emplace_back(range_start, range_end);
- } else if (range_start < delete_start) {
- /**
- * row_range
- * || --------|-------- || ----- |
- * delete_start delete_end
- */
- candidate_ranges.emplace_back(range_start, delete_start);
- }
- // range_end > delete_end && range_start > delete_start
- start_range++;
- } else {
- // delete_end < range_end，most of the time, we will use this branch
- if (processed_range_start_idx < delete_start) {
- /**
- * row_range_start row_range_end
- * || --- | --------- | --- ||
- * delete_range
- */
- candidate_ranges.emplace_back(processed_range_start_idx, delete_start);
- }
- // delete_end is in row_range, so it can assign to processed_range_start_idx
- processed_range_start_idx = delete_end;
- delete_range++;
- if (delete_range == delete_row_ranges.end()) {
- range_end = _row_ranges[_row_ranges.size() - 1].last_row;
- if (processed_range_start_idx != range_end) {
- candidate_ranges.emplace_back(processed_range_start_idx, range_end);
- }
- }
- }
- }
- _row_ranges.assign(candidate_ranges.begin(), candidate_ranges.end());
- }
- _current_group_reader->set_row_ranges(_row_ranges);
-}
-
Status ParquetReader::get_next_block(Block* block, size_t* read_rows, bool* eof) {
- int32_t num_of_readers = _row_group_readers.size();
- DCHECK(num_of_readers <= _read_row_groups.size());
- if (_read_row_groups.empty()) {
- *eof = true;
- return Status::OK();
+ if (_current_group_reader == nullptr) {
+ if (_read_row_groups.size() > 0) {
+ RETURN_IF_ERROR(_next_row_group_reader());
+ } else {
+ *read_rows = 0;
+ *eof = true;
+ return Status::OK();
+ }
}
- bool _batch_eof = false;
+ DCHECK(_current_group_reader != nullptr);
+ bool batch_eof = false;
{
SCOPED_RAW_TIMER(&_statistics.column_read_time);
RETURN_IF_ERROR(
- _current_group_reader->next_batch(block, _batch_size, read_rows, &_batch_eof));
+ _current_group_reader->next_batch(block, _batch_size, read_rows, &batch_eof));
}
- if (_batch_eof) {
+ if (batch_eof) {
auto column_st = _current_group_reader->statistics();
_column_statistics.merge(column_st);
_statistics.lazy_read_filtered_rows += _current_group_reader->lazy_read_filtered_rows();
- if (!_next_row_group_reader()) {
+ Status st = _next_row_group_reader();
+ if (st.is_end_of_file()) {
*eof = true;
+ } else if (!st.ok()) {
+ return st;
}
}
- VLOG_DEBUG << "ParquetReader::get_next_block: " << block->rows();
return Status::OK();
}
-bool ParquetReader::_next_row_group_reader() {
- if (_row_group_readers.empty()) {
- return false;
- }
- _current_group_reader = _row_group_readers.front();
- _row_group_readers.pop_front();
- return true;
+RowGroupReader::PositionDeleteContext ParquetReader::_get_position_delete_ctx(
+ const tparquet::RowGroup& row_group, const RowGroupReader::RowGroupIndex& row_group_index) {
+ if (_delete_rows == nullptr) {
+ return RowGroupReader::PositionDeleteContext(row_group.num_rows, row_group_index.first_row);
+ }
+ int64_t* delete_rows = const_cast<int64_t*>(&(*_delete_rows)[0]);
+ int64_t* delete_rows_end = delete_rows + _delete_rows->size();
+ int64_t* start_pos = std::lower_bound(delete_rows + _delete_rows_index, delete_rows_end,
+ row_group_index.first_row);
+ int64_t start_index = start_pos - delete_rows;
+ int64_t* end_pos = std::lower_bound(start_pos, delete_rows_end, row_group_index.last_row);
+ int64_t end_index = end_pos - delete_rows;
+ _delete_rows_index = end_index;
+ return RowGroupReader::PositionDeleteContext(*_delete_rows, row_group.num_rows,
+ row_group_index.first_row, start_index, end_index);
}
-Status ParquetReader::_init_row_group_readers(const bool& filter_groups) {
- std::vector<RowGroupIndex> group_indexes;
- RETURN_IF_ERROR(_filter_row_groups(filter_groups, group_indexes));
- DCHECK_EQ(group_indexes.size(), _read_row_groups.size());
- auto group_index = group_indexes.begin();
- for (auto row_group_id : _read_row_groups) {
- auto& row_group = _t_metadata->row_groups[row_group_id];
- std::shared_ptr<RowGroupReader> row_group_reader;
- row_group_reader.reset(new RowGroupReader(_file_reader.get(), _read_columns, *group_index,
- row_group, _ctz, _lazy_read_ctx));
- group_index++;
- RETURN_IF_ERROR(_process_page_index(row_group));
- if (_row_ranges.empty()) {
- _row_ranges.emplace_back(0, row_group.num_rows);
- _statistics.read_rows += row_group.num_rows;
- }
- RETURN_IF_ERROR(row_group_reader->init(_file_metadata->schema(), _col_offsets));
- row_group_reader->set_row_ranges(_row_ranges);
- _row_group_readers.emplace_back(row_group_reader);
- }
- if (!_next_row_group_reader()) {
- return Status::EndOfFile("No next reader");
- }
- return Status::OK();
+Status ParquetReader::_next_row_group_reader() {
+ if (_read_row_groups.empty()) {
+ _current_group_reader.reset(nullptr);
+ return Status::EndOfFile("No next RowGroupReader");
+ }
+ RowGroupReader::RowGroupIndex row_group_index = _read_row_groups.front();
+ _read_row_groups.pop_front();
+
+ // process page index and generate the ranges to read
+ auto& row_group = _t_metadata->row_groups[row_group_index.row_group_id];
+ std::vector<RowRange> candidate_row_ranges;
+ RETURN_IF_ERROR(_process_page_index(row_group, candidate_row_ranges));
+
+ RowGroupReader::PositionDeleteContext position_delete_ctx =
+ _get_position_delete_ctx(row_group, row_group_index);
+ _current_group_reader.reset(new RowGroupReader(_file_reader.get(), _read_columns,
+ row_group_index.row_group_id, row_group, _ctz,
+ position_delete_ctx, _lazy_read_ctx));
+ return _current_group_reader->init(_file_metadata->schema(), candidate_row_ranges,
+ _col_offsets);
}
-Status ParquetReader::_filter_row_groups(const bool& enabled,
- std::vector<RowGroupIndex>& group_indexes) {
+Status ParquetReader::_init_row_groups(const bool& is_filter_groups) {
SCOPED_RAW_TIMER(&_statistics.row_group_filter_time);
- if (enabled && (_total_groups == 0 || _t_metadata->num_rows == 0 || _range_size < 0)) {
- return Status::EndOfFile("No row group need read");
+ if (is_filter_groups && (_total_groups == 0 || _t_metadata->num_rows == 0 || _range_size < 0)) {
+ return Status::EndOfFile("No row group to read");
}
- int32_t start_row_id = 0;
+ int64_t row_index = 0;
for (int32_t row_group_idx = 0; row_group_idx < _total_groups; row_group_idx++) {
const tparquet::RowGroup& row_group = _t_metadata->row_groups[row_group_idx];
- if (enabled && _is_misaligned_range_group(row_group)) {
+ if (is_filter_groups && _is_misaligned_range_group(row_group)) {
+ row_index += row_group.num_rows;
continue;
}
bool filter_group = false;
- if (enabled) {
+ if (is_filter_groups) {
RETURN_IF_ERROR(_process_row_group_filter(row_group, &filter_group));
}
int64_t group_size = 0; // only calculate the needed columns
@@ -508,13 +433,12 @@ Status ParquetReader::_filter_row_groups(const bool& enabled,
group_size += row_group.columns[parquet_col_id].meta_data.total_compressed_size;
}
}
- // record row group physical id
- int32_t first_row_index = start_row_id;
- int32_t last_row_index = first_row_index + row_group.num_rows;
- start_row_id = last_row_index + 1;
if (!filter_group) {
- group_indexes.emplace_back(row_group_idx, first_row_index, last_row_index);
- _read_row_groups.emplace_back(row_group_idx);
+ _read_row_groups.emplace_back(row_group_idx, row_index, row_index + row_group.num_rows);
+ if (_statistics.read_row_groups == 0) {
+ _whole_range.first_row = row_index;
+ }
+ _whole_range.last_row = row_index + row_group.num_rows;
_statistics.read_row_groups++;
_statistics.read_bytes += group_size;
} else {
@@ -522,6 +446,11 @@ Status ParquetReader::_filter_row_groups(const bool& enabled,
_statistics.filtered_bytes += group_size;
_statistics.filtered_group_rows += row_group.num_rows;
}
+ row_index += row_group.num_rows;
+ }
+
+ if (_read_row_groups.empty()) {
+ return Status::EndOfFile("No row group to read");
}
return Status::OK();
}
@@ -545,13 +474,22 @@ bool ParquetReader::_has_page_index(const std::vector<tparquet::ColumnChunk>& co
return page_index.check_and_get_page_index_ranges(columns);
}
-Status ParquetReader::_process_page_index(const tparquet::RowGroup& row_group) {
+Status ParquetReader::_process_page_index(const tparquet::RowGroup& row_group,
+ std::vector<RowRange>& candidate_row_ranges) {
SCOPED_RAW_TIMER(&_statistics.page_index_filter_time);
+
+ std::function<void()> read_whole_row_group = [&]() {
+ candidate_row_ranges.emplace_back(0, row_group.num_rows);
+ _statistics.read_rows += row_group.num_rows;
+ };
+
if (_colname_to_value_range == nullptr || _colname_to_value_range->empty()) {
+ read_whole_row_group();
return Status::OK();
}
PageIndex page_index;
if (!_has_page_index(row_group.columns, page_index)) {
+ read_whole_row_group();
return Status::OK();
}
uint8_t col_index_buff[page_index._column_index_size];
@@ -573,12 +511,12 @@ Status ParquetReader::_process_page_index(const tparquet::RowGroup& row_group) {
auto& chunk = row_group.columns[read_col._parquet_col_id];
tparquet::ColumnIndex column_index;
if (chunk.column_index_offset == 0 && chunk.column_index_length == 0) {
- return Status::OK();
+ continue;
}
RETURN_IF_ERROR(page_index.parse_column_index(chunk, col_index_buff, &column_index));
const int num_of_pages = column_index.null_pages.size();
if (num_of_pages <= 0) {
- break;
+ continue;
}
auto& conjuncts = conjunct_iter->second;
std::vector<int> skipped_page_range;
@@ -600,6 +538,7 @@ Status ParquetReader::_process_page_index(const tparquet::RowGroup& row_group) {
_col_offsets.emplace(read_col._parquet_col_id, offset_index);
}
if (skipped_row_ranges.empty()) {
+ read_whole_row_group();
return Status::OK();
}
@@ -617,14 +556,14 @@ Status ParquetReader::_process_page_index(const tparquet::RowGroup& row_group) {
}
} else {
// read row with candidate ranges rather than skipped ranges
- _row_ranges.emplace_back(skip_end, skip_range.first_row);
+ candidate_row_ranges.emplace_back(skip_end, skip_range.first_row);
read_rows += skip_range.first_row - skip_end;
skip_end = skip_range.last_row;
}
}
DCHECK_LE(skip_end, row_group.num_rows);
if (skip_end != row_group.num_rows) {
- _row_ranges.emplace_back(skip_end, row_group.num_rows);
+ candidate_row_ranges.emplace_back(skip_end, row_group.num_rows);
read_rows += row_group.num_rows - skip_end;
}
_statistics.read_rows += read_rows;
@@ -676,13 +615,13 @@ Status ParquetReader::_process_column_stat_filter(const std::vector<tparquet::Co
void ParquetReader::_init_chunk_dicts() {}
Status ParquetReader::_process_dict_filter(bool* filter_group) {
- return Status();
+ return Status::OK();
}
void ParquetReader::_init_bloom_filter() {}
Status ParquetReader::_process_bloom_filter(bool* filter_group) {
- return Status();
+ return Status::OK();
}
int64_t ParquetReader::_get_column_start_offset(const tparquet::ColumnMetaData& column) {
diff --git a/be/src/vec/exec/format/parquet/vparquet_reader.h b/be/src/vec/exec/format/parquet/vparquet_reader.h
index f5891754e7..2bfc74f823 100644
--- a/be/src/vec/exec/format/parquet/vparquet_reader.h
+++ b/be/src/vec/exec/format/parquet/vparquet_reader.h
@@ -77,9 +77,10 @@ public:
void close();
- Status file_metadata(FileMetaData** metadata);
+ RowRange get_whole_range() { return _whole_range; }
- void merge_delete_row_ranges(const std::set<RowRange>& delete_row_ranges);
+ // set the delete rows in current parquet file
+ void set_delete_rows(const std::vector<int64_t>* delete_rows) { _delete_rows = delete_rows; }
int64_t size() const { return _file_reader->size(); }
@@ -87,8 +88,8 @@ public:
Status get_columns(std::unordered_map<std::string, TypeDescriptor>* name_to_type,
std::unordered_set<std::string>* missing_cols) override;
- Status get_parsered_schema(std::vector<std::string>* col_names,
- std::vector<TypeDescriptor>* col_types) override;
+ Status get_parsed_schema(std::vector<std::string>* col_names,
+ std::vector<TypeDescriptor>* col_types) override;
Statistics& statistics() { return _statistics; }
@@ -125,12 +126,16 @@ private:
Status _open_file();
void _init_profile();
- bool _next_row_group_reader();
+ Status _next_row_group_reader();
+ RowGroupReader::PositionDeleteContext _get_position_delete_ctx(
+ const tparquet::RowGroup& row_group,
+ const RowGroupReader::RowGroupIndex& row_group_index);
Status _init_read_columns();
- Status _init_row_group_readers(const bool& filter_groups);
+ Status _init_row_groups(const bool& is_filter_groups);
// Page Index Filter
bool _has_page_index(const std::vector<tparquet::ColumnChunk>& columns, PageIndex& page_index);
- Status _process_page_index(const tparquet::RowGroup& row_group);
+ Status _process_page_index(const tparquet::RowGroup& row_group,
+ std::vector<RowRange>& candidate_row_ranges);
// Row Group Filter
bool _is_misaligned_range_group(const tparquet::RowGroup& row_group);
@@ -141,28 +146,27 @@ private:
Status _process_dict_filter(bool* filter_group);
void _init_bloom_filter();
Status _process_bloom_filter(bool* filter_group);
- Status _filter_row_groups(const bool& enabled, std::vector<RowGroupIndex>& group_indexes);
int64_t _get_column_start_offset(const tparquet::ColumnMetaData& column_init_column_readers);
-private:
RuntimeProfile* _profile;
const TFileScanRangeParams& _scan_params;
const TFileRangeDesc& _scan_range;
std::unique_ptr<FileReader> _file_reader = nullptr;
- std::vector<RowRange> _row_ranges;
std::shared_ptr<FileMetaData> _file_metadata;
const tparquet::FileMetaData* _t_metadata;
- std::list<std::shared_ptr<RowGroupReader>> _row_group_readers;
- std::shared_ptr<RowGroupReader> _current_group_reader;
+ std::unique_ptr<RowGroupReader> _current_group_reader;
int32_t _total_groups; // num of groups(stripes) of a parquet(orc) file
std::map<std::string, int> _map_column; // column-name <---> column-index
std::unordered_map<std::string, ColumnValueRangeType>* _colname_to_value_range;
std::vector<ParquetReadColumn> _read_columns;
+ RowRange _whole_range = RowRange(0, 0);
+ const std::vector<int64_t>* _delete_rows = nullptr;
+ int64_t _delete_rows_index = 0;
// Used for column lazy read.
RowGroupReader::LazyReadContext _lazy_read_ctx;
- std::list<int32_t> _read_row_groups;
+ std::list<RowGroupReader::RowGroupIndex> _read_row_groups;
// parquet file reader object
size_t _batch_size;
int64_t _range_start_offset;
diff --git a/be/src/vec/exec/format/table/iceberg_reader.cpp b/be/src/vec/exec/format/table/iceberg_reader.cpp
index 9e2f659772..332aa44e54 100644
--- a/be/src/vec/exec/format/table/iceberg_reader.cpp
+++ b/be/src/vec/exec/format/table/iceberg_reader.cpp
@@ -17,12 +17,10 @@
#include "iceberg_reader.h"
-#include <vec/core/column_with_type_and_name.h>
-#include <vec/exec/format/parquet/vparquet_reader.h>
-
-#include <vec/data_types/data_type_factory.hpp>
-
#include "vec/common/assert_cast.h"
+#include "vec/core/column_with_type_and_name.h"
+#include "vec/data_types/data_type_factory.hpp"
+#include "vec/exec/format/parquet/vparquet_reader.h"
namespace doris::vectorized {
@@ -34,9 +32,11 @@ IcebergTableReader::IcebergTableReader(GenericReader* file_format_reader, Runtim
: TableFormatReader(file_format_reader), _profile(profile), _state(state), _params(params) {
static const char* iceberg_profile = "IcebergProfile";
ADD_TIMER(_profile, iceberg_profile);
- _iceberg_profile._delete_files_init_time =
- ADD_CHILD_TIMER(_profile, "DeleteFileInitTime", iceberg_profile);
- _iceberg_profile._delete_files_read_total_time =
+ _iceberg_profile.num_delete_files =
+ ADD_CHILD_COUNTER(_profile, "NumDeleteFiles", TUnit::UNIT, iceberg_profile);
+ _iceberg_profile.num_delete_rows =
+ ADD_CHILD_COUNTER(_profile, "NumDeleteRows", TUnit::UNIT, iceberg_profile);
+ _iceberg_profile.delete_files_read_time =
ADD_CHILD_TIMER(_profile, "DeleteFileReadTime", iceberg_profile);
}
@@ -63,162 +63,159 @@ Status IcebergTableReader::get_columns(
return _file_format_reader->get_columns(name_to_type, missing_cols);
}
-void IcebergTableReader::filter_rows(const TFileRangeDesc& range) {
- if (_cur_delete_file_reader == nullptr) {
- return;
- }
- SCOPED_TIMER(_iceberg_profile._delete_files_read_total_time);
+Status IcebergTableReader::init_row_filters(const TFileRangeDesc& range) {
auto& table_desc = range.table_format_params.iceberg_params;
auto& version = table_desc.format_version;
if (version < MIN_SUPPORT_DELETE_FILES_VERSION) {
- return;
+ return Status::OK();
}
- bool eof = false;
- std::set<RowRange> delete_row_ranges;
- while (!eof) {
- size_t read_rows = 0;
- Block block = Block();
- for (const FieldSchema& field : _column_schemas) {
- DataTypePtr data_type = DataTypeFactory::instance().create_data_type(field.type, true);
- MutableColumnPtr data_column = data_type->create_column();
- block.insert(ColumnWithTypeAndName(std::move(data_column), data_type, field.name));
- }
- Status st = _cur_delete_file_reader->get_next_block(&block, &read_rows, &eof);
- if (!st.ok() || eof) {
- if (!_delete_file_readers.empty()) {
- eof = false;
- _cur_delete_file_reader = std::move(_delete_file_readers.front());
- _delete_file_readers.pop_front();
- }
- }
- if (read_rows != 0) {
- auto& pos_type_column = block.get_by_name(ICEBERG_ROW_POS);
- ColumnPtr pos_column = pos_type_column.column;
- using ColumnType = typename PrimitiveTypeTraits<TYPE_BIGINT>::ColumnType;
- if (pos_type_column.type->is_nullable()) {
- pos_column =
- assert_cast<const ColumnNullable&>(*pos_column).get_nested_column_ptr();
- }
- auto& data = assert_cast<const ColumnType&>(*pos_column).get_data();
- std::vector<int64_t> delete_row_ids;
- for (int row_id = 0; row_id < read_rows; row_id++) {
- delete_row_ids.emplace_back(data[row_id]);
+ auto& delete_file_type = table_desc.content;
+ auto files = table_desc.delete_files;
+ if (files.empty()) {
+ return Status::OK();
+ }
+ if (delete_file_type == POSITION_DELETE) {
+ // position delete
+ SCOPED_TIMER(_iceberg_profile.delete_files_read_time);
+ auto row_desc = RowDescriptor(_state->desc_tbl(),
+ std::vector<TupleId>({table_desc.delete_table_tuple_id}),
+ std::vector<bool>({false}));
+ RETURN_IF_ERROR(VExpr::create_expr_tree(_state->obj_pool(), table_desc.file_select_conjunct,
+ &_data_path_conjunct_ctx));
+ RETURN_IF_ERROR(_data_path_conjunct_ctx->prepare(_state, row_desc));
+ RETURN_IF_ERROR(_data_path_conjunct_ctx->open(_state));
+
+ ParquetReader* parquet_reader = (ParquetReader*)(_file_format_reader.get());
+ RowRange whole_range = parquet_reader->get_whole_range();
+ bool init_schema = false;
+ std::vector<std::string> delete_file_col_names;
+ std::vector<TypeDescriptor> delete_file_col_types;
+ std::list<std::vector<int64_t>> delete_rows_list;
+ delete_rows_list.resize(files.size());
+ int64_t num_delete_rows = 0;
+ auto delete_rows_iter = delete_rows_list.begin();
+ for (auto& delete_file : files) {
+ if (whole_range.last_row <= delete_file.position_lower_bound ||
+ whole_range.first_row > delete_file.position_upper_bound) {
+ delete_rows_iter++;
+ continue;
}
- if (delete_row_ids.empty()) {
- return;
+ std::vector<int64_t>& delete_rows = *delete_rows_iter;
+ TFileRangeDesc delete_range;
+ delete_range.path = delete_file.path;
+ delete_range.start_offset = 0;
+ delete_range.size = -1;
+ delete_range.file_size = -1;
+ ParquetReader delete_reader(_profile, _params, delete_range, 102400,
+ const_cast<cctz::time_zone*>(&_state->timezone_obj()));
+ if (!init_schema) {
+ delete_reader.get_parsed_schema(&delete_file_col_names, &delete_file_col_types);
+ init_schema = true;
}
-
- int num_deleted_ids = delete_row_ids.size();
- int i = 0;
- while (i < num_deleted_ids) {
- int64_t row_id = delete_row_ids[i];
- int64_t row_range_start = row_id;
- int64_t row_range_end = row_id;
- while (i + 1 < num_deleted_ids) {
- if (delete_row_ids[i + 1] == delete_row_ids[i] + 1) {
- row_range_end = delete_row_ids[i + 1];
- i++;
- continue;
- } else {
- delete_row_ranges.emplace(row_range_start, row_range_end + 1);
- row_range_start = ++row_range_end;
- break;
- }
+ RETURN_IF_ERROR(delete_reader.init_reader(delete_file_col_names, nullptr,
+ _data_path_conjunct_ctx, false));
+ std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
+ partition_columns;
+ std::unordered_map<std::string, VExprContext*> missing_columns;
+ delete_reader.set_fill_columns(partition_columns, missing_columns);
+
+ bool eof = false;
+ while (!eof) {
+ Block block = Block();
+ for (int i = 0; i < delete_file_col_names.size(); ++i) {
+ DataTypePtr data_type = DataTypeFactory::instance().create_data_type(
+ delete_file_col_types[i], true);
+ MutableColumnPtr data_column = data_type->create_column();
+ block.insert(ColumnWithTypeAndName(std::move(data_column), data_type,
+ delete_file_col_names[i]));
}
- if (i == num_deleted_ids - 1) {
- delete_row_ranges.emplace(row_range_start,
- delete_row_ids[num_deleted_ids - 1] + 1);
+ eof = false;
+ size_t read_rows = 0;
+ RETURN_IF_ERROR(delete_reader.get_next_block(&block, &read_rows, &eof));
+ if (read_rows > 0) {
+ auto& pos_type_column = block.get_by_name(ICEBERG_ROW_POS);
+ ColumnPtr pos_column = pos_type_column.column;
+ using ColumnType = typename PrimitiveTypeTraits<TYPE_BIGINT>::ColumnType;
+ if (pos_type_column.type->is_nullable()) {
+ pos_column = assert_cast<const ColumnNullable&>(*pos_column)
+ .get_nested_column_ptr();
+ }
+ const int64_t* src_data =
+ assert_cast<const ColumnType&>(*pos_column).get_data().data();
+ const int64_t* src_data_end = src_data + read_rows;
+ const int64_t* cpy_start =
+ std::lower_bound(src_data, src_data_end, whole_range.first_row);
+ const int64_t* cpy_end =
+ std::lower_bound(cpy_start, src_data_end, whole_range.last_row);
+ int64_t cpy_count = cpy_end - cpy_start;
+
+ if (cpy_count > 0) {
+ int64_t origin_size = delete_rows.size();
+ delete_rows.resize(origin_size + cpy_count);
+ int64_t* dest_position = &delete_rows[origin_size];
+ memcpy(dest_position, cpy_start, cpy_count * sizeof(int64_t));
+ num_delete_rows += cpy_count;
+ }
}
- row_range_start = delete_row_ids[i + 1];
- i++;
}
+ delete_rows_iter++;
}
- }
- if (VLOG_IS_ON(3)) {
- if (!delete_row_ranges.empty()) {
- std::stringstream out;
- out << "[";
- for (const RowRange& delete_row_range : delete_row_ranges) {
- out << " " << delete_row_range.debug_string();
+ if (num_delete_rows > 0) {
+ for (auto iter = delete_rows_list.begin(); iter != delete_rows_list.end();) {
+ if (iter->empty()) {
+ delete_rows_list.erase(iter++);
+ } else {
+ iter++;
+ }
}
- out << " ]";
- VLOG_NOTICE << "Delete row range info: " << out.str();
+ _merge_sort(delete_rows_list, num_delete_rows);
+ parquet_reader->set_delete_rows(&_delete_rows);
+ COUNTER_UPDATE(_iceberg_profile.num_delete_rows, num_delete_rows);
}
}
- ParquetReader* parquet_reader = (ParquetReader*)(_file_format_reader.get());
- parquet_reader->merge_delete_row_ranges(delete_row_ranges);
+ // todo: equality delete
+ COUNTER_UPDATE(_iceberg_profile.num_delete_files, files.size());
+ return Status::OK();
}
-Status IcebergTableReader::init_row_filters(const TFileRangeDesc& range) {
- auto& table_desc = range.table_format_params.iceberg_params;
- auto& version = table_desc.format_version;
- if (version >= MIN_SUPPORT_DELETE_FILES_VERSION) {
- SCOPED_TIMER(_iceberg_profile._delete_files_init_time);
- auto& delete_file_type = table_desc.content;
- auto files = table_desc.delete_files;
- if (delete_file_type == POSITON_DELELE) {
- // position delete
- auto row_desc = RowDescriptor(_state->desc_tbl(),
- std::vector<TupleId>({table_desc.delete_table_tuple_id}),
- std::vector<bool>({false}));
- RETURN_IF_ERROR(VExpr::create_expr_tree(
- _state->obj_pool(), table_desc.file_select_conjunct, &_data_path_conjunct_ctx));
- RETURN_IF_ERROR(_data_path_conjunct_ctx->prepare(_state, row_desc));
- RETURN_IF_ERROR(_data_path_conjunct_ctx->open(_state));
- vector<std::string> names;
- for (auto& delete_file : files) {
- _position_delete_params.low_bound_index = delete_file.position_lower_bound;
- _position_delete_params.upper_bound_index = delete_file.position_upper_bound;
-
- TFileRangeDesc delete_range;
- delete_range.path = delete_file.path;
- delete_range.start_offset = 0;
- delete_range.size = -1;
- delete_range.file_size = -1;
- ParquetReader* delete_reader = new ParquetReader(
- _profile, _params, delete_range, _state->query_options().batch_size,
- const_cast<cctz::time_zone*>(&_state->timezone_obj()));
- if (_delete_file_schema == nullptr) {
- FileMetaData* metadata = nullptr;
- RETURN_IF_ERROR(delete_reader->file_metadata(&metadata));
- if (metadata == nullptr) {
- break;
- }
- _delete_file_schema = &metadata->schema();
- int num_of_col = _delete_file_schema->size();
- for (auto i = 0; i < num_of_col; ++i) {
- const FieldSchema* field = _delete_file_schema->get_column(i);
- _column_schemas.emplace_back(*field);
- names.emplace_back(field->name);
- }
- }
- DCHECK_EQ(_column_schemas.size(), _delete_file_schema->size());
- // The expr assure reading delete file data from current file range only
- Status d_st =
- delete_reader->init_reader(names, nullptr, _data_path_conjunct_ctx, false);
- std::unordered_map<std::string, std::tuple<std::string, const SlotDescriptor*>>
- partition_columns;
- std::unordered_map<std::string, VExprContext*> missing_columns;
- delete_reader->set_fill_columns(partition_columns, missing_columns);
-
- _delete_file_readers.emplace_back((GenericReader*)delete_reader);
-
- ParquetReader* parquet_reader = (ParquetReader*)(_file_format_reader.get());
- FileMetaData* file_metadata = nullptr;
- RETURN_IF_ERROR(parquet_reader->file_metadata(&file_metadata));
- _position_delete_params.total_file_rows = file_metadata->to_thrift().num_rows;
- }
- if (!_delete_file_readers.empty()) {
- _cur_delete_file_reader = std::move(_delete_file_readers.front());
- _delete_file_readers.pop_front();
- } else {
- _cur_delete_file_reader = nullptr;
- }
+void IcebergTableReader::_merge_sort(std::list<std::vector<int64_t>>& delete_rows_list,
+ int64_t num_delete_rows) {
+ if (delete_rows_list.empty()) {
+ return;
+ }
+ if (delete_rows_list.size() == 1) {
+ _delete_rows.resize(num_delete_rows);
+ memcpy(&_delete_rows[0], &(delete_rows_list.front()[0]), sizeof(int64_t) * num_delete_rows);
+ return;
+ }
+ if (delete_rows_list.size() == 2) {
+ _delete_rows.resize(num_delete_rows);
+ std::merge(delete_rows_list.front().begin(), delete_rows_list.front().end(),
+ delete_rows_list.back().begin(), delete_rows_list.back().end(),
+ _delete_rows.begin());
+ return;
+ }
+
+ // merge sort
+ using vec_pair =
+ std::pair<std::vector<int64_t>::iterator, std::vector<int64_t>::const_iterator>;
+ auto cmp = [](const vec_pair& a, const vec_pair& b) { return *a.first > *b.first; };
+ std::priority_queue<vec_pair, vector<vec_pair>, decltype(cmp)> pq(cmp);
+ for (auto iter = delete_rows_list.begin(); iter != delete_rows_list.end(); ++iter) {
+ if (iter->size() > 0) {
+ pq.push({iter->begin(), iter->end()});
+ }
+ }
+ _delete_rows.reserve(num_delete_rows);
+ while (!pq.empty()) {
+ vec_pair p = pq.top();
+ pq.pop();
+ _delete_rows.emplace_back(*p.first);
+ p.first++;
+ if (p.first != p.second) {
+ pq.push(p);
}
}
- // todo: equality delete
- filter_rows(range);
- return Status::OK();
}
-} // namespace doris::vectorized
\ No newline at end of file
+} // namespace doris::vectorized
diff --git a/be/src/vec/exec/format/table/iceberg_reader.h b/be/src/vec/exec/format/table/iceberg_reader.h
index c71ffdaa9c..0a9d4ef71f 100644
--- a/be/src/vec/exec/format/table/iceberg_reader.h
+++ b/be/src/vec/exec/format/table/iceberg_reader.h
@@ -17,12 +17,11 @@
#pragma once
-#include <vec/exec/format/parquet/parquet_common.h>
-
#include <queue>
#include "table_format_reader.h"
#include "vec/exec/format/generic_reader.h"
+#include "vec/exec/format/parquet/parquet_common.h"
#include "vec/exprs/vexpr.h"
namespace doris::vectorized {
@@ -32,8 +31,8 @@ public:
IcebergTableReader(GenericReader* file_format_reader, RuntimeProfile* profile,
RuntimeState* state, const TFileScanRangeParams& params);
~IcebergTableReader() override;
- Status init_row_filters(const TFileRangeDesc& range);
- void filter_rows(const TFileRangeDesc& range) override;
+
+ Status init_row_filters(const TFileRangeDesc& range) override;
Status get_next_block(Block* block, size_t* read_rows, bool* eof) override;
@@ -45,30 +44,23 @@ public:
Status get_columns(std::unordered_map<std::string, TypeDescriptor>* name_to_type,
std::unordered_set<std::string>* missing_cols) override;
-public:
- enum { DATA, POSITON_DELELE, EQULITY_DELELE };
- struct PositionDeleteParams {
- int64_t low_bound_index = -1;
- int64_t upper_bound_index = -1;
- int64_t last_delete_row_index = -1;
- int64_t total_file_rows = 0;
- };
+ enum { DATA, POSITION_DELETE, EQUALITY_DELETE };
private:
struct IcebergProfile {
- RuntimeProfile::Counter* _delete_files_init_time;
- RuntimeProfile::Counter* _delete_files_read_total_time;
+ RuntimeProfile::Counter* num_delete_files;
+ RuntimeProfile::Counter* num_delete_rows;
+ RuntimeProfile::Counter* delete_files_read_time;
};
+
+ void _merge_sort(std::list<std::vector<int64_t>>& delete_rows_list, int64_t num_delete_rows);
+
RuntimeProfile* _profile;
RuntimeState* _state;
const TFileScanRangeParams& _params;
- std::vector<FieldSchema> _column_schemas;
- std::deque<std::unique_ptr<GenericReader>> _delete_file_readers;
- std::unique_ptr<GenericReader> _cur_delete_file_reader;
- PositionDeleteParams _position_delete_params;
- const FieldDescriptor* _delete_file_schema = nullptr;
VExprContext* _data_path_conjunct_ctx = nullptr;
IcebergProfile _iceberg_profile;
+ std::vector<int64_t> _delete_rows;
};
} // namespace doris::vectorized
diff --git a/be/src/vec/exec/format/table/table_format_reader.h b/be/src/vec/exec/format/table/table_format_reader.h
index 188d41a2b9..1c2ebec7d3 100644
--- a/be/src/vec/exec/format/table/table_format_reader.h
+++ b/be/src/vec/exec/format/table/table_format_reader.h
@@ -17,12 +17,11 @@
#pragma once
-#include <vec/exec/format/parquet/parquet_common.h>
-
#include <string>
#include "runtime/runtime_state.h"
#include "vec/exec/format/generic_reader.h"
+#include "vec/exec/format/parquet/parquet_common.h"
namespace doris::vectorized {
@@ -38,7 +37,7 @@ public:
return _file_format_reader->get_columns(name_to_type, missing_cols);
}
- virtual void filter_rows(const TFileRangeDesc& range) = 0;
+ virtual Status init_row_filters(const TFileRangeDesc& range) = 0;
protected:
std::string _table_format; // hudi, iceberg
diff --git a/be/test/vec/exec/parquet/parquet_thrift_test.cpp b/be/test/vec/exec/parquet/parquet_thrift_test.cpp
index c8eeb5d00b..d1f59e9234 100644
--- a/be/test/vec/exec/parquet/parquet_thrift_test.cpp
+++ b/be/test/vec/exec/parquet/parquet_thrift_test.cpp
@@ -413,13 +413,13 @@ TEST_F(ParquetThriftReaderTest, group_reader) {
TimezoneUtils::find_cctz_time_zone(TimezoneUtils::default_time_zone, ctz);
auto row_group = t_metadata.row_groups[0];
std::shared_ptr<RowGroupReader> row_group_reader;
- row_group_reader.reset(new RowGroupReader(&file_reader, read_columns,
- RowGroupIndex(0, 0, 10000), row_group, &ctz,
- lazy_read_ctx));
+ RowGroupReader::PositionDeleteContext position_delete_ctx(row_group.num_rows, 0);
+ row_group_reader.reset(new RowGroupReader(&file_reader, read_columns, 0, row_group, &ctz,
+ position_delete_ctx, lazy_read_ctx));
+ std::vector<RowRange> row_ranges;
+ row_ranges.emplace_back(0, row_group.num_rows);
auto col_offsets = std::unordered_map<int, tparquet::OffsetIndex>();
- auto stg = row_group_reader->init(meta_data->schema(), col_offsets);
- std::vector<RowRange> row_ranges = std::vector<RowRange>();
- row_group_reader->set_row_ranges(row_ranges);
+ auto stg = row_group_reader->init(meta_data->schema(), row_ranges, col_offsets);
EXPECT_TRUE(stg.ok());
vectorized::Block block;
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 03/15: [Bug](Agg) fix crash when encountering not supported agg function like last_value(bitmap) (#15257)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit e8bb4a3bcc810e0c6fa3c4e3194630ed9c346a17
Author: AlexYue <yj...@gmail.com>
AuthorDate: Fri Dec 23 14:23:21 2022 +0800
[Bug](Agg) fix crash when encountering not supported agg function like last_value(bitmap) (#15257)
The former logic inside aggregate_function_window.cpp would shutdown BE once encountering agg function with complex type like BITMAP. This pr makes it don't crash and would return one more concrete error message which tells the unsupported function signature to user.
---
.../aggregate_function_window.cpp | 4 +-
be/src/vec/exec/vanalytic_eval_node.cpp | 4 ++
be/src/vec/exprs/vectorized_agg_fn.cpp | 2 +-
.../java/org/apache/doris/catalog/FunctionSet.java | 2 +-
.../function/test_not_supported_agg_fn.groovy | 43 ++++++++++++++++++++++
5 files changed, 51 insertions(+), 4 deletions(-)
diff --git a/be/src/vec/aggregate_functions/aggregate_function_window.cpp b/be/src/vec/aggregate_functions/aggregate_function_window.cpp
index 02b283ab2d..4714bd2609 100644
--- a/be/src/vec/aggregate_functions/aggregate_function_window.cpp
+++ b/be/src/vec/aggregate_functions/aggregate_function_window.cpp
@@ -78,8 +78,8 @@ static IAggregateFunction* create_function_lead_lag_first_last(const String& nam
TYPE_TO_BASIC_COLUMN_TYPE(DISPATCH)
#undef DISPATCH
- LOG(FATAL) << "with unknowed type, failed in create_aggregate_function_" << name
- << " and type is: " << argument_types[0]->get_name();
+ LOG(WARNING) << "with unknowed type, failed in create_aggregate_function_" << name
+ << " and type is: " << argument_types[0]->get_name();
return nullptr;
}
diff --git a/be/src/vec/exec/vanalytic_eval_node.cpp b/be/src/vec/exec/vanalytic_eval_node.cpp
index 8aa1f8708a..74625c14a6 100644
--- a/be/src/vec/exec/vanalytic_eval_node.cpp
+++ b/be/src/vec/exec/vanalytic_eval_node.cpp
@@ -30,6 +30,7 @@ namespace doris::vectorized {
VAnalyticEvalNode::VAnalyticEvalNode(ObjectPool* pool, const TPlanNode& tnode,
const DescriptorTbl& descs)
: ExecNode(pool, tnode, descs),
+ _fn_place_ptr(nullptr),
_intermediate_tuple_id(tnode.analytic_node.intermediate_tuple_id),
_output_tuple_id(tnode.analytic_node.output_tuple_id),
_window(tnode.analytic_node.window) {
@@ -649,6 +650,9 @@ Status VAnalyticEvalNode::_create_agg_status() {
}
Status VAnalyticEvalNode::_destroy_agg_status() {
+ if (UNLIKELY(_fn_place_ptr == nullptr)) {
+ return Status::OK();
+ }
for (size_t i = 0; i < _agg_functions_size; ++i) {
_agg_functions[i]->destroy(_fn_place_ptr + _offsets_of_aggregate_states[i]);
}
diff --git a/be/src/vec/exprs/vectorized_agg_fn.cpp b/be/src/vec/exprs/vectorized_agg_fn.cpp
index 8f7527a15e..0895eae775 100644
--- a/be/src/vec/exprs/vectorized_agg_fn.cpp
+++ b/be/src/vec/exprs/vectorized_agg_fn.cpp
@@ -129,7 +129,7 @@ Status AggFnEvaluator::prepare(RuntimeState* state, const RowDescriptor& desc, M
_fn.name.function_name, argument_types, {}, _data_type->is_nullable());
}
if (_function == nullptr) {
- return Status::InternalError("Agg Function {} is not implemented", _fn.name.function_name);
+ return Status::InternalError("Agg Function {} is not implemented", _fn.signature);
}
if (!_sort_description.empty()) {
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 532635cf36..af72fd6df8 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
@@ -2775,7 +2775,7 @@ public class FunctionSet<T> {
addBuiltin(AggregateFunction.createAnalyticBuiltin("ntile",
Collections.singletonList(Type.BIGINT), Type.BIGINT, Type.BIGINT, null, null, null, null, null, true));
- for (Type t : Type.getSupportedTypes()) {
+ for (Type t : Type.getTrivialTypes()) {
if (t.isNull()) {
continue; // NULL is handled through type promotion.
}
diff --git a/regression-test/suites/tpch_sf1_p1/function/test_not_supported_agg_fn.groovy b/regression-test/suites/tpch_sf1_p1/function/test_not_supported_agg_fn.groovy
new file mode 100644
index 0000000000..a7a6d89686
--- /dev/null
+++ b/regression-test/suites/tpch_sf1_p1/function/test_not_supported_agg_fn.groovy
@@ -0,0 +1,43 @@
+// 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.
+
+// There once exists a bug that when passing a too long string
+// along with an empty one to find_in_set, BE
+
+suite("test_not_supported_agg_fn") {
+
+ try {
+ test {
+ sql """
+ SELECT ref_0.`PS_SUPPKEY` AS c0,
+ ref_0.`PS_SUPPLYCOST` AS c1,
+ last_value(cast(bitmap_empty() AS BITMAP)) OVER (PARTITION BY ref_0.`PS_AVAILQTY`
+ ORDER BY ref_0.`PS_PARTKEY` DESC) AS c2,
+ CASE
+ WHEN FALSE THEN ref_0.`PS_COMMENT`
+ ELSE ref_0.`PS_COMMENT`
+ END AS c3
+ FROM regression_test_tpch_sf0_1_p1.partsupp AS ref_0
+ WHERE ref_0.`PS_PARTKEY` IS NULL
+ ORDER BY ref_0.`PS_COMMENT`
+ """
+ exception "errCode = 2, detailMessage = No matching function with signature: last_value(bitmap)"
+ }
+ } finally {
+ }
+
+}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org
[doris] 15/15: [fix](compile) fix compile error caused by `mysql_scan_node.cpp` not being found when enabling `WITH_MYSQL` (#15277)
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-1.2-lts
in repository https://gitbox.apache.org/repos/asf/doris.git
commit 9d637531e903c33903a2898a3d5a1b32dd3d0018
Author: luozenglin <37...@users.noreply.github.com>
AuthorDate: Fri Dec 23 16:25:28 2022 +0800
[fix](compile) fix compile error caused by `mysql_scan_node.cpp` not being found when enabling `WITH_MYSQL` (#15277)
---
be/src/exec/CMakeLists.txt | 7 -------
be/src/vec/CMakeLists.txt | 7 +++++++
2 files changed, 7 insertions(+), 7 deletions(-)
diff --git a/be/src/exec/CMakeLists.txt b/be/src/exec/CMakeLists.txt
index 2e020e2f92..003b613955 100644
--- a/be/src/exec/CMakeLists.txt
+++ b/be/src/exec/CMakeLists.txt
@@ -102,13 +102,6 @@ set(EXEC_FILES
assert_num_rows_node.cpp
)
-if (WITH_MYSQL)
- set(EXEC_FILES
- ${EXEC_FILES}
- mysql_scan_node.cpp
- mysql_scanner.cpp
- )
-endif()
if (WITH_LZO)
set(EXEC_FILES ${EXEC_FILES}
diff --git a/be/src/vec/CMakeLists.txt b/be/src/vec/CMakeLists.txt
index 0e83f98167..56cec51c0b 100644
--- a/be/src/vec/CMakeLists.txt
+++ b/be/src/vec/CMakeLists.txt
@@ -281,6 +281,13 @@ set(VEC_FILES
exec/format/table/table_format_reader.cpp
exec/format/table/iceberg_reader.cpp)
+if (WITH_MYSQL)
+ set(VEC_FILES
+ ${VEC_FILES}
+ exec/scan/mysql_scanner.cpp
+ )
+endif ()
+
add_library(Vec STATIC
${VEC_FILES}
)
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@doris.apache.org
For additional commands, e-mail: commits-help@doris.apache.org