Skip to content

Commit 52b8b1a

Browse files
deniskuzZdeniskuzZ
authored
HIVE-29619: Hive ACID: Long-running queries on one table shouldn't block cleanup operations on other tables. (apache#6497)
1 parent a1c1a12 commit 52b8b1a

10 files changed

Lines changed: 147 additions & 44 deletions

File tree

‎itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818

1919
package org.apache.hadoop.hive.ql;
2020

21+
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_RETENTION_TIME;
2122
import static org.junit.Assert.assertEquals;
2223
import static org.junit.Assert.assertTrue;
2324

@@ -28,6 +29,7 @@
2829
import java.util.HashSet;
2930
import java.util.List;
3031
import java.util.Set;
32+
import java.util.concurrent.TimeUnit;
3133

3234
import org.apache.hadoop.conf.Configuration;
3335
import org.apache.hadoop.fs.FileStatus;
@@ -127,6 +129,8 @@ public void setUp() throws Exception {
127129
"org.apache.hadoop.hive.ql.security.authorization.plugin.sqlstd.SQLStdHiveAuthorizerFactory");
128130
MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON, true);
129131
MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON, true);
132+
HiveConf.setTimeVar(hiveConf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, 0, TimeUnit.SECONDS);
133+
130134
TestTxnDbUtil.setConfValues(hiveConf);
131135
hiveConf.setInt(MRJobConfig.MAP_MEMORY_MB, 1024);
132136
hiveConf.setInt(MRJobConfig.REDUCE_MEMORY_MB, 1024);

‎itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorOnTezTest.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@
5656
import java.util.regex.Matcher;
5757
import java.util.regex.Pattern;
5858

59+
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_RETENTION_TIME;
5960
import static org.apache.hadoop.hive.ql.txn.compactor.CompactorTestUtil.executeStatementOnDriverAndReturnResults;
6061
import static org.apache.hadoop.hive.ql.txn.compactor.TestCompactor.executeStatementOnDriver;
6162
import static org.apache.hadoop.hive.ql.txn.compactor.TestCompactor.dropTables;
@@ -106,6 +107,7 @@ protected void setupWithConf(HiveConfForTest hiveConf) throws Exception {
106107
MetastoreConf.setTimeVar(hiveConf, MetastoreConf.ConfVars.TXN_OPENTXN_TIMEOUT, 2, TimeUnit.SECONDS);
107108
MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON, true);
108109
MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON, true);
110+
HiveConf.setTimeVar(hiveConf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, 0, TimeUnit.SECONDS);
109111

110112
TestTxnDbUtil.setConfValues(hiveConf);
111113
TestTxnDbUtil.cleanDb(hiveConf);

‎itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactorBase.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -46,6 +46,8 @@
4646
import java.util.concurrent.TimeUnit;
4747
import java.util.concurrent.atomic.AtomicInteger;
4848

49+
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_RETENTION_TIME;
50+
4951
class TestCompactorBase {
5052

5153
private static final Logger LOG = LoggerFactory.getLogger(TestCompactorBase.class);
@@ -92,6 +94,7 @@ public void setup() throws Exception {
9294
MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON, true);
9395
MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON, true);
9496
MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.COMPACTOR_CLEAN_ABORTS_USING_CLEANER, true);
97+
HiveConf.setTimeVar(hiveConf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, 0, TimeUnit.SECONDS);
9598

9699
TestTxnDbUtil.setConfValues(hiveConf);
97100
TestTxnDbUtil.cleanDb(hiveConf);

‎ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/handler/CompactionCleaner.java‎

Lines changed: 3 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -36,6 +36,7 @@
3636
import org.apache.hadoop.hive.metastore.metrics.Metrics;
3737
import org.apache.hadoop.hive.metastore.metrics.MetricsConstants;
3838
import org.apache.hadoop.hive.metastore.metrics.PerfLogger;
39+
import org.apache.hadoop.hive.metastore.txn.TxnHandler;
3940
import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
4041
import org.apache.hadoop.hive.metastore.txn.TxnStore;
4142
import org.apache.hadoop.hive.metastore.txn.TxnUtils;
@@ -73,7 +74,8 @@ public CompactionCleaner(HiveConf conf, TxnStore txnHandler,
7374
@Override
7475
public List<Runnable> getTasks(HiveConf conf) throws MetaException {
7576
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
76-
long retentionTime = HiveConf.getBoolVar(conf, HIVE_COMPACTOR_DELAYED_CLEANUP_ENABLED)
77+
long retentionTime = (HiveConf.getBoolVar(conf, HIVE_COMPACTOR_DELAYED_CLEANUP_ENABLED)
78+
|| TxnHandler.ConfVars.useMinHistoryWriteId())
7779
? HiveConf.getTimeVar(conf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, TimeUnit.MILLISECONDS)
7880
: 0;
7981
List<CompactionInfo> readyToClean = txnHandler.findReadyToClean(minOpenTxnId, retentionTime);

‎ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/service/AcidCompactionService.java‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -198,6 +198,10 @@ public Boolean compact(Table table, CompactionInfo ci) throws Exception {
198198
txnWriteIds.addTableValidWriteIdList(tblValidWriteIds);
199199
conf.set(ValidTxnWriteIdList.VALID_TABLES_WRITEIDS_KEY, txnWriteIds.toString());
200200

201+
// Register in MIN_HISTORY_WRITE_ID so the per-table cleaner admission blocks while open.
202+
msc.addWriteIdsToMinHistory(compactionTxn.getTxnId(),
203+
Map.of(fullTableName, txnWriteIds.getMinOpenWriteId(fullTableName)));
204+
201205
ci.highestWriteId = tblValidWriteIds.getHighWatermark();
202206
//this writes TXN_COMPONENTS to ensure that if compactorTxnId fails, we keep metadata about
203207
//it until after any data written by it are physically removed

‎ql/src/test/org/apache/hadoop/hive/ql/TxnCommandsBaseForTests.java‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,7 @@
6666
import org.slf4j.LoggerFactory;
6767

6868
import static org.apache.commons.lang3.StringUtils.isNotBlank;
69+
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_RETENTION_TIME;
6970
import static org.apache.hadoop.hive.metastore.DatabaseProduct.determineDatabaseProduct;
7071
import static org.apache.hadoop.hive.metastore.txn.TxnUtils.getEpochFn;
7172

@@ -152,6 +153,7 @@ void setUpInternal() throws Exception {
152153
hiveConf.setBoolean("mapred.input.dir.recursive", true);
153154
MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON, true);
154155
MetastoreConf.setBoolVar(hiveConf, MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON, true);
156+
HiveConf.setTimeVar(hiveConf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, 0, TimeUnit.SECONDS);
155157

156158
TestTxnDbUtil.setConfValues(hiveConf);
157159
TestTxnDbUtil.prepDb(hiveConf);

‎ql/src/test/org/apache/hadoop/hive/ql/lockmgr/DbTxnManagerEndToEndTestBase.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,9 @@
3838
import org.junit.BeforeClass;
3939

4040
import java.io.File;
41+
import java.util.concurrent.TimeUnit;
42+
43+
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_RETENTION_TIME;
4144

4245
/**
4346
* Base class for "end-to-end" tests for DbTxnManager and simulate concurrent queries.
@@ -59,6 +62,8 @@ public DbTxnManagerEndToEndTestBase() {
5962
MetastoreConf.setVar(conf, MetastoreConf.ConfVars.WAREHOUSE, getWarehouseDir());
6063
MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.COMPACTOR_INITIATOR_ON, true);
6164
MetastoreConf.setBoolVar(conf, MetastoreConf.ConfVars.COMPACTOR_CLEANER_ON, true);
65+
HiveConf.setTimeVar(conf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, 0, TimeUnit.SECONDS);
66+
6267
TestTxnDbUtil.setConfValues(conf);
6368
}
6469

‎ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/CompactorTest.java‎

Lines changed: 26 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
import org.apache.hadoop.hive.common.ServerUtils;
2727
import org.apache.hadoop.hive.common.ValidCompactorWriteIdList;
2828
import org.apache.hadoop.hive.common.ValidTxnList;
29+
import org.apache.hadoop.hive.common.ValidTxnWriteIdList;
2930
import org.apache.hadoop.hive.common.ValidWriteIdList;
3031
import org.apache.hadoop.hive.conf.HiveConf;
3132
import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
@@ -62,6 +63,7 @@
6263
import org.apache.hadoop.hive.metastore.conf.MetastoreConf;
6364
import org.apache.hadoop.hive.metastore.conf.MetastoreConf.ConfVars;
6465
import org.apache.hadoop.hive.metastore.metrics.AcidMetricService;
66+
import org.apache.hadoop.hive.metastore.txn.TxnHandler;
6567
import org.apache.hadoop.hive.metastore.txn.entities.CompactionInfo;
6668
import org.apache.hadoop.hive.metastore.txn.TxnCommonUtils;
6769
import org.apache.hadoop.hive.metastore.utils.TestTxnDbUtil;
@@ -113,6 +115,8 @@
113115
import javax.management.MBeanServer;
114116
import javax.management.ObjectName;
115117

118+
import static org.apache.hadoop.hive.conf.HiveConf.ConfVars.HIVE_COMPACTOR_CLEANER_RETENTION_TIME;
119+
116120
/**
117121
* Super class for all of the compactor test modules.
118122
*/
@@ -143,8 +147,12 @@ protected final void setup(HiveConf conf) throws Exception {
143147
MetastoreConf.setBoolVar(conf, ConfVars.COMPACTOR_INITIATOR_ON, true);
144148
MetastoreConf.setBoolVar(conf, ConfVars.COMPACTOR_CLEANER_ON, true);
145149
MetastoreConf.setBoolVar(conf, ConfVars.TXN_USE_MIN_HISTORY_WRITE_ID, useMinHistoryWriteId());
150+
TxnHandler.ConfVars.setUseMinHistoryWriteId(useMinHistoryWriteId());
146151
MetastoreConf.setVar(conf, ConfVars.COMPACTOR_INITIATOR_TABLE_OPTIMIZERS,
147152
"org.apache.hadoop.hive.ql.txn.compactor.AcidTableOptimizer");
153+
if (useMinHistoryWriteId()) {
154+
HiveConf.setTimeVar(conf, HIVE_COMPACTOR_CLEANER_RETENTION_TIME, 0, TimeUnit.SECONDS);
155+
}
148156
// Set this config to true in the base class, there are extended test classes which set this config to false.
149157
MetastoreConf.setBoolVar(conf, ConfVars.COMPACTOR_CLEAN_ABORTS_USING_CLEANER, true);
150158
TestTxnDbUtil.setConfValues(conf);
@@ -173,6 +181,12 @@ protected void startCleaner() throws Exception {
173181
runOneLoopOfCompactorThread(CompactorThreadType.CLEANER);
174182
}
175183

184+
void startCleaner(long retentionTime, TimeUnit timeUnit) throws Exception {
185+
HiveConf.setTimeVar(conf,
186+
HIVE_COMPACTOR_CLEANER_RETENTION_TIME, retentionTime, timeUnit);
187+
startCleaner();
188+
}
189+
176190
protected void runAcidMetricService() {
177191
TestTxnDbUtil.setConfValues(conf);
178192
AcidMetricService t = new AcidMetricService();
@@ -388,7 +402,7 @@ protected void burnThroughTransactions(String dbName, String tblName, int num, S
388402
txnHandler.commitTxn(new CommitTxnRequest(tid));
389403
} else if (open.contains(tid) && useMinHistoryWriteId()){
390404
txnHandler.addWriteIdsToMinHistory(tid,
391-
Collections.singletonMap(dbName + "." + tblName, minOpenWriteId));
405+
Map.of(dbName + "." + tblName, minOpenWriteId));
392406
}
393407
}
394408
}
@@ -754,21 +768,28 @@ long compactInTxn(CompactionRequest rqst, CommitAction commitAction) throws Exce
754768
ci.runAs = rqst.getRunas() == null ? System.getProperty("user.name") : rqst.getRunas();
755769

756770
long compactorTxnId = openTxn(TxnType.COMPACTION);
771+
String fullTableName = ci.getFullTableName().toLowerCase();
757772

758773
// Need to create a valid writeIdList to set the highestWriteId in ci
759774
ValidTxnList validTxnList = TxnCommonUtils.createValidReadTxnList(
760-
txnHandler.getOpenTxns(Collections.singletonList(TxnType.READ_ONLY)), compactorTxnId);
775+
txnHandler.getOpenTxns(List.of(TxnType.READ_ONLY)), compactorTxnId);
761776

762-
GetValidWriteIdsRequest writeIdsRequest = new GetValidWriteIdsRequest(
763-
Collections.singletonList(
764-
ci.getFullTableName().toLowerCase()));
777+
GetValidWriteIdsRequest writeIdsRequest = new GetValidWriteIdsRequest(List.of(fullTableName));
765778
writeIdsRequest.setValidTxnList(validTxnList.writeToString());
766779

767780
// with this ValidWriteIdList is capped at whatever HWM validTxnList has
768781
ValidCompactorWriteIdList tblValidWriteIds = TxnUtils.createValidCompactWriteIdList(
769782
txnHandler.getValidWriteIds(writeIdsRequest).getTblValidWriteIds()
770783
.getFirst());
771784

785+
if (useMinHistoryWriteId()) {
786+
ValidTxnWriteIdList txnWriteIds = new ValidTxnWriteIdList(compactorTxnId);
787+
txnWriteIds.addTableValidWriteIdList(tblValidWriteIds);
788+
789+
txnHandler.addWriteIdsToMinHistory(compactorTxnId,
790+
Map.of(fullTableName, txnWriteIds.getMinOpenWriteId(fullTableName)));
791+
}
792+
772793
ci.highestWriteId = tblValidWriteIds.getHighWatermark();
773794
txnHandler.updateCompactorState(ci, compactorTxnId);
774795

‎ql/src/test/org/apache/hadoop/hive/ql/txn/compactor/TestCleanerWithMinHistoryWriteId.java‎

Lines changed: 86 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@
2020

2121
import org.apache.hadoop.conf.Configuration;
2222
import org.apache.hadoop.fs.Path;
23+
import org.apache.hadoop.hive.metastore.api.AbortTxnsRequest;
2324
import org.apache.hadoop.hive.metastore.api.CommitTxnRequest;
2425
import org.apache.hadoop.hive.metastore.api.CompactionRequest;
2526
import org.apache.hadoop.hive.metastore.api.CompactionType;
@@ -32,9 +33,12 @@
3233
import org.junit.jupiter.api.BeforeEach;
3334
import org.junit.jupiter.api.Test;
3435

36+
import java.util.Collections;
3537
import java.util.List;
38+
import java.util.Map;
39+
import java.util.concurrent.TimeUnit;
3640

37-
import static org.apache.hadoop.hive.metastore.txn.TxnStore.FAILED_RESPONSE;
41+
import static org.apache.hadoop.hive.metastore.txn.TxnStore.CLEANING_RESPONSE;
3842
import static org.apache.hadoop.hive.metastore.txn.TxnStore.SUCCEEDED_RESPONSE;
3943
import static org.apache.hadoop.hive.metastore.txn.TxnStore.WORKING_RESPONSE;
4044
import static org.apache.hadoop.hive.metastore.txn.TxnStore.INITIATED_STATE;
@@ -59,7 +63,7 @@ protected boolean useMinHistoryWriteId() {
5963

6064
@Test
6165
public void cleanupAfterAbortedAndRetriedMajorCompaction() throws Exception {
62-
Table t = prepareTestTable();
66+
Table t = prepareTestTable("camtc");
6367
CompactionRequest rqst = new CompactionRequest("default", "camtc", CompactionType.MAJOR);
6468
long compactTxn = compactInTxn(rqst, CommitAction.ABORT);
6569
addBaseFile(t, null, 25L, 25, compactTxn);
@@ -83,10 +87,10 @@ public void cleanupAfterAbortedAndRetriedMajorCompaction() throws Exception {
8387

8488
@Test
8589
public void cleanupAfterKilledAndRetriedMajorCompaction() throws Exception {
86-
Table t = prepareTestTable();
90+
Table t = prepareTestTable("camtc");
8791
CompactionRequest rqst = new CompactionRequest("default", "camtc", CompactionType.MAJOR);
88-
long compactTxn = compactInTxn(rqst, CommitAction.NONE);
89-
addBaseFile(t, null, 25L, 25, compactTxn);
92+
long compactTxn1 = compactInTxn(rqst, CommitAction.NONE);
93+
addBaseFile(t, null, 25L, 25, compactTxn1);
9094

9195
txnHandler.revokeTimedoutWorkers(1L);
9296
// an open txn should prevent the retry
@@ -96,20 +100,32 @@ public void cleanupAfterKilledAndRetriedMajorCompaction() throws Exception {
96100

97101
// force retry
98102
revokeTimedoutWorkers(conf);
99-
compactTxn = compactInTxn(rqst);
100-
addBaseFile(t, null, 25L, 25, compactTxn);
103+
long compactTxn2 = compactInTxn(rqst);
104+
addBaseFile(t, null, 25L, 25, compactTxn2);
101105

102106
startCleaner();
103107

104-
// Validate that the cleanup attempt has failed.
108+
// Validate that the cleanup attempt was skipped.
105109
rsp = txnHandler.showCompact(new ShowCompactRequest());
106110
assertEquals(1, rsp.getCompactsSize());
107-
assertEquals(FAILED_RESPONSE, rsp.getCompacts().getFirst().getState());
108-
assertEquals("txnid:26 is open and <= hwm: 27", rsp.getCompacts().getFirst().getErrorMessage());
111+
assertEquals(CLEANING_RESPONSE, rsp.getCompacts().getFirst().getState());
109112

110113
// Check that the files are not removed
111114
List<Path> paths = getDirectories(conf, t, null);
112115
assertEquals(6, paths.size());
116+
117+
// Abort the open compaction txn, so that the Cleaner can proceed.
118+
txnHandler.abortTxns(
119+
new AbortTxnsRequest(Collections.singletonList(compactTxn1)));
120+
startCleaner();
121+
122+
rsp = txnHandler.showCompact(new ShowCompactRequest());
123+
assertEquals(1, rsp.getCompactsSize());
124+
assertEquals(SUCCEEDED_RESPONSE, rsp.getCompacts().getFirst().getState());
125+
126+
// Check that the files are removed
127+
paths = getDirectories(conf, t, null);
128+
assertEquals(1, paths.size());
113129
}
114130

115131
private static void revokeTimedoutWorkers(Configuration conf) throws Exception {
@@ -121,39 +137,88 @@ private static void revokeTimedoutWorkers(Configuration conf) throws Exception {
121137
}
122138

123139
@Test
124-
public void cleanupAfterMajorCompactionWithQueryWaitingToLockTheSnapshot() throws Exception {
125-
Table t = prepareTestTable();
140+
public void cleanupAndDanglingOpenTxnOnSameTable() throws Exception {
141+
Table t = prepareTestTable("camtc");
126142
CompactionRequest rqst = new CompactionRequest("default", "camtc", CompactionType.MAJOR);
127143
long compactTxn = compactInTxn(rqst, CommitAction.MARK_COMPACTED);
128144
addBaseFile(t, null, 25L, 25, compactTxn);
129145

130-
// Open a query during compaction,
146+
// Open a readerTxn during compaction,
131147
// Do not register minOpenWriteId (i.e. simulate delay locking the snapshot)
132-
openTxn();
148+
long readerTxn = openTxn();
133149

134150
txnHandler.commitTxn(new CommitTxnRequest(compactTxn));
135-
startCleaner();
151+
Thread.sleep(MetastoreConf.getTimeVar(
152+
conf, ConfVars.TXN_OPENTXN_TIMEOUT, TimeUnit.MILLISECONDS));
136153

137-
// Validate that the cleanup attempt has failed.
154+
startCleaner(10, TimeUnit.SECONDS);
155+
156+
// Validate that the cleanup attempt was delayed by retention time.
138157
ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
139158
assertEquals(1, rsp.getCompactsSize());
140-
assertEquals(FAILED_RESPONSE, rsp.getCompacts().getFirst().getState());
141-
assertEquals("txnid:27 is open and <= hwm: 27", rsp.getCompacts().getFirst().getErrorMessage());
159+
assertEquals(CLEANING_RESPONSE, rsp.getCompacts().getFirst().getState());
142160

143161
// Check that the files are not removed
144162
List<Path> paths = getDirectories(conf, t, null);
145163
assertEquals(5, paths.size());
164+
165+
// Register minOpenWriteId for the readerTxn.
166+
txnHandler.addWriteIdsToMinHistory(readerTxn, Map.of("default.camtc", 1L));
167+
168+
startCleaner(0, TimeUnit.SECONDS);
169+
170+
// Validate that the cleanup attempt was blocked by readerTxn.
171+
rsp = txnHandler.showCompact(new ShowCompactRequest());
172+
assertEquals(1, rsp.getCompactsSize());
173+
assertEquals(CLEANING_RESPONSE, rsp.getCompacts().getFirst().getState());
174+
175+
// Check that the files are not removed
176+
paths = getDirectories(conf, t, null);
177+
assertEquals(5, paths.size());
178+
}
179+
180+
@Test
181+
public void cleanupNotBlockedByOpenTxnOnAnotherTable() throws Exception {
182+
Table t1 = prepareTestTable("camtc1");
183+
Table t2 = prepareTestTable("camtc2");
184+
185+
// Open a readerTxn on t1 and register minOpenWriteId.
186+
long readerTxn = openTxn();
187+
txnHandler.addWriteIdsToMinHistory(readerTxn, Map.of("default.camtc1", 1L));
188+
189+
CompactionRequest rqstTbl1 = new CompactionRequest("default", "camtc1", CompactionType.MAJOR);
190+
long compactTxn = compactInTxn(rqstTbl1);
191+
addBaseFile(t1, null, 25L, 25, compactTxn);
192+
193+
CompactionRequest rqstTbl2 = new CompactionRequest("default", "camtc2", CompactionType.MAJOR);
194+
compactTxn = compactInTxn(rqstTbl2);
195+
addBaseFile(t2, null, 25L, 25, compactTxn);
196+
197+
startCleaner();
198+
199+
ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
200+
assertEquals(2, rsp.getCompactsSize());
201+
202+
assertEquals(SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
203+
assertEquals("camtc2", rsp.getCompacts().get(0).getTablename());
204+
// camtc2 was cleaned: only the new base remains.
205+
assertEquals(1, getDirectories(conf, t2, null).size());
206+
207+
assertEquals(CLEANING_RESPONSE, rsp.getCompacts().get(1).getState());
208+
assertEquals("camtc1", rsp.getCompacts().get(1).getTablename());
209+
// camtc1 wasn't actually cleaned (admission filter held it back).
210+
assertEquals(5, getDirectories(conf, t1, null).size());
146211
}
147212

148-
private Table prepareTestTable() throws Exception {
149-
Table t = newTable("default", "camtc", false);
213+
private Table prepareTestTable(String tblName) throws Exception {
214+
Table t = newTable("default", tblName, false);
150215

151216
addBaseFile(t, null, 20L, 20);
152217
addDeltaFile(t, null, 21L, 22L, 2);
153218
addDeltaFile(t, null, 23L, 24L, 2);
154219
addDeltaFile(t, null, 25L, 25, 2);
155220

156-
burnThroughTransactions("default", "camtc", 25);
221+
burnThroughTransactions("default", tblName, 25);
157222
return t;
158223
}
159224

0 commit comments

Comments
 (0)