zabetak commented on code in PR #4859:
URL: https://github.com/apache/hive/pull/4859#discussion_r1497657015
##########
ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/InitiatorBase.java:
##########
@@ -50,59 +49,53 @@
import java.io.IOException;
import java.security.PrivilegedExceptionAction;
-import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
public class InitiatorBase extends MetaStoreCompactorThread {
static final private String COMPACTOR_THRESHOLD_PREFIX =
"compactorthreshold.";
- private List<CompactionResponse>
initiateCompactionForMultiplePartitions(Table table,
- Map<String, Partition> partitions, CompactionRequest request) {
- List<CompactionResponse> compactionResponses = new ArrayList<>();
- partitions.entrySet().parallelStream().forEach(entry -> {
- try {
- StorageDescriptor sd = CompactorUtil.resolveStorageDescriptor(table,
entry.getValue());
- String runAs = TxnUtils.findUserToRunAs(sd.getLocation(), table, conf);
- CompactionInfo ci =
- new CompactionInfo(table.getDbName(), table.getTableName(),
entry.getKey(), request.getType());
- ci.initiatorId = request.getInitiatorId();
- ci.orderByClause = request.getOrderByClause();
- ci.initiatorVersion = request.getInitiatorVersion();
- if (request.getNumberOfBuckets() > 0) {
- ci.numberOfBuckets = request.getNumberOfBuckets();
- }
- ci.poolName = request.getPoolName();
- LOG.info(
- "Checking to see if we should compact partition " + entry.getKey()
+ " of table " + table.getDbName() + "."
- + table.getTableName());
- CollectionUtils.addIgnoreNull(compactionResponses,
- scheduleCompactionIfRequired(ci, table, entry.getValue(), runAs,
false));
- } catch (IOException | InterruptedException | MetaException e) {
- LOG.error(
- "Error occurred while Checking if we should compact partition " +
entry.getKey() + " of table " + table.getDbName() + "."
- + table.getTableName() + " Exception: " + e.getMessage());
- throw new RuntimeException(e);
- }
- });
- return compactionResponses;
- }
-
- public List<CompactionResponse> initiateCompactionForTable(CompactionRequest
request, Table table, Map<String, Partition> partitions) throws Exception {
+ public void initialize() throws Exception {
+ super.init(new AtomicBoolean());
Review Comment:
The latest patch changes the behavior of `Initiator` forcing the latter to
retrieve and set a `validTxnList` during initialization something that wasn't
done before. I reviewed related code in `Initiator` class and the change
appears safe but in general it is best to avoid such refactorings that
implicitly change the behavior of the code.
@tarak271 Did you double check that there are no undesired effects from this
change in behavior? If yes then we can proceed and merge this PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
