You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by st...@apache.org on 2025/04/29 07:44:35 UTC
(impala) branch master updated (96ae16b60 -> 56b465d91)
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 96ae16b60 IMPALA-13584: Add option to shows num row report in impala-shell
new d8e18fac6 IMPALA-13991: Skip CROSS_JOIN rewrite if subquery is in disjunctive
new c1aac4b3a IMPALA-13873: Missing equivalence conjunct in aggregation node with inline views
new 295f74ef1 IMPALA-13989: Invalidate table on rename failure
new 0e3ae5c33 IMPALA-13996: Deflake test_too_many_files by creating dedicate tables
new 56b465d91 IMPALA-13829: Postpone catalog deleteLog GC for waitForHmsEvent requests
The 5 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/catalog/catalog-server.cc | 13 +++
be/src/util/backend-gflag-util.cc | 2 +
common/thrift/BackendGflags.thrift | 2 +
common/thrift/CatalogService.thrift | 3 +
.../apache/impala/analysis/MultiAggregateInfo.java | 24 ++++
.../org/apache/impala/analysis/StmtRewriter.java | 9 +-
.../org/apache/impala/catalog/CatalogDeltaLog.java | 49 ++++++--
.../impala/catalog/CatalogServiceCatalog.java | 2 +
.../apache/impala/planner/SingleNodePlanner.java | 9 +-
.../apache/impala/service/CatalogOpExecutor.java | 60 ++++++----
.../java/org/apache/impala/service/FeSupport.java | 1 -
.../java/org/apache/impala/service/Frontend.java | 4 +-
.../java/org/apache/impala/service/JniCatalog.java | 5 -
.../java/org/apache/impala/util/DebugUtils.java | 5 +
.../queries/PlannerTest/nested-loop-join.test | 36 ++++++
.../queries/PlannerTest/subquery-rewrite.test | 17 +--
.../queries/PlannerTest/union.test | 125 +++++++++++++++++++++
.../queries/QueryTest/aggregation.test | 45 ++++++++
.../queries/QueryTest/single-node-nlj.test | 17 +++
tests/custom_cluster/test_concurrent_ddls.py | 16 +--
tests/custom_cluster/test_concurrent_rename.py | 77 +++++++++++++
tests/custom_cluster/test_local_catalog.py | 14 ++-
tests/metadata/test_event_processing.py | 64 ++++++++++-
23 files changed, 529 insertions(+), 70 deletions(-)
create mode 100644 tests/custom_cluster/test_concurrent_rename.py
(impala) 01/05: IMPALA-13991: Skip CROSS_JOIN rewrite if subquery is in disjunctive
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit d8e18fac619a129270634513359b3386212ff919
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Thu Apr 24 15:50:42 2025 -0700
IMPALA-13991: Skip CROSS_JOIN rewrite if subquery is in disjunctive
Inside StmtRewriter.mergeExpr() there is an optimization that set
JoinOperator.CROSS_JOIN under certain scenario. This patch adds criteria
to SKIP that rewrite if subquery is coming from inside disjunctive
expression, regardless of joinConjunct value. If joinConjunct is NOT
NULL, the inlineView maybe correlated through that joinConjunct. If
joinConjunct is NULL, then expr is a (NOT) EXISTS predicate. EXISTS
within a disjunct is not supported yet (see IMPALA-9931).
Testing:
- Add planner and query test for the corner case.
Before this patch, the query return wrong result.
- Fixed wrong testcase in subquery-rewrite.test.
Change-Id: Iac0deb0b2fb1536684cce2e004156a20b769b9ab
Reviewed-on: http://gerrit.cloudera.org:8080/22815
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../org/apache/impala/analysis/StmtRewriter.java | 9 ++++--
.../queries/PlannerTest/nested-loop-join.test | 36 ++++++++++++++++++++++
.../queries/PlannerTest/subquery-rewrite.test | 17 +++++-----
.../queries/QueryTest/single-node-nlj.test | 17 ++++++++++
4 files changed, 70 insertions(+), 9 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/analysis/StmtRewriter.java b/fe/src/main/java/org/apache/impala/analysis/StmtRewriter.java
index 5b0868d71..2ec24786e 100644
--- a/fe/src/main/java/org/apache/impala/analysis/StmtRewriter.java
+++ b/fe/src/main/java/org/apache/impala/analysis/StmtRewriter.java
@@ -22,7 +22,6 @@ import java.util.Collections;
import java.util.List;
import java.util.Set;
-import org.apache.impala.analysis.AnalysisContext.AnalysisResult;
import org.apache.impala.analysis.SetOperationStmt.SetOperand;
import org.apache.impala.analysis.SetOperationStmt.SetOperator;
import org.apache.impala.common.AnalysisException;
@@ -773,7 +772,13 @@ public class StmtRewriter {
break;
}
- if (!hasEqJoinPred && !inlineView.isCorrelated()) {
+ if (!hasEqJoinPred && !inlineView.isCorrelated() && !isDisjunctive) {
+ // IMPALA-13991: It is not safe to rewrite into CROSS_JOIN if
+ // isDisjunctive is True, regardless of joinConjunct value.
+ // If joinConjunct is NOT NULL, the inlineView maybe correlated through
+ // that joinConjunct.
+ // If joinConjunct is NULL, then expr is a (NOT) EXISTS predicate.
+ // EXISTS within a disjunct is not supported yet (see IMPALA-9931).
// TODO: IMPALA-9948: we could support non-equi joins here
// TODO: Remove this when independent subquery evaluation is implemented.
// TODO: IMPALA-5100 to cover all cases, we do let through runtime scalars with
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/nested-loop-join.test b/testdata/workloads/functional-planner/queries/PlannerTest/nested-loop-join.test
index 9ceac1560..18bb5bbbc 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/nested-loop-join.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/nested-loop-join.test
@@ -328,3 +328,39 @@ PLAN-ROOT SINK
HDFS partitions=24/24 files=24 size=478.45KB
row-size=4B cardinality=7.30K
====
+# IMPALA-13991: Disable CROSS_JOIN rewrite if there is a scalar subquery inside
+# disjunctive expression. right_tc is '2009-04-01 01:01:00'.
+select id, timestamp_col from functional.alltypes
+where (id in (1000, 1001))
+ or ((id in (1, 2))
+ and timestamp_col < (select hours_add(timestamp_col, 1) right_tc
+ from functional.alltypestiny
+ order by timestamp_col DESC limit 1));
+---- DISTRIBUTEDPLAN
+PLAN-ROOT SINK
+|
+06:EXCHANGE [UNPARTITIONED]
+|
+03:NESTED LOOP JOIN [LEFT OUTER JOIN, BROADCAST]
+| join predicates: timestamp_col < hours_add(timestamp_col, 1)
+| predicates: (id IN (1000, 1001)) OR ((id IN (1, 2)) AND hours_add(timestamp_col, 1) IS NOT NULL)
+| row-size=36B cardinality=7.30K
+|
+|--05:EXCHANGE [BROADCAST]
+| |
+| 04:MERGING-EXCHANGE [UNPARTITIONED]
+| | order by: timestamp_col DESC
+| | limit: 1
+| |
+| 02:TOP-N [LIMIT=1]
+| | order by: timestamp_col DESC
+| | row-size=16B cardinality=1
+| |
+| 01:SCAN HDFS [functional.alltypestiny]
+| HDFS partitions=4/4 files=4 size=460B
+| row-size=16B cardinality=8
+|
+00:SCAN HDFS [functional.alltypes]
+ HDFS partitions=24/24 files=24 size=478.45KB
+ row-size=20B cardinality=7.30K
+====
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/subquery-rewrite.test b/testdata/workloads/functional-planner/queries/PlannerTest/subquery-rewrite.test
index 8e83a774e..6946a5fee 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/subquery-rewrite.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/subquery-rewrite.test
@@ -4106,17 +4106,19 @@ PLAN-ROOT SINK
HDFS partitions=24/24 files=24 size=478.45KB
row-size=8B cardinality=7.30K
====
-# Uncorrelated subquery in BETWEEN in OR rewritten to INNER JOIN.
+# Uncorrelated subquery in BETWEEN in OR rewritten to LEFT OUTER JOIN.
+# Subquery will return 1.
select int_col from functional.alltypes t
-where (t.int_col is null or
+where (t.int_col = 0 or
(t.int_col between
(select min(int_col) from functional.alltypesagg t2)
and 2))
---- PLAN
PLAN-ROOT SINK
|
-03:NESTED LOOP JOIN [INNER JOIN]
-| predicates: t.int_col >= min(int_col), (t.int_col IS NULL OR min(int_col) IS NOT NULL AND t.int_col <= 2)
+03:NESTED LOOP JOIN [LEFT OUTER JOIN]
+| join predicates: t.int_col >= min(int_col)
+| predicates: (t.int_col = 0 OR min(int_col) IS NOT NULL AND t.int_col <= 2)
| row-size=8B cardinality=7.30K
|
|--02:AGGREGATE [FINALIZE]
@@ -4135,8 +4137,9 @@ PLAN-ROOT SINK
|
07:EXCHANGE [UNPARTITIONED]
|
-03:NESTED LOOP JOIN [INNER JOIN, BROADCAST]
-| predicates: t.int_col >= min(int_col), (t.int_col IS NULL OR min(int_col) IS NOT NULL AND t.int_col <= 2)
+03:NESTED LOOP JOIN [LEFT OUTER JOIN, BROADCAST]
+| join predicates: t.int_col >= min(int_col)
+| predicates: (t.int_col = 0 OR min(int_col) IS NOT NULL AND t.int_col <= 2)
| row-size=8B cardinality=7.30K
|
|--06:EXCHANGE [BROADCAST]
@@ -4149,7 +4152,7 @@ PLAN-ROOT SINK
| |
| 02:AGGREGATE
| | output: min(int_col)
-| | row-size=4B cardinality=1
+| | row-size=4B cardinality=3
| |
| 01:SCAN HDFS [functional.alltypesagg t2]
| HDFS partitions=11/11 files=11 size=814.73KB
diff --git a/testdata/workloads/functional-query/queries/QueryTest/single-node-nlj.test b/testdata/workloads/functional-query/queries/QueryTest/single-node-nlj.test
index 82370af53..143776538 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/single-node-nlj.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/single-node-nlj.test
@@ -222,4 +222,21 @@ limit 5
7295
---- TYPES
INT
+====
+---- QUERY
+# IMPALA-13991: Disable CROSS_JOIN rewrite if there is a scalar subquery inside
+# disjunctive expression. right_tc is '2009-04-01 01:01:00'.
+select id, timestamp_col from functional.alltypes
+where (id in (1000, 1001))
+ or ((id in (1, 2))
+ and timestamp_col < (select hours_add(timestamp_col, 1) right_tc
+ from functional.alltypestiny
+ order by timestamp_col DESC limit 1));
+---- RESULTS: VERIFY_IS_EQUAL_SORTED
+1,2009-01-01 00:01:00
+2,2009-01-01 00:02:00.100000000
+1000,2009-04-11 01:40:04.500000000
+1001,2009-04-11 01:41:04.500000000
+---- TYPES
+INT,TIMESTAMP
====
\ No newline at end of file
(impala) 05/05: IMPALA-13829: Postpone catalog deleteLog GC for waitForHmsEvent requests
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 56b465d91ff39d6c850a82ce87ca6115a0f3c37d
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Fri Apr 25 09:58:12 2025 +0800
IMPALA-13829: Postpone catalog deleteLog GC for waitForHmsEvent requests
When a db/table is removed in the catalog cache, catalogd assigns it a
new catalog version and put it into the deleteLog. This is used for the
catalog update thread to collect deletion updates. Once the catalog
update thread collects a range of updates, it triggers GC in the
deleteLog to clear items older than the last sent catalog version. The
deletions will be broadcasted by statestore to all the coordinators
eventually.
However, waitForHmsEvent requests is also a consumer of the deleteLog
and could be impacted by these GCs. waitForHmsEvent is a catalogd RPC
used by coordinators when a query wants to wait until the related
metadata is in synced with HMS. The response of waitForHmsEvent returns
the latest metadata including the deletions on related dbs/tables.
If the related deletions in deleteLog is GCed just before the
waitForHmsEvent request collects the results, they will be missing in
the response. Coordinator might keep using stale metadata of
non-existing dbs/tables.
This is a quick fix for the issue by postponing deleteLog GC in a
configurable number of topic updates, similar to what we have done on
the TopicUpdateLog. A thorough fix might need to carefully choose the
version to GC or let impalad waits for the deletions from statestore to
arrive.
A new flag, catalog_delete_log_ttl, is added for this. The deleteLog
items can survive for catalog_delete_log_ttl catalog updates. The
default is 60 so a deletion can survive for at least 120s. It should be
safe enough, i.e. the GCed deletions must have arrived in the impalad
side after 60 rounds of catalog updates, otherwise that's an abnormal
impalad and already has other more severe issues, e.g. lots of stale
tables due to metadata out of sync with catalogd.
Note that postponing deleteLog GCs might increase the memory
consumption. But since most of its memory is used by db/table/partition
names, the memory usage might still be trivial comparing to other
metadata like file descriptors and incremental stats in lived catalog
objects.
This patch also removed some unused imports.
Tests:
- Added e2e test with a debug action to reproduce the issue. Ran the
test 100 times. Without the fix, it consistently fails when runs for
2-3 times.
Change-Id: I2441440bca2b928205dd514047ba742a5e8bf05e
Reviewed-on: http://gerrit.cloudera.org:8080/22816
Reviewed-by: Riza Suminto <ri...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/catalog/catalog-server.cc | 13 +++++
be/src/util/backend-gflag-util.cc | 2 +
common/thrift/BackendGflags.thrift | 2 +
common/thrift/CatalogService.thrift | 3 +
.../org/apache/impala/catalog/CatalogDeltaLog.java | 49 +++++++++++++----
.../impala/catalog/CatalogServiceCatalog.java | 2 +
.../java/org/apache/impala/service/FeSupport.java | 1 -
.../java/org/apache/impala/service/Frontend.java | 4 +-
.../java/org/apache/impala/service/JniCatalog.java | 5 --
.../java/org/apache/impala/util/DebugUtils.java | 5 ++
tests/metadata/test_event_processing.py | 64 +++++++++++++++++++++-
11 files changed, 131 insertions(+), 19 deletions(-)
diff --git a/be/src/catalog/catalog-server.cc b/be/src/catalog/catalog-server.cc
index 26b5038e1..845812c71 100644
--- a/be/src/catalog/catalog-server.cc
+++ b/be/src/catalog/catalog-server.cc
@@ -174,6 +174,19 @@ DEFINE_int32(topic_update_log_gc_frequency, 1000, "Frequency at which the entrie
"of the catalog topic update log are garbage collected. An entry may survive "
"for (2 * TOPIC_UPDATE_LOG_GC_FREQUENCY) - 1 topic updates.");
+DEFINE_int32(catalog_delete_log_ttl, 60, "Number of catalog topic updates that an entry "
+ "in the catalog delete log can survive for.");
+
+static bool ValidatePositiveInt(const char* flagname, int32_t value) {
+ if (value > 0) return true;
+ LOG(ERROR) << Substitute(
+ "$0 must be a positive integer, value $1 is invalid", flagname, value);
+ return false;
+}
+
+DEFINE_validator(topic_update_log_gc_frequency, &ValidatePositiveInt);
+DEFINE_validator(catalog_delete_log_ttl, &ValidatePositiveInt);
+
DEFINE_bool(invalidate_metadata_on_event_processing_failure, true,
"This configuration is used to invalidate metadata for table(s) upon event process "
"failure other than HMS connection issues. The default value is true. When enabled, "
diff --git a/be/src/util/backend-gflag-util.cc b/be/src/util/backend-gflag-util.cc
index 3826b09bd..51ab03cb2 100644
--- a/be/src/util/backend-gflag-util.cc
+++ b/be/src/util/backend-gflag-util.cc
@@ -136,6 +136,7 @@ DECLARE_int32(dbcp_data_source_idle_timeout_s);
DECLARE_bool(enable_catalogd_ha);
DECLARE_string(injected_group_members_debug_only);
DECLARE_int32(hms_event_sync_sleep_interval_ms);
+DECLARE_int32(catalog_delete_log_ttl);
// HS2 SAML2.0 configuration
// Defined here because TAG_FLAG caused issues in global-flags.cc
@@ -517,6 +518,7 @@ Status PopulateThriftBackendGflags(TBackendGflags& cfg) {
#endif
cfg.__set_enable_catalogd_ha(FLAGS_enable_catalogd_ha);
cfg.__set_hms_event_sync_sleep_interval_ms(FLAGS_hms_event_sync_sleep_interval_ms);
+ cfg.__set_catalog_delete_log_ttl(FLAGS_catalog_delete_log_ttl);
return Status::OK();
}
diff --git a/common/thrift/BackendGflags.thrift b/common/thrift/BackendGflags.thrift
index ffa8c2d8a..37cefdc64 100644
--- a/common/thrift/BackendGflags.thrift
+++ b/common/thrift/BackendGflags.thrift
@@ -325,4 +325,6 @@ struct TBackendGflags {
146: required string catalog_config_dir
147: required i32 catalog_partial_fetch_max_files
+
+ 148: required i32 catalog_delete_log_ttl
}
diff --git a/common/thrift/CatalogService.thrift b/common/thrift/CatalogService.thrift
index ecf2c8f75..261fb7365 100644
--- a/common/thrift/CatalogService.thrift
+++ b/common/thrift/CatalogService.thrift
@@ -794,6 +794,9 @@ struct TWaitForHmsEventRequest {
// Whether to check tables used by views.
7: optional bool should_expand_views = false
+
+ // Passes the debug actions to catalogd if the query option is set.
+ 8: optional string debug_action
}
struct TWaitForHmsEventResponse {
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogDeltaLog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogDeltaLog.java
index 72eff61f5..2227569cc 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogDeltaLog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogDeltaLog.java
@@ -17,17 +17,22 @@
package org.apache.impala.catalog;
+import java.util.ArrayDeque;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Queue;
import java.util.SortedMap;
import java.util.TreeMap;
import org.apache.commons.lang3.StringUtils;
+import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TCatalogObject;
import org.apache.impala.thrift.TCatalogObjectType;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
/**
* Represents a log of deleted catalog objects.
@@ -53,11 +58,18 @@ import com.google.common.collect.ImmutableList;
*
* b) Building catalog topic updates in the catalogd
* The catalogd uses this log to identify deleted catalog objects that have been deleted
- * since the last catalog topic update. Once the catalog topic update is constructed,
- * the old entries in the log are garbage collected to prevent the log from growing
- * indefinitely.
+ * since the last deleteLogTtl_ catalog topic updates. Once the catalog topic update is
+ * constructed, the aged out entries in the log are garbage collected to prevent the log
+ * from growing indefinitely.
*/
public class CatalogDeltaLog {
+ private final static Logger LOG = LoggerFactory.getLogger(CatalogDeltaLog.class);
+ // Times of catalog updates an entry can survive for.
+ // BackendConfig.INSTANCE could be null in some FE tests that don't care what this is.
+ private final int deleteLogTtl_ = BackendConfig.INSTANCE == null ?
+ 60 : BackendConfig.INSTANCE.getBackendCfg().catalog_delete_log_ttl;
+ private final Queue<Long> catalogUpdateVersions_ = new ArrayDeque<>(deleteLogTtl_);
+
// Map of the catalog version an object was removed from the catalog
// to the catalog object, ordered by catalog version.
private SortedMap<Long, TCatalogObject> removedCatalogObjects_ = new TreeMap<>();
@@ -113,19 +125,34 @@ public class CatalogDeltaLog {
}
/**
- * Given the current catalog version, removes all items with catalogVersion <
- * currectCatalogVersion. Such objects do not need to be tracked in the delta
- * log anymore because they are consistent with the state store's view of the
- * catalog.
+ * Garbage-collects delete log entries older than the last deleteLogTtl_ topic updates,
+ * in case some deleteLog readers still need the item.
*/
- public synchronized void garbageCollect(long currentCatalogVersion) {
+ public synchronized void garbageCollect(long lastTopicUpdateVersion) {
+ if (!catalogUpdateVersions_.isEmpty()
+ && catalogUpdateVersions_.size() >= deleteLogTtl_) {
+ garbageCollectInternal(catalogUpdateVersions_.poll());
+ }
+ catalogUpdateVersions_.offer(lastTopicUpdateVersion);
+ }
+
+ /**
+ * Given a catalog version, removes all items with catalogVersion < it.
+ */
+ public void garbageCollectInternal(long catalogVersion) {
// Nothing will be garbage collected so avoid creating a new object.
if (!removedCatalogObjects_.isEmpty() &&
- removedCatalogObjects_.firstKey() < currentCatalogVersion) {
+ removedCatalogObjects_.firstKey() < catalogVersion) {
+ int originalSize = removedCatalogObjects_.size();
removedCatalogObjects_ = new TreeMap<>(
- removedCatalogObjects_.tailMap(currentCatalogVersion));
+ removedCatalogObjects_.tailMap(catalogVersion));
latestRemovedVersions_.entrySet().removeIf(
- e -> e.getValue() < currentCatalogVersion);
+ e -> e.getValue() < catalogVersion);
+ int numCleared = originalSize - removedCatalogObjects_.size();
+ if (numCleared > 0) {
+ LOG.info("Cleared {} removed items older than version {}", numCleared,
+ catalogVersion);
+ }
}
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index aef45a8fe..86c737b61 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -4364,6 +4364,8 @@ public class CatalogServiceCatalog extends Catalog {
res.setResult(new TCatalogUpdateResult());
res.getResult().setCatalog_service_id(JniCatalog.getServiceId());
+ DebugUtils.executeDebugAction(req.debug_action,
+ DebugUtils.COLLECT_CATALOG_RESULTS_DELAY);
// Collect catalog objects required by the query
boolean wantMinimalResponse = req.header.want_minimal_response;
if (req.isSetObject_descs()) {
diff --git a/fe/src/main/java/org/apache/impala/service/FeSupport.java b/fe/src/main/java/org/apache/impala/service/FeSupport.java
index baaf142eb..59bba2542 100644
--- a/fe/src/main/java/org/apache/impala/service/FeSupport.java
+++ b/fe/src/main/java/org/apache/impala/service/FeSupport.java
@@ -53,7 +53,6 @@ import org.apache.impala.thrift.TSymbolLookupResult;
import org.apache.impala.thrift.TTable;
import org.apache.impala.thrift.TUniqueId;
import org.apache.impala.thrift.TWaitForHmsEventRequest;
-import org.apache.impala.thrift.TWaitForHmsEventResponse;
import org.apache.impala.util.NativeLibUtil;
import org.apache.thrift.TDeserializer;
import org.apache.thrift.TException;
diff --git a/fe/src/main/java/org/apache/impala/service/Frontend.java b/fe/src/main/java/org/apache/impala/service/Frontend.java
index b3237b741..55315f303 100644
--- a/fe/src/main/java/org/apache/impala/service/Frontend.java
+++ b/fe/src/main/java/org/apache/impala/service/Frontend.java
@@ -238,7 +238,6 @@ import org.apache.impala.thrift.TUnit;
import org.apache.impala.thrift.TUpdateCatalogCacheRequest;
import org.apache.impala.thrift.TUpdateCatalogCacheResponse;
import org.apache.impala.thrift.TWaitForHmsEventRequest;
-import org.apache.impala.thrift.TWaitForHmsEventResponse;
import org.apache.impala.util.AcidUtils;
import org.apache.impala.util.DebugUtils;
import org.apache.impala.util.EventSequence;
@@ -2321,6 +2320,9 @@ public class Frontend {
req.setHeader(createCatalogServiceRequestHeader(
TSessionStateUtil.getEffectiveUser(queryCtx.session), queryCtx));
collectRequiredObjects(req, stmt, queryCtx.session.database);
+ if (queryOptions.isSetDebug_action()) {
+ req.setDebug_action(queryOptions.debug_action);
+ }
// TODO: share 'timeline' to BE so we know when the updates are applied
TStatus status = FeSupport.WaitForHmsEvents(req, queryOptions);
if (status.status_code != TErrorCode.OK) {
diff --git a/fe/src/main/java/org/apache/impala/service/JniCatalog.java b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
index 671d36e32..b7ef48b19 100644
--- a/fe/src/main/java/org/apache/impala/service/JniCatalog.java
+++ b/fe/src/main/java/org/apache/impala/service/JniCatalog.java
@@ -64,7 +64,6 @@ import org.apache.impala.thrift.TCatalogObject;
import org.apache.impala.thrift.TDatabase;
import org.apache.impala.thrift.TDdlExecRequest;
import org.apache.impala.thrift.TErrorCode;
-import org.apache.impala.thrift.TEventProcessorCmdParams;
import org.apache.impala.thrift.TGetCatalogDeltaRequest;
import org.apache.impala.thrift.TGetCatalogDeltaResponse;
import org.apache.impala.thrift.TGetCatalogServerMetricsResponse;
@@ -92,13 +91,10 @@ import org.apache.impala.thrift.TUpdateCatalogRequest;
import org.apache.impala.thrift.TUpdateTableUsageRequest;
import org.apache.impala.thrift.TGetAllHadoopConfigsResponse;
import org.apache.impala.thrift.TWaitForHmsEventRequest;
-import org.apache.impala.thrift.TWaitForHmsEventResponse;
import org.apache.impala.util.AuthorizationUtil;
import org.apache.impala.util.CatalogOpUtil;
-import org.apache.impala.util.EventSequence;
import org.apache.impala.util.GlogAppender;
import org.apache.impala.util.MetaStoreUtil;
-import org.apache.impala.util.NoOpEventSequence;
import org.apache.impala.util.PatternMatcher;
import org.apache.impala.util.TUniqueIdUtil;
import org.apache.impala.util.ThreadNameAnnotator;
@@ -107,7 +103,6 @@ import org.apache.thrift.TException;
import org.apache.thrift.TSerializer;
import org.apache.thrift.protocol.TBinaryProtocol;
import org.apache.thrift.protocol.TSimpleJSONProtocol;
-import org.apache.thrift.transport.TTransportException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
diff --git a/fe/src/main/java/org/apache/impala/util/DebugUtils.java b/fe/src/main/java/org/apache/impala/util/DebugUtils.java
index ffe506589..d18b3db17 100644
--- a/fe/src/main/java/org/apache/impala/util/DebugUtils.java
+++ b/fe/src/main/java/org/apache/impala/util/DebugUtils.java
@@ -111,6 +111,11 @@ public class DebugUtils {
// the write lock.
public static final String RESET_METADATA_LOOP_LOCKED = "reset_metadata_loop_locked";
+ // debug action label for introducing delay before collecting catalog results for
+ // TWaitForHmsEventResponse.
+ public static final String COLLECT_CATALOG_RESULTS_DELAY =
+ "collect_catalog_results_delay";
+
/**
* Returns true if the label of action is set in the debugActions
*/
diff --git a/tests/metadata/test_event_processing.py b/tests/metadata/test_event_processing.py
index 51bf2cf49..1f251d3b2 100644
--- a/tests/metadata/test_event_processing.py
+++ b/tests/metadata/test_event_processing.py
@@ -16,6 +16,7 @@
# under the License.
from __future__ import absolute_import, division, print_function
from subprocess import check_call
+import logging
import pytest
import re
import time
@@ -31,7 +32,7 @@ from tests.metadata.test_event_processing_base import TestEventProcessingBase
from tests.util.event_processor_utils import EventProcessorUtils
PROCESSING_TIMEOUT_S = 10
-
+LOG = logging.getLogger(__name__)
@SkipIfFS.hive
@SkipIfCatalogV2.hms_event_polling_disabled()
@@ -612,3 +613,64 @@ class TestEventSyncWaiting(ImpalaTestSuite):
finally:
self.run_stmt_in_hive(
"drop database if exists {0} cascade".format(db))
+
+ def test_hms_event_sync_on_deletion(self, vector, unique_name):
+ """Regression test for IMPALA-13829: TWaitForHmsEventResponse not able to collect
+ removed objects due to their items in deleteLog being GCed."""
+ client = self.create_impala_client_from_vector(vector)
+ # Set a sleep time so catalogd has time to GC the deleteLog.
+ client.execute("set debug_action='collect_catalog_results_delay:SLEEP@1000'")
+ db = unique_name + "_db"
+ tbl = db + ".foo"
+ self.execute_query("drop database if exists {} cascade".format(db))
+ self.execute_query("drop database if exists {}_2 cascade".format(db))
+ self.execute_query("create database {}".format(db))
+ self.execute_query("create database {}_2".format(db))
+ # Create HMS Thrift clients to drop db/tables in the fastest way
+ hive_clients = []
+ hive_transports = []
+ for _ in range(2):
+ c, t = ImpalaTestSuite.create_hive_client()
+ hive_clients.append(c)
+ hive_transports.append(t)
+
+ try:
+ # Drop 2 dbs concurrently. So their DROP_DATABASE events are processed together (in
+ # the same event batch). We need more than one db to be dropped so one of them in
+ # catalogd's deleteLog can be GCed since its version < latest catalog version.
+ # Note that this is no longer the way catalogd GCs the deleteLog after IMPALA-13829,
+ # but it can be used to trigger the issue before this fix.
+ def drop_db_in_hive(i, db_name):
+ hive_clients[i].drop_database(db_name, deleteData=True, cascade=True)
+ LOG.info("Dropped database {} in Hive".format(db_name))
+ ts = [threading.Thread(target=drop_db_in_hive, args=params)
+ for params in [[0, db], [1, db + "_2"]]]
+ for t in ts:
+ t.start()
+ for t in ts:
+ t.join()
+ client.execute("create database " + db)
+
+ self.execute_query("create table {}(i int)".format(tbl))
+ self.execute_query("create table {}_2(i int)".format(tbl))
+
+ # Drop 2 tables concurrently. So their DROP_TABLE events are processed together (in
+ # the same event batch). We need more than one table to be dropped so one of them in
+ # catalogd's deleteLog can be GCed since its version < latest catalog version.
+ # Note that this is no longer the way catalogd GCs the deleteLog after IMPALA-13829,
+ # but it can be used to trigger the issue before this fix.
+ def drop_table_in_hive(i, tbl_name):
+ hive_clients[i].drop_table(db, tbl_name, deleteData=True)
+ LOG.info("Dropped table {}.{} in Hive".format(db, tbl_name))
+ ts = [threading.Thread(target=drop_table_in_hive, args=params)
+ for params in [[0, "foo"], [1, "foo_2"]]]
+ for t in ts:
+ t.start()
+ for t in ts:
+ t.join()
+ client.execute("create table {}(i int)".format(tbl))
+ finally:
+ for t in hive_transports:
+ t.close()
+ self.execute_query("drop database if exists {} cascade".format(db))
+ self.execute_query("drop database if exists {}_2 cascade".format(db))
(impala) 03/05: IMPALA-13989: Invalidate table on rename failure
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 295f74ef12df0b851058efe2c43c9826b617d21f
Author: Michael Smith <mi...@cloudera.com>
AuthorDate: Wed Apr 23 15:50:30 2025 -0700
IMPALA-13989: Invalidate table on rename failure
Handles the error "Table/view rename succeeded in the Hive Metastore,
but failed in Impala's Catalog Server" rather than failing the table
rename. This error happens when catalog state catches up to the alter
event from our alter_table RPC to HMS before we renameTable explicitly
in the catalog. The catalog can update independently due to a concurrent
'invalidate metadata' call.
In that case we use the oldTbl definition we already have - updated from
the delete log if possible - and fetch the new table definition with
invalidateTable to continue, automating most of the work that the error
message suggested users do via 'invalidate metadata <tbl>'.
Updated the test_concurrent_ddls test to remove handle_rename_failure
and ran the tests a dozen times. Adds concurrency tests with
simultaneous rename and invalidate metadata that previously would fail.
Change-Id: Ic2a276b6e5ceb35b7f3ce788cc47052387ae8980
Reviewed-on: http://gerrit.cloudera.org:8080/22807
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../apache/impala/service/CatalogOpExecutor.java | 60 +++++++++++------
tests/custom_cluster/test_concurrent_ddls.py | 16 +----
tests/custom_cluster/test_concurrent_rename.py | 77 ++++++++++++++++++++++
3 files changed, 117 insertions(+), 36 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index f1ada478a..195605e6e 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -5742,28 +5742,46 @@ public class CatalogOpExecutor {
result.second.setCreateEventId(renamedTable.first);
}
}
- if (result.first == null || result.second == null) {
- // The rename succeeded in the HMS but failed in the catalog cache. The cache is in
- // an inconsistent state, but can likely be fixed by running "invalidate metadata".
- throw new ImpalaRuntimeException(String.format(
- "Table/view rename succeeded in the Hive Metastore, but failed in Impala's " +
- "Catalog Server. Running 'invalidate metadata <tbl>' on the old table name " +
- "'%s' and the new table name '%s' may fix the problem." , tableName,
- newTableName));
- }
- // TODO: call addVersionsForInflightEvents using InProgressTableModification object
- // that is passed into catalog_.renameTable()
- catalog_.addVersionsForInflightEvents(
- false, result.second, modification.newVersionNumber());
- if (wantMinimalResult) {
- response.result.addToRemoved_catalog_objects(result.first.toInvalidationObject());
- response.result.addToUpdated_catalog_objects(result.second.toInvalidationObject());
+ TCatalogObject oldTblDesc = null, newTblDesc = null;
+ if (result.first == null) {
+ // The old table object has been removed by a concurrent operation, e.g. INVALIDATE
+ // METADATA <table>. Fetch the latest delete from deleteLog.
+ oldTblDesc = wantMinimalResult ?
+ oldTbl.toInvalidationObject() : oldTbl.toMinimalTCatalogObject();
+ long version = catalog_.getDeleteLog().getLatestRemovedVersion(oldTblDesc);
+ if (version > 0) {
+ oldTblDesc.setCatalog_version(version);
+ } else {
+ LOG.warn("Deletion update on the old table {} not found. Impalad might still "
+ + "have its metadata until the deletion update arrives from statestore.",
+ tableName);
+ }
} else {
- response.result.addToRemoved_catalog_objects(
- result.first.toMinimalTCatalogObject());
- response.result.addToUpdated_catalog_objects(result.second.toTCatalogObject());
- }
- response.result.setVersion(result.second.getCatalogVersion());
+ oldTblDesc = wantMinimalResult ?
+ result.first.toInvalidationObject() : result.first.toMinimalTCatalogObject();
+ }
+ if (result.second == null) {
+ // The rename succeeded in HMS but failed in the catalog cache. The cache is in an
+ // inconsistent state, so invalidate the new table to reload it.
+ newTblDesc = catalog_.invalidateTable(newTableName.toThrift(),
+ new Reference<>(), new Reference<>(), catalogTimeline);
+ if (newTblDesc == null) {
+ throw new ImpalaRuntimeException(String.format(
+ "The new table/view %s was concurrently removed during rename.",
+ newTableName));
+ }
+ } else {
+ Preconditions.checkNotNull(result.first);
+ // TODO: call addVersionsForInflightEvents using InProgressTableModification object
+ // that is passed into catalog_.renameTable()
+ catalog_.addVersionsForInflightEvents(
+ false, result.second, modification.newVersionNumber());
+ newTblDesc = wantMinimalResult ?
+ result.second.toInvalidationObject() : result.second.toTCatalogObject();
+ }
+ response.result.addToRemoved_catalog_objects(oldTblDesc);
+ response.result.addToUpdated_catalog_objects(newTblDesc);
+ response.result.setVersion(newTblDesc.getCatalog_version());
addSummary(response, "Renaming was successful.");
}
diff --git a/tests/custom_cluster/test_concurrent_ddls.py b/tests/custom_cluster/test_concurrent_ddls.py
index aa19d2018..c94be7977 100644
--- a/tests/custom_cluster/test_concurrent_ddls.py
+++ b/tests/custom_cluster/test_concurrent_ddls.py
@@ -143,10 +143,7 @@ class TestConcurrentDdls(CustomClusterTestSuite):
break
except IMPALA_CONNECTION_EXCEPTION as e:
err = str(e)
- if self.handle_rename_failure(tls.client, tbl_name, err):
- # Table was successfully renamed, next case.
- break
- elif self.is_transient_error(err):
+ if self.is_transient_error(err):
# Retry the query.
continue
assert self.is_acceptable_error(err, sync_ddl), err
@@ -194,17 +191,6 @@ class TestConcurrentDdls(CustomClusterTestSuite):
return True
return False
- def handle_rename_failure(self, client, tbl_name, err):
- if "Table/view rename succeeded in the Hive Metastore, " \
- "but failed in Impala's Catalog Server." in err:
- # Invalidate the target table so we reload it from HMS.
- tbl_names = re.findall(r"{}[^']*".format(tbl_name), err)
- assert len(tbl_names) == 2
- self.execute_query_expect_success(
- client, "invalidate metadata {0}".format(tbl_names[1]))
- return True
- return False
-
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args="--use_local_catalog=true",
diff --git a/tests/custom_cluster/test_concurrent_rename.py b/tests/custom_cluster/test_concurrent_rename.py
new file mode 100644
index 000000000..e2dfb58ea
--- /dev/null
+++ b/tests/custom_cluster/test_concurrent_rename.py
@@ -0,0 +1,77 @@
+# 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.
+
+from __future__ import absolute_import, division, print_function
+import time
+
+from copy import deepcopy
+from tests.common.custom_cluster_test_suite import CustomClusterTestSuite
+
+
+@CustomClusterTestSuite.with_args(
+ catalogd_args="--hms_event_polling_interval_s=0",
+ # Requires longer update frequency so invalidate->drop can happen within an update.
+ # Still shorter than default so tests are quicker.
+ statestored_args="--statestore_update_frequency_ms=1000",
+ cluster_size=1)
+class TestConcurrentRename(CustomClusterTestSuite):
+ """Test concurrent rename with invalidate and other DDLs."""
+
+ def test_rename_drop(self, vector, unique_database):
+ name = "{}.tbl".format(unique_database)
+ self.execute_query("create table {} (s string)".format(name))
+ self.execute_query("describe {}".format(name))
+
+ new_vector = deepcopy(vector)
+ new_vector.get_value('exec_option')['debug_action'] = \
+ "catalogd_table_rename_delay:SLEEP@5000"
+ with self.create_impala_client_from_vector(new_vector) as alter_client, \
+ self.create_impala_client() as reset_client:
+ alter_handle = alter_client.execute_async(
+ "alter table {0} rename to {0}2".format(name))
+ time.sleep(0.1)
+ reset_handle = reset_client.execute_async("invalidate metadata {}".format(name))
+ self.execute_query("invalidate metadata {}2".format(name))
+ self.execute_query("drop table {}2".format(name))
+
+ try:
+ alter_client.wait_for_finished_timeout(alter_handle, timeout=10)
+ alter_client.close_query(alter_handle)
+ assert False, "Expected alter to fail"
+ except Exception as e:
+ assert "The new table/view {}2 was concurrently removed during rename."\
+ .format(name) in str(e)
+ finally:
+ reset_client.wait_for_finished_timeout(reset_handle, timeout=10)
+ reset_client.close_query(reset_handle)
+
+ def test_rename_invalidate(self, vector, unique_database):
+ name = "{}.tbl".format(unique_database)
+ self.execute_query("create table {} (s string)".format(name))
+ self.execute_query("describe {}".format(name))
+
+ new_vector = deepcopy(vector)
+ new_vector.get_value('exec_option')['debug_action'] = \
+ "catalogd_table_rename_delay:SLEEP@5000"
+ with self.create_impala_client_from_vector(new_vector) as alter_client:
+ alter_handle = alter_client.execute_async(
+ "alter table {0} rename to {0}2".format(name))
+ time.sleep(0.1)
+ self.execute_query("invalidate metadata {}".format(name))
+
+ alter_client.wait_for_finished_timeout(alter_handle, timeout=10)
+ alter_client.close_query(alter_handle)
(impala) 04/05: IMPALA-13996: Deflake test_too_many_files by creating dedicate tables
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 0e3ae5c339fad3183ba8ce5c0c7d954e50dee091
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Sun Apr 27 16:45:57 2025 +0800
IMPALA-13996: Deflake test_too_many_files by creating dedicate tables
TestAllowIncompleteData.test_too_many_files depends on
tpch_parquet.lineitem to have exactly 3 data files. This is false in
erasure coding builds in which tpch_parquet.lineitem has only 2 data
files.
This fixes the test to use dedicate tables created in the test.
Change-Id: I28cec8ec4bc59f066aa15a7243b7163639706cc7
Reviewed-on: http://gerrit.cloudera.org:8080/22824
Reviewed-by: Jason Fehr <jf...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
tests/custom_cluster/test_local_catalog.py | 14 +++++++++-----
1 file changed, 9 insertions(+), 5 deletions(-)
diff --git a/tests/custom_cluster/test_local_catalog.py b/tests/custom_cluster/test_local_catalog.py
index 3f7da98d8..c4e92e9ba 100644
--- a/tests/custom_cluster/test_local_catalog.py
+++ b/tests/custom_cluster/test_local_catalog.py
@@ -713,11 +713,15 @@ class TestAllowIncompleteData(CustomClusterTestSuite):
catalogd_args="--catalog_topic_mode=minimal --catalog_partial_fetch_max_files=1")
def test_too_many_files(self, unique_database):
"""Test the error reporting the limit is too small"""
- exception = self.execute_query_expect_failure(
- self.client, "show files in tpch_parquet.lineitem")
- err = ("Too many files to collect in table tpch_parquet.lineitem: 3. Current limit "
- "is 1 configured by startup flag 'catalog_partial_fetch_max_files'. Consider "
- "compacting files of the table.")
+ # Create a non-partitioned table with multiple files
+ tbl = unique_database + ".tbl"
+ self.execute_query("create table {0} (i int)".format(tbl))
+ self.execute_query("insert into {0} values (0)".format(tbl))
+ self.execute_query("insert into {0} values (1)".format(tbl))
+ exception = self.execute_query_expect_failure(self.client, "show files in " + tbl)
+ err = ("Too many files to collect in table {0}: 2. Current limit is 1 configured by "
+ "startup flag 'catalog_partial_fetch_max_files'. Consider compacting files of"
+ " the table.").format(tbl)
assert err in str(exception)
self.assert_catalogd_log_contains("ERROR", err)
(impala) 02/05: IMPALA-13873: Missing equivalence conjunct in aggregation node with inline views
Posted by st...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit c1aac4b3a4fb616763cedc59648cfde6e8f5ec70
Author: Daniel Becker <db...@apache.org>
AuthorDate: Tue Apr 1 10:18:51 2025 +0200
IMPALA-13873: Missing equivalence conjunct in aggregation node with inline views
Some queries involving plain (distinct) UNIONs miss conjuncts, leading
to incorrect results:
Example:
WITH u1 AS (select 10 a, 10 b),
t AS (select a, b, min(b) over (partition by a) min_b from u1 UNION
select 10, 10, 20)
select t.* from t where t.b = t.min_b;
Expected result:
+----+----+-------+
| a | b | min_b |
+----+----+-------+
| 10 | 10 | 10 |
+----+----+-------+
Actual result:
+----+----+-------+
| a | b | min_b |
+----+----+-------+
| 10 | 10 | 10 |
| 10 | 20 | 10 |
+----+----+-------+
This is caused by MultiAggregateInfo assuming that conjuncts bound by
grouping slots that are produced by SlotRef grouping expressions are
already evaluated below the AggregationNode. However, this is not true
in all cases: with UNIONs, there may be conjuncts that are unassigned
below the AggregationNode.
This may happen if a conjunct cannot be pushed into all operands of a
UNION, because the source tuples in the operands do not contain all of
the slots referenced by the predicate. In the example above, it happens
in the first operand:
select a, b, min(b) over (partition by a) min_b from u1
The source tuple, 'u1', contains only two slots ('a' and 'b'), but does
not contain a slot corresponding to 'min(b)' - therefore the predicate
't.b = t.min_b' is not bound by the tuple of 'u1'. In theory, the
predicate could still be evaluated directly after materialising the
tuple with 'min(b)', still inside the UNION operand, but Impala
currently does not work that way.
In these cases, the conjuncts need to be evaluated in the
AggregationNode (possibly in addition to some of the UNION operands).
This change fixes this problem by introducing a method in
MultiAggregateInfo: 'setConjunctsToKeep()', where the caller can pass a
list of conjuncts that will not be eliminated. This is called during the
planning of the UNION if there are unassigned conjuncts remaining.
Testing:
- Added a PlannerTest and an EE test for the case where a conjunct
was previously incorrectly removed from the AggregationNode.
- Existing tests cover the case when conjuncts can be safely removed
from an AggregationNode above a UnionNode because the conjuncts are
pushed into all union operands, see for example
https://github.com/apache/impala/blob/6f2d9a2/testdata/workloads/functional-planner/queries/PlannerTest/union.test#L3914
Change-Id: I67a59cd96d83181ce249fd6ca141906f549a09b3
Reviewed-on: http://gerrit.cloudera.org:8080/22746
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../apache/impala/analysis/MultiAggregateInfo.java | 24 ++++
.../apache/impala/planner/SingleNodePlanner.java | 9 +-
.../queries/PlannerTest/union.test | 125 +++++++++++++++++++++
.../queries/QueryTest/aggregation.test | 45 ++++++++
4 files changed, 202 insertions(+), 1 deletion(-)
diff --git a/fe/src/main/java/org/apache/impala/analysis/MultiAggregateInfo.java b/fe/src/main/java/org/apache/impala/analysis/MultiAggregateInfo.java
index 24c12fda9..9530b7aae 100644
--- a/fe/src/main/java/org/apache/impala/analysis/MultiAggregateInfo.java
+++ b/fe/src/main/java/org/apache/impala/analysis/MultiAggregateInfo.java
@@ -133,6 +133,11 @@ public class MultiAggregateInfo {
// Result of substituting 'groupingExprs_' with the output smap of the AggregationNode.
private List<Expr> substGroupingExprs_;
+ // These conjuncts are kept even if they would otherwise be removed as redundant. This
+ // can be used for example when a conjunct cannot be pushed into all operands of a UNION
+ // node below the AggregationNode, so it needs to be evaluated by the AggregationNode.
+ private List<Expr> conjunctsToKeep_;
+
// Results of analyze():
// Aggregation classes and their AggregateInfos. If there is a class with non-distinct
@@ -220,6 +225,10 @@ public class MultiAggregateInfo {
return new MultiAggregateInfo(distinctAggInfo);
}
+ public void setConjunctsToKeep(List<Expr> conjuncts) {
+ conjunctsToKeep_ = conjuncts;
+ }
+
public void analyze(Analyzer analyzer) throws AnalysisException {
if (isAnalyzed_) return;
@@ -1024,7 +1033,22 @@ public class MultiAggregateInfo {
if (markAssigned) analyzer.markConjunctsAssigned(unassignedConjuncts);
result.addAll(bindingPredicates);
result.addAll(unassignedConjuncts);
+
+ // Select those conjuncts in 'result' that we need to keep.
+ List<Expr> origConjunctsToKeep = new ArrayList<>();
+ if (conjunctsToKeep_ != null) {
+ for (Expr conj : result) {
+ if (conjunctsToKeep_.contains(conj)) {
+ origConjunctsToKeep.add(conj);
+ }
+ }
+ }
+
analyzer.createEquivConjuncts(tid, result, groupBySids);
+
+ // Add back conjuncts that we need to keep but may have been removed.
+ if (origConjunctsToKeep != null) result.addAll(origConjunctsToKeep);
+
Expr.removeDuplicates(result);
return result;
}
diff --git a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
index 921cdd8cf..07333895e 100644
--- a/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/SingleNodePlanner.java
@@ -2346,8 +2346,15 @@ public class SingleNodePlanner implements SingleNodePlannerIntf {
if (unionStmt.hasUnionDistinctOps()) {
result = createUnionPlan(
analyzer, unionStmt, unionStmt.getUnionDistinctOperands(), null);
+
+ MultiAggregateInfo unionDistinctAggInfo = unionStmt.getDistinctAggInfo();
+
+ // Make sure MultiAggregateInfo does not remove conjuncts that are unassigned at
+ // this point as redundant.
+ unionDistinctAggInfo.setConjunctsToKeep(analyzer.getUnassignedConjuncts(result));
+
result = new AggregationNode(
- ctx_.getNextNodeId(), result, unionStmt.getDistinctAggInfo(), AggPhase.FIRST);
+ ctx_.getNextNodeId(), result, unionDistinctAggInfo, AggPhase.FIRST);
result.init(analyzer);
}
// create ALL tree
diff --git a/testdata/workloads/functional-planner/queries/PlannerTest/union.test b/testdata/workloads/functional-planner/queries/PlannerTest/union.test
index e004dfc25..fc5ed39a0 100644
--- a/testdata/workloads/functional-planner/queries/PlannerTest/union.test
+++ b/testdata/workloads/functional-planner/queries/PlannerTest/union.test
@@ -4634,3 +4634,128 @@ Per-Host Resources: mem-estimate=80.00MB mem-reservation=128.00KB thread-reserva
tuple-ids=0 row-size=4B cardinality=11.00K
in pipelines: 00(GETNEXT)
====
+# Regression test for IMPALA-13873: the conjunct 'b = min_b' in the aggregation node
+# should not be discarded.
+WITH u1 AS (select 10 a, 10 b),
+t AS (select a, b, min(b) over (partition by a) min_b from u1 UNION select 10, 10, 20)
+select t.* from t where t.b = t.min_b;
+---- QUERYOPTIONS
+explain_level=2
+---- PLAN
+F00:PLAN FRAGMENT [UNPARTITIONED] hosts=1 instances=1
+| Per-Host Resources: mem-estimate=20.00MB mem-reservation=11.94MB thread-reservation=1
+PLAN-ROOT SINK
+| output exprs: a, b, min_b
+| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
+|
+04:AGGREGATE [FINALIZE]
+| group by: a, b, min_b
+| having: b = min_b
+| mem-estimate=10.00MB mem-reservation=1.94MB spill-buffer=64.00KB thread-reservation=0
+| tuple-ids=2 row-size=3B cardinality=1
+| in pipelines: 04(GETNEXT), 02(OPEN)
+|
+00:UNION
+| constant-operands=1
+| mem-estimate=0B mem-reservation=0B thread-reservation=0
+| tuple-ids=2 row-size=3B cardinality=2
+| in pipelines: 02(GETNEXT)
+|
+03:ANALYTIC
+| functions: min(b)
+| partition by: a
+| mem-estimate=4.00MB mem-reservation=4.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=5,4 row-size=3B cardinality=1
+| in pipelines: 02(GETNEXT)
+|
+02:SORT
+| order by: a ASC NULLS LAST
+| mem-estimate=6.00MB mem-reservation=6.00MB spill-buffer=2.00MB thread-reservation=0
+| tuple-ids=5 row-size=2B cardinality=1
+| in pipelines: 02(GETNEXT)
+|
+01:UNION
+ constant-operands=1
+ mem-estimate=0B mem-reservation=0B thread-reservation=0
+ tuple-ids=0 row-size=2B cardinality=1
+ in pipelines: <none>
+====
+# Regression test for IMPALA-13873: the conjunct 'b = min_b' in the aggregation node
+# should not be discarded.
+# Querying from tables, not inline expressions.
+WITH t AS (
+ select d2 a, d3 b, min(d3) over (partition by d2) min_b from functional.decimal_tbl
+ UNION
+ select smallint_col, int_col, bigint_col from functional.alltypestiny
+)
+select t.* from t where t.b = t.min_b;
+---- QUERYOPTIONS
+explain_level=1
+---- PLAN
+PLAN-ROOT SINK
+|
+05:AGGREGATE [FINALIZE]
+| group by: a, b, min_b
+| having: b = min_b
+| row-size=40B cardinality=1
+|
+00:UNION
+| row-size=40B cardinality=4
+|
+|--04:SCAN HDFS [functional.alltypestiny]
+| HDFS partitions=4/4 files=4 size=460B
+| predicates: functional.alltypestiny.int_col = functional.alltypestiny.bigint_col
+| row-size=14B cardinality=1
+|
+03:ANALYTIC
+| functions: min(d3)
+| partition by: d2
+| row-size=40B cardinality=3
+|
+02:SORT
+| order by: d2 ASC NULLS LAST
+| row-size=24B cardinality=3
+|
+01:SCAN HDFS [functional.decimal_tbl]
+ HDFS partitions=1/1 files=1 size=195B
+ row-size=24B cardinality=3
+====
+# Regression test for IMPALA-13873: the conjunct 'b = min_b' in the aggregation node
+# should not be discarded.
+# The predicate cannot be pushed down to any UNION operand.
+WITH u1 AS (select tinyint_col, id from functional.alltypestiny),
+t AS (
+ select tinyint_col, id, min(id) over (partition by tinyint_col) min_id from u1
+ UNION
+ select tinyint_col, id, id+100 as min_id from u1)
+select t.* from t where t.id = t.min_id;
+---- QUERYOPTIONS
+explain_level=1
+---- PLAN
+PLAN-ROOT SINK
+|
+05:AGGREGATE [FINALIZE]
+| group by: tinyint_col, id, min_id
+| having: id = min_id
+| row-size=13B cardinality=2
+|
+00:UNION
+| row-size=13B cardinality=16
+|
+|--04:SCAN HDFS [functional.alltypestiny]
+| HDFS partitions=4/4 files=4 size=460B
+| row-size=5B cardinality=8
+|
+03:ANALYTIC
+| functions: min(id)
+| partition by: tinyint_col
+| row-size=9B cardinality=8
+|
+02:SORT
+| order by: tinyint_col ASC NULLS LAST
+| row-size=5B cardinality=8
+|
+01:SCAN HDFS [functional.alltypestiny]
+ HDFS partitions=4/4 files=4 size=460B
+ row-size=5B cardinality=8
+====
diff --git a/testdata/workloads/functional-query/queries/QueryTest/aggregation.test b/testdata/workloads/functional-query/queries/QueryTest/aggregation.test
index cb20e8c6d..8839edba0 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/aggregation.test
+++ b/testdata/workloads/functional-query/queries/QueryTest/aggregation.test
@@ -2911,3 +2911,48 @@ select s_store_sk, regr_count(s_number_employees, s_floor_space) over (partition
---- TYPES
int,bigint
====
+---- QUERY
+# Regression test for IMPALA-13873: the conjunct 'b = min_b' in the aggregation node
+# should not be discarded.
+WITH u1 AS (select 10 a, 10 b),
+t AS (select a, b, min(b) over (partition by a) min_b from u1 UNION select 10, 10, 20)
+select t.* from t where t.b = t.min_b;
+---- RESULTS
+10,10,10
+---- TYPES
+TINYINT,TINYINT,TINYINT
+====
+---- QUERY
+# Regression test for IMPALA-13873: the conjunct 'b = min_b' in the aggregation node
+# should not be discarded.
+# Querying from tables, not inline expressions.
+WITH t AS (
+ select d2 a, d3 b, min(d3) over (partition by d2) min_b from functional.decimal_tbl
+ UNION
+ select smallint_col, int_col, bigint_col from functional.alltypestiny
+)
+select t.* from t where t.b = t.min_b;
+---- RESULTS
+2222,1.2345678900,1.2345678900
+0,0.0000000000,0.0000000000
+333,123.4567890000,123.4567890000
+111,12.3456789000,12.3456789000
+---- TYPES
+decimal,decimal,decimal
+====
+---- QUERY
+# Regression test for IMPALA-13873: the conjunct 'b = min_b' in the aggregation node
+# should not be discarded.
+# The predicate cannot be pushed down to any UNION operand.
+WITH u1 AS (select tinyint_col, id from functional.alltypestiny),
+t AS (
+ select tinyint_col, id, min(id) over (partition by tinyint_col) min_id from u1
+ UNION
+ select tinyint_col, id, id+100 as min_id from u1)
+select t.* from t where t.id = t.min_id;
+---- RESULTS
+1,1,1
+0,0,0
+---- TYPES
+tinyint,int,bigint
+====