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 b4517e1a991 HIVE-30026: (ACID Compaction) Extend the query for finding
potential compactions with ordering (#6768)
b4517e1a991 is described below
commit b4517e1a99160f4ed6cc3a8920f05413d096f926
Author: Marta Kuczora <[email protected]>
AuthorDate: Mon Sep 21 09:36:39 2026 +0200
HIVE-30026: (ACID Compaction) Extend the query for finding potential
compactions with ordering (#6768)
---
.../metastore/txn/TestCompactionTxnHandler.java | 80 ++++++++++++++
.../hive/ql/txn/compactor/TestInitiator.java | 117 +++++++++++++++++++++
.../FindPotentialCompactionsFunction.java | 7 +-
.../txn/jdbc/queries/AbortedTxnHandler.java | 3 +-
.../jdbc/queries/CompactionCandidateHandler.java | 8 +-
5 files changed, 208 insertions(+), 7 deletions(-)
diff --git
a/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestCompactionTxnHandler.java
b/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestCompactionTxnHandler.java
index 52ebac46c79..2693914f7e3 100644
---
a/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestCompactionTxnHandler.java
+++
b/ql/src/test/org/apache/hadoop/hive/metastore/txn/TestCompactionTxnHandler.java
@@ -998,6 +998,86 @@ public void testFindPotentialCompactions_limitFetchSize()
throws Exception {
assertEquals(1, potentials.size());
}
+ /**
+ * If the number of candidates is bigger than the compactor.fetch.size
value, it can happen that the
+ * Initiator doesn't schedule all tables for compaction.
+ * For example: we have 4 tables, which are all eligible for compaction, but
the fetch size is 2.
+ * In the first initiator run, it will pick two of these tables. It would be
expected that during the second
+ * run of the initiator it would pick the other two tables, so all 4 tables
have been scheduled for compaction.
+ * Since there was no ordering when selecting the tables for compaction,
this was not always the case.
+ * It could happen that some tables never got compacted.
+ *
+ * Changing the initiator to pick the tables ordered by CTC_TXNID, so the
tables with the oldest TXNID would be
+ * picked first. This way we can be sure that all tables will be scheduled
for compaction at some point.
+ */
+ @Test
+ public void testFindPotentialCompactionsFairness() throws Exception {
+ MetastoreConf.setLongVar(conf, COMPACTOR_FETCH_SIZE, 3);
+
+ commitOrAbortOneTableUpdate("table_c", false);
+ commitOrAbortOneTableUpdate("table_a", false);
+ commitOrAbortOneTableUpdate("table_d", false);
+ commitOrAbortOneTableUpdate("table_b", false); // newest — must be dropped
by the LIMIT
+
+ Set<CompactionInfo> potentials = txnHandler.findPotentialCompactions(100,
-1L);
+ assertEquals(3, potentials.size());
+
+ Set<String> names = potentials.stream().map(ci ->
ci.tableName).collect(Collectors.toSet());
+ assertTrue("oldest (table_c) must be picked", names.contains("table_c"));
+ assertTrue("second-oldest (table_a) must be picked",
names.contains("table_a"));
+ assertTrue("third-oldest (table_d) must be picked",
names.contains("table_d"));
+ assertFalse("newest (table_b) must be excluded by the fetch.size window",
+ names.contains("table_b"));
+ }
+
+ /**
+ * If the number of candidates is bigger than the compactor.fetch.size
limit, the aborted
+ * txn candidates will never be picked up by the initiator. Because the
"remaining budget" will always
+ * be 0 in the FindPotentialCompactionsFunction. This affects only the cases
when the aborted txn clean-up
+ * is done by the compaction cycle and not just the cleaner.
+ *
+ * Changing how the compaction candidates are fetched to be sure that the
aborted txn are also picked up.
+ */
+ @Test
+ public void
testFindPotentialCompactionsIncludesAbortsUnderFetchSizePressure() throws
Exception {
+ // Route aborted-txn cleanup through the compaction path
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.COMPACTOR_CLEAN_ABORTS_USING_CLEANER, false);
+ MetastoreConf.setLongVar(conf, COMPACTOR_FETCH_SIZE, 2);
+
+ // Two committed candidates — fills the entire fetch.size budget.
+ commitOrAbortOneTableUpdate("committed_a", false);
+ commitOrAbortOneTableUpdate("committed_b", false);
+
+ // One aborted candidate with enough aborts to exceed abortedThreshold=0.
+ commitOrAbortOneTableUpdate("aborted_only", true);
+
+ Set<CompactionInfo> potentials = txnHandler.findPotentialCompactions(0,
-1L);
+
+ Set<String> names = potentials.stream().map(ci ->
ci.tableName).collect(Collectors.toSet());
+ assertTrue("committed candidates must be picked",
names.contains("committed_a")
+ && names.contains("committed_b"));
+ assertTrue("aborted candidate must not be starved by fetch.size pressure",
+ names.contains("aborted_only"));
+ }
+
+ /**
+ * Opens one txn, takes a SHARED_WRITE lock on the (db,table), and commits
or aborts.
+ * This produces exactly one COMPLETED_TXN_COMPONENTS row.
+ */
+ private void commitOrAbortOneTableUpdate(String table, boolean abort) throws
Exception {
+ long txnId = openTxn();
+ LockComponent lc = createLockComponent(
+ LockType.SHARED_WRITE, LockLevel.DB, "default", table, null,
DataOperationType.UPDATE);
+ LockRequest req = new LockRequest(Collections.singletonList(lc), "me",
"localhost");
+ req.setTxnid(txnId);
+ assertSame(LockState.ACQUIRED, txnHandler.lock(req).getState());
+ if (abort) {
+ txnHandler.abortTxn(new AbortTxnRequest(txnId));
+ } else {
+ txnHandler.commitTxn(new CommitTxnRequest(txnId));
+ }
+ }
+
@Test
public void testFindNextToClean_limitFetchSize() throws Exception {
MetastoreConf.setLongVar(conf, COMPACTOR_FETCH_SIZE, 1);
diff --git
a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestInitiator.java
b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestInitiator.java
index 53b9d0881f2..b34b2c96da3 100644
--- a/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestInitiator.java
+++ b/ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestInitiator.java
@@ -49,11 +49,15 @@
import java.util.Arrays;
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 java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.stream.Collectors;
+import static org.junit.Assert.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doReturn;
import static org.mockito.Mockito.doThrow;
@@ -1114,6 +1118,119 @@ public void testMetaCache() throws Exception {
Mockito.verify(initiator, times(2)).resolvePartition(Mockito.any());
}
+ @Test
+ public void testCompactionInitWithSmallFetchSize() throws Exception {
+ MetastoreConf.setLongVar(conf,
MetastoreConf.ConfVars.COMPACTOR_FETCH_SIZE, 2);
+ Set<String> compactedTables = new HashSet<>();
+ Map<String, Table> tables = new HashMap<>();
+ String[] tableNames = {"testtablea", "testtableb", "testtablec",
"testtabled"};
+ for (int i = 0; i < 4; i++) {
+ Table t = newTable("default", tableNames[i], false);
+ tables.put(t.getTableName(), t);
+ prepareTable(t, true, 1L);
+ }
+
+ startInitiator();
+ startWorker();
+ startWorker();
+ startCleaner();
+ startCleaner();
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ compactedTables.addAll(rsp.getCompacts().stream().map(c ->
c.getTablename()).collect(Collectors.toSet()));
+
+ for (String tableName: compactedTables) {
+ prepareTable(tables.get(tableName), false, 5L);
+ }
+
+ startInitiator();
+ startWorker();
+ startWorker();
+ startCleaner();
+ startCleaner();
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ compactedTables.addAll(rsp.getCompacts().stream().map(c ->
c.getTablename()).collect(Collectors.toSet()));
+
+ Assert.assertEquals(4, compactedTables.size());
+ for (String tableName: tableNames) {
+ assertTrue(tableName + " isn't scheduled for compaction",
compactedTables.contains(tableName));
+ }
+ }
+
+ @Test
+ public void testCompactionInitWithSmallFetchSizeWithAbortedTxn() throws
Exception{
+ MetastoreConf.setLongVar(conf,
MetastoreConf.ConfVars.COMPACTOR_FETCH_SIZE, 2);
+ MetastoreConf.setBoolVar(conf,
MetastoreConf.ConfVars.COMPACTOR_CLEAN_ABORTS_USING_CLEANER, false);
+ HiveConf.setIntVar(conf,
HiveConf.ConfVars.HIVE_COMPACTOR_ABORTEDTXN_THRESHOLD, 2);
+
+ Set<String> compactedTables = new HashSet<>();
+ String[] tableNames = {"testtablea", "testtableb", "testtablec",
"testtabled"};
+ List<Table> tables = new ArrayList<>();
+ for (int i = 0; i < 2; i++) {
+ Table t = newTable("default", tableNames[i], false);
+ tables.add(t);
+ prepareTable(t, true, 1L);
+ }
+ long txnId = 10L;
+ for (int i = 2; i < 4; i++) {
+ Table t = newTable("default", tableNames[i], false);
+ tables.add(t);
+ Set<Long> abortedTxns = new HashSet<>();
+ for (int j = 1; j < 6; j++) {
+ abortedTxns.add(txnId + j);
+ }
+ prepareTable(t, true, 1L, abortedTxns);
+ txnId = 15L;
+ }
+
+ startInitiator();
+ startWorker();
+ startWorker();
+ startCleaner();
+ startCleaner();
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ compactedTables.addAll(rsp.getCompacts().stream().map(c ->
c.getTablename()).collect(Collectors.toSet()));
+
+ for (int i = 0; i < 2; i++) {
+ prepareTable(tables.get(i), false, 5L);
+ }
+
+ startInitiator();
+ startWorker();
+ startWorker();
+ startCleaner();
+ startCleaner();
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ compactedTables.addAll(rsp.getCompacts().stream().map(c ->
c.getTablename()).collect(Collectors.toSet()));
+
+ Assert.assertEquals(4, compactedTables.size());
+ for (String tableName: tableNames) {
+ assertTrue(tableName + " isn't scheduled for compaction",
compactedTables.contains(tableName));
+ }
+ }
+
+ private void prepareTable(Table table, boolean addBase, long writeId) throws
Exception {
+ prepareTable(table, addBase, writeId, null);
+ }
+
+ private void prepareTable(Table table, boolean addBase, long writeId,
Set<Long> abortedTxns) throws Exception {
+ int txnNum = 0;
+ if (addBase) {
+ addBaseFile(table, null, writeId, 20);
+ txnNum++;
+ }
+ for (int i = 1; i < 5; i++) {
+ addDeltaFile(table, null, writeId + i, writeId + i, 2);
+ txnNum++;
+ }
+ LockComponent comp = new LockComponent(LockType.SHARED_WRITE,
LockLevel.TABLE, table.getDbName());
+ comp.setTablename(table.getTableName());
+ comp.setOperationType(DataOperationType.INSERT);
+ List<LockComponent> components = new ArrayList<LockComponent>(1);
+ components.add(comp);
+ LockRequest req = new LockRequest(components, "me", "localhost");
+ burnThroughTransactions(table.getDbName(), table.getTableName(), txnNum,
null, abortedTxns, req);
+ }
+
private static FindNextCompactRequest aFindNextCompactRequest(String
workerId, String workerVersion) {
FindNextCompactRequest request = new FindNextCompactRequest();
request.setWorkerId(workerId);
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/FindPotentialCompactionsFunction.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/FindPotentialCompactionsFunction.java
index c24ea97efdb..86b953b27d5 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/FindPotentialCompactionsFunction.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/functions/FindPotentialCompactionsFunction.java
@@ -48,11 +48,12 @@ public FindPotentialCompactionsFunction(Configuration conf,
int abortedThreshold
@Override
public Set<CompactionInfo> execute(MultiDataSourceJdbcResource jdbcResource)
throws MetaException {
+ // Each source gets its own fetchSize budget instead of sharing a single
pool.
+ // Sharing the pool starved aborted-txn cleanup whenever there were >=
fetchSize committed candidates.
Set<CompactionInfo> candidates = new HashSet<>(jdbcResource.execute(
new CompactionCandidateHandler(lastChecked, fetchSize)));
- int remaining = fetchSize - candidates.size();
- if (collectAbortedTxns && remaining > 0) {
- candidates.addAll(jdbcResource.execute(new
AbortedTxnHandler(abortedTimeThreshold, abortedThreshold, remaining)));
+ if (collectAbortedTxns) {
+ candidates.addAll(jdbcResource.execute(new
AbortedTxnHandler(abortedTimeThreshold, abortedThreshold, fetchSize)));
}
return candidates;
}
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/queries/AbortedTxnHandler.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/queries/AbortedTxnHandler.java
index 861c6998aad..a8187052466 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/queries/AbortedTxnHandler.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/queries/AbortedTxnHandler.java
@@ -47,7 +47,8 @@ public String getParameterizedQueryString(DatabaseProduct
databaseProduct) throw
return databaseProduct.addLimitClause(fetchSize, " \"TC_DATABASE\",
\"TC_TABLE\", \"TC_PARTITION\", " +
"MIN(\"TXN_STARTED\"), COUNT(*) FROM \"TXNS\", \"TXN_COMPONENTS\" " +
" WHERE \"TXN_ID\" = \"TC_TXNID\" AND \"TXN_STATE\" = :state " +
- "GROUP BY \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\" ");
+ "GROUP BY \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\" " +
+ "ORDER BY MIN(\"TXN_STARTED\") ASC ");
}
@Override
diff --git
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/queries/CompactionCandidateHandler.java
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/queries/CompactionCandidateHandler.java
index eb290167799..4561ea523b9 100644
---
a/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/queries/CompactionCandidateHandler.java
+++
b/standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/jdbc/queries/CompactionCandidateHandler.java
@@ -43,8 +43,8 @@ public class CompactionCandidateHandler implements
QueryHandler<Set<CompactionIn
//language=SQL
@Override
public String getParameterizedQueryString(DatabaseProduct databaseProduct)
throws MetaException {
- return databaseProduct.addLimitClause(fetchSize,
- "DISTINCT \"TC\".\"CTC_DATABASE\", \"TC\".\"CTC_TABLE\",
\"TC\".\"CTC_PARTITION\" " +
+ return databaseProduct.addLimitClause(fetchSize,
+ "\"TC\".\"CTC_DATABASE\", \"TC\".\"CTC_TABLE\",
\"TC\".\"CTC_PARTITION\" " +
"FROM \"COMPLETED_TXN_COMPONENTS\" \"TC\" " + (checkInterval > 0 ?
"LEFT JOIN ( " +
" SELECT \"C1\".* FROM \"COMPLETED_COMPACTIONS\" \"C1\" " +
@@ -57,7 +57,9 @@ public String getParameterizedQueryString(DatabaseProduct
databaseProduct) throw
") \"C\" " +
"ON \"TC\".\"CTC_DATABASE\" = \"C\".\"CC_DATABASE\" AND
\"TC\".\"CTC_TABLE\" = \"C\".\"CC_TABLE\" " +
" AND (\"TC\".\"CTC_PARTITION\" = \"C\".\"CC_PARTITION\" OR
(\"TC\".\"CTC_PARTITION\" IS NULL AND \"C\".\"CC_PARTITION\" IS NULL)) " +
- "WHERE \"C\".\"CC_ID\" IS NOT NULL OR " +
databaseProduct.isWithinCheckInterval("\"TC\".\"CTC_TIMESTAMP\"",
checkInterval) : ""));
+ "WHERE \"C\".\"CC_ID\" IS NOT NULL OR " +
databaseProduct.isWithinCheckInterval("\"TC\".\"CTC_TIMESTAMP\"",
checkInterval) : "") +
+ " GROUP BY \"TC\".\"CTC_DATABASE\", \"TC\".\"CTC_TABLE\",
\"TC\".\"CTC_PARTITION\" " +
+ " ORDER BY MIN(\"TC\".\"CTC_TXNID\") ASC");
}
@Override
