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 2021/06/27 13:30:33 UTC
[impala] 04/04: IMPALA-10765: Allow firing INSERT events with empty
results
This is an automated email from the ASF dual-hosted git repository.
stigahuang pushed a commit to branch branch-4.0.0
in repository https://gitbox.apache.org/repos/asf/impala.git
commit a702d2d4cf7e2062476e1057ebcbafd5e6fdcf02
Author: stiga-huang <hu...@gmail.com>
AuthorDate: Thu Jun 24 10:59:41 2021 +0800
IMPALA-10765: Allow firing INSERT events with empty results
When INSERT or the INSERT part of a CTAS statement produces empty
results, catalogd will hit a Precondition check in createInsertEvents()
if the statement is not specified with OVERWRITE. This patch removes the
checks to relax the restriction. This is consistent with Hive since Hive
also fires insert events on empty insert results.
Previously when event processor is disabled, we won't generate INSERT
events. This is inconsistent with other DDL events which will always be
generated by Hive. This patch removes the check on event processor's
status. So firing INSERT events is independent to the event processor.
This fixes IMPALA-10744.
Tests
- Add e2e test
Change-Id: Id708f38302b37d7e09435e7b8b0c9fc071659606
Reviewed-on: http://gerrit.cloudera.org:8080/17629
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../apache/impala/service/CatalogOpExecutor.java | 7 +----
tests/custom_cluster/test_event_processing.py | 35 ++++++++++++++++++++++
2 files changed, 36 insertions(+), 6 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 8e10077..acf369e 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -4922,7 +4922,6 @@ public class CatalogOpExecutor {
// see HdfsTable.DEFAULT_PARTITION_NAME.
List<String> newFiles = updatedPartitions.get("").getFiles();
List<String> partVals = new ArrayList<>();
- Preconditions.checkState(!newFiles.isEmpty() || isInsertOverwrite);
LOG.info(String.format("%s new files detected for table %s", newFiles.size(),
table.getFullName()));
insertEventReqDatas.add(
@@ -4933,7 +4932,6 @@ public class CatalogOpExecutor {
for (HdfsPartition part : existingPartitions) {
List<String> newFiles = updatedPartitions.get(part.getPartitionName()).getFiles();
List<String> partVals = part.getPartitionValuesAsStrings(true);
- Preconditions.checkState(!newFiles.isEmpty() || isInsertOverwrite);
Preconditions.checkState(!partVals.isEmpty());
LOG.info(String.format("%s new files detected for table %s partition %s",
newFiles.size(), table.getFullName(), part.getPartitionName()));
@@ -4946,7 +4944,6 @@ public class CatalogOpExecutor {
for (Map.Entry<String, List<String>> part : addedPartitionNames.entrySet()) {
List<String> newFiles = updatedPartitions.get(part.getKey()).getFiles();
List<String> partVals = part.getValue();
- Preconditions.checkState(!newFiles.isEmpty() || isInsertOverwrite);
Preconditions.checkState(!partVals.isEmpty());
LOG.info(String.format("%s new files detected for table %s new partition %s",
newFiles.size(), table.getFullName(), part.getKey()));
@@ -4989,9 +4986,7 @@ public class CatalogOpExecutor {
private boolean shouldGenerateInsertEvents(FeFsTable table) {
if (table instanceof FeIcebergTable) return false;
- if (!BackendConfig.INSTANCE.isInsertEventsEnabled()) return false;
- if (!catalog_.isEventProcessingActive()) return false;
- return true;
+ return BackendConfig.INSTANCE.isInsertEventsEnabled();
}
private InsertEventRequestData makeInsertEventData(FeFsTable tbl, List<String> partVals,
diff --git a/tests/custom_cluster/test_event_processing.py b/tests/custom_cluster/test_event_processing.py
index 9927566..3ef2795 100644
--- a/tests/custom_cluster/test_event_processing.py
+++ b/tests/custom_cluster/test_event_processing.py
@@ -119,6 +119,12 @@ class TestEventProcessing(CustomClusterTestSuite):
self.run_stmt_in_hive("drop table if exists %s.%s" % (db_name, test_tbl_name))
self.run_stmt_in_hive("create table %s.%s (id int, val int) %s"
% (db_name, test_tbl_name, tblproperties))
+ EventProcessorUtils.wait_for_event_processing(self)
+ # Test CTAS and insert by Impala with empty results (IMPALA-10765).
+ self.execute_query("create table {db}.ctas_tbl {prop} as select * from {db}.{tbl}"
+ .format(db=db_name, tbl=test_tbl_name, prop=tblproperties))
+ self.execute_query("insert into {db}.ctas_tbl select * from {db}.{tbl}"
+ .format(db=db_name, tbl=test_tbl_name))
# Test insert into table, this will fire an insert event.
self.run_stmt_in_hive("insert into %s.%s values(101, 200)"
% (db_name, test_tbl_name))
@@ -137,6 +143,12 @@ class TestEventProcessing(CustomClusterTestSuite):
# Verify that the data is present in Impala.
data = self.execute_scalar("select * from %s.%s" % (db_name, test_tbl_name))
assert data.split('\t') == ['101', '201']
+ # Test insert overwrite by Impala with empty results (IMPALA-10765).
+ self.execute_query("insert overwrite {db}.{tbl} select * from {db}.ctas_tbl"
+ .format(db=db_name, tbl=test_tbl_name))
+ result = self.execute_query("select * from {db}.{tbl}"
+ .format(db=db_name, tbl=test_tbl_name))
+ assert len(result.data) == 0
# Test partitioned table.
test_part_tblname = 'tbl_insert_part'
@@ -144,6 +156,14 @@ class TestEventProcessing(CustomClusterTestSuite):
self.run_stmt_in_hive("create table %s.%s (id int, name string) "
"partitioned by(day int, month int, year int) %s"
% (db_name, test_part_tblname, tblproperties))
+ EventProcessorUtils.wait_for_event_processing(self)
+ # Test insert overwrite by Impala with empty results (IMPALA-10765).
+ self.execute_query("create table {db}.ctas_part partitioned by (day, month, year) "
+ "{prop} as select * from {db}.{tbl}"
+ .format(db=db_name, tbl=test_part_tblname, prop=tblproperties))
+ self.execute_query("insert into {db}.ctas_part partition(day=0, month=0, year=0) "
+ "select id, name from {db}.{tbl}"
+ .format(db=db_name, tbl=test_part_tblname))
# Insert data into partitions.
self.run_stmt_in_hive("insert into %s.%s partition(day=28, month=03, year=2019)"
"values(101, 'x')" % (db_name, test_part_tblname))
@@ -161,6 +181,11 @@ class TestEventProcessing(CustomClusterTestSuite):
data = self.execute_scalar("select count(*) from %s.%s where day=28 and month=3 "
"and year=2019" % (db_name, test_part_tblname))
assert data.split('\t') == ['2']
+ # Test inserting into existing partitions by Impala with empty results
+ # (IMPALA-10765).
+ self.execute_query("insert into {db}.{tbl} partition(day=28, month=03, year=2019) "
+ "select id, name from {db}.ctas_part"
+ .format(db=db_name, tbl=test_part_tblname))
# Test insert overwrite into existing partitions
self.run_stmt_in_hive("insert overwrite table %s.%s partition(day=28, month=03, "
@@ -170,6 +195,16 @@ class TestEventProcessing(CustomClusterTestSuite):
data = self.execute_scalar("select * from %s.%s where day=28 and month=3 and"
" year=2019 and id=101" % (db_name, test_part_tblname))
assert data.split('\t') == ['101', 'z', '28', '3', '2019']
+ # Test insert overwrite into existing partitions by Impala with empty results
+ # (IMPALA-10765).
+ self.execute_query("insert overwrite {db}.{tbl} "
+ "partition(day=28, month=03, year=2019) "
+ "select id, name from {db}.ctas_part"
+ .format(db=db_name, tbl=test_part_tblname))
+ result = self.execute_query("select * from {db}.{tbl} "
+ "where day=28 and month=3 and year=2019"
+ .format(db=db_name, tbl=test_part_tblname))
+ assert len(result.data) == 0
@CustomClusterTestSuite.with_args(catalogd_args="--hms_event_polling_interval_s=1")
def test_iceberg_inserts(self):