You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@impala.apache.org by jo...@apache.org on 2019/11/10 19:59:21 UTC
[impala] 01/04: IMPALA-9045: Filter base directories of
open/aborted compactions
This is an automated email from the ASF dual-hosted git repository.
joemcdonnell pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/impala.git
commit 52c774d2ae9bb9d37a10f51fe5721d11a3ec7416
Author: Zoltan Borok-Nagy <bo...@cloudera.com>
AuthorDate: Fri Oct 25 17:49:34 2019 +0200
IMPALA-9045: Filter base directories of open/aborted compactions
Base directories are in the format of base_<write_id>_<transaction_id>.
The <transaction_id> part helps to decide whether a base directory
is fully written, or is it still being written by a compaction job.
Compaction jobs don't increase the write id of a table, hence the
<write_id> part cannot be used for that.
Before this commit Impala didn't check the validity of <transaction_id>,
therefore it might read the contents of a half-written base directory.
With this change Impala retrieves the valid transaction list from HMS
and checks if <transaction_id> is committed.
Testing
* Added an e2e test that simulates in-progress compactions
* Added frontend test that filters based on custom valid txn list
Change-Id: Idb895df38bc075e4767e44a6887dbe3000a19ea6
Reviewed-on: http://gerrit.cloudera.org:8080/14547
Reviewed-by: Impala Public Jenkins <im...@cloudera.com>
Tested-by: Impala Public Jenkins <im...@cloudera.com>
---
.../org/apache/impala/compat/MetastoreShim.java | 8 +++
.../org/apache/impala/compat/MetastoreShim.java | 7 +++
.../impala/catalog/CatalogServiceCatalog.java | 2 +-
.../apache/impala/catalog/FileMetadataLoader.java | 20 ++++--
.../java/org/apache/impala/catalog/HdfsTable.java | 52 ++++++++++-----
.../impala/catalog/local/DirectMetaProvider.java | 1 +
.../apache/impala/service/CatalogOpExecutor.java | 9 ++-
.../java/org/apache/impala/util/AcidUtils.java | 49 ++++++++++++---
.../impala/catalog/FileMetadataLoaderTest.java | 10 +--
.../apache/impala/catalog/HdfsPartitionTest.java | 2 +-
.../java/org/apache/impala/util/AcidUtilsTest.java | 73 +++++++++++++++++++++-
tests/query_test/test_acid.py | 54 ++++++++++++++++
12 files changed, 245 insertions(+), 42 deletions(-)
diff --git a/fe/src/compat-hive-2/java/org/apache/impala/compat/MetastoreShim.java b/fe/src/compat-hive-2/java/org/apache/impala/compat/MetastoreShim.java
index 5f47d9d..16f9a9e 100644
--- a/fe/src/compat-hive-2/java/org/apache/impala/compat/MetastoreShim.java
+++ b/fe/src/compat-hive-2/java/org/apache/impala/compat/MetastoreShim.java
@@ -29,6 +29,7 @@ import java.util.List;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.common.FileUtils;
import org.apache.hadoop.hive.common.StatsSetupConst;
+import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.common.ValidWriteIdList;
import org.apache.hadoop.hive.ql.metadata.formatting.MetaDataFormatUtils;
import org.apache.hadoop.hive.conf.HiveConf;
@@ -342,6 +343,13 @@ public class MetastoreShim {
/**
* Hive-3 only function
+ */
+ public static ValidTxnList getValidTxns(IMetaStoreClient client) throws TException {
+ throw new UnsupportedOperationException("getValidTxns not supported");
+ }
+
+ /**
+ * Hive-3 only function
* -1 means undefined
*/
public static long getWriteIdFromMSPartition(Partition partition) {
diff --git a/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java b/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java
index 4cfa538..8fda9dc 100644
--- a/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java
+++ b/fe/src/compat-hive-3/java/org/apache/impala/compat/MetastoreShim.java
@@ -574,6 +574,13 @@ public class MetastoreShim {
return new ValidReaderWriteIdList(validWriteIds);
}
+ /**
+ * Returns a ValidTxnList object that helps to identify in-progress and aborted
+ * transactions.
+ */
+ public static ValidTxnList getValidTxns(IMetaStoreClient client) throws TException {
+ return client.getValidTxns();
+ }
/**
* Get validWriteIds in string with txnId and table name
diff --git a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
index 7e91cbd..f6d95f4 100644
--- a/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
+++ b/fe/src/main/java/org/apache/impala/catalog/CatalogServiceCatalog.java
@@ -2561,7 +2561,7 @@ public class CatalogServiceCatalog extends Catalog {
throw new CatalogException("Error loading metadata for partition: "
+ hdfsTable.getFullName() + " " + partitionName, e);
}
- hdfsTable.reloadPartition(hdfsPartition, hmsPartition);
+ hdfsTable.reloadPartition(msClient.getHiveClient(), hdfsPartition, hmsPartition);
}
hdfsTable.setCatalogVersion(newCatalogVersion);
LOG.info(String.format("Refreshed partition metadata: %s %s",
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 85281c3..ceb4343 100644
--- a/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
+++ b/fe/src/main/java/org/apache/impala/catalog/FileMetadataLoader.java
@@ -20,6 +20,7 @@ import com.google.common.base.Objects;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableMap;
import com.google.common.collect.Maps;
+
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.BlockLocation;
import org.apache.hadoop.fs.FileStatus;
@@ -27,6 +28,7 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.common.ValidWriteIdList;
import org.apache.impala.catalog.HdfsPartition.FileDescriptor;
import org.apache.impala.common.FileSystemUtil;
@@ -58,6 +60,8 @@ public class FileMetadataLoader {
private final ListMap<TNetworkAddress> hostIndex_;
@Nullable
private final ValidWriteIdList writeIds_;
+ @Nullable
+ private final ValidTxnList validTxnList_;
private boolean forceRefreshLocations = false;
@@ -70,18 +74,24 @@ public class FileMetadataLoader {
* @param oldFds any pre-existing file descriptors loaded for this table, used
* to optimize refresh if available.
* @param hostIndex the host index with which to associate the file descriptors
+ * @param validTxnList if non-null, it can tell whether a given transaction id is
+ * committed or not. We need it to ignore base directories of in-progress
+ * compactions.
* @param writeIds if non-null, a write-id list which will filter the returned
* file descriptors to only include those indicated to be valid.
- * TODO(todd) we also likely need to pass an open transaction list here to deal
- * with ignoring in-progress (not-yet-visible) compactions.
*/
public FileMetadataLoader(Path partDir, boolean recursive, List<FileDescriptor> oldFds,
- ListMap<TNetworkAddress> hostIndex, @Nullable ValidWriteIdList writeIds) {
+ ListMap<TNetworkAddress> hostIndex, @Nullable ValidTxnList validTxnList,
+ @Nullable ValidWriteIdList writeIds) {
+ // Either both validTxnList and writeIds are null, or none of them.
+ Preconditions.checkState((validTxnList == null && writeIds == null) ||
+ (validTxnList != null && writeIds != null));
partDir_ = Preconditions.checkNotNull(partDir);
recursive_ = recursive;
hostIndex_ = Preconditions.checkNotNull(hostIndex);
oldFdsByRelPath_ = Maps.uniqueIndex(oldFds, FileDescriptor::getRelativePath);
writeIds_ = writeIds;
+ validTxnList_ = validTxnList;
if (writeIds_ != null) {
Preconditions.checkArgument(recursive_, "ACID tables must be listed recursively");
@@ -168,8 +178,8 @@ public class FileMetadataLoader {
}
if (writeIds_ != null) {
- stats = AcidUtils.filterFilesForAcidState(stats, partDir_, writeIds_,
- loadStats_);
+ stats = AcidUtils.filterFilesForAcidState(stats, partDir_, validTxnList_,
+ writeIds_, loadStats_);
}
for (FileStatus fileStatus : stats) {
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 d2039f1..b310b45 100644
--- a/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
+++ b/fe/src/main/java/org/apache/impala/catalog/HdfsTable.java
@@ -39,6 +39,7 @@ import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
+import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.common.ValidWriteIdList;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.IMetaStoreClient;
@@ -85,6 +86,7 @@ import org.apache.impala.util.MetaStoreUtil;
import org.apache.impala.util.TAccessLevelUtil;
import org.apache.impala.util.TResultRowBuilder;
import org.apache.impala.util.ThreadNameAnnotator;
+import org.apache.thrift.TException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -511,7 +513,7 @@ public class HdfsTable extends Table implements FeFsTable {
* If there are no partitions in the Hive metadata, a single partition is added with no
* partition keys.
*/
- private long loadAllPartitions(
+ private long loadAllPartitions(IMetaStoreClient client,
List<org.apache.hadoop.hive.metastore.api.Partition> msPartitions,
org.apache.hadoop.hive.metastore.api.Table msTbl) throws IOException,
CatalogException {
@@ -544,10 +546,20 @@ public class HdfsTable extends Table implements FeFsTable {
}
}
// Load the file metadata from scratch.
- loadFileMetadataForPartitions(partitionMap_.values(), /*isRefresh=*/false);
+ loadFileMetadataForPartitions(client, partitionMap_.values(), /*isRefresh=*/false);
return clock.getTick() - startTime;
}
+ /**
+ * Loads valid txn list from HMS. Re-throws exceptions as CatalogException.
+ */
+ private ValidTxnList loadValidTxns(IMetaStoreClient client) throws CatalogException {
+ try {
+ return MetastoreShim.getValidTxns(client);
+ } catch (TException exception) {
+ throw new CatalogException(exception.getMessage());
+ }
+ }
/**
* Helper method to load the block locations for each partition in 'parts'.
@@ -556,8 +568,8 @@ public class HdfsTable extends Table implements FeFsTable {
* @param isRefresh whether this is a refresh operation or an initial load. This only
* affects logging.
*/
- private void loadFileMetadataForPartitions(Iterable<HdfsPartition> parts,
- boolean isRefresh) throws CatalogException {
+ private void loadFileMetadataForPartitions(IMetaStoreClient client,
+ Iterable<HdfsPartition> parts, boolean isRefresh) throws CatalogException {
// Group the partitions by their path (multiple partitions may point to the same
// path).
Map<Path, List<HdfsPartition>> partsByPath = Maps.newHashMap();
@@ -569,6 +581,13 @@ public class HdfsTable extends Table implements FeFsTable {
ValidWriteIdList writeIds = validWriteIds_ != null
? MetastoreShim.getValidWriteIdListFromString(validWriteIds_) : null;
+ //TODO: maybe it'd be better to load the valid txn list in the context of a
+ // transaction to have consistent valid write ids and valid transaction ids.
+ // Currently tables are loaded when they are first referenced and stay in catalog
+ // until certain actions occur (refresh, invalidate, insert, etc.). However,
+ // Impala doesn't notice when HMS's cleaner removes old transactional directories,
+ // which might lead to FileNotFound exceptions.
+ ValidTxnList validTxnList = writeIds != null ? loadValidTxns(client) : null;
// Create a FileMetadataLoader for each path.
Map<Path, FileMetadataLoader> loadersByPath = Maps.newHashMap();
@@ -576,7 +595,7 @@ public class HdfsTable extends Table implements FeFsTable {
List<FileDescriptor> oldFds = e.getValue().get(0).getFileDescriptors();
FileMetadataLoader loader = new FileMetadataLoader(e.getKey(),
Utils.shouldRecursivelyListPartitions(this),
- oldFds, hostIndex_, writeIds);
+ oldFds, hostIndex_, validTxnList, writeIds);
// 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.
@@ -687,7 +706,7 @@ public class HdfsTable extends Table implements FeFsTable {
* Throws CatalogException if one of the supplied storage descriptors contains metadata
* that Impala can't understand.
*/
- public List<HdfsPartition> createAndLoadPartitions(
+ public List<HdfsPartition> createAndLoadPartitions(IMetaStoreClient client,
List<org.apache.hadoop.hive.metastore.api.Partition> msPartitions)
throws CatalogException {
List<HdfsPartition> addedParts = new ArrayList<>();
@@ -698,7 +717,7 @@ public class HdfsTable extends Table implements FeFsTable {
Preconditions.checkNotNull(hdfsPartition);
addedParts.add(hdfsPartition);
}
- loadFileMetadataForPartitions(addedParts, /* isRefresh = */ false);
+ loadFileMetadataForPartitions(client, addedParts, /* isRefresh = */ false);
return addedParts;
}
@@ -956,7 +975,7 @@ public class HdfsTable extends Table implements FeFsTable {
storageMetadataLoadTime_ += updateMdFromHmsTable(msTbl);
if (msTbl.getPartitionKeysSize() == 0) {
if (loadParitionFileMetadata) {
- storageMetadataLoadTime_ += updateUnpartitionedTableFileMd();
+ storageMetadataLoadTime_ += updateUnpartitionedTableFileMd(client);
}
} else {
storageMetadataLoadTime_ += updatePartitionsFromHms(
@@ -970,7 +989,7 @@ public class HdfsTable extends Table implements FeFsTable {
MetaStoreUtil.fetchAllPartitions(
client, db_.getName(), name_, NUM_PARTITION_FETCH_RETRIES);
LOG.info("Fetched partition metadata from the Metastore: " + getFullName());
- storageMetadataLoadTime_ = loadAllPartitions(msPartitions, msTbl);
+ storageMetadataLoadTime_ = loadAllPartitions(client, msPartitions, msTbl);
}
if (loadTableSchema) setAvroSchema(client, msTbl);
setTableStats(msTbl);
@@ -1017,7 +1036,8 @@ public class HdfsTable extends Table implements FeFsTable {
* This is optimized for the case where few files have changed. See
* {@link #refreshFileMetadata(Path, List)} above for details.
*/
- private long updateUnpartitionedTableFileMd() throws CatalogException {
+ private long updateUnpartitionedTableFileMd(IMetaStoreClient client)
+ throws CatalogException {
Preconditions.checkState(getNumClusteringCols() == 0);
if (LOG.isTraceEnabled()) {
LOG.trace("update unpartitioned table: " + getFullName());
@@ -1040,7 +1060,7 @@ public class HdfsTable extends Table implements FeFsTable {
part.setFileDescriptors(oldPartition.getFileDescriptors());
addPartition(part);
if (isMarkedCached_) part.markCached();
- loadFileMetadataForPartitions(ImmutableList.of(part), /*isRefresh=*/true);
+ loadFileMetadataForPartitions(client, ImmutableList.of(part), /*isRefresh=*/true);
return clock.getTick() - startTime;
}
@@ -1128,7 +1148,7 @@ public class HdfsTable extends Table implements FeFsTable {
// Only reload file metadata of partitions specified in 'partitionsToUpdate'
partitionsToLoadFiles = getPartitionsForNames(partitionsToUpdate);
}
- loadFileMetadataForPartitions(partitionsToLoadFiles, /* isRefresh=*/true);
+ loadFileMetadataForPartitions(client, partitionsToLoadFiles, /* isRefresh=*/true);
fileLoadMdTime = clock.getTick() - startTime;
}
return fileLoadMdTime;
@@ -1289,7 +1309,7 @@ public class HdfsTable extends Table implements FeFsTable {
if (partition == null) continue;
partitions.add(partition);
}
- loadFileMetadataForPartitions(partitions, /* isRefresh=*/false);
+ loadFileMetadataForPartitions(client, partitions, /* isRefresh=*/false);
for (HdfsPartition partition : partitions) addPartition(partition);
}
@@ -1858,8 +1878,8 @@ public class HdfsTable extends Table implements FeFsTable {
* 'hmsPartition'. If old partition is null then nothing is removed and
* and partition constructed from 'hmsPartition' is simply added.
*/
- public void reloadPartition(HdfsPartition oldPartition, Partition hmsPartition)
- throws CatalogException {
+ public void reloadPartition(IMetaStoreClient client, HdfsPartition oldPartition,
+ Partition hmsPartition) throws CatalogException {
// Instead of updating the existing partition in place, we create a new one
// so that we reflect any changes in the hmsPartition object and also assign a new
// ID. This is one step towards eventually implementing IMPALA-7533.
@@ -1870,7 +1890,7 @@ public class HdfsTable extends Table implements FeFsTable {
if (oldPartition != null) {
refreshedPartition.setFileDescriptors(oldPartition.getFileDescriptors());
}
- loadFileMetadataForPartitions(ImmutableList.of(refreshedPartition),
+ loadFileMetadataForPartitions(client, ImmutableList.of(refreshedPartition),
/*isRefresh=*/true);
dropPartition(oldPartition, false);
addPartition(refreshedPartition);
diff --git a/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
index c732d8b..a92a750 100644
--- a/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
+++ b/fe/src/main/java/org/apache/impala/catalog/local/DirectMetaProvider.java
@@ -297,6 +297,7 @@ class DirectMetaProvider implements MetaProvider {
/* recursive= */BackendConfig.INSTANCE.recursivelyListPartitions(),
/* oldFds= */Collections.emptyList(),
hostIndex,
+ /* validTxnList=*/null,
/* writeIds=*/null);
try {
diff --git a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
index 0abae8f..ce1c481 100644
--- a/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
+++ b/fe/src/main/java/org/apache/impala/service/CatalogOpExecutor.java
@@ -867,9 +867,12 @@ public class CatalogOpExecutor {
throw new CatalogException("Table " + tbl.getFullName() + " is not an HDFS table");
}
HdfsTable hdfsTable = (HdfsTable) tbl;
- List<HdfsPartition> hdfsPartitions = hdfsTable.createAndLoadPartitions(partitions);
- for (HdfsPartition hdfsPartition: hdfsPartitions) {
- catalog_.addPartition(hdfsPartition);
+ try (MetaStoreClient msClient = catalog_.getMetaStoreClient()) {
+ List<HdfsPartition> hdfsPartitions = hdfsTable.createAndLoadPartitions(
+ msClient.getHiveClient(), partitions);
+ for (HdfsPartition hdfsPartition: hdfsPartitions) {
+ catalog_.addPartition(hdfsPartition);
+ }
}
return hdfsTable;
}
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 29db92f..3b856ad 100644
--- a/fe/src/main/java/org/apache/impala/util/AcidUtils.java
+++ b/fe/src/main/java/org/apache/impala/util/AcidUtils.java
@@ -22,6 +22,7 @@ import com.google.errorprone.annotations.Immutable;
import org.apache.hadoop.fs.FileStatus;
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.FileMetadataLoader.LoadStats;
import org.apache.impala.common.FileSystemUtil;
@@ -142,16 +143,19 @@ public class AcidUtils {
*/
private static class WriteListBasedPredicate implements Predicate<String> {
+ private final ValidTxnList validTxnList;
private final ValidWriteIdList writeIdList;
- WriteListBasedPredicate(ValidWriteIdList writeIdList) {
+ WriteListBasedPredicate(ValidTxnList validTxnList, ValidWriteIdList writeIdList) {
+ this.validTxnList = Preconditions.checkNotNull(validTxnList);
this.writeIdList = Preconditions.checkNotNull(writeIdList);
}
public boolean test(String dirPath) {
- long baseWriteId = getBaseWriteId(dirPath);
- if (baseWriteId != SENTINEL_BASE_WRITE_ID) {
- return writeIdList.isValidBase(baseWriteId);
+ ParsedBase parsedBase = parseBase(dirPath);
+ if (parsedBase.writeId != SENTINEL_BASE_WRITE_ID) {
+ return writeIdList.isValidBase(parsedBase.writeId) &&
+ isTxnValid(parsedBase.visibilityTxnId);
} else {
ParsedDelta pd = parseDelta(dirPath);
if (pd != null) {
@@ -165,15 +169,41 @@ public class AcidUtils {
// TODO(todd) add an e2e test for this.
return true;
}
+
+ private boolean isTxnValid(long visibilityTxnId) {
+ return visibilityTxnId == -1 || validTxnList.isTxnValid(visibilityTxnId);
+ }
+ }
+
+ @Immutable
+ private static final class ParsedBase {
+ final long writeId;
+ final long visibilityTxnId;
+
+ ParsedBase(long writeId, long visibilityTxnId) {
+ this.writeId = writeId;
+ this.visibilityTxnId = visibilityTxnId;
+ }
}
@VisibleForTesting
- static long getBaseWriteId(String relPath) {
+ static ParsedBase parseBase(String relPath) {
Matcher baseMatcher = BASE_PATTERN.matcher(relPath);
if (baseMatcher.matches()) {
- return Long.valueOf(baseMatcher.group("writeId"));
+ long writeId = Long.valueOf(baseMatcher.group("writeId"));
+ long visibilityTxnId = -1;
+ String visibilityTxnIdStr = baseMatcher.group("visibilityTxnId");
+ if (visibilityTxnIdStr != null) {
+ visibilityTxnId = Long.valueOf(visibilityTxnIdStr);
+ }
+ return new ParsedBase(writeId, visibilityTxnId);
}
- return SENTINEL_BASE_WRITE_ID;
+ return new ParsedBase(SENTINEL_BASE_WRITE_ID, -1);
+ }
+
+ @VisibleForTesting
+ static long getBaseWriteId(String relPath) {
+ return parseBase(relPath).writeId;
}
@Immutable
@@ -215,12 +245,13 @@ public class AcidUtils {
* must be used.
*/
public static List<FileStatus> filterFilesForAcidState(List<FileStatus> stats,
- Path baseDir, ValidWriteIdList writeIds, @Nullable LoadStats loadStats) {
+ Path baseDir, ValidTxnList validTxnList, ValidWriteIdList writeIds,
+ @Nullable LoadStats loadStats) {
List<FileStatus> validStats = new ArrayList<>(stats);
// First filter out any paths that are not considered valid write IDs.
// At the same time, calculate the max valid base write ID.
- Predicate<String> pred = new WriteListBasedPredicate(writeIds);
+ Predicate<String> pred = new WriteListBasedPredicate(validTxnList, writeIds);
long maxBaseWriteId = Long.MIN_VALUE;
for (Iterator<FileStatus> it = validStats.iterator(); it.hasNext(); ) {
FileStatus stat = it.next();
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 ffc0481..d9c61c3 100644
--- a/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/FileMetadataLoaderTest.java
@@ -42,7 +42,7 @@ public class FileMetadataLoaderTest {
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
Path tablePath = new Path("hdfs://localhost:20500/test-warehouse/alltypes/");
FileMetadataLoader fml = new FileMetadataLoader(tablePath, /* recursive=*/true,
- /* oldFds = */Collections.emptyList(), hostIndex, null);
+ /* oldFds = */Collections.emptyList(), hostIndex, null, null);
fml.load();
assertEquals(24, fml.getStats().loadedFiles);
assertEquals(24, fml.getLoadedFds().size());
@@ -56,7 +56,7 @@ public class FileMetadataLoaderTest {
// Test that refreshing is properly incremental if no files changed.
FileMetadataLoader refreshFml = new FileMetadataLoader(tablePath, /* recursive=*/true,
- /* oldFds = */fml.getLoadedFds(), hostIndex, null);
+ /* oldFds = */fml.getLoadedFds(), hostIndex, null, null);
refreshFml.load();
assertEquals(24, refreshFml.getStats().skippedFiles);
assertEquals(0, refreshFml.getStats().loadedFiles);
@@ -69,7 +69,7 @@ public class FileMetadataLoaderTest {
fs.setTimes(filePath, fd.getModificationTime() + 1, /* atime= */-1);
refreshFml = new FileMetadataLoader(tablePath, /* recursive=*/true,
- /* oldFds = */fml.getLoadedFds(), hostIndex, null);
+ /* oldFds = */fml.getLoadedFds(), hostIndex, null, null);
refreshFml.load();
assertEquals(1, refreshFml.getStats().loadedFiles);
}
@@ -80,7 +80,7 @@ public class FileMetadataLoaderTest {
ListMap<TNetworkAddress> hostIndex = new ListMap<>();
Path tablePath = new Path("hdfs://localhost:20500/test-warehouse/does-not-exist/");
FileMetadataLoader fml = new FileMetadataLoader(tablePath, recursive,
- /* oldFds = */Collections.emptyList(), hostIndex, null);
+ /* oldFds = */Collections.emptyList(), hostIndex, null, null);
fml.load();
assertEquals(0, fml.getLoadedFds().size());
}
@@ -106,7 +106,7 @@ public class FileMetadataLoaderTest {
dstFs.createNewFile(new Path(hiveStaging, ".hidden-tmp-stats"));
FileMetadataLoader fml = new FileMetadataLoader(tmpTestPath, true,
- Collections.emptyList(), new ListMap <>(), null);
+ Collections.emptyList(), new ListMap <>(), null, null);
fml.load();
assertEquals(24, fml.getStats().loadedFiles);
assertEquals(24, fml.getLoadedFds().size());
diff --git a/fe/src/test/java/org/apache/impala/catalog/HdfsPartitionTest.java b/fe/src/test/java/org/apache/impala/catalog/HdfsPartitionTest.java
index 51706c1..5f9eb30 100644
--- a/fe/src/test/java/org/apache/impala/catalog/HdfsPartitionTest.java
+++ b/fe/src/test/java/org/apache/impala/catalog/HdfsPartitionTest.java
@@ -151,7 +151,7 @@ public class HdfsPartitionTest {
Path p = new Path("hdfs://localhost:20500/test-warehouse/schemas");
ListMap<TNetworkAddress> origIndex = new ListMap<>();
FileMetadataLoader fml = new FileMetadataLoader(p, /* recursive= */false,
- Collections.emptyList(), origIndex, /*writeIds=*/null);
+ Collections.emptyList(), origIndex, /*validTxnList=*/null, /*writeIds=*/null);
fml.load();
List<FileDescriptor> fileDescriptors = fml.getLoadedFds();
assertTrue(!fileDescriptors.isEmpty());
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 86a47d7..aa24878 100644
--- a/fe/src/test/java/org/apache/impala/util/AcidUtilsTest.java
+++ b/fe/src/test/java/org/apache/impala/util/AcidUtilsTest.java
@@ -25,6 +25,7 @@ import java.util.stream.Collectors;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.ValidReadTxnList;
import org.apache.hadoop.hive.common.ValidWriteIdList;
import org.apache.impala.compat.MetastoreShim;
import org.hamcrest.Matchers;
@@ -51,14 +52,19 @@ public class AcidUtilsTest {
private void assertFiltering(String[] relPaths, String validWriteIdListStr,
String[] expectedRelPaths) {
+ assertFiltering(relPaths, "", validWriteIdListStr, expectedRelPaths);
+ }
+ private void assertFiltering(String[] relPaths, String validTxnListStr,
+ String validWriteIdListStr, String[] expectedRelPaths) {
ValidWriteIdList writeIds = MetastoreShim.getValidWriteIdListFromString(
validWriteIdListStr);
List<FileStatus> stats = createMockStats(relPaths);
List<FileStatus> expectedStats = createMockStats(expectedRelPaths);
- assertThat(AcidUtils.filterFilesForAcidState(stats, BASE_PATH, writeIds, null),
- Matchers.containsInAnyOrder(expectedStats.toArray()));
+ assertThat(AcidUtils.filterFilesForAcidState(stats, BASE_PATH,
+ new ValidReadTxnList(validTxnListStr), writeIds, null),
+ Matchers.containsInAnyOrder(expectedStats.toArray()));
}
@Test
@@ -123,6 +129,69 @@ public class AcidUtilsTest {
}
@Test
+ public void testInProgressCompaction() {
+ assertFiltering(new String[]{
+ "base_01.txt",
+ "post_upgrade.txt",
+ "base_0000005_v900/",
+ "base_0000005_v900/abc.txt",
+ "base_0000005_v900/0000/",
+ "base_0000005_v900/0000/abc.txt",
+ "base_0000008_v1000/",
+ "base_0000008_v1000/abc.txt",
+ "delta_0000006_0000006_0000/",
+ "delta_0000006_0000006_0000/000000_0",
+ "delta_0000007_0000007_0000/",
+ "delta_0000007_0000007_0000/000000_0",
+ "delta_0000008_0000008_0000/",
+ "delta_0000008_0000008_0000/000000_0",
+ "delta_0000009_0000009_0000/",
+ "delta_0000009_0000009_0000/000000_0",
+ "delta_0000009_0000009_0000/0000/def.txt"},
+ "1100:1000:1000:", // txn 1000 is open
+ "default.test:10:10::", // write ids are committed
+ new String[]{
+ "base_0000005_v900/abc.txt",
+ "base_0000005_v900/0000/abc.txt",
+ "delta_0000006_0000006_0000/000000_0",
+ "delta_0000007_0000007_0000/000000_0",
+ "delta_0000008_0000008_0000/000000_0",
+ "delta_0000009_0000009_0000/000000_0",
+ "delta_0000009_0000009_0000/0000/def.txt"});
+ }
+
+ @Test
+ public void testAbortedCompaction() {
+ assertFiltering(new String[]{
+ "base_01.txt",
+ "post_upgrade.txt",
+ "base_0000005_v900/",
+ "base_0000005_v900/abc.txt",
+ "base_0000005_v900/0000/",
+ "base_0000005_v900/0000/abc.txt",
+ "base_0000007_v950/",
+ "base_0000007_v950/abc.txt",
+ "base_0000008_v1000/",
+ "base_0000008_v1000/abc.txt",
+ "delta_0000006_0000006_0000/",
+ "delta_0000006_0000006_0000/000000_0",
+ "delta_0000007_0000007_0000/",
+ "delta_0000007_0000007_0000/000000_0",
+ "delta_0000008_0000008_0000/",
+ "delta_0000008_0000008_0000/000000_0",
+ "delta_0000009_0000009_0000/",
+ "delta_0000009_0000009_0000/000000_0",
+ "delta_0000009_0000009_0000/0000/def.txt"},
+ "1100:1100::1000", // txn 1000 is aborted
+ "default.test:10:10::", // write ids are committed
+ new String[]{
+ "base_0000007_v950/abc.txt",
+ "delta_0000008_0000008_0000/000000_0",
+ "delta_0000009_0000009_0000/000000_0",
+ "delta_0000009_0000009_0000/0000/def.txt"});
+ }
+
+ @Test
public void testAbortedTransactions() {
assertFiltering(new String[]{
"base_01.txt",
diff --git a/tests/query_test/test_acid.py b/tests/query_test/test_acid.py
index 2468911..19df4e5 100644
--- a/tests/query_test/test_acid.py
+++ b/tests/query_test/test_acid.py
@@ -20,10 +20,14 @@
import pytest
import time
+from hive_metastore.ttypes import CommitTxnRequest, OpenTxnRequest
+from subprocess import check_call
from tests.common.impala_test_suite import ImpalaTestSuite
from tests.common.skip import (SkipIfHive2, SkipIfCatalogV2, SkipIfS3, SkipIfABFS,
SkipIfADLS, SkipIfIsilon, SkipIfLocal)
from tests.common.test_dimensions import create_single_exec_option_dimension
+
+
class TestAcid(ImpalaTestSuite):
@classmethod
def get_workload(self):
@@ -204,3 +208,53 @@ class TestAcid(ImpalaTestSuite):
if txn_dict['user'] != 'Impala':
continue
yield txn_dict
+
+ def _open_txn(self):
+ open_txn_req = OpenTxnRequest()
+ open_txn_req.num_txns = 1
+ open_txn_req.user = "AcidTest"
+ open_txn_req.hostname = "localhost"
+ open_txn_resp = self.hive_client.open_txns(open_txn_req)
+ return open_txn_resp.txn_ids[0]
+
+ def _commit_txn(self, txn_id):
+ commit_req = CommitTxnRequest()
+ commit_req.txnid = txn_id
+ return self.hive_client.commit_txn(commit_req)
+
+ @SkipIfHive2.acid
+ @SkipIfS3.hive
+ @SkipIfABFS.hive
+ @SkipIfADLS.hive
+ @SkipIfIsilon.hive
+ @SkipIfLocal.hive
+ def test_in_progress_compactions(self, vector, unique_database):
+ """Checks that in-progress compactions are not visible. The test mimics
+ in-progress compactions by opening a transaction and creating a new base
+ directory. The new base directory is empty and must not have an effect
+ on query results until the transaction is committed."""
+ tbl_name = "{}.{}".format(unique_database, "test_compaction")
+ self.execute_query("create table {} (i int) tblproperties"
+ "('transactional'='true',"
+ "'transactional_properties'='insert_only')".format(tbl_name))
+ self.execute_query("insert into {} values (1)".format(tbl_name))
+
+ # Create new base directory with a valid write id.
+ txn_id = self._open_txn()
+ tbl_file = self.execute_query(
+ "show files in {}".format(tbl_name)).data[0].split("\t")[0]
+ tbl_dir = tbl_file[tbl_file.find("/test-warehouse"):tbl_file.rfind("delta_")]
+ new_base_dir_with_old_write_id = tbl_dir + "base_1_v" + str(txn_id)
+ check_call(['hdfs', 'dfs', '-mkdir', '-p', new_base_dir_with_old_write_id])
+
+ # Transaction is not committed so the new empty base directory must not have
+ # any effect on query results.
+ self.execute_query("refresh {}".format(tbl_name))
+ assert len(self.execute_query("select * from {}".format(tbl_name)).data) != 0
+
+ # Transaction is committed, now the query should see the table as empty. Of course,
+ # real compactions don't remove data, but that verifies that the query reads the
+ # new directory.
+ self._commit_txn(txn_id)
+ self.execute_query("refresh {}".format(tbl_name))
+ assert len(self.execute_query("select * from {}".format(tbl_name)).data) == 0