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 909b22cf8a47..af93bd506301 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 bd6cacbde805..0a3f7534aa7e 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 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 30815442ff25..df73adf26ed3 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 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 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 expectedDirs) throws Exception { + List paths = getDirectories(conf, table, null); + Set 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 f58fe57a9f2f..d4d85f467d96 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 af32ad3308b6..56a5353d0d1e 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); } }