Skip to content

Commit 341b863

Browse files
deniskuzZdeniskuzZ
authored
HIVE-29571: ACID Compaction: Refused compaction Txn should not be aborted (#6517)
1 parent 59b36ea commit 341b863

1 file changed

Lines changed: 63 additions & 73 deletions

File tree

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

Lines changed: 63 additions & 73 deletions
Original file line numberDiff line numberDiff line change
@@ -126,57 +126,56 @@ public void cleanupResultDirs(CompactionInfo ci) {
126126

127127
public Boolean compact(Table table, CompactionInfo ci) throws Exception {
128128

129-
try (CompactionTxn compactionTxn = new CompactionTxn()) {
130-
131-
if (ci.isRebalanceCompaction() && table.getSd().getNumBuckets() > 0) {
132-
LOG.error("Cannot execute rebalancing compaction on bucketed tables.");
133-
ci.errorMessage = "Cannot execute rebalancing compaction on bucketed tables.";
134-
msc.markRefused(CompactionInfo.compactionInfoToStruct(ci));
135-
return false;
136-
}
129+
if (ci.isRebalanceCompaction() && table.getSd().getNumBuckets() > 0) {
130+
LOG.error("Cannot execute rebalancing compaction on bucketed tables.");
131+
ci.errorMessage = "Cannot execute rebalancing compaction on bucketed tables.";
132+
msc.markRefused(CompactionInfo.compactionInfoToStruct(ci));
133+
return false;
134+
}
137135

138-
if (!ci.type.equals(CompactionType.REBALANCE) && ci.numberOfBuckets > 0) {
139-
if (LOG.isWarnEnabled()) {
140-
LOG.warn("Only the REBALANCE compaction accepts the number of buckets clause (CLUSTERED INTO {N} BUCKETS). " +
141-
"Since the compaction request is {}, it will be ignored.", ci.type);
142-
}
136+
if (!ci.type.equals(CompactionType.REBALANCE) && ci.numberOfBuckets > 0) {
137+
if (LOG.isWarnEnabled()) {
138+
LOG.warn("Only the REBALANCE compaction accepts the number of buckets clause (CLUSTERED INTO {N} BUCKETS). " +
139+
"Since the compaction request is {}, it will be ignored.", ci.type);
143140
}
141+
}
144142

145-
String fullTableName = TxnUtils.getFullTableName(table.getDbName(), table.getTableName());
143+
String fullTableName = TxnUtils.getFullTableName(table.getDbName(), table.getTableName());
146144

147-
// Find the partition we will be working with, if there is one.
148-
Partition p;
149-
try {
150-
p = CompactorUtil.resolvePartition(conf, msc, ci.dbname, ci.tableName, ci.partName,
151-
CompactorUtil.METADATA_FETCH_MODE.REMOTE);
152-
if (p == null && ci.partName != null) {
153-
ci.errorMessage = "Unable to find partition " + ci.getFullPartitionName() + ", assuming it was dropped and moving on.";
154-
LOG.warn(ci.errorMessage + " Compaction info: {}", ci);
155-
msc.markRefused(CompactionInfo.compactionInfoToStruct(ci));
156-
return false;
157-
}
158-
} catch (Exception e) {
159-
LOG.error("Unexpected error during resolving partition.", e);
160-
ci.errorMessage = e.getMessage();
161-
msc.markFailed(CompactionInfo.compactionInfoToStruct(ci));
145+
// Find the partition we will be working with, if there is one.
146+
Partition p;
147+
try {
148+
p = CompactorUtil.resolvePartition(conf, msc, ci.dbname, ci.tableName, ci.partName,
149+
CompactorUtil.METADATA_FETCH_MODE.REMOTE);
150+
if (p == null && ci.partName != null) {
151+
ci.errorMessage = "Unable to find partition " + ci.getFullPartitionName() + ", assuming it was dropped and moving on.";
152+
LOG.warn(ci.errorMessage + " Compaction info: {}", ci);
153+
msc.markRefused(CompactionInfo.compactionInfoToStruct(ci));
162154
return false;
163155
}
156+
} catch (Exception e) {
157+
LOG.error("Unexpected error during resolving partition.", e);
158+
ci.errorMessage = e.getMessage();
159+
msc.markFailed(CompactionInfo.compactionInfoToStruct(ci));
160+
return false;
161+
}
164162

165-
CompactorUtil.checkInterrupt(CLASS_NAME);
163+
CompactorUtil.checkInterrupt(CLASS_NAME);
166164

167-
// Find the appropriate storage descriptor
168-
sd = CompactorUtil.resolveStorageDescriptor(table, p);
165+
// Find the appropriate storage descriptor
166+
sd = CompactorUtil.resolveStorageDescriptor(table, p);
169167

170-
if (isTableSorted(sd, ci)) {
171-
return false;
172-
}
168+
if (isTableSorted(sd, ci)) {
169+
return false;
170+
}
173171

174-
if (ci.runAs == null) {
175-
ci.runAs = TxnUtils.findUserToRunAs(sd.getLocation(), table, conf);
176-
}
172+
if (ci.runAs == null) {
173+
ci.runAs = TxnUtils.findUserToRunAs(sd.getLocation(), table, conf);
174+
}
177175

178-
CompactorUtil.checkInterrupt(CLASS_NAME);
176+
CompactorUtil.checkInterrupt(CLASS_NAME);
179177

178+
try (CompactionTxn compactionTxn = new CompactionTxn()) {
180179
/*
181180
* we cannot have Worker use HiveTxnManager (which is on ThreadLocal) since
182181
* then the Driver would already have the an open txn but then this txn would have
@@ -212,40 +211,43 @@ public Boolean compact(Table table, CompactionInfo ci) throws Exception {
212211

213212
// Don't start compaction or cleaning if not necessary
214213
if (isDynPartAbort(table, ci)) {
215-
compactionTxn.markForCommit(() -> msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci)));
214+
compactionTxn.commit(() ->
215+
msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci)));
216216
return false;
217217
}
218218
dir = getAcidStateForWorker(ci, sd, tblValidWriteIds);
219219
if (!isEnoughToCompact(ci, dir, sd)) {
220220
if (needsCleaning(dir, sd)) {
221-
compactionTxn.markForCommit(() -> msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci)));
221+
compactionTxn.commit(() ->
222+
msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci)));
222223
} else {
223224
// do nothing
224225
ci.errorMessage = "None of the compaction thresholds met, compaction request is refused!";
225226
LOG.debug(ci.errorMessage + " Compaction info: {}", ci);
226-
compactionTxn.markForCommit(() -> msc.markRefused(CompactionInfo.compactionInfoToStruct(ci)));
227-
227+
compactionTxn.commit(() ->
228+
msc.markRefused(CompactionInfo.compactionInfoToStruct(ci)));
228229
}
229230
return false;
230231
}
231232
if (!ci.isMajorCompaction() && !CompactorUtil.isMinorCompactionSupported(conf, table.getParameters(), dir)) {
232233
ci.errorMessage = "Query based Minor compaction is not possible for full acid tables having raw format " +
233234
"(non-acid) data in them.";
234235
LOG.error(ci.errorMessage + " Compaction info: {}", ci);
235-
compactionTxn.markForAbort(() -> msc.markRefused(CompactionInfo.compactionInfoToStruct(ci)));
236+
compactionTxn.commit(() ->
237+
msc.markRefused(CompactionInfo.compactionInfoToStruct(ci)));
236238
return false;
237239
}
238240
CompactorUtil.checkInterrupt(CLASS_NAME);
239241

240242
try {
241243
failCompactionIfSetForTest();
242244

243-
/*
244-
First try to run compaction via HiveQL queries.
245-
Compaction for MM tables happens here, or run compaction for Crud tables if query-based compaction is enabled.
246-
todo Find a more generic approach to collecting files in the same logical bucket to compact within the same
247-
task (currently we're using Tez split grouping).
248-
*/
245+
/*
246+
First try to run compaction via HiveQL queries.
247+
Compaction for MM tables happens here, or run compaction for Crud tables if query-based compaction is enabled.
248+
todo Find a more generic approach to collecting files in the same logical bucket to compact within the same
249+
task (currently we're using Tez split grouping).
250+
*/
249251
CompactorPipeline compactorPipeline = compactorFactory.getCompactorPipeline(table, conf, ci, msc);
250252
computeStats = (compactorPipeline.isMRCompaction() && collectMrStats) || collectGenericStats;
251253

@@ -257,7 +259,8 @@ public Boolean compact(Table table, CompactionInfo ci) throws Exception {
257259

258260
LOG.info("Completed " + ci.type.toString() + " compaction for " + ci.getFullPartitionName() + " in "
259261
+ compactionTxn + ", marking as compacted.");
260-
compactionTxn.markForCommit(() -> msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci)));
262+
compactionTxn.commit(() ->
263+
msc.markCompacted(CompactionInfo.compactionInfoToStruct(ci)));
261264

262265
AcidMetricService.updateMetricsFromWorker(ci.dbname, ci.tableName, ci.partName, ci.type,
263266
dir.getCurrentDirectories().size(), dir.getDeleteDeltas().size(), conf, msc);
@@ -267,9 +270,6 @@ public Boolean compact(Table table, CompactionInfo ci) throws Exception {
267270
}
268271

269272
return true;
270-
} catch (Exception e) {
271-
LOG.error("Caught exception in " + CLASS_NAME + " while trying to compact " + ci, e);
272-
throw e;
273273
}
274274
}
275275

@@ -341,11 +341,8 @@ class CompactionTxn implements AutoCloseable {
341341
private long lockId = 0;
342342

343343
private TxnStatus status = TxnStatus.UNKNOWN;
344-
345-
private ThrowingRunnable onCommitSuccess;
346-
private ThrowingRunnable onAbortSuccess;
347-
348344
private boolean rollbackOnly = true;
345+
private ThrowingRunnable<?> onCommitSuccess;
349346

350347
/**
351348
* Try to open a new txn.
@@ -376,13 +373,9 @@ private LockRequest createLockRequest(CompactionInfo ci) {
376373
return CompactorUtil.createLockRequest(conf, ci, txnId, lockAndOpType.getKey(), lockAndOpType.getValue());
377374
}
378375

379-
void markForCommit(ThrowingRunnable action) {
376+
void commit(ThrowingRunnable<?> postAction) {
380377
this.rollbackOnly = false;
381-
this.onCommitSuccess = action;
382-
}
383-
384-
void markForAbort(ThrowingRunnable action) {
385-
this.onAbortSuccess = action;
378+
this.onCommitSuccess = postAction;
386379
}
387380

388381
/**
@@ -397,16 +390,13 @@ public void close() throws Exception {
397390
//the transaction is about to close, we can stop heartbeating regardless of it's state
398391
CompactionHeartbeatService.getInstance(conf).stopHeartbeat(txnId);
399392
} finally {
400-
if (rollbackOnly) {
401-
abort();
402-
if (onAbortSuccess != null) {
403-
onAbortSuccess.run();
393+
if (!rollbackOnly) {
394+
commit();
395+
if (onCommitSuccess != null) {
396+
onCommitSuccess.run();
404397
}
405-
return;
406-
}
407-
commit();
408-
if (onCommitSuccess != null) {
409-
onCommitSuccess.run();
398+
} else {
399+
abort();
410400
}
411401
}
412402
}

0 commit comments

Comments
 (0)