This is an automated email from the ASF dual-hosted git repository.
dkuzmenko 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 a1b2d5717cd HIVE-29567: Data loss in query-based major compaction
(#6431)
a1b2d5717cd is described below
commit a1b2d5717cd03c5880dc0396407e3166e4c703ae
Author: Denys Kuzmenko <[email protected]>
AuthorDate: Wed Apr 15 20:31:32 2026 +0300
HIVE-29567: Data loss in query-based major compaction (#6431)
---
.../org/apache/hadoop/hive/ql/io/AcidUtils.java | 4 ++++
.../compactor/service/AcidCompactionService.java | 11 +++++++----
.../hadoop/hive/metastore/txn/TxnCommonUtils.java | 18 +++++++++++++++---
.../apache/hadoop/hive/metastore/txn/TxnUtils.java | 22 ++++++++++++++++++----
4 files changed, 44 insertions(+), 11 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 c0e3b25001f..909b22cf8a4 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
@@ -1558,6 +1558,10 @@ public static Map<Path, HdfsDirSnapshot>
getHdfsDirSnapshots(final FileSystem fs
FileStatus fStatus = itr.next();
Path fPath = fStatus.getPath();
if (fStatus.isDirectory()) {
+ if (baseFileFilter.accept(fPath) || deltaFileFilter.accept(fPath)
+ || deleteEventDeltaDirFilter.accept(fPath)) {
+ addToSnapshot(dirToSnapshots, fPath);
+ }
stack.push(FileUtils.listStatusIterator(fs, fPath,
acidHiddenFileFilter));
} else {
Path parentDirPath = fPath.getParent();
diff --git
a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/service/AcidCompactionService.java
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/service/AcidCompactionService.java
index ab1581c836e..3538b07c169 100644
---
a/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/service/AcidCompactionService.java
+++
b/ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/service/AcidCompactionService.java
@@ -57,7 +57,7 @@
import java.io.IOException;
import java.security.PrivilegedExceptionAction;
-import java.util.Collections;
+import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
@@ -184,10 +184,13 @@ public Boolean compact(Table table, CompactionInfo ci)
throws Exception {
* multi-stmt txn. {@link Driver#setCompactionWriteIds(ValidWriteIdList,
long)} */
compactionTxn.open(ci);
- final ValidTxnList validTxnList =
msc.getValidTxns(compactionTxn.getTxnId());
+ final ValidTxnList validTxnList =
+ TxnUtils.createValidTxnListForCompactor(msc.getOpenTxns(),
compactionTxn.getTxnId());
//with this ValidWriteIdList is capped at whatever HWM validTxnList has
- tblValidWriteIds =
TxnUtils.createValidCompactWriteIdList(msc.getValidWriteIds(
- Collections.singletonList(fullTableName),
validTxnList.writeToString()).get(0));
+ tblValidWriteIds =
+ TxnUtils.createValidCompactWriteIdList(
+ msc.getValidWriteIds(List.of(fullTableName),
validTxnList.writeToString()).get(0)
+ );
LOG.debug("ValidCompactWriteIdList: " +
tblValidWriteIds.writeToString());
conf.set(ValidTxnList.VALID_TXNS_KEY, validTxnList.writeToString());
diff --git
a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnCommonUtils.java
b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnCommonUtils.java
index 582b41c546a..8d875e81200 100644
---
a/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnCommonUtils.java
+++
b/standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnCommonUtils.java
@@ -41,6 +41,11 @@ public class TxnCommonUtils {
* @return a valid txn list.
*/
public static ValidTxnList createValidReadTxnList(GetOpenTxnsResponse txns,
long currentTxn) {
+ return createValidReadTxnList(txns, currentTxn, true);
+ }
+
+ static ValidTxnList createValidReadTxnList(GetOpenTxnsResponse txns, long
currentTxn,
+ boolean excludeCurrentTxn) {
assert currentTxn <= txns.getTxn_high_water_mark();
/*
* The highWaterMark should be
min(currentTxn,txns.getTxn_high_water_mark()) assuming currentTxn>0
@@ -61,7 +66,13 @@ public static ValidTxnList
createValidReadTxnList(GetOpenTxnsResponse txns, long
// txn is read-only or aborted by AcidHouseKeeperService and compactor
actually cleans up the aborted txns.
// So, for such cases, we get negative value for sizeToHwm with found
position for currentTxn, and so,
// we just negate it to get the size.
- int sizeToHwm = (currentTxn > 0) ?
Math.abs(Collections.binarySearch(openTxns, currentTxn)) : openTxns.size();
+ int sizeToHwm;
+ if (currentTxn > 0) {
+ int pos = Collections.binarySearch(openTxns, currentTxn);
+ sizeToHwm = (pos >= 0 && !excludeCurrentTxn) ? pos + 1 : Math.abs(pos);
+ } else {
+ sizeToHwm = openTxns.size();
+ }
sizeToHwm = Math.min(sizeToHwm, openTxns.size());
long[] exceptions = new long[sizeToHwm];
BitSet inAbortedBits = BitSet.valueOf(txns.getAbortedBits());
@@ -70,8 +81,9 @@ public static ValidTxnList
createValidReadTxnList(GetOpenTxnsResponse txns, long
int i = 0;
for (long txn : openTxns) {
// For snapshot isolation, we don't care about txns greater than current
txn and so stop here.
- // Also, we need not include current txn to exceptions list.
- if ((currentTxn > 0) && (txn >= currentTxn)) {
+ // When excludeCurrentTxn is true, we also exclude current txn from the
exceptions list
+ // (own-txn exclusion for regular reads). When false, we include it
(compaction worker).
+ if ((currentTxn > 0) && (excludeCurrentTxn ? txn >= currentTxn : txn >
currentTxn)) {
break;
}
if (inAbortedBits.get(i)) {
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java
index cd56311abc0..525d6654a27 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnUtils.java
@@ -67,10 +67,12 @@ public class TxnUtils {
private static final Logger LOG = LoggerFactory.getLogger(TxnUtils.class);
/**
- * Returns a valid txn list for cleaner.
- * @param txns Response containing open txns list.
- * @param minOpenTxn Minimum open txn which is min open write txn on the
table in the case of abort cleanup.
- * @param isAbortCleanup Whether the request is for abort cleanup.
+ * Returns a valid txn list for the cleaner. The high watermark is set to
{@code minOpenTxn - 1}
+ * so the cleaner only sees transactions that committed before the oldest
open write txn on the table.
+ *
+ * @param txns response containing the currently open txns list
+ * @param minOpenTxn minimum open write txn id on the table being cleaned
+ * @param isAbortCleanup whether the request is for abort cleanup
* @return a valid txn list
*/
public static ValidTxnList createValidTxnListForCleaner(GetOpenTxnsResponse
txns, long minOpenTxn, boolean isAbortCleanup) {
@@ -106,6 +108,18 @@ public static ValidTxnList
createValidTxnListForCleaner(GetOpenTxnsResponse txns
}
}
+ /**
+ * Returns a valid txn list for the compaction worker. The high watermark is
capped at {@code compactionTxnId}
+ * and the compaction transaction itself is kept in the exceptions list as
OPEN.
+ *
+ * @param txns open txns response from the metastore
+ * @param compactionTxnId the transaction ID of the compaction
+ * @return a valid txn list
+ */
+ public static ValidTxnList
createValidTxnListForCompactor(GetOpenTxnsResponse txns, long compactionTxnId) {
+ return TxnCommonUtils.createValidReadTxnList(txns, compactionTxnId, false);
+ }
+
/**
* Transform a {@link
org.apache.hadoop.hive.metastore.api.TableValidWriteIds} to a
* {@link org.apache.hadoop.hive.common.ValidCompactorWriteIdList}. This
assumes that the caller intends to
