This is an automated email from the ASF dual-hosted git repository.
kuczoram pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hive.git
The following commit(s) were added to refs/heads/master by this push:
new 3d96d70fc1a HIVE-29601: ACID: Cleaner finds base directories valid
with writeId above the cleaner highWaterMark (#6470)
3d96d70fc1a is described below
commit 3d96d70fc1ace73506079fcfc063199a74dd66a2
Author: Marta Kuczora <[email protected]>
AuthorDate: Tue May 12 17:40:18 2026 +0200
HIVE-29601: ACID: Cleaner finds base directories valid with writeId above
the cleaner highWaterMark (#6470)
---
.../org/apache/hadoop/hive/ql/io/AcidUtils.java | 9 +-
.../apache/hadoop/hive/ql/io/TestAcidUtils.java | 226 +++++++++++++++++++++
.../hadoop/hive/ql/txn/compactor/TestCleaner.java | 174 +++++++++++++++-
.../hive/common/ValidCleanerWriteIdList.java | 13 ++
.../hadoop/hive/common/ValidReaderWriteIdList.java | 1 +
5 files changed, 417 insertions(+), 6 deletions(-)
diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
b/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
index 909b22cf8a4..af93bd50630 100644
--- a/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
+++ b/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
@@ -1781,7 +1781,7 @@ public String toString() {
* causes anything written previously to be ignored (hence the overwrite).
In this case, base_x
* is visible if writeid:x is committed for current reader.
*/
- private static boolean isValidBase(ParsedBaseLight parsedBase,
ValidWriteIdList writeIdList, FileSystem fs,
+ static boolean isValidBase(ParsedBaseLight parsedBase, ValidWriteIdList
writeIdList, FileSystem fs,
HdfsDirSnapshot dirSnapshot) throws IOException {
boolean isValidBase;
if (dirSnapshot != null && dirSnapshot.isValidBase() != null) {
@@ -1791,13 +1791,12 @@ private static boolean isValidBase(ParsedBaseLight
parsedBase, ValidWriteIdList
//such base is created by 1st compaction in case of non-acid to acid
table conversion
//By definition there are no open txns with id < 1.
isValidBase = true;
- } else if (writeIdList.getMinOpenWriteId() != null &&
parsedBase.getWriteId() <= writeIdList
- .getMinOpenWriteId()) {
- isValidBase = true;
} else if (isCompactedBase(parsedBase, fs, dirSnapshot)) {
+ // Compacted base covers range [1..writeId]; require no open writeId
in that range
+ // and writeId within reader's high watermark.
isValidBase = writeIdList.isValidBase(parsedBase.getWriteId());
} else {
- // if here, it's a result of IOW
+ // IOW base: a single writeId. Visible only if that writeId is
committed for this reader.
isValidBase = writeIdList.isWriteIdValid(parsedBase.getWriteId());
}
if (dirSnapshot != null) {
diff --git a/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java
b/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java
index bd6cacbde80..0a3f7534aa7 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/io/TestAcidUtils.java
@@ -32,12 +32,15 @@
import java.util.Set;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.ValidCleanerWriteIdList;
import org.apache.hadoop.hive.common.ValidCompactorWriteIdList;
import org.apache.hadoop.hive.common.ValidReadTxnList;
import org.apache.hadoop.hive.common.ValidReaderWriteIdList;
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.TableType;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
@@ -749,4 +752,227 @@ public void
testShouldNotThrowFNFEWhenHiveStagingDirectoryIsRemovedWhileFetching
Assert.fail("Should not throw FileNotFoundException when a directory is
removed while fetching HDFSSnapshots");
}
}
+
+ @Test
+ public void testIsValidBaseCompactedBaseOnlyHighWaterMarkSet() throws
Exception {
+ // Only highWaterMark is set in the validWriteIdList. minOpenWriteId is
not set and there are no exceptions
+ ValidWriteIdList cleanerWriteIdList = new ValidCleanerWriteIdList(
+ new ValidReaderWriteIdList("testBaseTable", new long[] {}, new
BitSet(), 10L));
+
+ checkBase(0L, Long.MIN_VALUE, cleanerWriteIdList, true,
+ "Base after first compaction in case of non-acid to acid table
conversion is always valid");
+ checkBase(2L, 2L, cleanerWriteIdList, true,
+ "Base with writeId smaller than highWaterMark should be valid");
+ checkBase(10L, 10L, cleanerWriteIdList, true,
+ "Base with writeId equals to highWaterMark should be valid");
+ checkBase(11L, 11L, cleanerWriteIdList, false,
+ "Base with writeId greater than highWaterMark should not be valid");
+ }
+
+ @Test
+ public void testIsValidBaseNotCompactedBaseOnlyHighWaterMarkSet() throws
Exception {
+ // Only highWaterMark is set in the validWriteIdList. minOpenWriteId is
not set and there are no exceptions
+ ValidWriteIdList cleanerWriteIdList = new ValidCleanerWriteIdList(
+ new ValidReaderWriteIdList("testBaseTable", new long[] {}, new
BitSet(), 10L));
+
+ checkNotCompactedBase(2L, cleanerWriteIdList, true,
+ "Non-compacted base with writeId smaller than highWaterMark should be
valid");
+ checkNotCompactedBase(10L, cleanerWriteIdList, true,
+ "Non-compacted base with writeId equals to highWaterMark should be
valid");
+ checkNotCompactedBase(11L, cleanerWriteIdList, false,
+ "Non-compacted base with writeId greater than highWaterMark should not
be valid");
+ }
+
+ @Test
+ public void
testIsValidBaseCompactedBaseMinOpenWriteIdGreaterThanHighWaterMark() throws
Exception {
+ // Both highWaterMark and the minOpenWriteId are set and minOpenWriteId >
highWaterMark
+ ValidWriteIdList cleanerWriteIdList = new ValidCleanerWriteIdList(
+ new ValidReaderWriteIdList("testBaseTable", new long[] {13L}, new
BitSet(), 10L, 13L));
+
+ checkBase(5L, 5L, cleanerWriteIdList, true,
+ "Base with writeId smaller than highWaterMark and minOpenWriteId
should be valid");
+ checkBase(10L, 10L, cleanerWriteIdList, true,
+ "Base with writeId equals to highWaterMark and smaller than
minOpenWriteId should be valid");
+ checkBase(11L, 11L, cleanerWriteIdList, false,
+ "Base with writeId greater than highWaterMark and smaller than
minOpenWriteId should not be valid");
+ }
+
+ @Test
+ public void
testIsValidBaseNotCompactedBaseMinOpenWriteIdGreaterThanHighWaterMark() throws
Exception {
+ // Both highWaterMark and the minOpenWriteId are set and minOpenWriteId >
highWaterMark
+ ValidWriteIdList cleanerWriteIdList = new ValidCleanerWriteIdList(
+ new ValidReaderWriteIdList("testBaseTable", new long[] {13L}, new
BitSet(), 10L, 13L));
+
+ checkNotCompactedBase(5L, cleanerWriteIdList, true,
+ "Non-compacted base with writeId smaller than highWaterMark and
minOpenWriteId should be valid");
+ checkNotCompactedBase(10L, cleanerWriteIdList, true,
+ "Non-compacted base with writeId equals to highWaterMark and smaller
than minOpenWriteId should be valid");
+ checkNotCompactedBase(11L, cleanerWriteIdList, false,
+ "Non-compacted base with writeId greater than highWaterMark and
smaller than minOpenWriteId should not be valid");
+ }
+
+ @Test
+ public void
testIsValidBaseCompactedBaseMinOpenWriteIdSmallerThanHighWaterMark() throws
Exception {
+ // Both highWaterMark and minOpenWriteId are set and minOpenWriteId <
highWaterMark
+ ValidWriteIdList cleanerWriteIdList = new ValidCleanerWriteIdList(
+ new ValidReaderWriteIdList("testBaseTable", new long[] {4L}, new
BitSet(), 10L, 4L));
+
+ checkBase(2L, 2L, cleanerWriteIdList, true,
+ "Base with writeId smaller than highWaterMark and minOpenWriteId
should be valid");
+ checkBase(4L, 4L, cleanerWriteIdList, false,
+ "Base with writeId smaller than highWaterMark and equals to
minOpenWriteId should not be valid");
+ checkBase(5L, 5L, cleanerWriteIdList, false,
+ "Base with writeId smaller than highWaterMark and greater than
minOpenWriteId should not be valid");
+ }
+
+ @Test
+ public void
testIsValidBaseNotCompactedBaseMinOpenWriteIdSmallerThanHighWaterMark() throws
Exception {
+ // Both highWaterMark and minOpenWriteId are set and minOpenWriteId <
highWaterMark
+ ValidWriteIdList cleanerWriteIdList = new ValidCleanerWriteIdList(
+ new ValidReaderWriteIdList("testBaseTable", new long[] {4L}, new
BitSet(), 10L, 4L));
+
+ checkNotCompactedBase(2L, cleanerWriteIdList, true,
+ "Non-compacted base with writeId smaller than highWaterMark and
minOpenWriteId should be valid");
+ checkNotCompactedBase(4L, cleanerWriteIdList, false,
+ "Non-compacted base with writeId smaller than highWaterMark and equals
to minOpenWriteId should not be valid");
+ checkNotCompactedBase(5L, cleanerWriteIdList, true,
+ "Non-compacted base with writeId smaller than highWaterMark and
greater than minOpenWriteId should be valid");
+ }
+
+ @Test
+ public void testIsValidBaseCompactedBaseWithAbortedWriteIds() throws
Exception {
+ // minOpenWriteId is not set, but there are aborted writeIds
+ // For compacted base directories, the writeId is not checked against the
exception list, as it is checked earlier
+ BitSet aborted = new BitSet();
+ aborted.set(0);
+ aborted.set(1);
+ ValidWriteIdList cleanerWriteIdList = new ValidCleanerWriteIdList(
+ new ValidReaderWriteIdList("testBaseTable", new long[] {3L, 5L},
aborted, 10L));
+
+ checkBase(2L, 2L, cleanerWriteIdList, true,
+ "Base with writeId smaller than highWaterMark should be valid");
+ checkBase(3L, 3L, cleanerWriteIdList, true,
+ "Base with writeId smaller than highWaterMark should be valid");
+ checkBase(5L, 5L, cleanerWriteIdList, true,
+ "Base with writeId smaller than highWaterMark should be valid");
+ checkBase(10L, 10L, cleanerWriteIdList, true,
+ "Base with writeId equals to highWaterMark should be valid");
+ checkBase(12L, 12L, cleanerWriteIdList, false,
+ "Base with writeId greater than highWaterMark should not be valid");
+ }
+
+ @Test
+ public void testIsValidBaseNotCompactedBaseWithAbortedWriteIds() throws
Exception {
+ // minOpenWriteId is not set, but there are aborted writeIds
+ // For not compacted base directories, the writeId is checked if it is
present in the exception list. If it is
+ // listed as exception, it should not be valid.
+ BitSet aborted = new BitSet();
+ aborted.set(0);
+ aborted.set(1);
+ ValidWriteIdList cleanerWriteIdList = new ValidCleanerWriteIdList(
+ new ValidReaderWriteIdList("testBaseTable", new long[] {3L, 5L},
aborted, 10L));
+
+ checkNotCompactedBase(0L, cleanerWriteIdList, true,
+ "Non-compacted base after first compaction in case of non-acid to acid
table conversion is always valid");
+ checkNotCompactedBase(2L, cleanerWriteIdList, true,
+ "Non-compacted base with writeId smaller than highWaterMark should be
valid");
+ checkNotCompactedBase(3L, cleanerWriteIdList, false,
+ "Non-compacted base with aborted writeId should not be valid");
+ checkNotCompactedBase(5L, cleanerWriteIdList, false,
+ "Non-compacted base with aborted writeId should not be valid");
+ checkNotCompactedBase(10L, cleanerWriteIdList, true,
+ "Non-compacted base with writeId equals to highWaterMark should be
valid");
+ checkNotCompactedBase(12L, cleanerWriteIdList, false,
+ "Non-compacted base with writeId greater than highWaterMark should not
be valid");
+ }
+
+ private void checkNotCompactedBase(long writeId, ValidWriteIdList
cleanerWriteIdList, boolean valid,
+ String errorMessage) throws IOException {
+ checkBase(writeId, 0L, cleanerWriteIdList, valid, errorMessage);
+ }
+
+ private void checkBase(long writeId, long visibilityTxnId, ValidWriteIdList
cleanerWriteIdList, boolean valid,
+ String errorMessage) throws IOException {
+ AcidUtils.ParsedBaseLight p = new AcidUtils.ParsedBaseLight(writeId,
visibilityTxnId,
+ new Path("testpath"));
+ if (valid) {
+ Assert.assertTrue(errorMessage, AcidUtils.isValidBase(p,
cleanerWriteIdList, null,
+ new MockHdfsDirSnapshotImpl()));
+ } else {
+ Assert.assertFalse(errorMessage, AcidUtils.isValidBase(p,
cleanerWriteIdList, null,
+ new MockHdfsDirSnapshotImpl()));
+ }
+ }
+
+ private final class MockHdfsDirSnapshotImpl implements
AcidUtils.HdfsDirSnapshot {
+ @Override
+ public Path getPath() {
+ return null;
+ }
+ @Override
+ public void addOrcAcidFormatFile(FileStatus fStatus) {
+ // this method is not used by the tests, no need to add an implementation
+ }
+ @Override
+ public FileStatus getOrcAcidFormatFile() {
+ return null;
+ }
+ @Override
+ public void addMetadataFile(FileStatus fStatus) {
+ // this method is not used by the tests, no need to add an implementation
+ }
+ @Override
+ public FileStatus getMetadataFile() {
+ return null;
+ }
+ @Override
+ public List<FileStatus> getFiles() {
+ return null;
+ }
+ @Override
+ public void addFile(FileStatus file) {
+ // this method is not used by the tests, no need to add an implementation
+ }
+ @Override
+ public Long getFileId() {
+ return null;
+ }
+ @Override
+ public Boolean isRawFormat() {
+ return null;
+ }
+ @Override
+ public void setIsRawFormat(boolean isRawFormat) {
+ // this method is not used by the tests, no need to add an implementation
+ }
+ @Override
+ public Boolean isBase() {
+ return null;
+ }
+ @Override
+ public void setIsBase(boolean isBase) {
+ // this method is not used by the tests, no need to add an implementation
+ }
+ @Override
+ public Boolean isValidBase() {
+ return null;
+ }
+ @Override
+ public void setIsValidBase(boolean isValidBase) {
+ // this method is not used by the tests, no need to add an implementation
+ }
+ @Override
+ public Boolean isCompactedBase() {
+ return null;
+ }
+ @Override
+ public void setIsCompactedBase(boolean isCompactedBase) {
+ // this method is not used by the tests, no need to add an implementation
+ }
+ @Override
+ public boolean contains(Path path) {
+ return false;
+ }
+ }
+
}
diff --git
a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java
b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java
index 30815442ff2..df73adf26ed 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleaner.java
@@ -20,6 +20,7 @@
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.metastore.ReplChangeManager;
+import org.apache.hadoop.hive.metastore.api.AbortTxnRequest;
import org.apache.hadoop.hive.metastore.api.CommitTxnRequest;
import org.apache.hadoop.hive.metastore.api.CompactionRequest;
import org.apache.hadoop.hive.metastore.api.CompactionResponse;
@@ -30,14 +31,17 @@
import org.apache.hadoop.hive.metastore.api.ShowCompactResponse;
import org.apache.hadoop.hive.metastore.api.ShowCompactResponseElement;
import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.metastore.api.TxnType;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
import org.apache.hadoop.hive.metastore.txn.TxnStore;
+import org.apache.hadoop.hive.ql.io.AcidUtils;
import org.apache.hadoop.hive.ql.testutil.TxnStoreHelper;
import org.apache.hadoop.hive.ql.txn.compactor.handler.TaskHandler;
import org.apache.hadoop.hive.ql.txn.compactor.handler.TaskHandlerFactory;
import org.apache.hive.common.util.ReflectionUtil;
+import org.junit.Assert;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Test;
@@ -49,6 +53,7 @@
import java.util.List;
import java.util.Map;
import java.util.Optional;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
@@ -1195,4 +1200,171 @@ public void testCleanupOnConcurrentMinorCompactions()
throws Exception {
assertTrue(sawBase);
assertTrue(sawDelta);
}
-}
+
+ @Test
+ public void testCompactionHwmIsHonoredWithMinOpenWriteIdSet() throws
Exception {
+
+ String dbName = "default";
+ String tableName = "campcnb";
+
+ Table table = newTable(dbName, tableName, false);
+
+ // Create deltas and run 3 MAJOR compactions
+ String baseName1 = createDeltasAndRunMajorCompaction(table, 1L, 3);
+ String baseName2 = createDeltasAndRunMajorCompaction(table, 4L, 3);
+ String baseName3 = createDeltasAndRunMajorCompaction(table, 7L, 3);
+
+ // Before the cleaner starts, the house keeper service cleans up the
TXN_TO_WRITE_ID table
+ // The purpose of this is to hit the code path in
GetValidWriteIdsForTableFunction
+ // where there is no writeIds allocated by txns under txnHwm, so the
writeHwm will be get from NEXT_WRITE_ID.
+ txnHandler.cleanTxnToWriteIdTable();
+
+ // Then a new insert is started on the table and it will be open when the
cleaner starts.
+ // With this open transaction, the minOpenWriteId will be set in the
GetValidWriteIdsForTableFunction
+ long txnId = openTxn(TxnType.DEFAULT);
+ long writeId = allocateWriteId(dbName, tableName, txnId);
+ addDeltaFile(table, null, writeId, writeId, 1);
+
+ Set<String> expectedDirs = new HashSet<>();
+ expectedDirs.add(baseName1);
+ expectedDirs.add(baseName2);
+ expectedDirs.add(baseName3);
+ for (int i = 1; i < 11; i++) {
+ expectedDirs.add(makeDeltaDirName(i, i));
+ }
+
+ // This cleaner picks up the first compaction, and it should delete the
directories made obsolete by the first
+ // compaction. It should not delete anything above the first compactions's
highWaterMark.
+ // It means, it should delete delta_0000001_0000001, delta_0000002_0000002
and delta_0000003_0000003 only.
+ startCleaner();
+ for (int i = 1; i < 4; i++) {
+ expectedDirs.remove(makeDeltaDirName(i, i));
+ }
+ verifyDirectories(table, expectedDirs);
+
+ // Commit the open transaction, so the cleaner would run after that
+ txnHandler.commitTxn(new CommitTxnRequest(txnId));
+
+ // This cleaner picks up the second compaction, and it should delete
base_3_v0000004, delta_0000004_0000004,
+ // delta_0000005_0000005 and delta_0000006_0000006.
+ startCleaner();
+ for (int i = 4; i < 7; i++) {
+ expectedDirs.remove(makeDeltaDirName(i, i));
+ }
+ expectedDirs.remove(baseName1);
+ verifyDirectories(table, expectedDirs);
+
+ // This cleaner picks up the third compaction, and it should delete
base_6_v0000008,
+ // .delta_0000007_0000007, delta_0000008_0000008 and delta_0000009_0000009
+ startCleaner();
+ for (int i = 7; i < 10; i++) {
+ expectedDirs.remove(makeDeltaDirName(i, i));
+ }
+ expectedDirs.remove(baseName2);
+ verifyDirectories(table, expectedDirs);
+
+ ShowCompactResponse scr = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(3, scr.getCompactsSize());
+ for (ShowCompactResponseElement scre: scr.getCompacts()) {
+ Assert.assertEquals("succeeded", scre.getState());
+ }
+ }
+
+ @Test
+ public void testCompactionHwmIsHonoredWithMinOpenWriteIdSetAndAbortedIOW()
throws Exception {
+ String dbName = "default";
+ String tableName = "campcnb";
+
+ Table table = newTable(dbName, tableName, false);
+
+ // Create deltas and run 2 MAJOR compactions
+ String baseName1 = createDeltasAndRunMajorCompaction(table, 1L, 3);
+ String baseName2 = createDeltasAndRunMajorCompaction(table, 4L, 3);
+
+ // Create an aborted insert overwrite, this will create a base directory.
+ long txnId1 = openTxn(TxnType.DEFAULT);
+ long writeId1 = allocateWriteId(dbName, tableName, txnId1);
+ addBaseFile(table, null, writeId1, 1, 0);
+ txnHandler.abortTxn(new AbortTxnRequest(txnId1));
+
+ // Before the cleaner starts, the house keeper service cleans up the
TXN_TO_WRITE_ID table
+ // The purpose of this is to hit the code path in
GetValidWriteIdsForTableFunction
+ // where there is no writeIds allocated by txns under txnHwm, so the
writeHwm will be fetched from NEXT_WRITE_ID.
+ txnHandler.cleanTxnToWriteIdTable();
+
+ // Then a new insert is started on the table, and it will be open when the
cleaner starts.
+ // With this open transaction, the minOpenWriteId will be set in the
GetValidWriteIdsForTableFunction
+ long txnId = openTxn(TxnType.DEFAULT);
+ long writeId = allocateWriteId(dbName, tableName, txnId);
+ addDeltaFile(table, null, writeId, writeId, 1);
+
+ Set<String> expectedDirs = new HashSet<>();
+ expectedDirs.add(baseName1);
+ expectedDirs.add(baseName2);
+ for (int i = 1; i < 7; i++) {
+ expectedDirs.add(makeDeltaDirName(i, i));
+ }
+ expectedDirs.add("base_7");
+ expectedDirs.add(makeDeltaDirName(8, 8));
+ verifyDirectories(table, expectedDirs);
+
+ // This cleaner picks up the first compaction, and it should delete the
directories made obsolete by the first
+ // compaction. It should not delete anything above the first compactions's
highWaterMark.
+ // It means, it should delete delta_0000001_0000001, delta_0000002_0000002
and delta_0000003_0000003 only.
+ startCleaner();
+ for (int i = 1; i < 4; i++) {
+ expectedDirs.remove(makeDeltaDirName(i, i));
+ }
+ verifyDirectories(table, expectedDirs);
+
+ // Commit the open transaction, so the cleaner would run after that
+ txnHandler.commitTxn(new CommitTxnRequest(txnId));
+
+ // This cleaner picks up the second compaction, and it should delete
base_3_v0000004, delta_0000004_0000004,
+ // delta_0000005_0000005 and delta_0000006_0000006.
+ // base_6_v0000008 is the latest valid base
+ // base_7 is the directory of the aborted insert overwrite. This shouldn't
been deleted by this cleaner
+ // because its writeId (7) > cleaner's watermark (6)
+ // delta_0000008_0000008 is created after the second compaction
+ startCleaner();
+ for (int i = 4; i < 7; i++) {
+ expectedDirs.remove(makeDeltaDirName(i, i));
+ }
+ expectedDirs.remove(baseName1);
+ verifyDirectories(table, expectedDirs);
+
+ createDeltasAndRunMajorCompaction(table, 9L, 2);
+ // This cleaner picks up the third compaction and it should delete base_7
as well
+ startCleaner();
+ expectedDirs.clear();
+ expectedDirs.add("base_10_v0000013");
+ verifyDirectories(table, expectedDirs);
+
+ ShowCompactResponse scr = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(3, scr.getCompactsSize());
+ for (ShowCompactResponseElement scre: scr.getCompacts()) {
+ Assert.assertEquals("succeeded", scre.getState());
+ }
+ }
+
+ private String createDeltasAndRunMajorCompaction(Table table, long minTxnId,
int numberOfDeltas) throws Exception {
+ String dbName = table.getDbName();
+ String tableName = table.getTableName();
+ for (int i = 0; i < numberOfDeltas; i++) {
+ addDeltaFile(table, null, minTxnId + i, minTxnId + i, 1);
+ }
+ burnThroughTransactions(dbName, tableName, numberOfDeltas, null, null);
+ CompactionRequest rqst = new CompactionRequest(dbName, tableName,
CompactionType.MAJOR);
+ long compactTxn = compactInTxn(rqst);
+ long maxTxnId = minTxnId + numberOfDeltas - 1;
+ addBaseFile(table, null, maxTxnId, (int)maxTxnId, compactTxn);
+ return AcidUtils.addVisibilitySuffix(AcidUtils.BASE_PREFIX + maxTxnId,
compactTxn);
+ }
+
+ private void verifyDirectories(Table table, Set<String> expectedDirs) throws
Exception {
+ List<Path> paths = getDirectories(conf, table, null);
+ Set<String> actualDirs =
paths.stream().map(Path::getName).collect(Collectors.toSet());
+ Assert.assertEquals(expectedDirs, actualDirs);
+ }
+
+}
\ No newline at end of file
diff --git
a/storage-api/src/java/org/apache/hadoop/hive/common/ValidCleanerWriteIdList.java
b/storage-api/src/java/org/apache/hadoop/hive/common/ValidCleanerWriteIdList.java
index f58fe57a9f2..d4d85f467d9 100644
---
a/storage-api/src/java/org/apache/hadoop/hive/common/ValidCleanerWriteIdList.java
+++
b/storage-api/src/java/org/apache/hadoop/hive/common/ValidCleanerWriteIdList.java
@@ -49,4 +49,17 @@ public RangeResponse isWriteIdRangeValid(long minWriteId,
long maxWriteId) {
}
return super.isWriteIdRangeValid(minWriteId, maxWriteId);
}
+
+ /**
+ * Only consider the writeIds below the highWaterMark. This is to prevent
the cleaner
+ * from deleting above its highWaterMark, even in case of aborted
directories.
+ * otherwise uses {@link ValidReaderWriteIdList#isWriteIdAborted(long)}
+ */
+ @Override
+ public boolean isWriteIdAborted(long writeId) {
+ if (writeId > highWatermark) {
+ return false;
+ }
+ return super.isWriteIdAborted(writeId);
+ }
}
diff --git
a/storage-api/src/java/org/apache/hadoop/hive/common/ValidReaderWriteIdList.java
b/storage-api/src/java/org/apache/hadoop/hive/common/ValidReaderWriteIdList.java
index af32ad3308b..56a5353d0d1 100644
---
a/storage-api/src/java/org/apache/hadoop/hive/common/ValidReaderWriteIdList.java
+++
b/storage-api/src/java/org/apache/hadoop/hive/common/ValidReaderWriteIdList.java
@@ -267,6 +267,7 @@ public RangeResponse isWriteIdRangeAborted(long minWriteId,
long maxWriteId) {
public void setHighWatermark(long value) {
this.highWatermark = value;
+ this.minOpenWriteId = Math.min(minOpenWriteId, value + 1);
}
}
