You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by cs...@apache.org on 2025/02/20 08:06:06 UTC
(impala) branch master updated (7373ee27d -> 0f17ae8fc)
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
from 7373ee27d IMPALA-13752: Remove option value quoting in __restore_query_options
new 37e409059 IMPALA-13737: Directly load file metadata via IcebergFileMetadataLoader
new c8c64fff3 IMPALA-13759: Fix Hive ACID INSERT OVERWRITE base detection
new 9cb9bae84 IMPALA-13758: Use context manager in ImpalaTestSuite.change_database
new 0f17ae8fc IMPALA-13771: Fix heap-use-after-free in Cluster Membership Manager
The 4 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/scheduling/cluster-membership-mgr.cc | 5 +-
.../org/apache/impala/analysis/OptimizeStmt.java | 4 +-
.../org/apache/impala/catalog/FeIcebergTable.java | 17 ++--
.../apache/impala/catalog/FileMetadataLoader.java | 13 ++--
.../java/org/apache/impala/catalog/HdfsTable.java | 42 +++++-----
.../impala/catalog/IcebergContentFileStore.java | 68 ++++++----------
.../impala/catalog/IcebergFileMetadataLoader.java | 49 ++++++------
.../org/apache/impala/catalog/IcebergTable.java | 19 +++--
.../impala/catalog/ParallelFileMetadataLoader.java | 21 +----
.../apache/impala/planner/IcebergScanPlanner.java | 7 +-
.../java/org/apache/impala/util/AcidUtils.java | 20 ++++-
.../java/org/apache/impala/util/IcebergUtil.java | 23 +++---
.../impala/catalog/FileMetadataLoaderTest.java | 55 +++++++------
.../java/org/apache/impala/util/AcidUtilsTest.java | 18 +++++
.../nested-types-subplan-single-node.test | 3 -
tests/common/impala_test_suite.py | 90 ++++++++++++++++------
tests/common/test_vector.py | 1 +
tests/conftest.py | 3 -
tests/custom_cluster/test_admission_controller.py | 20 ++---
.../test_blacklisted_dbs_and_tables.py | 10 +--
tests/custom_cluster/test_data_cache.py | 43 ++++++-----
tests/failure/test_failpoints.py | 15 ++--
tests/query_test/test_acid.py | 25 ++++++
tests/query_test/test_aggregation.py | 14 ++--
tests/query_test/test_cancellation.py | 42 +++++-----
tests/query_test/test_mem_usage_scaling.py | 14 ++--
tests/query_test/test_mt_dop.py | 13 ++--
tests/query_test/test_nested_types.py | 5 +-
tests/query_test/test_runtime_filters.py | 20 +++--
tests/query_test/test_scanners.py | 29 +++----
tests/query_test/test_tablesample.py | 15 ++--
tests/util/cancel_util.py | 14 +++-
32 files changed, 420 insertions(+), 317 deletions(-)
rename testdata/workloads/{functional-query => tpch_nested}/queries/QueryTest/nested-types-subplan-single-node.test (97%)
(impala) 04/04: IMPALA-13771: Fix heap-use-after-free in Cluster Membership Manager
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 0f17ae8fc7e89d4a8bd6bc27b816a1a7213c1737
Author: jasonmfehr <jf...@cloudera.com>
AuthorDate: Wed Feb 19 13:33:54 2025 -0800
IMPALA-13771: Fix heap-use-after-free in Cluster Membership Manager
The ASAN tests identified a heap-use-after-free issue. This patch
fixes that issue by moving the memory release to after its last use.
Local execution of custom-cluster-mgr-test ASAN replicated the same
heap-use-after-free failure.
Testing:
Existing custom-cluster-mgr-test passed locally using debug build.
Existing custom-cluster-mgr-test passed locally using ASAN build.
Change-Id: I4fd2c9faa6daba9274f38238b952c377a07794e9
Reviewed-on: http://gerrit.cloudera.org:8080/22503
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
be/src/scheduling/cluster-membership-mgr.cc | 5 ++++-
1 file changed, 4 insertions(+), 1 deletion(-)
diff --git a/be/src/scheduling/cluster-membership-mgr.cc b/be/src/scheduling/cluster-membership-mgr.cc
index ab8bf9a61..ce3c211db 100644
--- a/be/src/scheduling/cluster-membership-mgr.cc
+++ b/be/src/scheduling/cluster-membership-mgr.cc
@@ -300,12 +300,15 @@ void ClusterMembershipMgr::UpdateMembership(
RemoveExecutorAndGroup(be_desc, group, new_executor_groups);
}
}
- new_backend_map->erase(item.key);
// If a coordinator is not shutdown gracefully, then it will be deleted here.
if (be_desc.is_coordinator()) {
_removeCoordIfExists(new_state, be_desc);
}
+
+ // Note: be_desc is a reference to item.key, thus this erase must come at the end
+ // of the loop where be_desc is initialized.
+ new_backend_map->erase(item.key);
}
continue;
}
(impala) 02/04: IMPALA-13759: Fix Hive ACID INSERT OVERWRITE base detection
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit c8c64fff3a1bbeb05a6869b91a945b2730f62083
Author: Csaba Ringhofer <cs...@cloudera.com>
AuthorDate: Wed Feb 19 13:32:56 2025 +0100
IMPALA-13759: Fix Hive ACID INSERT OVERWRITE base detection
Base directory created by INSERT OVERWRITE / TRUNCATE should be
treated differently than bases created by compaction because
IOW/TRUNCATE bases must be accepted even if there is an earlier
open writeId. This scenario can easily occur if there is
a pending write to a single partition, as this doesn't block
an IOW/TRUNCATE to another partition, while the global
minOpenWrite affects whether the base is accepted.
This change updates Impala logic to consider these bases
valid similarly to Hive.
Note that differentiating IOW/TRUNCATE from compaction is
different than in Hive, as metadata files are not considered
in Impala (IMPALA-13769). This can only cause problems when
interacting with earlier Hive versions that did not use
visibilityTxnId in the base path. I don't consider this
to be a significant regression that should block the current
critical fix.
Testing:
- added regression EE/FE tests
Change-Id: I838eaf4f41bae148e558f64288a1370c0908efa4
Reviewed-on: http://gerrit.cloudera.org:8080/22499
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../java/org/apache/impala/util/AcidUtils.java | 20 +++++++++++++++--
.../java/org/apache/impala/util/AcidUtilsTest.java | 18 ++++++++++++++++
tests/query_test/test_acid.py | 25 ++++++++++++++++++++++
3 files changed, 61 insertions(+), 2 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/util/AcidUtils.java b/fe/src/main/java/org/apache/impala/util/AcidUtils.java
index 8b369a1a4..fccf06bc9 100644
--- a/fe/src/main/java/org/apache/impala/util/AcidUtils.java
+++ b/fe/src/main/java/org/apache/impala/util/AcidUtils.java
@@ -257,9 +257,25 @@ public class AcidUtils {
public boolean check(String dirPath) throws CatalogException {
ParsedBase parsedBase = parseBase(dirPath);
+ // The logic below is not completely in sync with the way Hive works currently:
+ // https://github.com/apache/hive/blob/0759352ddddc793c0e717c460f0e08eb3f14c1e9/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java#L1774-L1797
+ // The main difference is that metadata files are ignored, so Impala cannot
+ // differentiate between bases created by INSERT OVERWRITE / TRUNCATE and compacted
+ // bases created by older Hives (that did not use visibiliyTxnId). See IMPALA-13759
+ // for more info.
if (parsedBase.writeId != SENTINEL_BASE_WRITE_ID) {
- boolean isValid = writeIdList.isValidBase(parsedBase.writeId) &&
- isTxnValid(parsedBase.visibilityTxnId);
+ boolean isValid = false;
+ if (parsedBase.visibilityTxnId != -1) {
+ // Assume that the base is produced by a compaction.
+ isValid = writeIdList.isValidBase(parsedBase.writeId) &&
+ isTxnValid(parsedBase.visibilityTxnId);
+ } else {
+ // Assume that the base is produced by INSERT OVERWRITE or TRUNCATE. It is
+ // possible for isValidBase() to fail in this case, for example if there is
+ // an earlier open writeId in a different partition, so only check whether
+ // the writeId is committed.
+ isValid = writeIdList.isWriteIdValid(parsedBase.writeId);
+ }
if (doStrictCheck && !isValid) {
throw new CatalogException("Invalid base file found " + dirPath);
}
diff --git a/fe/src/test/java/org/apache/impala/util/AcidUtilsTest.java b/fe/src/test/java/org/apache/impala/util/AcidUtilsTest.java
index 6a17b82d2..04644bf94 100644
--- a/fe/src/test/java/org/apache/impala/util/AcidUtilsTest.java
+++ b/fe/src/test/java/org/apache/impala/util/AcidUtilsTest.java
@@ -597,6 +597,24 @@ public class AcidUtilsTest {
"delta_0000012_0000012_0000/0000_1"});
}
+ @Test
+ public void testOpenWriteIdBeforeInsertOverwrite() {
+ assertFiltering(new String[]{
+ "base_000001/",
+ "base_000001/0000_0",
+ "delta_000002_000002_0000/",
+ "delta_000002_000002_0000/0000",
+ "base_000003/",
+ "base_000003/0000_0"},
+ // <tbl>:<hwm>:<minOpenWriteId>:<openWriteIds>:<abortedWriteIds>
+ "default.test:3:2:1:",
+ new String[]{
+ // As there is no visibilityTxnId, it is assumed that base_000003 was produced
+ // by an insert overwrite and is accepted even though there is an earlier open
+ // write id (regression test for IMPALA-13759).
+ "base_000003/0000_0"});
+ }
+
@Test
public void testWriteIdListCompare() {
ValidWriteIdList a =
diff --git a/tests/query_test/test_acid.py b/tests/query_test/test_acid.py
index 98931119f..860aee168 100644
--- a/tests/query_test/test_acid.py
+++ b/tests/query_test/test_acid.py
@@ -231,6 +231,12 @@ class TestAcid(ImpalaTestSuite):
commit_req.txnid = txn_id
return self.hive_client.commit_txn(commit_req)
+ def _create_insert_only_acid_table(self, tbl, cols, part_cols=None):
+ part_part = ("partitioned by (%s)" % part_cols) if part_cols else ""
+ self.execute_query("create table {} ({}) {} tblproperties"
+ "('transactional'='true',"
+ "'transactional_properties'='insert_only')".format(tbl, cols, part_part))
+
@SkipIfFS.hive
def test_lock_timings(self, vector, unique_database):
def elapsed_time_for_query(query):
@@ -360,3 +366,22 @@ class TestAcid(ImpalaTestSuite):
# Create a new table and load it in catalogd. Catalogd should not hang.
self.client.execute("CREATE TABLE {0}.tbl (i int)".format(unique_database))
self.client.execute("DESCRIBE {0}.tbl".format(unique_database))
+
+ def test_insert_overwrite_base_detection(self, unique_database):
+ """"Regression test for IMPALA-13759. Checks that the base directory created by
+ INSERT OVERWRITE is detected correctly even if there is an earlier open writeId
+ in another partition."""
+ tbl = unique_database + ".insert_only_table"
+ self._create_insert_only_acid_table(tbl, "i int", "p string")
+ sleep_time_ms = 15000
+ other_partition_insert_handle = self.execute_query_async(
+ 'insert into %s partition (p="a") values (sleep(%d))' % (tbl, sleep_time_ms))
+ self.execute_query('insert into %s partition (p="b") values (2)' % tbl)
+ self.execute_query('insert overwrite %s partition (p="b") values (3)' % tbl)
+ self.execute_query('insert into %s partition (p="b") values (4)' % tbl)
+ result = self.execute_query("select * from %s order by i" % tbl)
+ assert result.data == ['3\tb', '4\tb']
+ self.client.wait_for_finished_timeout(
+ other_partition_insert_handle, 0.001 * sleep_time_ms)
+ result = self.execute_query("select * from %s order by i" % tbl)
+ assert result.data == ['1\ta', '3\tb', '4\tb']
(impala) 01/04: IMPALA-13737: Directly load file metadata via IcebergFileMetadataLoader
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 37e409059437279c960eba71b3bce69ffbd65f2e
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
AuthorDate: Wed Feb 5 18:38:38 2025 +0100
IMPALA-13737: Directly load file metadata via IcebergFileMetadataLoader
Currently we let HdfsTable to drive file metadata loading of Iceberg
tables. To have better control over file loading, IcebergTable should
use IcebergFileMetadataLoader directly. The underlying HdfsTable can be
empty, which will make it easier to remove this dependency completely.
Also, it solves the de-duplication of file descriptors in Local Catalog
mode.
This patch also clarifies the responsibilities of
IcebergFileMetadataLoader and IcebergContentFileStore. The former
is in charge of loading the file descriptors and decorating them
with Iceberg metadata. The latter is only responsible for grouping
and storing them in an efficient manner.
This patch removes the dependency of IcebergContentFileStore on
FeIcebergTable which will make the REST Catalog implementation
cleaner.
Measurements
(Thanks to Gabor Kaszab for the numbers)
As mentioned above, this patch de-duplicates the file descriptors
in local catalog mode. I.e. it greatly reduces the memory footprint
(IMPALA-11265) in the Coordinator when local catalog is being used.
The measured table had 110k files, 16400 partitions, 1000 manifests,
1000 snapshots. The memory footprint:
Before this patch: 107MB
After this patch: 74MB
Testing:
* no new functionalities added, existing tests should work
Change-Id: Iaf7e23ec21b65036b47edadcb4cbe4b64be3baee
Reviewed-on: http://gerrit.cloudera.org:8080/22458
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../org/apache/impala/analysis/OptimizeStmt.java | 4 +-
.../org/apache/impala/catalog/FeIcebergTable.java | 17 ++++--
.../apache/impala/catalog/FileMetadataLoader.java | 13 +++--
.../java/org/apache/impala/catalog/HdfsTable.java | 42 ++++++-------
.../impala/catalog/IcebergContentFileStore.java | 68 ++++++++--------------
.../impala/catalog/IcebergFileMetadataLoader.java | 49 ++++++++--------
.../org/apache/impala/catalog/IcebergTable.java | 19 +++---
.../impala/catalog/ParallelFileMetadataLoader.java | 21 +------
.../apache/impala/planner/IcebergScanPlanner.java | 7 ++-
.../java/org/apache/impala/util/IcebergUtil.java | 23 ++++----
.../impala/catalog/FileMetadataLoaderTest.java | 55 +++++++++--------
11 files changed, 148 insertions(+), 170 deletions(-)
diff --git a/fe/src/main/java/org/apache/impala/analysis/OptimizeStmt.java b/fe/src/main/java/org/apache/impala/analysis/OptimizeStmt.java
index 3fecd954b..de120dc56 100644
--- a/fe/src/main/java/org/apache/impala/analysis/OptimizeStmt.java
+++ b/fe/src/main/java/org/apache/impala/analysis/OptimizeStmt.java
@@ -238,7 +238,9 @@ public class OptimizeStmt extends DmlStatementBase {
GroupedContentFiles selectedContentFiles = new GroupedContentFiles();
selectedContentFiles.dataFilesWithoutDeletes = contentFiles;
IcebergContentFileStore selectedFiles =
- new IcebergContentFileStore(iceTable, selectedContentFiles);
+ new IcebergContentFileStore(iceTable.getIcebergApiTable(),
+ iceTable.getContentFileStore().getDataFilesWithoutDeletes(),
+ selectedContentFiles);
return selectedFiles.getDataFilesWithoutDeletes();
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
index 8acc4ca41..c6597ef80 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FeIcebergTable.java
@@ -708,7 +708,9 @@ public interface FeIcebergTable extends FeFsTable {
* Get FileDescriptor by data file location
*/
public static FileDescriptor getFileDescriptor(
- ContentFile<?> contentFile, FeIcebergTable table) throws IOException {
+ ContentFile<?> contentFile, Table iceApiTable,
+ boolean requiresDataFilesInTableLocation,
+ ListMap<TNetworkAddress> hostIndex) throws IOException {
Path fileLoc = FileSystemUtil.createFullyQualifiedPath(
new Path(contentFile.path().toString()));
FileSystem fsForPath = FileSystemUtil.getFileSystemForPath(fileLoc);
@@ -719,19 +721,22 @@ public interface FeIcebergTable extends FeFsTable {
// For OSS service (e.g. S3A, COS, OSS, etc), we create FileStatus ourselves.
fileStatus = createFileStatus(contentFile, fileLoc);
}
- return getFileDescriptor(fsForPath, fileStatus, table);
+ return getFileDescriptor(fsForPath, fileStatus, iceApiTable,
+ requiresDataFilesInTableLocation, hostIndex);
}
private static FileDescriptor getFileDescriptor(FileSystem fs,
- FileStatus fileStatus, FeIcebergTable table) throws IOException {
+ FileStatus fileStatus, Table iceApiTable,
+ boolean requiresDataFilesInTableLocation,
+ ListMap<TNetworkAddress> hostIndex) throws IOException {
Reference<Long> numUnknownDiskIds = new Reference<>(0L);
String absPath = null;
- Path tableLoc = new Path(table.getIcebergTableLocation());
+ Path tableLoc = new Path(iceApiTable.location());
String relPath = FileSystemUtil.relativizePathNoThrow(
fileStatus.getPath(), tableLoc);
if (relPath == null) {
- if (Utils.requiresDataFilesInTableLocation(table)) {
+ if (requiresDataFilesInTableLocation) {
throw new RuntimeException(fileStatus.getPath()
+ " is outside of the Iceberg table location " + tableLoc);
}
@@ -751,7 +756,7 @@ public interface FeIcebergTable extends FeFsTable {
}
return FileDescriptor.create(fileStatus, relPath, locations,
- table.getHostIndex(), fileStatus.isEncrypted(), fileStatus.isErasureCoded(),
+ hostIndex, fileStatus.isEncrypted(), fileStatus.isErasureCoded(),
numUnknownDiskIds, absPath);
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
index f94114d9a..a6e0534cc 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
@@ -96,9 +96,10 @@ public class FileMetadataLoader {
* @param fileFormat if non-null and equal to HdfsFileFormat.HUDI_PARQUET,
* this loader will filter files based on Hudi's HoodieROTablePathFilter method
*/
- public FileMetadataLoader(Path partDir, boolean recursive, List<FileDescriptor> oldFds,
- ListMap<TNetworkAddress> hostIndex, @Nullable ValidTxnList validTxnList,
- @Nullable ValidWriteIdList writeIds, @Nullable HdfsFileFormat fileFormat) {
+ public FileMetadataLoader(Path partDir, boolean recursive,
+ Iterable<FileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
+ @Nullable ValidTxnList validTxnList, @Nullable ValidWriteIdList writeIds,
+ @Nullable HdfsFileFormat fileFormat) {
// Either both validTxnList and writeIds are null, or none of them.
Preconditions.checkState((validTxnList == null && writeIds == null)
|| (validTxnList != null && writeIds != null));
@@ -116,9 +117,9 @@ public class FileMetadataLoader {
TOTAL_TASKS.incrementAndGet();
}
- public FileMetadataLoader(Path partDir, boolean recursive, List<FileDescriptor> oldFds,
- ListMap<TNetworkAddress> hostIndex, @Nullable ValidTxnList validTxnList,
- @Nullable ValidWriteIdList writeIds) {
+ public FileMetadataLoader(Path partDir, boolean recursive,
+ Iterable<FileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
+ @Nullable ValidTxnList validTxnList, @Nullable ValidWriteIdList writeIds) {
this(partDir, recursive, oldFds, hostIndex, validTxnList, writeIds, null);
}
diff --git a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
index 0498cef0e..141f38286 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -58,7 +58,6 @@ import org.apache.impala.analysis.NullLiteral;
import org.apache.impala.analysis.NumericLiteral;
import org.apache.impala.analysis.PartitionKeyValue;
import org.apache.impala.catalog.events.MetastoreEventsProcessor;
-import org.apache.impala.catalog.iceberg.GroupedContentFiles;
import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.ImpalaException;
import org.apache.impala.common.Pair;
@@ -244,10 +243,11 @@ public class HdfsTable extends Table implements FeFsTable {
// Declared as protected to allow third party extension visibility.
protected final Map<Long, HdfsPartition> partitionMap_ = new HashMap<>();
- // The data and delete files of the table, for Iceberg tables only.
- private GroupedContentFiles icebergFiles_;
- // Whether the data and delete files of The Iceberg tables can be outside the location.
- private boolean canDataBeOutsideOfTableLocation_;
+ // Whether to skip file metadata loading. Iceberg tables take care of file metadata
+ // loading themselves, so it is true iff this HdfsTable is part of an
+ // IcebergTable object.
+ // TODO(IMPALA-13734/IMPALA-13738): Remove this flag.
+ private boolean skipIcebergFileMetadataLoading_ = false;
// Map of partition name to HdfsPartition object. Used for speeding up
// table metadata loading. It is only populated if this table object is stored in
@@ -339,13 +339,8 @@ public class HdfsTable extends Table implements FeFsTable {
// to determine if we can skip the table from the topic update.
private long lastVersionSeenByTopicUpdate_ = -1;
- public void setIcebergFiles(GroupedContentFiles icebergFiles) {
- icebergFiles_ = icebergFiles;
- }
-
- public void setCanDataBeOutsideOfTableLocation(
- boolean canDataBeOutsideOfTableLocation) {
- canDataBeOutsideOfTableLocation_ = canDataBeOutsideOfTableLocation;
+ public void setSkipIcebergFileMetadataLoading(boolean skipIcebergFileMetadataLoading) {
+ skipIcebergFileMetadataLoading_ = skipIcebergFileMetadataLoading;
}
// Represents a set of storage-related statistics aggregated at the table or partition
@@ -425,8 +420,6 @@ public class HdfsTable extends Table implements FeFsTable {
super(msTbl, db, name, owner);
partitionLocationCompressor_ =
new HdfsPartitionLocationCompressor(numClusteringCols_);
- icebergFiles_ = new GroupedContentFiles();
- canDataBeOutsideOfTableLocation_ = false;
}
@Override // FeFsTable
@@ -814,15 +807,18 @@ public class HdfsTable extends Table implements FeFsTable {
isRefresh ? "Refreshing" : "Loading", partBuilders.size(),
getFullName());
- // Actually load the partitions.
- // TODO(IMPALA-8406): if this fails to load files from one or more partitions, then
- // we'll throw an exception here and end up bailing out of whatever catalog operation
- // we're in the middle of. This could cause a partial metadata update -- eg we may
- // have refreshed the top-level table properties without refreshing the files.
- new ParallelFileMetadataLoader(getFileSystem(), partBuilders, validWriteIds_,
- validTxnList, Utils.shouldRecursivelyListPartitions(this),
- getHostIndex(), debugActions, logPrefix, icebergFiles_,
- canDataBeOutsideOfTableLocation_).load();
+
+ if (!skipIcebergFileMetadataLoading_) {
+ // Actually load the partitions.
+ // TODO(IMPALA-8406): if this fails to load files from one or more partitions, then
+ // we'll throw an exception here and end up bailing out of whatever catalog
+ // operation we're in the middle of. This could cause a partial metadata update
+ // -- eg we may have refreshed the top-level table properties without refreshing the
+ // files.
+ new ParallelFileMetadataLoader(getFileSystem(), partBuilders, validWriteIds_,
+ validTxnList, Utils.shouldRecursivelyListPartitions(this),
+ getHostIndex(), debugActions, logPrefix).load();
+ }
// TODO(todd): would be good to log a summary of the loading process:
// - how many block locations did we reuse/load individually/load via batch
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java b/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java
index 562a85d50..1878eab12 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergContentFileStore.java
@@ -20,10 +20,8 @@ package org.apache.impala.catalog;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
-import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.ArrayList;
-import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -35,12 +33,13 @@ import org.apache.hadoop.fs.Path;
import org.apache.iceberg.ContentFile;
import org.apache.iceberg.DataFile;
import org.apache.iceberg.DeleteFile;
-import org.apache.iceberg.FileContent;
+import org.apache.iceberg.Table;
import org.apache.impala.catalog.iceberg.GroupedContentFiles;
import org.apache.impala.common.Pair;
import org.apache.impala.fb.FbFileDesc;
import org.apache.impala.fb.FbFileMetadata;
import org.apache.impala.fb.FbIcebergDataFileFormat;
+import org.apache.impala.fb.FbIcebergMetadata;
import org.apache.impala.thrift.THdfsFileDesc;
import org.apache.impala.thrift.TIcebergContentFileStore;
import org.apache.impala.thrift.TNetworkAddress;
@@ -191,38 +190,36 @@ public class IcebergContentFileStore {
public IcebergContentFileStore() {}
public IcebergContentFileStore(
- FeIcebergTable icebergTable, GroupedContentFiles icebergFiles) {
- Preconditions.checkNotNull(icebergTable);
+ Table iceApiTable, List<FileDescriptor> fileDescriptors,
+ GroupedContentFiles icebergFiles) {
+ Preconditions.checkNotNull(iceApiTable);
+ Preconditions.checkNotNull(fileDescriptors);
Preconditions.checkNotNull(icebergFiles);
Map<String, FileDescriptor> hdfsFileDescMap = new HashMap<>();
- Collection<? extends FeFsPartition> partitions =
- FeCatalogUtils.loadAllPartitions(icebergTable.getFeFsTable());
- for (FeFsPartition partition : partitions) {
- for (FileDescriptor fileDesc : partition.getFileDescriptors()) {
- Path path = new Path(fileDesc.getAbsolutePath(icebergTable.getHdfsBaseDir()));
- hdfsFileDescMap.put(path.toUri().getPath(), fileDesc);
- }
+ for (FileDescriptor fileDesc : fileDescriptors) {
+ Path path = new Path(fileDesc.getAbsolutePath(iceApiTable.location()));
+ hdfsFileDescMap.put(path.toUri().getPath(), fileDesc);
}
for (DataFile dataFile : icebergFiles.dataFilesWithoutDeletes) {
Pair<String, EncodedFileDescriptor> pathHashAndFd =
- getPathHashAndFd(icebergTable, dataFile, hdfsFileDescMap);
+ getPathHashAndFd(dataFile, hdfsFileDescMap);
dataFilesWithoutDeletes_.add(pathHashAndFd.first, pathHashAndFd.second);
}
for (DataFile dataFile : icebergFiles.dataFilesWithDeletes) {
Pair<String, EncodedFileDescriptor> pathHashAndFd =
- getPathHashAndFd(icebergTable, dataFile, hdfsFileDescMap);
+ getPathHashAndFd(dataFile, hdfsFileDescMap);
dataFilesWithDeletes_.add(pathHashAndFd.first, pathHashAndFd.second);
}
for (DeleteFile deleteFile : icebergFiles.positionDeleteFiles) {
Pair<String, EncodedFileDescriptor> pathHashAndFd =
- getPathHashAndFd(icebergTable, deleteFile, hdfsFileDescMap);
+ getPathHashAndFd(deleteFile, hdfsFileDescMap);
positionDeleteFiles_.add(pathHashAndFd.first, pathHashAndFd.second);
}
for (DeleteFile deleteFile : icebergFiles.equalityDeleteFiles) {
Pair<String, EncodedFileDescriptor> pathHashAndFd =
- getPathHashAndFd(icebergTable, deleteFile, hdfsFileDescMap);
+ getPathHashAndFd(deleteFile, hdfsFileDescMap);
equalityDeleteFiles_.add(pathHashAndFd.first, pathHashAndFd.second);
}
}
@@ -291,11 +288,10 @@ public class IcebergContentFileStore {
public boolean hasOrc() { return hasOrc_; }
public boolean hasParquet() { return hasParquet_; }
- private void updateFileFormats(FbFileMetadata icebergMetadata) {
+ private void updateFileFormats(FbIcebergMetadata icebergMetadata) {
Preconditions.checkNotNull(icebergMetadata);
- Preconditions.checkNotNull(icebergMetadata.icebergMetadata());
- byte fileFormat = icebergMetadata.icebergMetadata().fileFormat();
+ byte fileFormat = icebergMetadata.fileFormat();
if (fileFormat == FbIcebergDataFileFormat.PARQUET) {
hasParquet_ = true;
} else if (fileFormat == FbIcebergDataFileFormat.ORC) {
@@ -306,37 +302,22 @@ public class IcebergContentFileStore {
}
private Pair<String, EncodedFileDescriptor> getPathHashAndFd(
- FeIcebergTable icebergTable,
- ContentFile<?> contentFile,
- Map<String, FileDescriptor> hdfsFileDescMap) {
+ ContentFile<?> contentFile, Map<String, FileDescriptor> hdfsFileDescMap) {
return new Pair<>(
IcebergUtil.getFilePathHash(contentFile),
- getOrCreateIcebergFd(icebergTable, hdfsFileDescMap, contentFile));
+ getIcebergFd(hdfsFileDescMap, contentFile));
}
- private EncodedFileDescriptor getOrCreateIcebergFd(
- FeIcebergTable icebergTable,
+ private EncodedFileDescriptor getIcebergFd(
Map<String, FileDescriptor> hdfsFileDescMap,
ContentFile<?> contentFile) {
Path path = new Path(contentFile.path().toString());
- FileDescriptor fileDesc = null;
- if (hdfsFileDescMap.containsKey(path.toUri().getPath())) {
- fileDesc = hdfsFileDescMap.get(path.toUri().getPath());
- } else {
- if (FeIcebergTable.Utils.requiresDataFilesInTableLocation(icebergTable)) {
- LOG.warn("Iceberg file '{}' cannot be found in the HDFS recursive"
- + "file listing results.", path);
- }
- try {
- fileDesc = FeIcebergTable.Utils.getFileDescriptor(contentFile, icebergTable);
- } catch (IOException e) {
- throw new IllegalArgumentException(e.getMessage());
- }
- }
- Preconditions.checkNotNull(fileDesc);
+ FileDescriptor fileDesc = hdfsFileDescMap.get(path.toUri().getPath());
- FbFileMetadata icebergMetadata =
- IcebergUtil.createIcebergMetadata(icebergTable, contentFile);
+ FbFileMetadata fileMetadata = fileDesc.getFbFileMetadata();
+ Preconditions.checkState(fileMetadata != null);
+ FbIcebergMetadata icebergMetadata = fileMetadata.icebergMetadata();
+ Preconditions.checkState(icebergMetadata != null);
updateFileFormats(icebergMetadata);
@@ -357,9 +338,6 @@ public class IcebergContentFileStore {
return ret;
}
- // TODO IMPALA-11265: After converting to/from thrift the byte arrays representing the
- // file descriptors won't be shared between the HdfsTable and IcebergContentFileStore.
- // This redundancy causes unnecessary JVM heap memory usage on the coordinator.
public static IcebergContentFileStore fromThrift(TIcebergContentFileStore tFileStore,
List<TNetworkAddress> networkAddresses,
ListMap<TNetworkAddress> hostIndex) {
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
index d4f6edcf2..af338372e 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergFileMetadataLoader.java
@@ -53,6 +53,7 @@ import org.apache.impala.common.Reference;
import org.apache.impala.common.Pair;
import org.apache.impala.service.BackendConfig;
import org.apache.impala.thrift.TNetworkAddress;
+import org.apache.impala.util.IcebergUtil;
import org.apache.impala.util.ListMap;
import org.apache.impala.util.ThreadNameAnnotator;
@@ -70,32 +71,32 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
// invalid value.
private final int NEW_FILES_THRESHOLD_DEFAULT = 100;
+ private final org.apache.iceberg.Table iceTbl_;
+
// If there are more new files than 'newFilesThreshold_', we should fall back
// to regular file metadata loading.
private final int newFilesThreshold_;
private final GroupedContentFiles icebergFiles_;
- private final boolean canDataBeOutsideOfTableLocation_;
+ private final boolean requiresDataFilesInTableLocation_;
private boolean useParallelListing_;
- public IcebergFileMetadataLoader(Path partDir, boolean recursive,
- List<FileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
- ValidTxnList validTxnList, ValidWriteIdList writeIds,
- GroupedContentFiles icebergFiles, boolean canDataBeOutsideOfTableLocation) {
- this(partDir, recursive, oldFds, hostIndex, validTxnList, writeIds,
- icebergFiles, canDataBeOutsideOfTableLocation,
+ public IcebergFileMetadataLoader(org.apache.iceberg.Table iceTbl,
+ Iterable<FileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
+ GroupedContentFiles icebergFiles, boolean requiresDataFilesInTableLocation) {
+ this(iceTbl, oldFds, hostIndex, icebergFiles, requiresDataFilesInTableLocation,
BackendConfig.INSTANCE.icebergReloadNewFilesThreshold());
}
- public IcebergFileMetadataLoader(Path partDir, boolean recursive,
- List<FileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
- ValidTxnList validTxnList, ValidWriteIdList writeIds,
- GroupedContentFiles icebergFiles, boolean canDataBeOutsideOfTableLocation,
+ public IcebergFileMetadataLoader(org.apache.iceberg.Table iceTbl,
+ Iterable<FileDescriptor> oldFds, ListMap<TNetworkAddress> hostIndex,
+ GroupedContentFiles icebergFiles, boolean requiresDataFilesInTableLocation,
int newFilesThresholdParam) {
- super(partDir, recursive, oldFds, hostIndex, validTxnList, writeIds,
- HdfsFileFormat.ICEBERG);
+ super(FileSystemUtil.createFullyQualifiedPath(new Path(iceTbl.location())), true,
+ oldFds, hostIndex, null, null, HdfsFileFormat.ICEBERG);
+ iceTbl_ = iceTbl;
icebergFiles_ = icebergFiles;
- canDataBeOutsideOfTableLocation_ = canDataBeOutsideOfTableLocation;
+ requiresDataFilesInTableLocation_ = requiresDataFilesInTableLocation;
if (newFilesThresholdParam >= 0) {
newFilesThreshold_ = newFilesThresholdParam;
} else {
@@ -131,9 +132,9 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
FileSystem defaultFs = FileSystemUtil.getDefaultFileSystem();
for (ContentFile<?> contentFile : newContentFiles) {
FileSystem fsForPath = fsForTable;
- // If canDataBeOutsideOfTableLocation is not true, we assume that the file system
+ // If requiresDataFilesInTableLocation_ is true, we assume that the file system
// for all ContentFiles is the same as fsForTable
- if (canDataBeOutsideOfTableLocation_) {
+ if (!requiresDataFilesInTableLocation_) {
Path path = new Path(contentFile.path().toString());
fsForPath = path.toUri().getScheme() != null ?
FileSystemUtil.getFileSystemForPath(path) : defaultFs;
@@ -218,14 +219,16 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
String absPath = null;
String relPath = FileSystemUtil.relativizePathNoThrow(stat.getPath(), partDir_);
if (relPath == null) {
- if (canDataBeOutsideOfTableLocation_) {
- absPath = stat.getPath().toString();
- } else {
+ if (requiresDataFilesInTableLocation_) {
throw new IOException(String.format("Failed to load Iceberg datafile %s, because "
+ "it's outside of the table location", stat.getPath().toUri()));
+ } else {
+ absPath = stat.getPath().toString();
}
}
- return createFd(fs, stat, relPath, numUnknownDiskIds, absPath);
+ FileDescriptor fsFd = createFd(fs, stat, relPath, numUnknownDiskIds, absPath);
+ return fsFd.cloneWithFileMetadata(
+ IcebergUtil.createIcebergMetadata(iceTbl_, contentFile));
}
/**
@@ -293,11 +296,11 @@ public class IcebergFileMetadataLoader extends FileMetadataLoader {
new Path(contentFile.path().toString()));
String lookupPath = FileSystemUtil.relativizePathNoThrow(contentFilePath, partDir_);
if (lookupPath == null) {
- if (canDataBeOutsideOfTableLocation_) {
- lookupPath = contentFilePath.toString();
- } else {
+ if (requiresDataFilesInTableLocation_) {
throw new IOException(String.format("Failed to load Iceberg datafile %s, because "
+ "it's outside of the table location", contentFilePath));
+ } else {
+ lookupPath = contentFilePath.toString();
}
}
return oldFdsByPath_.get(lookupPath);
diff --git a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
index 525525592..31730c66b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/IcebergTable.java
@@ -20,12 +20,15 @@ package org.apache.impala.catalog;
import com.codahale.metrics.Timer;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
+
import java.util.ArrayList;
+import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
import org.apache.hadoop.hive.metastore.TableType;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
@@ -443,19 +446,21 @@ public class IcebergTable extends Table implements FeIcebergTable {
GroupedContentFiles icebergFiles = IcebergUtil.getIcebergFiles(this,
new ArrayList<>(), /*timeTravelSpec=*/null);
catalogTimeline.markEvent("Loaded Iceberg files");
- hdfsTable_.setIcebergFiles(icebergFiles);
- hdfsTable_.setCanDataBeOutsideOfTableLocation(
- !Utils.requiresDataFilesInTableLocation(this));
+ hdfsTable_.setSkipIcebergFileMetadataLoading(true);
hdfsTable_.load(reuseMetadata, msClient, msTable_, reason, catalogTimeline);
- fileStore_ = new IcebergContentFileStore(this, icebergFiles);
+ IcebergFileMetadataLoader loader = new IcebergFileMetadataLoader(
+ icebergApiTable_,
+ fileStore_ == null ? Collections.emptyList() : fileStore_.getAllFiles(),
+ getHostIndex(), Preconditions.checkNotNull(icebergFiles),
+ Utils.requiresDataFilesInTableLocation(this));
+ loader.load();
+ fileStore_ = new IcebergContentFileStore(
+ icebergApiTable_, loader.getLoadedFds(), icebergFiles);
partitionStats_ = Utils.loadPartitionStats(this, icebergFiles);
setIcebergTableStats();
loadAllColumnStats(msClient, catalogTimeline);
applyPuffinNdvStats(catalogTimeline);
setAvroSchema(msClient, msTbl, fileStore_, catalogTimeline);
-
- // We no longer need to keep Iceberg's content files in memory.
- hdfsTable_.setIcebergFiles(null);
} catch (Exception e) {
throw new IcebergTableLoadingException("Error loading metadata for Iceberg table "
+ icebergTableLocation_, e);
diff --git a/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java b/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
index b3362da03..6f35f146b 100644
--- a/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/ParallelFileMetadataLoader.java
@@ -34,7 +34,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.common.ValidWriteIdList;
import org.apache.impala.catalog.HdfsPartition.Builder;
-import org.apache.impala.catalog.iceberg.GroupedContentFiles;
import org.apache.impala.common.FileSystemUtil;
import org.apache.impala.common.Pair;
import org.apache.impala.service.BackendConfig;
@@ -81,15 +80,6 @@ public class ParallelFileMetadataLoader {
ValidWriteIdList writeIdList, ValidTxnList validTxnList, boolean isRecursive,
@Nullable ListMap<TNetworkAddress> hostIndex, String debugAction,
String logPrefix) {
- this(fs, partBuilders, writeIdList, validTxnList, isRecursive, hostIndex, debugAction,
- logPrefix, new GroupedContentFiles(), false);
- }
-
- public ParallelFileMetadataLoader(FileSystem fs,
- Collection<Builder> partBuilders,
- ValidWriteIdList writeIdList, ValidTxnList validTxnList, boolean isRecursive,
- @Nullable ListMap<TNetworkAddress> hostIndex, String debugAction, String logPrefix,
- GroupedContentFiles icebergFiles, boolean canDataBeOutsideOfTableLocation) {
if (writeIdList != null || validTxnList != null) {
// make sure that both either both writeIdList and validTxnList are set or both
// of them are not.
@@ -109,14 +99,9 @@ public class ParallelFileMetadataLoader {
List<FileDescriptor> oldFds = e.getValue().get(0).getFileDescriptors();
FileMetadataLoader loader;
HdfsFileFormat format = e.getValue().get(0).getFileFormat();
- if (format.equals(HdfsFileFormat.ICEBERG)) {
- loader = new IcebergFileMetadataLoader(e.getKey(), isRecursive, oldFds, hostIndex,
- validTxnList, writeIdList, Preconditions.checkNotNull(icebergFiles),
- canDataBeOutsideOfTableLocation);
- } else {
- loader = new FileMetadataLoader(e.getKey(), isRecursive, oldFds, hostIndex,
- validTxnList, writeIdList, format);
- }
+ Preconditions.checkState(!HdfsFileFormat.ICEBERG.equals(format));
+ loader = new FileMetadataLoader(e.getKey(), isRecursive, oldFds, hostIndex,
+ validTxnList, writeIdList, format);
// If there is a cached partition mapped to this path, we recompute the block
// locations even if the underlying files have not changed.
// This is done to keep the cached block metadata up to date.
diff --git a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
index 7e20f0f5a..8cd74c7ff 100644
--- a/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
+++ b/fe/src/main/java/org/apache/impala/planner/IcebergScanPlanner.java
@@ -730,7 +730,10 @@ public class IcebergScanPlanner {
}
cachehit = false;
try {
- fileDesc = FeIcebergTable.Utils.getFileDescriptor(cf, getIceTable());
+ fileDesc = FeIcebergTable.Utils.getFileDescriptor(cf,
+ getIceTable().getIcebergApiTable(),
+ FeIcebergTable.Utils.requiresDataFilesInTableLocation(getIceTable()),
+ getIceTable().getHostIndex());
} catch (IOException ex) {
throw new ImpalaRuntimeException(
"Cannot load file descriptor for " + cf.path(), ex);
@@ -741,7 +744,7 @@ public class IcebergScanPlanner {
}
// Add file descriptor to the cache.
fileDesc = fileDesc.cloneWithFileMetadata(
- IcebergUtil.createIcebergMetadata(getIceTable(), cf));
+ IcebergUtil.createIcebergMetadata(getIceTable().getIcebergApiTable(), cf));
fileStore.addOldFileDescriptor(pathHash, fileDesc);
}
return new Pair<>(fileDesc, cachehit);
diff --git a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
index 5ad64f71e..78a99a72f 100644
--- a/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
+++ b/fe/src/main/java/org/apache/impala/util/IcebergUtil.java
@@ -817,6 +817,7 @@ public class IcebergUtil {
public static Object getPartitionValue(Type type,
TIcebergPartitionTransformType transformType, String stringValue)
throws ImpalaRuntimeException {
+ // This needs to be consistent with createPartitionTransformValue().
String HIVE_NULL = MetaStoreUtil.DEFAULT_NULL_PARTITION_KEY_VALUE;
if (stringValue == null || stringValue.equals(HIVE_NULL)) return null;
@@ -1040,15 +1041,14 @@ public class IcebergUtil {
}
/**
- * Extracts metadata from Iceberg data file object 'df'. Such metadata is the file
+ * Extracts metadata from Iceberg data file object 'cf'. Such metadata is the file
* format of the data file, also the partition information the data file belongs.
* It creates a flatbuffer so it can be passed between machines and processes without
* further de/serialization.
*/
- public static FbFileMetadata createIcebergMetadata(FeIcebergTable feTbl,
- ContentFile cf) {
+ public static FbFileMetadata createIcebergMetadata(Table iceApiTbl, ContentFile cf) {
FlatBufferBuilder fbb = new FlatBufferBuilder(1);
- int iceOffset = createIcebergMetadata(feTbl, fbb, cf);
+ int iceOffset = createIcebergMetadata(iceApiTbl, fbb, cf);
fbb.finish(FbFileMetadata.createFbFileMetadata(fbb, iceOffset));
ByteBuffer bb = fbb.dataBuffer().slice();
ByteBuffer compressedBb = ByteBuffer.allocate(bb.capacity());
@@ -1056,12 +1056,12 @@ public class IcebergUtil {
return FbFileMetadata.getRootAsFbFileMetadata((ByteBuffer)compressedBb.flip());
}
- private static int createIcebergMetadata(FeIcebergTable feTbl, FlatBufferBuilder fbb,
+ private static int createIcebergMetadata(Table iceApiTbl, FlatBufferBuilder fbb,
ContentFile cf) {
int partKeysOffset = -1;
- PartitionSpec spec = feTbl.getIcebergApiTable().specs().get(cf.specId());
+ PartitionSpec spec = iceApiTbl.specs().get(cf.specId());
if (spec != null && !spec.fields().isEmpty()) {
- partKeysOffset = createPartitionKeys(feTbl, fbb, spec, cf);
+ partKeysOffset = createPartitionKeys(fbb, spec, cf);
}
int eqFieldIdsOffset = -1;
List<Integer> eqFieldIds = cf.equalityFieldIds();
@@ -1101,18 +1101,18 @@ public class IcebergUtil {
return FbIcebergMetadata.endFbIcebergMetadata(fbb);
}
- private static int createPartitionKeys(FeIcebergTable feTbl, FlatBufferBuilder fbb,
+ private static int createPartitionKeys(FlatBufferBuilder fbb,
PartitionSpec spec, ContentFile cf) {
Preconditions.checkState(spec.fields().size() == cf.partition().size());
int[] partitionKeyOffsets = new int[spec.fields().size()];
for (int i = 0; i < spec.fields().size(); ++i) {
partitionKeyOffsets[i] =
- createPartitionTransformValue(feTbl, fbb, spec, cf, i);
+ createPartitionTransformValue(fbb, spec, cf, i);
}
return FbIcebergMetadata.createPartitionKeysVector(fbb, partitionKeyOffsets);
}
- private static int createPartitionTransformValue(FeIcebergTable feTbl,
+ private static int createPartitionTransformValue(
FlatBufferBuilder fbb, PartitionSpec spec, ContentFile cf, int fieldIndex) {
PartitionField field = spec.fields().get(fieldIndex);
Pair<Byte, Integer> transform = getFbTransform(spec.schema(), field);
@@ -1123,7 +1123,8 @@ public class IcebergUtil {
if (partValue != null) {
partValueString = partValue.toString();
} else {
- partValueString = feTbl.getNullPartitionKeyValue();
+ // This needs to be consistent with getPartitionValue().
+ partValueString = MetaStoreUtil.DEFAULT_NULL_PARTITION_KEY_VALUE;
}
valueOffset = fbb.createString(partValueString);
}
diff --git a/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java b/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
index 30dc8c75b..397f79d74 100644
--- a/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
@@ -121,7 +121,7 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ Collections.emptyList(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml1.load();
assertEquals(20, fml1.getStats().loadedFiles);
assertEquals(0, fml1.getStats().skippedFiles);
@@ -136,7 +136,7 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml2 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ Collections.emptyList(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml2.load();
assertEquals(20, fml2.getStats().loadedFiles);
assertEquals(0, fml2.getStats().skippedFiles);
@@ -155,13 +155,13 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ Collections.emptyList(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml1.load();
IcebergFileMetadataLoader fml1Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ fml1.getLoadedFds(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml1Refresh.load();
assertFalse(fml1Refresh.useParallelListing());
assertEquals(0, fml1Refresh.getStats().loadedFiles);
@@ -178,13 +178,13 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml2 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ Collections.emptyList(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml2.load();
IcebergFileMetadataLoader fml2Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ fml2.getLoadedFds(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml2Refresh.load();
assertFalse(fml2Refresh.useParallelListing());
assertEquals(0, fml2Refresh.getStats().loadedFiles);
@@ -205,13 +205,13 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ Collections.emptyList(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml1.load();
IcebergFileMetadataLoader fml1Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ fml1.getLoadedFds().subList(0, 10),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml1Refresh.load();
assertFalse(fml1Refresh.useParallelListing());
assertEquals(10, fml1Refresh.getStats().loadedFiles);
@@ -221,13 +221,13 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml2 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ Collections.emptyList(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml2.load();
IcebergFileMetadataLoader fml2Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ fml2.getLoadedFds().subList(0, 10),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml2Refresh.load();
assertFalse(fml2Refresh.useParallelListing());
assertEquals(10, fml2Refresh.getStats().loadedFiles);
@@ -241,13 +241,13 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ Collections.emptyList(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml1.load();
IcebergFileMetadataLoader fml1ForceRefresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ fml1.getLoadedFds().subList(0, 10),
- /* canDataBeOutsideOfTableLocation = */ false, 10);
+ /* requiresDataFilesInTableLocation = */ true, 10);
fml1ForceRefresh.setForceRefreshBlockLocations(true);
fml1ForceRefresh.load();
assertTrue(fml1ForceRefresh.useParallelListing());
@@ -255,7 +255,7 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml1Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ fml1.getLoadedFds().subList(0, 10),
- /* canDataBeOutsideOfTableLocation = */ false, 10);
+ /* requiresDataFilesInTableLocation = */ true, 10);
fml1Refresh.setForceRefreshBlockLocations(false);
fml1Refresh.load();
assertFalse(fml1Refresh.useParallelListing());
@@ -263,13 +263,13 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml1Refresh10 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ fml1.getLoadedFds().subList(0, 10),
- /* canDataBeOutsideOfTableLocation = */ false, 10);
+ /* requiresDataFilesInTableLocation = */ true, 10);
fml1Refresh10.load();
assertFalse(fml1Refresh10.useParallelListing());
IcebergFileMetadataLoader fml1Refresh9 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_partitioned",
/* oldFds = */ fml1.getLoadedFds().subList(0, 10),
- /* canDataBeOutsideOfTableLocation = */ false, 9);
+ /* requiresDataFilesInTableLocation = */ true, 9);
fml1Refresh9.load();
assertTrue(fml1Refresh9.useParallelListing());
@@ -277,23 +277,23 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml2 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ Collections.emptyList(),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
fml2.load();
IcebergFileMetadataLoader fml2Refresh = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ fml2.getLoadedFds().subList(0, 10),
- /* canDataBeOutsideOfTableLocation = */ false);
+ /* requiresDataFilesInTableLocation = */ true);
IcebergFileMetadataLoader fml2Refresh10 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ fml2.getLoadedFds().subList(0, 10),
- /* canDataBeOutsideOfTableLocation = */ false, 10);
+ /* requiresDataFilesInTableLocation = */ true, 10);
fml2Refresh10.load();
assertFalse(fml2Refresh10.useParallelListing());
IcebergFileMetadataLoader fml2Refresh9 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_non_partitioned",
/* oldFds = */ fml2.getLoadedFds().subList(0, 10),
- /* canDataBeOutsideOfTableLocation = */ false, 9);
+ /* requiresDataFilesInTableLocation = */ true, 9);
fml2Refresh9.load();
assertTrue(fml2Refresh9.useParallelListing());
}
@@ -305,13 +305,13 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_multiple_storage_locations",
/* oldFds = */ Collections.emptyList(),
- /* canDataBeOutsideOfTableLocation = */ true);
+ /* requiresDataFilesInTableLocation = */ false);
fml1.load();
IcebergFileMetadataLoader fml1Refresh1 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_multiple_storage_locations",
/* oldFds = */ fml1.getLoadedFds().subList(0, 1),
- /* canDataBeOutsideOfTableLocation = */ true);
+ /* requiresDataFilesInTableLocation = */ false);
fml1Refresh1.load();
assertFalse(fml1Refresh1.useParallelListing());
assertEquals(5, fml1Refresh1.getStats().loadedFiles);
@@ -321,7 +321,7 @@ public class FileMetadataLoaderTest {
IcebergFileMetadataLoader fml1Refresh5 = getLoaderForIcebergTable(catalog,
"functional_parquet", "iceberg_multiple_storage_locations",
/* oldFds = */ fml1.getLoadedFds().subList(0, 5),
- /* canDataBeOutsideOfTableLocation = */ true);
+ /* requiresDataFilesInTableLocation = */ false);
fml1Refresh5.load();
assertFalse(fml1Refresh5.useParallelListing());
assertEquals(1, fml1Refresh5.getStats().loadedFiles);
@@ -331,15 +331,15 @@ public class FileMetadataLoaderTest {
private IcebergFileMetadataLoader getLoaderForIcebergTable(
CatalogServiceCatalog catalog, String dbName, String tblName,
- List<FileDescriptor> oldFds, boolean canDataBeOutsideOfTableLocation)
+ List<FileDescriptor> oldFds, boolean requiresDataFilesInTableLocation)
throws CatalogException {
return getLoaderForIcebergTable(catalog, dbName, tblName, oldFds,
- canDataBeOutsideOfTableLocation, -1);
+ requiresDataFilesInTableLocation, -1);
}
private IcebergFileMetadataLoader getLoaderForIcebergTable(
CatalogServiceCatalog catalog, String dbName, String tblName,
- List<FileDescriptor> oldFds, boolean canDataBeOutsideOfTableLocation,
+ List<FileDescriptor> oldFds, boolean requiresDataFilesInTableLocation,
int newFilesThreshold)
throws CatalogException {
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
@@ -348,9 +348,8 @@ public class FileMetadataLoaderTest {
Path location = new Path(iceT.getLocation());
GroupedContentFiles iceFiles = IcebergUtil.getIcebergFiles(iceT,
/*predicates=*/Collections.emptyList(), /*timeTravelSpec=*/null);
- return new IcebergFileMetadataLoader(location, /* recursive=*/true,
- oldFds, hostIndex, null, null, iceFiles, canDataBeOutsideOfTableLocation,
- newFilesThreshold);
+ return new IcebergFileMetadataLoader(iceT.getIcebergApiTable(),
+ oldFds, hostIndex, iceFiles, requiresDataFilesInTableLocation, newFilesThreshold);
}
private FileMetadataLoader getLoaderForAcidTable(
(impala) 03/04: IMPALA-13758: Use context manager in ImpalaTestSuite.change_database
Posted by cs...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
csringhofer pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 9cb9bae84e8888203e4bdfd3e20ee3e61c6059fe
Author: Riza Suminto <ri...@cloudera.com>
AuthorDate: Thu Feb 13 11:19:31 2025 -0800
IMPALA-13758: Use context manager in ImpalaTestSuite.change_database
ImpalaTestSuite.change_database is responsible to point impala client to
database under test. However, it left client pointing to that database
after the test without reverting them back to default database. This
patch does the reversal by changing ImpalaTestSuite.change_database to
use context manager.
This patch change the behavior of execute_query_using_client() and
execute_query_async_using_client(). They used to change database
according to the given vector parameter, but not anymore after this
patch. In practice, this behavior change does not affect many tests
because most queries going through these functions already use fully
qualified table name. Going forward, querying through function other
than run_test_case() should try to use fully qualified table name as
much as possible.
Retain behavior of ImpalaTestSuite._get_table_location() since there are
considerable number of tests relies on it (changing database when
called).
Removed unused test fixtures and fixed several flake8 issues in modified
test files.
Testing:
- Moved nested-types-subplan-single-node.test. This allows the test
framework to point to the right tpch_nested* database.
- Pass exhaustive test except IMPALA-13752 and IMPALA-13761. They will
be fixed in separate patch.
Change-Id: I75bec7403cc302728a630efe3f95e852a84594e2
Reviewed-on: http://gerrit.cloudera.org:8080/22487
Reviewed-by: Csaba Ringhofer <cs...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../nested-types-subplan-single-node.test | 3 -
tests/common/impala_test_suite.py | 90 ++++++++++++++++------
tests/common/test_vector.py | 1 +
tests/conftest.py | 3 -
tests/custom_cluster/test_admission_controller.py | 20 ++---
.../test_blacklisted_dbs_and_tables.py | 10 +--
tests/custom_cluster/test_data_cache.py | 43 ++++++-----
tests/failure/test_failpoints.py | 15 ++--
tests/query_test/test_aggregation.py | 14 ++--
tests/query_test/test_cancellation.py | 42 +++++-----
tests/query_test/test_mem_usage_scaling.py | 14 ++--
tests/query_test/test_mt_dop.py | 13 ++--
tests/query_test/test_nested_types.py | 5 +-
tests/query_test/test_runtime_filters.py | 20 +++--
tests/query_test/test_scanners.py | 29 +++----
tests/query_test/test_tablesample.py | 15 ++--
tests/util/cancel_util.py | 14 +++-
17 files changed, 207 insertions(+), 144 deletions(-)
diff --git a/testdata/workloads/functional-query/queries/QueryTest/nested-types-subplan-single-node.test b/testdata/workloads/tpch_nested/queries/QueryTest/nested-types-subplan-single-node.test
similarity index 97%
rename from testdata/workloads/functional-query/queries/QueryTest/nested-types-subplan-single-node.test
rename to testdata/workloads/tpch_nested/queries/QueryTest/nested-types-subplan-single-node.test
index ece9c7007..3a9ab06ad 100644
--- a/testdata/workloads/functional-query/queries/QueryTest/nested-types-subplan-single-node.test
+++ b/testdata/workloads/tpch_nested/queries/QueryTest/nested-types-subplan-single-node.test
@@ -1,8 +1,5 @@
====
---- QUERY
-use tpch_nested_parquet
-====
----- QUERY
# IMPALA-2289: Test proper handling of AtCapacity() inside the subplan node.
# num_nodes is set to 1 in the python test to make it very likely to hit the once buggy
# code path because a single scan node instance must process all input files.
diff --git a/tests/common/impala_test_suite.py b/tests/common/impala_test_suite.py
index 7c5495f4f..0cc5d4ea5 100644
--- a/tests/common/impala_test_suite.py
+++ b/tests/common/impala_test_suite.py
@@ -19,6 +19,7 @@
from __future__ import absolute_import, division, print_function
from builtins import range, round
+import contextlib
import glob
import grp
import hashlib
@@ -63,7 +64,7 @@ from tests.common.test_result_verifier import (
verify_raw_results,
verify_runtime_profile)
from tests.common.test_vector import (
- EXEC_OPTION, PROTOCOL, TABLE_FORMAT,
+ EXEC_OPTION, PROTOCOL, TABLE_FORMAT, VECTOR,
BEESWAX, HS2, HS2_HTTP,
ImpalaTestDimension)
from tests.performance.query import Query
@@ -752,8 +753,9 @@ class ImpalaTestSuite(BaseTestSuite):
# Change the database to reflect the file_format, compression codec etc, or the
# user specified database for all targeted impalad.
for impalad_client in target_impalad_clients:
- ImpalaTestSuite.change_database(impalad_client,
- table_format_info, use_db, pytest.config.option.scale_factor)
+ ImpalaTestSuite.__change_client_database(
+ impalad_client, table_format=table_format_info, db_name=use_db,
+ scale_factor=pytest.config.option.scale_factor)
impalad_client.set_configuration(exec_options)
def __exec_in_impala(query, user=None):
@@ -984,6 +986,10 @@ class ImpalaTestSuite(BaseTestSuite):
os.makedirs(output_dir)
write_test_file(output_file, sections, encoding=encoding)
+ # Revert target_impalad_clients back to default database.
+ for impalad_client in target_impalad_clients:
+ ImpalaTestSuite.__change_client_database(impalad_client, db_name='default')
+
def get_query_lineage(self, query_id, lineage_dir):
"""Walks through the lineage files in lineage_dir to look for a given query_id.
This is an expensive operation is lineage_dir is large, so use carefully."""
@@ -1010,18 +1016,50 @@ class ImpalaTestSuite(BaseTestSuite):
def get_db_name_from_format(table_format, scale_factor=''):
return QueryTestSectionReader.get_db_name(table_format, scale_factor)
- @classmethod
- def change_database(cls, impala_client, table_format=None,
- db_name=None, scale_factor=None):
+ @staticmethod
+ def __change_client_database(impala_client, table_format=None, db_name=None,
+ scale_factor=None):
+ """Change 'impala_client' to point to either 'db_name' or workload-specific
+ database described by 'table_format' and 'scale_vector'.
+ Restore client configuration back to its default. This is intended for internal use
+ within ImpalaTestSuite. For changing database in ImpalaTestSuite subclasses, please
+ refer to ImpalaTestSuite.change_database(), which provide a more consistent way to
+ temporarily change database and reverting back to default database.
+ """
if db_name is None:
assert table_format is not None
- db_name = QueryTestSectionReader.get_db_name(table_format,
- scale_factor if scale_factor else '')
- query = 'use %s' % db_name
+ db_name = ImpalaTestSuite.get_db_name_from_format(
+ table_format, scale_factor if scale_factor else '')
# Clear the exec_options before executing a USE statement.
# The USE statement should not fail for negative exec_option tests.
impala_client.clear_configuration()
- impala_client.execute(query)
+ impala_client.execute('use ' + db_name)
+
+ @classmethod
+ @contextlib.contextmanager
+ def change_database(cls, impala_client, table_format=None, db_name=None,
+ scale_factor=None):
+ """Change impala_client to point to approriate database under test and revert
+ it to default database once it exit the with scope.
+ Restore client configuration back to its default when getting in and getting out of
+ 'with' scope. Test method should try to use fully qualified table name in the test
+ query as much as possible and only use this context manager function when it is
+ not practical to do so.
+ Sample usage:
+
+ with ImpalaTestSuite.change_database(client, db_name='functional_parquet'):
+ # client clear configuration and pointing to 'functional_parquet' database here.
+ client.execute('show tables')
+ # client clear configuration and pointing to 'default' database after it exit the
+ # 'with' scope.
+ client.execute('show tables')
+ """
+ try:
+ cls.__change_client_database(impala_client, table_format=table_format,
+ db_name=db_name, scale_factor=scale_factor)
+ yield
+ finally:
+ cls.__change_client_database(impala_client, db_name='default')
def execute_wrapper(function):
"""
@@ -1031,6 +1069,8 @@ class ImpalaTestSuite(BaseTestSuite):
remaining the same. A use database is issued before query execution. As such,
database names need to be build pre execution, this method wraps around the different
execute methods and provides a common interface to issue the proper use command.
+ IMPALA-13766: Remove this function decorator and let test method to change database
+ by themself.
"""
@wraps(function)
def wrapper(*args, **kwargs):
@@ -1038,13 +1078,16 @@ class ImpalaTestSuite(BaseTestSuite):
if kwargs.get(TABLE_FORMAT):
table_format = kwargs.get(TABLE_FORMAT)
del kwargs[TABLE_FORMAT]
- if kwargs.get('vector'):
- table_format = kwargs.get('vector').get_table_format()
- del kwargs['vector']
+ if kwargs.get(VECTOR):
+ ImpalaTestSuite.validate_exec_option_dimension(kwargs.get(VECTOR))
+ table_format = kwargs.get(VECTOR).get_table_format()
+ del kwargs[VECTOR]
# self is the implicit first argument
if table_format is not None:
- args[0].change_database(args[0].client, table_format)
- return function(*args, **kwargs)
+ with ImpalaTestSuite.change_database(args[0].client, table_format):
+ return function(*args, **kwargs)
+ else:
+ return function(*args, **kwargs)
return wrapper
@classmethod
@@ -1101,16 +1144,16 @@ class ImpalaTestSuite(BaseTestSuite):
def execute_query_using_client(self, client, query, vector):
self.validate_exec_option_dimension(vector)
- self.change_database(client, vector.get_table_format())
query_options = vector.get_value(EXEC_OPTION)
- if query_options is not None: client.set_configuration(query_options)
+ if query_options is not None:
+ client.set_configuration(query_options)
return client.execute(query)
def execute_query_async_using_client(self, client, query, vector):
self.validate_exec_option_dimension(vector)
- self.change_database(client, vector.get_table_format())
query_options = vector.get_value(EXEC_OPTION)
- if query_options is not None: client.set_configuration(query_options)
+ if query_options is not None:
+ client.set_configuration(query_options)
return client.execute_async(query)
def close_query_using_client(self, client, query):
@@ -1248,7 +1291,10 @@ class ImpalaTestSuite(BaseTestSuite):
assert abs(a - b) / float(max(abs(a), abs(b))) <= diff_perc
def _get_table_location(self, table_name, vector):
- """ Returns the HDFS location of the table """
+ """ Returns the HDFS location of the table.
+ This method changes self.client to point to the dabatase described by 'vector'."""
+ db_name = self.get_db_name_from_format(vector.get_table_format())
+ self.__change_client_database(self.client, db_name=db_name)
result = self.execute_query_using_client(self.client,
"describe formatted %s" % table_name, vector)
for row in result.data:
@@ -1626,7 +1672,8 @@ class ImpalaTestSuite(BaseTestSuite):
str(e))
time.sleep(1)
- def validate_exec_option_dimension(self, vector):
+ @staticmethod
+ def validate_exec_option_dimension(vector):
"""Validate that test dimension with name matching query option name is
also registered in 'exec_option' dimension."""
option_dim_names = []
@@ -1641,7 +1688,6 @@ class ImpalaTestSuite(BaseTestSuite):
return
for name in option_dim_names:
- # TODO: enforce these warnings by changing them into pytest.fail()
if name not in exec_option:
pytest.fail("Exec option {} declared as independent dimension but not inserted "
"into {} dimension. Consider using helper function "
diff --git a/tests/common/test_vector.py b/tests/common/test_vector.py
index 900f3a072..12f87bcb6 100644
--- a/tests/common/test_vector.py
+++ b/tests/common/test_vector.py
@@ -63,6 +63,7 @@ import logging
LOG = logging.getLogger(__name__)
+VECTOR = 'vector'
# Literal constants to refer to some standard dimension names.
EXEC_OPTION = 'exec_option'
PROTOCOL = 'protocol'
diff --git a/tests/conftest.py b/tests/conftest.py
index aa179c365..626d68b2d 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -402,9 +402,6 @@ def unique_database(request, testid_checksum):
request.instance.filesystem_client.delete_file_dir(db_location, recursive=True)
def cleanup():
- # Make sure we don't try to drop the current session database
- # TODO: clean this up via IMPALA-13758.
- request.instance.execute_query_expect_success(request.instance.client, "use default")
with request.cls.create_impala_client(protocol=HS2) as client:
client.set_configuration({'sync_ddl': sync_ddl})
for db_name in db_names:
diff --git a/tests/custom_cluster/test_admission_controller.py b/tests/custom_cluster/test_admission_controller.py
index 7c6c9aed4..5223a421d 100644
--- a/tests/custom_cluster/test_admission_controller.py
+++ b/tests/custom_cluster/test_admission_controller.py
@@ -41,7 +41,6 @@ from tests.common.custom_cluster_test_suite import (
START_ARGS,
CustomClusterTestSuite)
from tests.common.environ import build_flavor_timeout, ImpalaTestClusterProperties
-from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.resource_pool_config import ResourcePoolConfig
from tests.common.skip import SkipIfFS, SkipIfEC, SkipIfNotHdfsMinicluster
from tests.common.test_dimensions import (
@@ -74,8 +73,8 @@ SLOW_QUERY = "select count(*) from functional.alltypes where int_col = sleep(200
# The unpartitioned fragments are both interior fragments that consume input
# from a scan fragment and non-interior fragments with a constant UNION.
QUERY_WITH_UNPARTITIONED_FRAGMENTS = """
- select *, (select count(distinct int_col) from alltypestiny) subquery1,
- (select count(distinct int_col) from alltypes) subquery2,
+ select *, (select count(distinct int_col) from functional.alltypestiny) subquery1,
+ (select count(distinct int_col) from functional.alltypes) subquery2,
(select 1234) subquery3
from (""" + QUERY + """) v"""
@@ -579,7 +578,6 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
def test_sanity_checks_dedicated_coordinator(self, vector, unique_database):
"""Sanity tests for verifying targeted dedicated coordinator memory estimations and
behavior."""
- ImpalaTestSuite.change_database(self.client, vector.get_value('table_format'))
self.client.set_configuration_option('request_pool', "root.regularPool")
exec_options = vector.get_value('exec_option')
# Make sure query option MAX_MEM_ESTIMATE_FOR_ADMISSION is enforced on the dedicated
@@ -625,8 +623,8 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
the actual vs expected values for mem admitted and mem limit for both coord and
executor. Also verifies that those memory values are different if
'using_dedicated_coord_estimates' is true."""
- ImpalaTestSuite.change_database(self.client, vector.get_value('table_format'))
- self.client.set_configuration_option('request_pool', "root.regularPool")
+ vector.set_exec_option('request_pool', 'root.regularPool')
+ self.client.set_configuration(vector.get_exec_option_dict())
# Use a test query that has unpartitioned non-coordinator fragments to make
# sure those are handled correctly (IMPALA-10036).
for query in [QUERY, QUERY_WITH_UNPARTITIONED_FRAGMENTS]:
@@ -707,10 +705,9 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
@SkipIfNotHdfsMinicluster.tuned_for_minicluster
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(num_exclusive_coordinators=1, cluster_size=2)
- def test_mem_limit_executors(self, vector):
+ def test_mem_limit_executors(self):
"""Verify that the query option mem_limit_executors is only enforced on the
executors."""
- ImpalaTestSuite.change_database(self.client, vector.get_value('table_format'))
expected_exec_mem_limit = "999999999"
self.client.set_configuration({"MEM_LIMIT_EXECUTORS": expected_exec_mem_limit})
handle = self.client.execute_async(QUERY.format(1))
@@ -726,10 +723,9 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
@CustomClusterTestSuite.with_args(num_exclusive_coordinators=1, cluster_size=2,
impalad_args=impalad_admission_ctrl_flags(max_requests=1, max_queued=1,
pool_max_mem=2 * 1024 * 1024 * 1024, proc_mem_limit=3 * 1024 * 1024 * 1024))
- def test_mem_limit_coordinators(self, vector):
+ def test_mem_limit_coordinators(self):
"""Verify that the query option mem_limit_coordinators is only enforced on the
coordinators."""
- ImpalaTestSuite.change_database(self.client, vector.get_value('table_format'))
expected_exec_mem_limit = "999999999"
expected_coord_mem_limit = "111111111"
self.client.set_configuration({"MEM_LIMIT_EXECUTORS": expected_exec_mem_limit,
@@ -747,10 +743,9 @@ class TestAdmissionController(TestAdmissionControllerBase, HS2TestSuite):
@CustomClusterTestSuite.with_args(num_exclusive_coordinators=1, cluster_size=2,
impalad_args=impalad_admission_ctrl_flags(max_requests=1, max_queued=1,
pool_max_mem=2 * 1024 * 1024 * 1024, proc_mem_limit=3 * 1024 * 1024 * 1024))
- def test_mem_limits(self, vector):
+ def test_mem_limits(self):
"""Verify that the query option mem_limit_coordinators and mem_limit_executors are
ignored when mem_limit is set."""
- ImpalaTestSuite.change_database(self.client, vector.get_value('table_format'))
exec_mem_limit = "999999999"
coord_mem_limit = "111111111"
mem_limit = "888888888"
@@ -2475,7 +2470,6 @@ class TestAdmissionControllerStress(TestAdmissionControllerBase):
query = QUERY.format(self.query_num)
self.query_state = 'SUBMITTING'
client = self.impalad.service.create_beeswax_client()
- ImpalaTestSuite.change_database(client, self.vector.get_value('table_format'))
client.set_configuration(exec_options)
LOG.info("Submitting query %s with ending behavior %s",
diff --git a/tests/custom_cluster/test_blacklisted_dbs_and_tables.py b/tests/custom_cluster/test_blacklisted_dbs_and_tables.py
index 8ce0e5969..f5ea32da6 100644
--- a/tests/custom_cluster/test_blacklisted_dbs_and_tables.py
+++ b/tests/custom_cluster/test_blacklisted_dbs_and_tables.py
@@ -95,7 +95,7 @@ class TestBlacklistedDbsAndTables(CustomClusterTestSuite):
"--blacklisted_tables=functional.alltypes,functional_parquet.alltypes",
catalogd_args="--blacklisted_dbs=functional_rc,functional_seq "
"--blacklisted_tables=functional.alltypes,functional_parquet.alltypes")
- def test_blacklisted_dbs_and_tables(self, vector):
+ def test_blacklisted_dbs_and_tables(self):
self.__check_db_not_visible("functional_rc")
self.__check_db_not_visible("functional_seq")
self.__check_table_not_visible("functional", "alltypes")
@@ -128,7 +128,7 @@ class TestBlacklistedDbsAndTables(CustomClusterTestSuite):
@CustomClusterTestSuite.with_args(
impalad_args="--blacklisted_tables=alltypes_def,functional.alltypes",
catalogd_args="--blacklisted_tables=alltypes_def,functional.alltypes")
- def test_resolving_default_database(self, vector):
+ def test_resolving_default_database(self):
# Check that "alltypes_def" is resolved as "default.alltypes_def"
table = self.hive_client.get_table("functional", "alltypes")
table.dbName = "default"
@@ -138,8 +138,8 @@ class TestBlacklistedDbsAndTables(CustomClusterTestSuite):
self.hive_client.drop_table("default", "alltypes_def", True)
# Check non fully qualified table names are recognized correctly
- self.change_database(self.client, db_name="functional")
- self.__check_create_drop_table(use_fully_qualified_table_name=False)
+ with self.change_database(self.client, db_name="functional"):
+ self.__check_create_drop_table(use_fully_qualified_table_name=False)
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
@@ -147,7 +147,7 @@ class TestBlacklistedDbsAndTables(CustomClusterTestSuite):
"--blacklisted_tables=functional.alltypes",
impalad_args="--blacklisted_dbs=functional_seq "
"--blacklisted_tables=functional.alltypestiny")
- def test_inconsistent_blacklist(self, vector):
+ def test_inconsistent_blacklist(self):
"""Test the error handling when blacklists are accidentally set differently between
coordinators and the catalogd"""
self.__expect_error_in_query(
diff --git a/tests/custom_cluster/test_data_cache.py b/tests/custom_cluster/test_data_cache.py
index 734d4c049..a020bfc68 100644
--- a/tests/custom_cluster/test_data_cache.py
+++ b/tests/custom_cluster/test_data_cache.py
@@ -82,7 +82,8 @@ class TestDataCache(CustomClusterTestSuite):
# Expect all cache hits results in no opened files.
opened_file_handles_metric = 'impala-server.io.mgr.cached-file-handles-miss-count'
baseline = self.get_metric(opened_file_handles_metric)
- self.execute_query("select count(distinct l_orderkey) from test_parquet")
+ self.execute_query("select count(distinct l_orderkey) from {0}.test_parquet".format(
+ unique_database))
assert self.get_metric(opened_file_handles_metric) == baseline
@pytest.mark.execute_serially
@@ -106,7 +107,7 @@ class TestDataCache(CustomClusterTestSuite):
def test_data_cache_deterministic_no_file_handle_cache(self, vector, unique_database):
self.__test_data_cache_deterministic(vector, unique_database)
- def __test_data_cache(self, vector):
+ def __test_data_cache(self):
""" This test scans the same table twice and verifies the cache hit count metrics
are correct. The exact number of bytes hit is non-deterministic between runs due
to different mtime of files and multiple shards in the cache.
@@ -132,16 +133,16 @@ class TestDataCache(CustomClusterTestSuite):
impalad_args=get_impalad_args("LRU", high_write_concurrency=False,
force_single_shard=False),
start_args=CACHE_START_ARGS, cluster_size=1)
- def test_data_cache_lru(self, vector):
- self.__test_data_cache(vector)
+ def test_data_cache_lru(self):
+ self.__test_data_cache()
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args=get_impalad_args("LIRS", high_write_concurrency=False,
force_single_shard=False),
start_args=CACHE_START_ARGS, cluster_size=1)
- def test_data_cache_lirs(self, vector):
- self.__test_data_cache(vector)
+ def test_data_cache_lirs(self):
+ self.__test_data_cache()
def __test_data_cache_disablement(self, vector):
# Verifies that the cache metrics are all zero.
@@ -185,7 +186,7 @@ class TestDataCache(CustomClusterTestSuite):
impalad_args=get_impalad_args("LIRS", high_write_concurrency=False),
start_args="--data_cache_dir=/tmp --data_cache_size=9MB",
cluster_size=1)
- def test_data_cache_lirs_instant_evictions(self, vector):
+ def test_data_cache_lirs_instant_evictions(self):
# The setup for this test is intricate. For Allocate() to succeed, the request
# needs to be smaller than the protected size (95% of the cache). For Insert() to
# fail, the request needs to be larger than the unprotected size (5% of the cache).
@@ -215,7 +216,7 @@ class TestDataCache(CustomClusterTestSuite):
assert self.get_data_cache_metric('num-writes') >= 0
assert self.get_data_cache_metric('total-bytes') >= 0
- def __test_data_cache_keep_across_restarts(self, vector, test_reduce_size=False):
+ def __test_data_cache_keep_across_restarts(self, test_reduce_size=False):
QUERY = "select * from tpch_parquet.lineitem"
# Execute a query, record the total bytes and the number of entries of cache before
# cache dump.
@@ -260,31 +261,31 @@ class TestDataCache(CustomClusterTestSuite):
@CustomClusterTestSuite.with_args(
impalad_args=get_impalad_args("LRU", keep_across_restarts=True),
start_args=CACHE_START_ARGS, cluster_size=1)
- def test_data_cache_keep_across_restarts_lru(self, vector):
- self.__test_data_cache_keep_across_restarts(vector)
+ def test_data_cache_keep_across_restarts_lru(self):
+ self.__test_data_cache_keep_across_restarts()
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args=get_impalad_args("LIRS", keep_across_restarts=True),
start_args=CACHE_START_ARGS, cluster_size=1)
- def test_data_cache_keep_across_restarts_lirs(self, vector):
- self.__test_data_cache_keep_across_restarts(vector)
+ def test_data_cache_keep_across_restarts_lirs(self):
+ self.__test_data_cache_keep_across_restarts()
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args=get_impalad_args("LRU", keep_across_restarts=True),
start_args=CACHE_START_ARGS, cluster_size=1)
- def test_data_cache_reduce_size_restarts_lru(self, vector):
- self.__test_data_cache_keep_across_restarts(vector, test_reduce_size=True)
+ def test_data_cache_reduce_size_restarts_lru(self):
+ self.__test_data_cache_keep_across_restarts(test_reduce_size=True)
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args=get_impalad_args("LIRS", keep_across_restarts=True),
start_args=CACHE_START_ARGS, cluster_size=1)
- def test_data_cache_reduce_size_restarts_lirs(self, vector):
- self.__test_data_cache_keep_across_restarts(vector, test_reduce_size=True)
+ def test_data_cache_reduce_size_restarts_lirs(self):
+ self.__test_data_cache_keep_across_restarts(test_reduce_size=True)
- def __test_data_cache_readonly(self, vector):
+ def __test_data_cache_readonly(self):
QUERY = "select * from tpch_parquet.lineitem"
# Execute the query asynchronously, wait a short while, and do gracefully shutdown
# immediately to test the race between cache writes and setting cache read-only.
@@ -310,12 +311,12 @@ class TestDataCache(CustomClusterTestSuite):
@CustomClusterTestSuite.with_args(
impalad_args=get_impalad_args("LRU", keep_across_restarts=True),
start_args=CACHE_START_ARGS, cluster_size=1, disable_log_buffering=True)
- def test_data_cache_readonly_lru(self, vector):
- self.__test_data_cache_readonly(vector)
+ def test_data_cache_readonly_lru(self):
+ self.__test_data_cache_readonly()
@pytest.mark.execute_serially
@CustomClusterTestSuite.with_args(
impalad_args=get_impalad_args("LIRS", keep_across_restarts=True),
start_args=CACHE_START_ARGS, cluster_size=1, disable_log_buffering=True)
- def test_data_cache_readonly_lirs(self, vector):
- self.__test_data_cache_readonly(vector)
+ def test_data_cache_readonly_lirs(self):
+ self.__test_data_cache_readonly()
diff --git a/tests/failure/test_failpoints.py b/tests/failure/test_failpoints.py
index 38f0919e5..d5bb1bf9e 100644
--- a/tests/failure/test_failpoints.py
+++ b/tests/failure/test_failpoints.py
@@ -97,6 +97,10 @@ class TestFailpoints(ImpalaTestSuite):
# killer on machines with 30GB RAM. This makes the test run in 4 minutes instead of 1-2.
@pytest.mark.execute_serially
def test_failpoints(self, vector):
+ with self.change_database(self.client, vector.get_table_format()):
+ self.__run_failpoints(vector)
+
+ def __run_failpoints(self, vector):
query = vector.get_value('query')
action = vector.get_value('action')
location = vector.get_value('location')
@@ -131,9 +135,8 @@ class TestFailpoints(ImpalaTestSuite):
def __parse_plan_nodes_from_explain(self, query, vector):
"""Parses the EXPLAIN <query> output and returns a list of node ids.
Expects format of <ID>:<NAME>"""
- explain_result =\
- self.execute_query("explain " + query, vector.get_value('exec_option'),
- table_format=vector.get_value('table_format'))
+ explain_result = self.execute_query("explain " + query,
+ vector.get_value('exec_option'))
node_ids = []
for row in explain_result.data:
match = re.search(r'\s*(?P<node_id>\d+)\:(?P<node_type>\S+\s*\S+)', row)
@@ -190,8 +193,7 @@ class TestFailpoints(ImpalaTestSuite):
def __execute_fail_action(self, query, vector):
try:
- self.execute_query(query, vector.get_value('exec_option'),
- table_format=vector.get_value('table_format'))
+ self.execute_query(query, vector.get_value('exec_option'))
assert 'Expected Failure'
except ImpalaBeeswaxException as e:
LOG.debug(e)
@@ -200,8 +202,7 @@ class TestFailpoints(ImpalaTestSuite):
def __execute_cancel_action(self, query, vector):
LOG.info('Starting async query execution')
- handle = self.execute_query_async(query, vector.get_value('exec_option'),
- table_format=vector.get_value('table_format'))
+ handle = self.execute_query_async(query, vector.get_value('exec_option'))
LOG.info('Sleeping')
sleep(3)
cancel_result = self.client.cancel(handle)
diff --git a/tests/query_test/test_aggregation.py b/tests/query_test/test_aggregation.py
index 2e98f4737..bb5711a53 100644
--- a/tests/query_test/test_aggregation.py
+++ b/tests/query_test/test_aggregation.py
@@ -147,11 +147,11 @@ class TestAggregation(ImpalaTestSuite):
exec_option = vector.get_value('exec_option')
disable_codegen = exec_option['disable_codegen']
data_type, agg_func = (vector.get_value('data_type'), vector.get_value('agg_func'))
+ db_name = self.get_db_name_from_format(vector.get_table_format())
- query = 'select %s(%s_col) from alltypesagg where day is not null' % (agg_func,
- data_type)
- result = self.execute_query(query, exec_option,
- table_format=vector.get_value('table_format'))
+ query = 'select {0}({1}_col) from {2}.alltypesagg where day is not null'.format(
+ agg_func, data_type, db_name)
+ result = self.execute_query(query, exec_option)
assert len(result.data) == 1
self.verify_agg_result(agg_func, data_type, False, result.data[0])
@@ -160,9 +160,9 @@ class TestAggregation(ImpalaTestSuite):
# It is deliberately disabled for the merge aggregation.
assert_codegen_enabled(result.runtime_profile, [1])
- query = 'select %s(DISTINCT(%s_col)) from alltypesagg where day is not null' % (
- agg_func, data_type)
- result = self.execute_query(query, vector.get_value('exec_option'))
+ query = ('select {0}(DISTINCT({1}_col)) from {2}.alltypesagg '
+ 'where day is not null').format(agg_func, data_type, db_name)
+ result = self.execute_query(query, exec_option)
assert len(result.data) == 1
self.verify_agg_result(agg_func, data_type, True, result.data[0])
diff --git a/tests/query_test/test_cancellation.py b/tests/query_test/test_cancellation.py
index 342a39f03..49efa053f 100644
--- a/tests/query_test/test_cancellation.py
+++ b/tests/query_test/test_cancellation.py
@@ -21,10 +21,8 @@
from __future__ import absolute_import, division, print_function
from builtins import range
import pytest
-import threading
from time import sleep
from RuntimeProfile.ttypes import TRuntimeProfileFormat
-from tests.beeswax.impala_beeswax import ImpalaBeeswaxException
from tests.common.test_dimensions import add_mandatory_exec_option
from tests.common.test_vector import ImpalaTestDimension
from tests.common.impala_test_suite import ImpalaTestSuite
@@ -57,9 +55,6 @@ QUERY_TYPE = ["SELECT", "CTAS"]
# are prone to occur more often when the time between RPCs is small.
CANCEL_DELAY_IN_SECONDS = [0, 0.01, 0.1, 1, 4]
-# Number of times to execute/cancel each query under test
-NUM_CANCELATION_ITERATIONS = 1
-
# Test cancellation on both running and hung queries. Node ID 0 is the scan node
WAIT_ACTIONS = [None, '0:GETNEXT:WAIT']
@@ -77,7 +72,7 @@ JOIN_BEFORE_CLOSE = [False, True]
# Extra dimensions to test order by without limit
SORT_QUERY = 'select * from lineitem order by l_orderkey'
SORT_CANCEL_DELAY = list(range(6, 10))
-SORT_BUFFER_POOL_LIMIT = ['0', '300m'] # Test spilling and non-spilling sorts.
+SORT_BUFFER_POOL_LIMIT = ['0', '300m'] # Test spilling and non-spilling sorts.
# Test with and without multithreading
MT_DOP_VALUES = [0, 4]
@@ -87,6 +82,7 @@ MT_DOP_VALUES = [0, 4]
# True: Execute a KILL QUERY statement.
USE_KILL_QUERY_STATEMENT = [False, True]
+
class TestCancellation(ImpalaTestSuite):
@classmethod
def get_workload(self):
@@ -115,17 +111,21 @@ class TestCancellation(ImpalaTestSuite):
ImpalaTestDimension('mt_dop', *MT_DOP_VALUES))
cls.ImpalaTestMatrix.add_dimension(
ImpalaTestDimension('use_kill_query_statement', *USE_KILL_QUERY_STATEMENT))
+ # Number of times to execute/cancel each query under test
+ num_iterations = 1 if cls.exploration_strategy() == 'core' else 3
+ cls.ImpalaTestMatrix.add_dimension(
+ ImpalaTestDimension('num_cancellation_iterations', num_iterations))
cls.ImpalaTestMatrix.add_constraint(
- lambda v: v.get_value('query_type') != 'CTAS' or (\
+ lambda v: v.get_value('query_type') != 'CTAS' or (
v.get_value('table_format').file_format in ['text', 'parquet', 'kudu', 'json']
and v.get_value('table_format').compression_codec == 'none'))
cls.ImpalaTestMatrix.add_constraint(
lambda v: v.get_value('exec_option')['batch_size'] == 0)
# Ignore 'compute stats' queries for the CTAS query type.
cls.ImpalaTestMatrix.add_constraint(
- lambda v: not (v.get_value('query_type') == 'CTAS' and
- v.get_value('query').startswith('compute stats')))
+ lambda v: not (v.get_value('query_type') == 'CTAS'
+ and v.get_value('query').startswith('compute stats')))
# 'use_kill_query_statement' and 'join_before_close' cannot be both True, since
# the KILL QUERY statement will also close the query.
cls.ImpalaTestMatrix.add_constraint(
@@ -134,16 +134,14 @@ class TestCancellation(ImpalaTestSuite):
# Ignore CTAS on Kudu if there is no PRIMARY KEY specified.
cls.ImpalaTestMatrix.add_constraint(
- lambda v: not (v.get_value('query_type') == 'CTAS' and
- v.get_value('table_format').file_format == 'kudu' and
- QUERIES[v.get_value('query')] is None))
+ lambda v: not (v.get_value('query_type') == 'CTAS'
+ and v.get_value('table_format').file_format == 'kudu'
+ and QUERIES[v.get_value('query')] is None))
# tpch tables are not generated for hbase as the data loading takes a very long time.
# TODO: Add cancellation tests for hbase.
- cls.ImpalaTestMatrix.add_constraint(lambda v:\
+ cls.ImpalaTestMatrix.add_constraint(lambda v:
v.get_value('table_format').file_format != 'hbase')
- if cls.exploration_strategy() != 'core':
- NUM_CANCELATION_ITERATIONS = 3
def cleanup_test_table(self, table_format):
self.execute_query("drop table if exists ctas_cancel", table_format=table_format)
@@ -176,7 +174,7 @@ class TestCancellation(ImpalaTestSuite):
vector.get_value('exec_option')['mt_dop'] = vector.get_value('mt_dop')
# Execute the query multiple times, cancelling it each time.
- for i in range(NUM_CANCELATION_ITERATIONS):
+ for i in range(vector.get_value('num_cancellation_iterations')):
cancel_query_and_validate_state(self.client, query,
vector.get_value('exec_option'), vector.get_value('table_format'),
vector.get_value('cancel_delay'), vector.get_value('join_before_close'),
@@ -188,7 +186,8 @@ class TestCancellation(ImpalaTestSuite):
# Executing the same query without canceling should work fine. Only do this if the
# query has a limit or aggregation
if not debug_action and ('count' in query or 'limit' in query):
- self.execute_query(query, vector.get_value('exec_option'))
+ with self.change_database(self.client, vector.get_value('table_format')):
+ self.execute_query(query, vector.get_value('exec_option'))
@pytest.mark.execute_serially
def test_misformatted_profile_text(self):
@@ -226,6 +225,7 @@ class TestCancellation(ImpalaTestSuite):
# fail. Introducing a small delay allows everything to quiesce.
# TODO: Figure out a better way to address this
sleep(1)
+ super(TestCancellation, self).teardown_method(method)
class TestCancellationParallel(TestCancellation):
@@ -237,6 +237,7 @@ class TestCancellationParallel(TestCancellation):
def test_cancel_select(self, vector):
self.execute_cancel_test(vector)
+
class TestCancellationSerial(TestCancellation):
@classmethod
def add_test_dimensions(cls):
@@ -268,6 +269,7 @@ class TestCancellationSerial(TestCancellation):
metric_verifier = MetricVerifier(self.impalad_test_service)
metric_verifier.verify_no_open_files(timeout=10)
+
class TestCancellationFullSort(TestCancellation):
@classmethod
def add_test_dimensions(cls):
@@ -283,9 +285,9 @@ class TestCancellationFullSort(TestCancellation):
ImpalaTestDimension('buffer_pool_limit', *SORT_BUFFER_POOL_LIMIT))
cls.ImpalaTestMatrix.add_constraint(
lambda v: v.get_value('fail_rpc_action') == FAIL_RPC_ACTIONS[0])
- cls.ImpalaTestMatrix.add_constraint(lambda v:\
- v.get_value('table_format').file_format =='parquet' and\
- v.get_value('table_format').compression_codec == 'none')
+ cls.ImpalaTestMatrix.add_constraint(lambda v:
+ v.get_value('table_format').file_format == 'parquet'
+ and v.get_value('table_format').compression_codec == 'none')
def test_cancel_sort(self, vector):
self.execute_cancel_test(vector)
diff --git a/tests/query_test/test_mem_usage_scaling.py b/tests/query_test/test_mem_usage_scaling.py
index ab4e1af32..07c35a588 100644
--- a/tests/query_test/test_mem_usage_scaling.py
+++ b/tests/query_test/test_mem_usage_scaling.py
@@ -327,7 +327,7 @@ class TestScanMemLimit(ImpalaTestSuite):
cls.ImpalaTestMatrix.add_dimension(create_single_exec_option_dimension())
cls.ImpalaTestMatrix.add_dimension(create_avro_snappy_dimension(cls.get_workload()))
- def test_wide_avro_mem_usage(self, vector, unique_database):
+ def test_wide_avro_mem_usage(self, unique_database):
"""Create a wide avro table with large strings and test scans that can cause OOM."""
if self.exploration_strategy() != 'exhaustive':
pytest.skip("only run resource-intensive query on exhaustive")
@@ -409,12 +409,12 @@ class TestHashJoinMemLimit(ImpalaTestSuite):
"""Selective scan with hash join and aggregate above it. Regression test for
IMPALA-9712 - before the fix this ran out of memory."""
OPTS = {'mem_limit': "80MB", 'mt_dop': 1}
- self.change_database(self.client, vector.get_value('table_format'))
- result = self.execute_query_expect_success(self.client,
- """select sum(l_extendedprice * (1 - l_discount)) as revenue
- from lineitem join part on p_partkey = l_partkey
- where l_comment like 'ab%'""", query_options=OPTS)
- assert result.data[0] == '440443181.0505'
+ with self.change_database(self.client, vector.get_value('table_format')):
+ result = self.execute_query_expect_success(self.client,
+ """select sum(l_extendedprice * (1 - l_discount)) as revenue
+ from lineitem join part on p_partkey = l_partkey
+ where l_comment like 'ab%'""", query_options=OPTS)
+ assert result.data[0] == '440443181.0505'
@SkipIfNotHdfsMinicluster.tuned_for_minicluster
diff --git a/tests/query_test/test_mt_dop.py b/tests/query_test/test_mt_dop.py
index 535adf5c9..dc8ff9347 100644
--- a/tests/query_test/test_mt_dop.py
+++ b/tests/query_test/test_mt_dop.py
@@ -85,11 +85,12 @@ class TestMtDop(ImpalaTestSuite):
expected_results = "Updated 24 partition(s) and 11 column(s)."
else:
# Create a second table in the same format pointing to the same data files.
- # This function switches to the format-specific DB in 'vector'.
- table_loc = self._get_table_location("alltypes", vector)
- self.execute_query_using_client(self.client,
- "create external table %s like alltypes location '%s'"
- % (fq_table_name, table_loc), vector)
+ # This switches the client momentarily to the format-specific DB in 'vector'.
+ with self.change_database(self.client, vector.get_table_format()):
+ table_loc = self._get_table_location("alltypes", vector)
+ self.execute_query_using_client(self.client,
+ "create external table %s like alltypes location '%s'"
+ % (fq_table_name, table_loc), vector)
# Recover partitions for HDFS tables.
self.execute_query("alter table %s recover partitions" % fq_table_name)
expected_results = "Updated 24 partition(s) and 11 column(s)."
@@ -152,7 +153,7 @@ class TestMtDopNonZeroParquet(ImpalaTestSuite):
test_file_vars={'$ORIGINAL_DB': ImpalaTestSuite
.get_db_name_from_format(vector.get_value('table_format'))})
- def test_mt_dop_only_joins(self, vector, unique_database):
+ def test_mt_dop_only_joins(self, vector):
"""MT_DOP specific tests for joins."""
new_vector = deepcopy(vector)
# Allow test to override num_nodes.
diff --git a/tests/query_test/test_nested_types.py b/tests/query_test/test_nested_types.py
index 0d4f5ba19..6f541c780 100644
--- a/tests/query_test/test_nested_types.py
+++ b/tests/query_test/test_nested_types.py
@@ -91,9 +91,10 @@ class TestNestedTypes(ImpalaTestSuite):
class TestNestedTypesSingleNode(ImpalaTestSuite):
"""Functional tests for nested types, run for all file formats that support nested
types. All tests here runs with single node only."""
+
@classmethod
- def get_workload(self):
- return 'functional-query'
+ def get_workload(cls):
+ return 'tpch_nested'
@classmethod
def add_test_dimensions(cls):
diff --git a/tests/query_test/test_runtime_filters.py b/tests/query_test/test_runtime_filters.py
index 83a0031b7..6e5b4d394 100644
--- a/tests/query_test/test_runtime_filters.py
+++ b/tests/query_test/test_runtime_filters.py
@@ -116,8 +116,10 @@ class TestRuntimeFilters(ImpalaTestSuite):
get woken up and exit promptly when the query is cancelled."""
# Make sure the cluster is quiesced before we start this test
self._verify_no_fragments_running()
+ with self.change_database(self.client, vector.get_value('table_format')):
+ self.__run_wait_time_cancellation(vector)
- self.change_database(self.client, vector.get_value('table_format'))
+ def __run_wait_time_cancellation(self, vector):
# Set up a query where a scan (plan node 0, scanning alltypes) will wait
# indefinitely for a filter to arrive. The filter arrival is delayed
# by adding a wait to the scan of alltypestime (plan node 0).
@@ -150,8 +152,11 @@ class TestRuntimeFilters(ImpalaTestSuite):
def test_file_filtering(self, vector):
if 'kudu' in str(vector.get_value('table_format')):
- return
- self.change_database(self.client, vector.get_value('table_format'))
+ pytest.skip('Skip test against kudu')
+ with self.change_database(self.client, vector.get_value('table_format')):
+ self.__run_file_filtering(vector)
+
+ def __run_file_filtering(self, vector):
self.execute_query("SET RUNTIME_FILTER_MODE=GLOBAL")
self.execute_query("SET RUNTIME_FILTER_WAIT_TIME_MS=10000")
result = self.execute_query("""select STRAIGHT_JOIN * from alltypes inner join
@@ -165,8 +170,11 @@ class TestRuntimeFilters(ImpalaTestSuite):
"""Test that late-arriving filters are applied to files when the scanner starts processing
each scan range."""
if 'kudu' in str(vector.get_value('table_format')):
- return
- self.change_database(self.client, vector.get_value('table_format'))
+ pytest.skip('Skip test against kudu')
+ with self.change_database(self.client, vector.get_value('table_format')):
+ self.__run_file_filtering_late_arriving_filter(vector)
+
+ def __run_file_filtering_late_arriving_filter(self, vector):
self.execute_query("SET RUNTIME_FILTER_MODE=GLOBAL")
self.execute_query("SET RUNTIME_FILTER_WAIT_TIME_MS=1")
self.execute_query("SET NUM_SCANNER_THREADS=1")
@@ -357,7 +365,7 @@ class TestOverlapMinMaxFilters(ImpalaTestSuite):
unique_database, test_file_vars=DEFAULT_VARS)
@SkipIfLocal.hdfs_client
- def test_partition_column_in_parquet_data_file(self, vector, unique_database):
+ def test_partition_column_in_parquet_data_file(self, unique_database):
"""IMPALA-11147: Test that runtime min/max filters still work on data files that
contain the partitioning columns."""
tbl_name = "part_col_in_data_file"
diff --git a/tests/query_test/test_scanners.py b/tests/query_test/test_scanners.py
index 85b53951e..825cb09fa 100644
--- a/tests/query_test/test_scanners.py
+++ b/tests/query_test/test_scanners.py
@@ -323,9 +323,10 @@ class TestUnmatchedSchema(ImpalaTestSuite):
'invalidate metadata {0}.jointbl_test'.format(unique_database),
vector)
else:
+ origin_db = self.get_db_name_from_format(vector.get_table_format())
self.execute_query_using_client(self.client,
- "create external table {0}.jointbl_test like jointbl".format(
- unique_database), vector)
+ "create external table {0}.jointbl_test like {1}.jointbl".format(
+ unique_database, origin_db), vector)
# Update the location of the new table to point the same location as the old table
location = self._get_table_location('jointbl', vector)
@@ -439,8 +440,8 @@ class TestHdfsScannerSkew(ImpalaTestSuite):
super(TestHdfsScannerSkew, cls).add_test_dimensions()
add_mandatory_exec_option(cls, 'mt_dop', 2)
cls.ImpalaTestMatrix.add_constraint(lambda v:
- v.get_value('table_format').file_format in ('text') and
- v.get_value('table_format').compression_codec == 'none')
+ v.get_value('table_format').file_format in ('text')
+ and v.get_value('table_format').compression_codec == 'none')
@SkipIfLocal.multiple_impalad
@pytest.mark.execute_serially
@@ -1499,9 +1500,11 @@ class TestTextScanRangeLengths1(ImpalaTestSuite):
# the count(col) result because if the scan range is split right after the escape
# char, the escape char has no effect because we cannot scan backwards to the
# previous scan range.
- for t in self.ESCAPE_TABLE_LIST:
- expected_result = self.client.execute("select count(col) from " + t)
- result = self.client.execute("select count(*) from " + t)
+ db_name = self.get_db_name_from_format(vector.get_table_format())
+ for table_name in self.ESCAPE_TABLE_LIST:
+ fq_table_name = db_name + '.' + table_name
+ expected_result = self.client.execute("select count(col) from " + fq_table_name)
+ result = self.client.execute("select count(*) from " + fq_table_name)
assert result.data == expected_result.data
@@ -1548,10 +1551,10 @@ class TestTextSplitDelimiters(ImpalaTestSuite):
in the main text parsing algorithm. The second scan range exercises correctly
identifying a split delimiter as the first in a scan range."""
DEFAULT_IO_BUFFER_SIZE = 8 * 1024 * 1024
- data = ('a' * (DEFAULT_IO_BUFFER_SIZE - 1) + "\r\n" + # first scan range
- 'b' * (DEFAULT_IO_BUFFER_SIZE - 3) + "\r\n" +
- 'a' * (DEFAULT_IO_BUFFER_SIZE - 1) + "\r\n" + # second scan range
- 'b' * (DEFAULT_IO_BUFFER_SIZE - 1))
+ data = ('a' * (DEFAULT_IO_BUFFER_SIZE - 1) + "\r\n" # first scan range
+ + 'b' * (DEFAULT_IO_BUFFER_SIZE - 3) + "\r\n"
+ + 'a' * (DEFAULT_IO_BUFFER_SIZE - 1) + "\r\n" # second scan range
+ + 'b' * (DEFAULT_IO_BUFFER_SIZE - 1))
assert len(data) == DEFAULT_IO_BUFFER_SIZE * 4
max_scan_range_length = DEFAULT_IO_BUFFER_SIZE * 2
@@ -1594,8 +1597,8 @@ class TestTextScanRangeLengths2(ImpalaTestSuite):
def add_test_dimensions(cls):
super(TestTextScanRangeLengths2, cls).add_test_dimensions()
cls.ImpalaTestMatrix.add_constraint(lambda v:
- v.get_value('table_format').file_format == 'text' and
- v.get_value('table_format').compression_codec in ['none', 'gzip'])
+ v.get_value('table_format').file_format == 'text'
+ and v.get_value('table_format').compression_codec in ['none', 'gzip'])
def test_text_scanner_with_header(self, vector, unique_database):
# Remove to allow .test file to set abort_on_error.
diff --git a/tests/query_test/test_tablesample.py b/tests/query_test/test_tablesample.py
index 08e73ccb7..3cab33be3 100644
--- a/tests/query_test/test_tablesample.py
+++ b/tests/query_test/test_tablesample.py
@@ -18,12 +18,12 @@
# Tests the TABLESAMPLE clause.
from __future__ import absolute_import, division, print_function
-import pytest
import subprocess
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.test_vector import ImpalaTestDimension
+
class TestTableSample(ImpalaTestSuite):
@classmethod
def get_workload(cls):
@@ -36,19 +36,23 @@ class TestTableSample(ImpalaTestSuite):
cls.ImpalaTestMatrix.add_dimension(ImpalaTestDimension('filtered', *[True, False]))
# Tablesample is only supported on HDFS tables.
cls.ImpalaTestMatrix.add_constraint(lambda v:
- v.get_value('table_format').file_format != 'kudu' and
- v.get_value('table_format').file_format != 'hbase')
+ v.get_value('table_format').file_format != 'kudu'
+ and v.get_value('table_format').file_format != 'hbase')
if cls.exploration_strategy() != 'exhaustive':
# Cut down on core testing time by limiting the file formats.
cls.ImpalaTestMatrix.add_constraint(lambda v:
- v.get_value('table_format').file_format == 'parquet' or
- v.get_value('table_format').file_format == 'text')
+ v.get_value('table_format').file_format == 'parquet'
+ or v.get_value('table_format').file_format == 'text')
def test_tablesample(self, vector):
# Do not use a .test to avoid making this test flaky.
# 1. Queries without the repeatable clause are non-deterministic.
# 2. The results of queries without a repeatable clause could change due to
# changes in data loading that affect the number or size of files.
+ with self.change_database(self.client, vector.get_value('table_format')):
+ self.__run_tablesample(vector)
+
+ def __run_tablesample(self, vector):
repeatable = vector.get_value('repeatable')
filtered = vector.get_value('filtered')
@@ -56,7 +60,6 @@ class TestTableSample(ImpalaTestSuite):
if filtered:
where_clause = "where month between 1 and 6"
- ImpalaTestSuite.change_database(self.client, vector.get_value('table_format'))
result = self.client.execute("select count(*) from alltypes %s" % where_clause)
baseline_count = int(result.data[0])
prev_count = None
diff --git a/tests/util/cancel_util.py b/tests/util/cancel_util.py
index 5957ce757..fc1ed4e8b 100644
--- a/tests/util/cancel_util.py
+++ b/tests/util/cancel_util.py
@@ -102,9 +102,17 @@ def cancel_query_and_validate_state(client, query, exec_option, table_format,
statement will be executed to cancel and close the query, instead of sending the Thrift
RPCs directly.
"""
+ assert table_format is not None
+ with ImpalaTestSuite.change_database(client, table_format):
+ __run_cancel_query_and_validate_state(
+ client, query, exec_option, cancel_delay, join_before_close,
+ use_kill_query_statement)
+
+
+def __run_cancel_query_and_validate_state(client, query, exec_option,
+ cancel_delay, join_before_close=False, use_kill_query_statement=False):
assert not (join_before_close and use_kill_query_statement)
- if table_format: ImpalaTestSuite.change_database(client, table_format)
if exec_option: client.set_configuration(exec_option)
handle = client.execute_async(query)
@@ -160,8 +168,8 @@ def cancel_query_and_validate_state(client, query, exec_option, table_format,
# that the synchronous phase of query unregistration has finished and the profile
# is final.
profile = client.get_runtime_profile(handle)
- if ("- Completed admission: " in profile and
- ("- First row fetched:" in profile or "- Request finished:" in profile)):
+ if ("- Completed admission: " in profile
+ and ("- First row fetched:" in profile or "- Request finished:" in profile)):
# TotalBytesRead is a sentinel that will only be created if ComputeQuerySummary()
# has been run by the cancelling thread.
assert "- TotalBytesRead:" in profile, profile