You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2024/02/15 02:07:54 UTC
(impala) branch master updated (4b4f210c0 -> 1dbde2c19)
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 4b4f210c0 IMPALA-12803: Fix missing exchange lines in query timeline
new 13030f840 IMPALA-12796: Add is_footer_only in TFileSplitGeneratorSpec
new 1dbde2c19 IMPALA-12790: Fix overestimation in ScanNode.getInputCardinality
The 2 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/hdfs-columnar-scanner.cc | 5 +
be/src/exec/hdfs-columnar-scanner.h | 4 +
be/src/exec/parquet/hdfs-parquet-scanner.cc | 1 +
be/src/scheduling/scheduler-test-util.cc | 1 +
be/src/scheduling/scheduler-test-util.h | 10 +-
be/src/scheduling/scheduler-test.cc | 21 +++
be/src/scheduling/scheduler.cc | 13 +-
common/thrift/PlanNodes.thrift | 4 +
.../org/apache/impala/planner/HdfsScanNode.java | 31 ++--
.../java/org/apache/impala/planner/ScanNode.java | 32 +++-
.../impala/util/MaxRowsProcessedVisitor.java | 5 +-
.../org/apache/impala/planner/PlannerTest.java | 9 +-
.../queries/PlannerTest/data-source-tables.test | 15 --
.../queries/PlannerTest/kudu.test | 29 ---
.../queries/PlannerTest/small-query-opt.test | 199 +++++++++++++++++++++
.../queries/PlannerTest/tpcds-processing-cost.test | 61 +++++++
tests/query_test/test_codegen.py | 3 +-
17 files changed, 364 insertions(+), 79 deletions(-)
(impala) 02/02: IMPALA-12790: Fix overestimation in ScanNode.getInputCardinality
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 1dbde2c197a26971aabed7d6bb387aac7c64d83f
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Mon Feb 5 12:10:27 2024 -0800
IMPALA-12790: Fix overestimation in ScanNode.getInputCardinality
ScanNode.getInputCardinality() can overestimate if LIMIT is large. If
table cardinality stats exist and it is less than the LIMIT, then the
table cardinality should be returned.
This patch adds those fixes and more test cases under
small-query-opt.test. This also moves tests added by IMPALA-5602 to
small-query-opt.test since only
PlannerTest.testSmallQueryOptimization runs with
EXEC_SINGLE_NODE_ROWS_THRESHOLD > 0.
Testing:
- Pass FE tests.
Change-Id: Icc5b39a7684fb8748185349d0b80baf8dcd6b126
Reviewed-on: http://gerrit.cloudera.org:8080/21005
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../java/org/apache/impala/planner/ScanNode.java | 32 +++-
.../impala/util/MaxRowsProcessedVisitor.java | 5 +-
.../org/apache/impala/planner/PlannerTest.java | 9 +-
.../queries/PlannerTest/data-source-tables.test | 15 --
.../queries/PlannerTest/kudu.test | 29 ---
.../queries/PlannerTest/small-query-opt.test | 199 +++++++++++++++++++++
.../queries/PlannerTest/tpcds-processing-cost.test | 61 +++++++
tests/query_test/test_codegen.py | 3 +-
8 files changed, 293 insertions(+), 60 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/planner/ScanNode.java b/fe/src/main/java/org/apache/impala/planner/ScanNode.java
index ea9b6e17c..0dbef697a 100644
--- a/fe/src/main/java/org/apache/impala/planner/ScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/ScanNode.java
@@ -324,20 +324,36 @@ abstract public class ScanNode extends PlanNode {
return result;
}
+ /**
+ * Return true if this scan node has limit, no scan conjunct,
+ * and no storage layer conjunct. Otherwise, return false.
+ * This is mainly used to determine whether a scan's input cardinality can be bounded by
+ * the LIMIT clause or not.
+ */
+ public boolean hasSimpleLimit() {
+ return hasLimit() && !hasScanConjuncts() && !hasStorageLayerConjuncts();
+ }
+
+ private long capInputCardinalityWithLimit(long inputCardinality) {
+ if (hasSimpleLimit()) {
+ if (inputCardinality < 0) {
+ return getLimit();
+ } else {
+ return Math.min(getLimit(), inputCardinality);
+ }
+ }
+ return inputCardinality;
+ }
+
@Override
public long getInputCardinality() {
- if (!hasScanConjuncts() && !hasStorageLayerConjuncts() && hasLimit()) {
- return getLimit();
- }
- return inputCardinality_;
+ return capInputCardinalityWithLimit(inputCardinality_);
}
// TODO: merge this with getInputCardinality().
public long getFilteredInputCardinality() {
- if (!hasScanConjuncts() && !hasStorageLayerConjuncts() && hasLimit()) {
- return getLimit();
- }
- return filteredInputCardinality_ > -1 ? filteredInputCardinality_ : inputCardinality_;
+ return capInputCardinalityWithLimit(
+ filteredInputCardinality_ > -1 ? filteredInputCardinality_ : inputCardinality_);
}
@Override
diff --git a/fe/src/main/java/org/apache/impala/util/MaxRowsProcessedVisitor.java b/fe/src/main/java/org/apache/impala/util/MaxRowsProcessedVisitor.java
index fb10a74f1..de4592a80 100644
--- a/fe/src/main/java/org/apache/impala/util/MaxRowsProcessedVisitor.java
+++ b/fe/src/main/java/org/apache/impala/util/MaxRowsProcessedVisitor.java
@@ -50,12 +50,11 @@ public class MaxRowsProcessedVisitor implements Visitor<PlanNode> {
boolean missingStats = scan.isTableMissingStats() || scan.hasCorruptTableStats();
// In the absence of collection stats, treat scans on collections as if they
// have no limit.
- if (scan.isAccessingCollectionType() ||
- (missingStats && !(scan.hasLimit() && !scan.hasScanConjuncts() &&
- !scan.hasStorageLayerConjuncts()))) {
+ if (scan.isAccessingCollectionType() || (missingStats && !scan.hasSimpleLimit())) {
valid_ = false;
return;
}
+ Preconditions.checkState(numRows > -1);
maxRowsProcessed_ = Math.max(maxRowsProcessed_, numRows);
maxRowsProcessedPerNode_ = Math.max(maxRowsProcessedPerNode_,
(long)Math.ceil(numRows / (double)numNodes));
diff --git a/fe/src/test/java/org/apache/impala/planner/PlannerTest.java b/fe/src/test/java/org/apache/impala/planner/PlannerTest.java
index 8a3d077cc..b28d489eb 100644
--- a/fe/src/test/java/org/apache/impala/planner/PlannerTest.java
+++ b/fe/src/test/java/org/apache/impala/planner/PlannerTest.java
@@ -577,7 +577,11 @@ public class PlannerTest extends PlannerTestBase {
public void testSmallQueryOptimization() {
TQueryOptions options = new TQueryOptions();
options.setExec_single_node_rows_threshold(8);
- runPlannerTestFile("small-query-opt", options);
+ addTestDb("kudu_planner_test", "Test DB for Kudu Planner.");
+ addTestTable("CREATE EXTERNAL TABLE kudu_planner_test.no_stats STORED AS KUDU "
+ + "TBLPROPERTIES ('kudu.table_name' = 'impala::functional_kudu.alltypes');");
+ runPlannerTestFile("small-query-opt", options,
+ ImmutableSet.of(PlannerTestOption.VALIDATE_CARDINALITY));
}
@Test
@@ -679,9 +683,6 @@ public class PlannerTest extends PlannerTestBase {
options.unsetEnabled_runtime_filter_types();
options.addToEnabled_runtime_filter_types(TRuntimeFilterType.BLOOM);
options.addToEnabled_runtime_filter_types(TRuntimeFilterType.MIN_MAX);
- addTestDb("kudu_planner_test", "Test DB for Kudu Planner.");
- addTestTable("CREATE EXTERNAL TABLE kudu_planner_test.no_stats STORED AS KUDU " +
- "TBLPROPERTIES ('kudu.table_name' = 'impala::functional_kudu.alltypes');");
runPlannerTestFile("kudu", options);
}
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/data-source-tables.test b/testdata/workloads/functional-planner/queries/PlannerTest/data-source-tables.test
index 5fcbdf732..cbbebcd94 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/data-source-tables.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/data-source-tables.test
@@ -103,18 +103,3 @@ PLAN-ROOT SINK
|
00:EMPTYSET
====
-# IMPALA-5602: If a query contains predicates that are all pushed to the datasource and
-# there is a limit, then the query should not incorrectly run with 'small query'
-# optimization.
-select * from functional.alltypes_datasource where id = 1 limit 15
----- DISTRIBUTEDPLAN
-PLAN-ROOT SINK
-|
-01:EXCHANGE [UNPARTITIONED]
-| limit: 15
-|
-00:SCAN DATA SOURCE [functional.alltypes_datasource]
-data source predicates: id = 1
- limit: 15
- row-size=116B cardinality=15
-====
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test b/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
index ce20acd33..69fb70ce1 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/kudu.test
@@ -511,35 +511,6 @@ INSERT INTO KUDU [default.t]
partitions=24/24 files=24 size=478.45KB
row-size=89B cardinality=7.30K
====
-# IMPALA-5602: If a query contains predicates that are all pushed to kudu and there is a
-# limit, then the query should not incorrectly run with 'small query' optimization.
-select * from functional_kudu.alltypesagg where tinyint_col = 9 limit 10;
----- DISTRIBUTEDPLAN
-PLAN-ROOT SINK
-|
-01:EXCHANGE [UNPARTITIONED]
-| limit: 10
-|
-00:SCAN KUDU [functional_kudu.alltypesagg_idx]
- kudu predicates: functional_kudu.alltypesagg_idx.tinyint_col = 9
- limit: 10
- row-size=103B cardinality=10
-====
-# IMPALA-5602: If a query contains predicates that are all pushed to kudu, there is a
-# limit, and no table stats, then the query should not incorrectly run with 'small query'
-# optimization.
-select * from kudu_planner_test.no_stats where tinyint_col = 9 limit 10;
----- DISTRIBUTEDPLAN
-PLAN-ROOT SINK
-|
-01:EXCHANGE [UNPARTITIONED]
-| limit: 10
-|
-00:SCAN KUDU [kudu_planner_test.no_stats]
- kudu predicates: tinyint_col = 9
- limit: 10
- row-size=88B cardinality=10
-====
# Insert into an unpartitioned table, shouldn't partition/sort
insert into tpch_kudu.nation
select * from tpch_parquet.nation
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/small-query-opt.test b/testdata/workloads/functional-planner/queries/PlannerTest/small-query-opt.test
index 3817a82f4..9ce15f4ef 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/small-query-opt.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/small-query-opt.test
@@ -449,3 +449,202 @@ PLAN-ROOT SINK
partition key scan
row-size=4B cardinality=1.82K
====
+# testbl is an empty table. It does not have stats computed and query has no limit.
+# Query should run without 'small query' optimization.
+select * from functional.testtbl;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+|
+00:SCAN HDFS [functional.testtbl]
+ HDFS partitions=1/1 files=0 size=0B
+ row-size=24B cardinality=0
+====
+# testbl is an empty table, but has limit less than 8.
+# 'small query' optimization is enabled.
+select * from functional.testtbl limit 4;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional.testtbl]
+ HDFS partitions=1/1 files=0 size=0B
+ limit: 4
+ row-size=24B cardinality=0
+====
+# testbl does not have stats computed.
+# But its input cardinality is estimated to be 0 and query has limit.
+# 'small query' optimization is enabled.
+select * from functional.testtbl limit 1000;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional.testtbl]
+ HDFS partitions=1/1 files=0 size=0B
+ limit: 1000
+ row-size=24B cardinality=0
+====
+# functional.tinytable has no stats and not empty.
+# Query should run without 'small query' optimization.
+select * from functional.tinytable;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+|
+00:SCAN HDFS [functional.tinytable]
+ HDFS partitions=1/1 files=1 size=38B
+ row-size=24B cardinality=2
+====
+# tinytable has no stats, but has limit less than 8.
+# 'small query' optimization is enabled.
+select * from functional.tinytable limit 4;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional.tinytable]
+ HDFS partitions=1/1 files=1 size=38B
+ limit: 4
+ row-size=24B cardinality=2
+====
+# tinytable does not have stats computed.
+# But its input cardinality is estimated to be 2 and query has limit.
+# 'small query' optimization is enabled.
+select * from functional.tinytable limit 1000;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+00:SCAN HDFS [functional.tinytable]
+ HDFS partitions=1/1 files=1 size=38B
+ limit: 1000
+ row-size=24B cardinality=2
+====
+# functional_kudu.tinytable has stats and its input cardinality is 3.
+# Query should run with 'small query' optimization even without limit.
+select * from functional_kudu.tinytable;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+00:SCAN KUDU [functional_kudu.tinytable]
+ row-size=43B cardinality=3
+====
+# If input cardinality < EXEC_SINGLE_NODE_ROWS_THRESHOLD < limit,
+# then the query should run with 'small query' optimization.
+select * from functional_kudu.tinytable limit 1000;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+00:SCAN KUDU [functional_kudu.tinytable]
+ limit: 1000
+ row-size=43B cardinality=3
+====
+# kudu_planner_test.no_stats has no stats computed and query has no limit.
+# Query should run without 'small query' optimization.
+select * from kudu_planner_test.no_stats;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+|
+00:SCAN KUDU [kudu_planner_test.no_stats]
+ row-size=88B cardinality=unavailable
+====
+# kudu_planner_test.no_stats has no stats computed and
+# query has limit less than 8. 'small query' optimization is enabled.
+select * from kudu_planner_test.no_stats limit 7;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+00:SCAN KUDU [kudu_planner_test.no_stats]
+ limit: 7
+ row-size=88B cardinality=7
+====
+# kudu_planner_test.no_stats has no stats computed and query has limit more than 8.
+# 'small query' optimization is disabled.
+select * from kudu_planner_test.no_stats limit 1000;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+| limit: 1000
+|
+00:SCAN KUDU [kudu_planner_test.no_stats]
+ limit: 1000
+ row-size=88B cardinality=1.00K
+====
+# IMPALA-5602: If a query contains predicates that are all pushed to kudu and there is a
+# limit, then the query should not incorrectly run with 'small query' optimization.
+select * from functional_kudu.alltypesagg where tinyint_col = 9 limit 7;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+| limit: 7
+|
+00:SCAN KUDU [functional_kudu.alltypesagg_idx]
+ kudu predicates: functional_kudu.alltypesagg_idx.tinyint_col = 9
+ limit: 7
+ row-size=103B cardinality=7
+====
+# IMPALA-5602: If a query contains predicates that are all pushed to kudu, there is a
+# limit, and no table stats, then the query should not incorrectly run with 'small query'
+# optimization.
+select * from kudu_planner_test.no_stats where tinyint_col = 9 limit 7;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+| limit: 7
+|
+00:SCAN KUDU [kudu_planner_test.no_stats]
+ kudu predicates: tinyint_col = 9
+ limit: 7
+ row-size=88B cardinality=0
+====
+# alltypes_datasource has estimated input cardinality of 5000.
+# Query should run without 'small query' optimization.
+select * from functional.alltypes_datasource;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+|
+00:SCAN DATA SOURCE [functional.alltypes_datasource]
+ row-size=116B cardinality=5.00K
+====
+# limit is less than 8. 'small query' optimization is enabled.
+select * from functional.alltypes_datasource limit 7;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+00:SCAN DATA SOURCE [functional.alltypes_datasource]
+ limit: 7
+ row-size=116B cardinality=7
+====
+# limit is more than 8. 'small query' optimization is disabled.
+select * from functional.alltypes_datasource limit 1000;
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+| limit: 1000
+|
+00:SCAN DATA SOURCE [functional.alltypes_datasource]
+ limit: 1000
+ row-size=116B cardinality=1.00K
+====
+# IMPALA-5602: If a query contains predicates that are all pushed to the datasource and
+# there is a limit, then the query should not incorrectly run with 'small query'
+# optimization.
+select * from functional.alltypes_datasource where id = 1 limit 7
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+01:EXCHANGE [UNPARTITIONED]
+| limit: 7
+|
+00:SCAN DATA SOURCE [functional.alltypes_datasource]
+data source predicates: id = 1
+ limit: 7
+ row-size=116B cardinality=7
+====
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds-processing-cost.test b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds-processing-cost.test
index 4c2104f26..4a71ca2a7 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/tpcds-processing-cost.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/tpcds-processing-cost.test
@@ -139,3 +139,64 @@ max-parallelism=21 segment-costs=[194962048, 2276043]
tuple-ids=0 row-size=24B cardinality=2.88M cost=91267504
in pipelines: 00(GETNEXT)
====
+select * from income_band;
+---- PARALLELPLANS
+Max Per-Host Resource Reservation: Memory=4.02MB Threads=2
+Per-Host Resource Estimates: Memory=20MB
+F01:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Instance Resources: mem-estimate=4.02MB mem-reservation=4.00MB thread-reservation=1
+| max-parallelism=1 segment-costs=[61]
+PLAN-ROOT SINK
+| output exprs: tpcds_partitioned_parquet_snap.income_band.ib_income_band_sk, tpcds_partitioned_parquet_snap.income_band.ib_lower_bound, tpcds_partitioned_parquet_snap.income_band.ib_upper_bound
+| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0 cost=60
+|
+01:EXCHANGE [UNPARTITIONED]
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=0 row-size=12B cardinality=20 cost=1
+| in pipelines: 00(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
+Per-Instance Resources: mem-estimate=16.06MB mem-reservation=24.00KB thread-reservation=1
+max-parallelism=1 segment-costs=[50002]
+00:SCAN HDFS [tpcds_partitioned_parquet_snap.income_band, RANDOM]
+ HDFS partitions=1/1 files=1 size=1.21KB
+ stored statistics:
+ table: rows=20 size=1.21KB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=20
+ mem-estimate=16.00MB mem-reservation=24.00KB thread-reservation=0
+ tuple-ids=0 row-size=12B cardinality=20 cost=50001
+ in pipelines: 00(GETNEXT)
+====
+# Scan cost should be exactly the same as select star without limit.
+select * from income_band limit 1000000000;
+---- PARALLELPLANS
+Max Per-Host Resource Reservation: Memory=4.02MB Threads=2
+Per-Host Resource Estimates: Memory=20MB
+F01:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Instance Resources: mem-estimate=4.02MB mem-reservation=4.00MB thread-reservation=1
+| max-parallelism=1 segment-costs=[61]
+PLAN-ROOT SINK
+| output exprs: tpcds_partitioned_parquet_snap.income_band.ib_income_band_sk, tpcds_partitioned_parquet_snap.income_band.ib_lower_bound, tpcds_partitioned_parquet_snap.income_band.ib_upper_bound
+| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0 cost=60
+|
+01:EXCHANGE [UNPARTITIONED]
+| limit: 1000000000
+| mem-estimate=16.00KB mem-reservation=0B thread-reservation=0
+| tuple-ids=0 row-size=12B cardinality=20 cost=1
+| in pipelines: 00(GETNEXT)
+|
+F00:PLAN FRAGMENT [RANDOM] hosts=1 instances=1
+Per-Instance Resources: mem-estimate=16.06MB mem-reservation=24.00KB thread-reservation=1
+max-parallelism=1 segment-costs=[50002]
+00:SCAN HDFS [tpcds_partitioned_parquet_snap.income_band, RANDOM]
+ HDFS partitions=1/1 files=1 size=1.21KB
+ stored statistics:
+ table: rows=20 size=1.21KB
+ columns: all
+ extrapolated-rows=disabled max-scan-range-rows=20
+ limit: 1000000000
+ mem-estimate=16.00MB mem-reservation=24.00KB thread-reservation=0
+ tuple-ids=0 row-size=12B cardinality=20 cost=50001
+ in pipelines: 00(GETNEXT)
+====
diff --git a/tests/query_test/test_codegen.py b/tests/query_test/test_codegen.py
index 81a56b3b0..3ef4ad9c9 100644
--- a/tests/query_test/test_codegen.py
+++ b/tests/query_test/test_codegen.py
@@ -47,7 +47,8 @@ class TestCodegen(ImpalaTestSuite):
def test_select_node_codegen(self, vector):
"""Test that select node is codegened"""
result = self.execute_query('select * from (select * from functional.alltypes '
- 'limit 1000000) t1 where int_col > 10 limit 10')
+ 'limit 1000000) t1 where int_col > 10 limit 10',
+ {'disable_codegen_rows_threshold': 7000})
exec_options = get_node_exec_options(result.runtime_profile, 1)
# Make sure test fails if there are no exec options in the profile for the node
assert len(exec_options) > 0
(impala) 01/02: IMPALA-12796: Add is_footer_only in TFileSplitGeneratorSpec
Posted by jo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 13030f840a23e67c5e9923e8b1abab3b717c106a
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Thu Feb 8 12:14:50 2024 -0800
IMPALA-12796: Add is_footer_only in TFileSplitGeneratorSpec
Several tests in test_scanners.py failed by wrong row counts with S3
target filesystem after IMPALA-12631. S3 filesystem does not have block.
Planner will produce TFileSplitGeneratorSpec instead of
TScanRangeLocationList, and IMPALA-12631 miss to address necessary
changes in TFileSplitGeneratorSpec. Meanwhile, it already changed the
behavior of hdfs-parquet-scanner.cc. For each scan range, the new code
will loop file_metadata_.row_groups, while the old code just take one
entry of file_metadata_.row_groups after calling NextRowGroup().
This patch address the issue by adding is_footer_only field in
TFileSplitGeneratorSpec schedule accordingly in schedule.cc. This also
add field 'is_footer_scanner_' in hdfs-columnar-scanner.h to check that
optimized count star only applied with footer range.
Testing:
- Pass core tests with S3 target filesystem.
Change-Id: Iaa6e3c14debe68cf601131c6594774c8c695923e
Reviewed-on: http://gerrit.cloudera.org:8080/21021
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/exec/hdfs-columnar-scanner.cc | 5 ++++
be/src/exec/hdfs-columnar-scanner.h | 4 +++
be/src/exec/parquet/hdfs-parquet-scanner.cc | 1 +
be/src/scheduling/scheduler-test-util.cc | 1 +
be/src/scheduling/scheduler-test-util.h | 10 +++++--
be/src/scheduling/scheduler-test.cc | 21 +++++++++++++++
be/src/scheduling/scheduler.cc | 13 +++++++--
common/thrift/PlanNodes.thrift | 4 +++
.../org/apache/impala/planner/HdfsScanNode.java | 31 +++++++++++-----------
9 files changed, 71 insertions(+), 19 deletions(-)
diff --git a/be/src/exec/hdfs-columnar-scanner.cc b/be/src/exec/hdfs-columnar-scanner.cc
index 518091612..7a1ad7744 100644
--- a/be/src/exec/hdfs-columnar-scanner.cc
+++ b/be/src/exec/hdfs-columnar-scanner.cc
@@ -81,6 +81,11 @@ HdfsColumnarScanner::~HdfsColumnarScanner() {}
Status HdfsColumnarScanner::Open(ScannerContext* context) {
RETURN_IF_ERROR(HdfsScanner::Open(context));
+ // Memorize 'is_footer_scanner_' here since 'stream_' can be released early.
+ const io::ScanRange* range = stream_->scan_range();
+ is_footer_scanner_ =
+ range->offset() + range->bytes_to_read() >= stream_->file_desc()->file_length;
+
RuntimeProfile* profile = scan_node_->runtime_profile();
num_cols_counter_ = PROFILE_NumColumns.Instantiate(profile);
num_scanners_with_no_reads_counter_ =
diff --git a/be/src/exec/hdfs-columnar-scanner.h b/be/src/exec/hdfs-columnar-scanner.h
index a1d0b2903..00276d5be 100644
--- a/be/src/exec/hdfs-columnar-scanner.h
+++ b/be/src/exec/hdfs-columnar-scanner.h
@@ -60,6 +60,10 @@ class HdfsColumnarScanner : public HdfsScanner {
/// top-level tuples. See AssembleRows() in the derived classes.
boost::scoped_ptr<ScratchTupleBatch> scratch_batch_;
+ /// Indicate whether this is a footer scanner or not.
+ /// Assigned in HdfsColumnarScanner::Open().
+ bool is_footer_scanner_ = false;
+
/// Scan range for the metadata.
const io::ScanRange* metadata_range_ = nullptr;
diff --git a/be/src/exec/parquet/hdfs-parquet-scanner.cc b/be/src/exec/parquet/hdfs-parquet-scanner.cc
index d69e985d4..6d41dc147 100644
--- a/be/src/exec/parquet/hdfs-parquet-scanner.cc
+++ b/be/src/exec/parquet/hdfs-parquet-scanner.cc
@@ -435,6 +435,7 @@ Status HdfsParquetScanner::GetNextInternal(RowBatch* row_batch) {
DCHECK(parse_status_.ok()) << parse_status_.GetDetail();
if (scan_node_->optimize_parquet_count_star()) {
// Populate the single slot with the Parquet num rows statistic.
+ DCHECK(is_footer_scanner_);
int64_t tuple_buf_size;
uint8_t* tuple_buf;
int capacity = 1;
diff --git a/be/src/scheduling/scheduler-test-util.cc b/be/src/scheduling/scheduler-test-util.cc
index 14e7eb620..97c32e724 100644
--- a/be/src/scheduling/scheduler-test-util.cc
+++ b/be/src/scheduling/scheduler-test-util.cc
@@ -411,6 +411,7 @@ void Plan::BuildScanRangeSpec(const TableName& table_name,
thrift_spec->__set_file_desc(thrift_file);
thrift_spec->__set_max_block_size(spec.block_size);
thrift_spec->__set_is_splittable(spec.is_splittable);
+ thrift_spec->__set_is_footer_only(spec.is_footer_only);
int32_t partition_path_hash = static_cast<int32_t>(HashUtil::Hash(partition_path.data(),
partition_path.length(), 0));
thrift_spec->__set_partition_path_hash(partition_path_hash);
diff --git a/be/src/scheduling/scheduler-test-util.h b/be/src/scheduling/scheduler-test-util.h
index 2b6cab078..7696e2c2b 100644
--- a/be/src/scheduling/scheduler-test-util.h
+++ b/be/src/scheduling/scheduler-test-util.h
@@ -202,8 +202,12 @@ struct Block {
struct FileSplitGeneratorSpec {
FileSplitGeneratorSpec() {}
- FileSplitGeneratorSpec(int64_t length, int64_t block, bool splittable)
- : length(length), block_size(block), is_splittable(splittable) {}
+ FileSplitGeneratorSpec(
+ int64_t length, int64_t block, bool splittable, bool is_footer_only = false)
+ : length(length),
+ block_size(block),
+ is_splittable(splittable),
+ is_footer_only(is_footer_only) {}
/// Length of file for which to generate file splits.
int64_t length = DEFAULT_FILE_SIZE;
@@ -213,6 +217,8 @@ struct FileSplitGeneratorSpec {
bool is_splittable = true;
+ bool is_footer_only = false;
+
static const int64_t DEFAULT_FILE_SIZE;
static const int64_t DEFAULT_BLOCK_SIZE;
};
diff --git a/be/src/scheduling/scheduler-test.cc b/be/src/scheduling/scheduler-test.cc
index 3a132a784..b1ee871df 100644
--- a/be/src/scheduling/scheduler-test.cc
+++ b/be/src/scheduling/scheduler-test.cc
@@ -642,6 +642,27 @@ TEST_F(SchedulerTest, TestGeneratedVariableSizeSplit) {
EXPECT_EQ(300, result.NumTotalAssignedBytes());
}
+TEST_F(SchedulerTest, TestGeneratedVariableSizeSplitFooterOnly) {
+ Cluster cluster;
+
+ cluster.AddHosts(3, true, true);
+
+ Schema schema(cluster);
+ schema.AddFileSplitGeneratorSpecs(
+ "T", {{100, 100, false, false}, {100, 1, true, true}, {100, 10, true, true}});
+
+ Plan plan(schema);
+ plan.AddTableScan("T");
+ plan.SetRandomReplica(true);
+
+ Result result(plan);
+ SchedulerWrapper scheduler(plan);
+ ASSERT_OK(scheduler.Compute(&result));
+
+ EXPECT_EQ(3, result.NumTotalAssignments());
+ EXPECT_EQ(111, result.NumTotalAssignedBytes());
+}
+
TEST_F(SchedulerTest, TestBlockAndGenerateSplit) {
Cluster cluster;
diff --git a/be/src/scheduling/scheduler.cc b/be/src/scheduling/scheduler.cc
index 7f16a4058..287407389 100644
--- a/be/src/scheduling/scheduler.cc
+++ b/be/src/scheduling/scheduler.cc
@@ -121,8 +121,17 @@ Status Scheduler::GenerateScanRanges(const vector<TFileSplitGeneratorSpec>& spec
long scan_range_offset = 0;
long remaining = fb_desc->length();
- long scan_range_length = std::min(spec.max_block_size, fb_desc->length());
- if (!spec.is_splittable) scan_range_length = fb_desc->length();
+ long scan_range_length = fb_desc->length();
+
+ if (spec.is_splittable) {
+ scan_range_length = std::min(spec.max_block_size, fb_desc->length());
+ if (spec.is_footer_only) {
+ scan_range_offset = fb_desc->length() - scan_range_length;
+ remaining = scan_range_length;
+ }
+ } else {
+ DCHECK(!spec.is_footer_only);
+ }
while (remaining > 0) {
THdfsFileSplit hdfs_scan_range;
diff --git a/common/thrift/PlanNodes.thrift b/common/thrift/PlanNodes.thrift
index c957a5af7..cfee32a5a 100644
--- a/common/thrift/PlanNodes.thrift
+++ b/common/thrift/PlanNodes.thrift
@@ -271,6 +271,10 @@ struct TFileSplitGeneratorSpec {
// Hash of the partition path
5: required i32 partition_path_hash
+
+ // True if only footer range (the last block in file) is needed.
+ // If True, is_splittable must also be True as well.
+ 6: required bool is_footer_only
}
// Specification of an individual data range which is held in its entirety
diff --git a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
index ed926adef..60e4abc04 100644
--- a/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
+++ b/fe/src/main/java/org/apache/impala/planner/HdfsScanNode.java
@@ -1253,23 +1253,24 @@ public class HdfsScanNode extends ScanNode {
totalBytesPerFsEC_.merge(fsType, fileDesc.getFileLength(), Long::sum);
}
+ // If parquet count star optimization is enabled, we only need the
+ // 'RowGroup.num_rows' in file metadata, thus only the scan range that contains
+ // a file footer is required.
+ // IMPALA-8834 introduced the optimization for partition key scan by generating
+ // one scan range for each HDFS file. With Parquet and ORC, we only need to get
+ // the scan range that contains a file footer for short-circuiting.
+ boolean isFooterOnly = countStarSlot_ != null
+ || (isPartitionKeyScan_
+ && (partition.getFileFormat().isParquetBased()
+ || partition.getFileFormat() == HdfsFileFormat.ORC));
+
if (!fsHasBlocks) {
Preconditions.checkState(fileDesc.getNumFileBlocks() == 0);
generateScanRangeSpecs(
- partition, partitionLocation, fileDesc, scanRangeBytesLimit);
+ partition, partitionLocation, fileDesc, scanRangeBytesLimit, isFooterOnly);
} else {
// Skips files that have no associated blocks.
if (fileDesc.getNumFileBlocks() == 0) continue;
- // If parquet count star optimization is enabled, we only need the
- // 'RowGroup.num_rows' in file metadata, thus only the scan range that contains
- // a file footer is required.
- // IMPALA-8834 introduced the optimization for partition key scan by generating
- // one scan range for each HDFS file. With Parquet and ORC, we only need to get
- // the scan range that contains a file footer for short-circuiting.
- boolean isFooterOnly = countStarSlot_ != null
- || (isPartitionKeyScan_
- && (partition.getFileFormat().isParquetBased()
- || partition.getFileFormat() == HdfsFileFormat.ORC));
Pair<Boolean, Long> result =
transformBlocksToScanRanges(partition, partitionLocation, fsType, fileDesc,
fsHasBlocks, scanRangeBytesLimit, analyzer, isFooterOnly);
@@ -1371,17 +1372,17 @@ public class HdfsScanNode extends ScanNode {
* FeFsPartition can be expensive.
*/
private void generateScanRangeSpecs(FeFsPartition partition, String partitionLocation,
- FileDescriptor fileDesc, long maxBlockSize) {
+ FileDescriptor fileDesc, long maxBlockSize, boolean isFooterOnly) {
Preconditions.checkArgument(fileDesc.getNumFileBlocks() == 0);
Preconditions.checkArgument(maxBlockSize > 0);
if (fileDesc.getFileLength() <= 0) return;
boolean splittable = partition.getFileFormat().isSplittable(
HdfsCompression.fromFileName(fileDesc.getPath()));
+ isFooterOnly &= splittable;
// Hashing must use String.hashCode() for consistency.
int partitionHash = partitionLocation.hashCode();
- TFileSplitGeneratorSpec splitSpec = new TFileSplitGeneratorSpec(
- fileDesc.toThrift(), maxBlockSize, splittable, partition.getId(),
- partitionHash);
+ TFileSplitGeneratorSpec splitSpec = new TFileSplitGeneratorSpec(fileDesc.toThrift(),
+ maxBlockSize, splittable, partition.getId(), partitionHash, isFooterOnly);
scanRangeSpecs_.addToSplit_specs(splitSpec);
long scanRangeBytes = Math.min(maxBlockSize, fileDesc.getFileLength());
if (splittable && !isPartitionKeyScan_) {