You are viewing a plain text version of this content. The canonical link for it is here.
Posted to gitbox@hive.apache.org by GitBox <gi...@apache.org> on 2022/12/20 08:35:26 UTC
[GitHub] [hive] rkirtir opened a new pull request, #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
rkirtir opened a new pull request, #3880:
URL: https://github.com/apache/hive/pull/3880
<!--
Thanks for sending a pull request! Here are some tips for you:
1. If this is your first time, please read our contributor guidelines: https://cwiki.apache.org/confluence/display/Hive/HowToContribute
2. Ensure that you have created an issue on the Hive project JIRA: https://issues.apache.org/jira/projects/HIVE/summary
3. Ensure you have added or run the appropriate tests for your PR:
4. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP]HIVE-XXXXX: Your PR title ...'.
5. Be sure to keep the PR description updated to reflect all changes.
6. Please write your PR title to summarize what this PR proposes.
7. If possible, provide a concise example to reproduce the issue for a faster review.
-->
### What changes were proposed in this pull request?
Initial classes to provide ability to cancel compaction and ability to cancel compaction in initialed state
### Why are the changes needed?
To provide new ability of cancelling compaction
### Does this PR introduce _any_ user-facing change?
Yes
### How was this patch tested?
junit
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] sonarcloud[bot] commented on pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #3880:
URL: https://github.com/apache/hive/pull/3880#issuecomment-1368662009
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=3880)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL) [5 Code Smells](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=3880&metric=coverage&view=list) No Coverage information
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=3880&metric=duplicated_lines_density&view=list) No Duplication information
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1085143309
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3418,6 +3422,100 @@ public void testShowCompactionOrder() throws Exception {
Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
}
+
+ @Test
+ public void testAbortCompaction() throws Exception {
+
+ d.destroy();
+ hiveConf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, "nonstrict");
+ d = new Driver(hiveConf);
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl1 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl1" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("drop table if exists T1");
+ runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+
+ //create failed compaction attempt so that compactor txn is aborted
+ HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
+ runStatementOnDriver("alter table T1 compact 'minor'");
+ TestTxnCommands2.runWorker(hiveConf);
+ // Verify compaction order
+ List<ShowCompactResponseElement> compacts =
+ txnHandler.showCompact(new ShowCompactRequest()).getCompacts();
+ Assert.assertEquals(6, compacts.size());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, compacts.get(0).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(1).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(2).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(3).getState());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, compacts.get(4).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
+
+ long initiatedStateCompId = compacts.get(0).getId();
+ List<Long> refusedStateCompIds = Arrays.asList(compacts.get(1).getId(),compacts.get(5).getId());
+ List<Long> compactionsToAbort = Arrays.asList(Long.parseLong("12"), compacts.get(0).getId(),
+ compacts.get(1).getId(), compacts.get(2).getId(), compacts.get(3).getId(), compacts.get(4).getId(),
+ compacts.get(5).getId());
Review Comment:
fixed
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,110 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ addAbortCompactionResponse(abortCompactionResponseElements,x, "No Such Compaction Id Available","Error");
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(abortCompactionResponseElements, eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements,
+ CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ compactionInfo.state = TxnStore.ABORTED_STATE;
+ compactionInfo.errorMessage = "Comapction Aborted by Abort Comapction request.";
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:Unable to update compaction record in COMPLETED_COMPACTIONS", "Error");
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction: Unable to update compaction record in COMPACTION_QUEUE", "Error");
+ } else {
+ dbConn.commit();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id, "Successfully aborted compaction",
+ "Success");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:"+e.getMessage(), "Error");
+ }
+ }
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database: " + e.getMessage());
+ checkRetryable(e, "abortCompaction(" + compactionInfo + ")");
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:"+ e.getMessage(), "Error" );
+ }
+ } catch (RetryException e) {
+ abortCompaction(abortCompactionResponseElements, compactionInfo);
+ }
+ }
+
+ private List<CompactionInfo> findEligibleCompactionsToAbort(Map<Long,
+ AbortCompactionResponseElement> abortCompactionResponseElements, List<Long> requestedCompId) throws MetaException {
+
+ List<CompactionInfo> compactionInfoList = new ArrayList<>();
+ String queryText = TxnQueries.SELECT_COMPACTION_QUEUE_BY_COMPID + " WHERE \"CC_ID\" IN " +
+ "(" + Joiner.on(',').join(requestedCompId) + ") ";
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ PreparedStatement pStmt = dbConn.prepareStatement(queryText)) {
+ try (ResultSet rs = pStmt.executeQuery()) {
+ while (rs.next()) {
+ if (checkIfCompactionEligibleToAbort(rs.getString(5).charAt(0))) {
+ compactionInfoList.add(CompactionInfo.loadFullFromCompactionQueue(rs));
+ } else {
+ addAbortCompactionResponse(abortCompactionResponseElements, rs.getLong(1),
+ "Error while aborting compaction as compaction is in state-" +
+ CompactionState.fromSqlConst(rs.getString(5).charAt(0)), "Error");
+ }
+ }
+ }
+ } catch (SQLException e) {
+ throw new MetaException("Unable to select from transaction database-" + StringUtils.stringifyException(e));
+ }
+ return compactionInfoList;
+ }
+
+ private boolean checkIfCompactionEligibleToAbort(char state) {
+
+ return CompactionState.INITIATED.equals(CompactionState.fromSqlConst(state));
+ }
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] veghlaci05 commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "veghlaci05 (via GitHub)" <gi...@apache.org>.
veghlaci05 commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1088761769
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3458,14 +3458,14 @@ public void testAbortCompaction() throws Exception {
runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
TestTxnCommands2.runWorker(hiveConf);
- runStatementOnDriver("drop table if exists T1");
- runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
- runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
- runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+ runStatementOnDriver("drop table if exists myT1");
+ runStatementOnDriver("create table myT1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into myT1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into myT1 values(1,4)");//makes delta_2_2 in T2
//create failed compaction attempt so that compactor txn is aborted
HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
- runStatementOnDriver("alter table T1 compact 'minor'");
+ runStatementOnDriver("alter table myT1 compact 'minor'");
Review Comment:
Is this change intended? If yes, why?
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,105 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error",
+ "No Such Compaction Id Available"));
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,
+ compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
Review Comment:
You could replace it with
`for (CompactionInfo compactionInfo : eligibleCompactionsToAbort) {`
##########
standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnQueries.java:
##########
@@ -18,36 +18,70 @@
package org.apache.hadoop.hive.metastore.txn;
public class TxnQueries {
- public static final String SHOW_COMPACTION_ORDERBY_CLAUSE =
- " ORDER BY CASE " +
- " WHEN \"CC_END\" > \"CC_START\" and \"CC_END\" > \"CC_COMMIT_TIME\" " +
- " THEN \"CC_END\" " +
- " WHEN \"CC_START\" > \"CC_COMMIT_TIME\" " +
- " THEN \"CC_START\" " +
- " ELSE \"CC_COMMIT_TIME\" " +
- " END desc ," +
- " \"CC_ENQUEUE_TIME\" asc";
+ public static final String SHOW_COMPACTION_ORDERBY_CLAUSE =
+ " ORDER BY CASE " +
+ " WHEN \"CC_END\" > \"CC_START\" and \"CC_END\" > \"CC_COMMIT_TIME\" " +
+ " THEN \"CC_END\" " +
+ " WHEN \"CC_START\" > \"CC_COMMIT_TIME\" " +
+ " THEN \"CC_START\" " +
+ " ELSE \"CC_COMMIT_TIME\" " +
+ " END desc ," +
+ " \"CC_ENQUEUE_TIME\" asc";
- public static final String SHOW_COMPACTION_QUERY =
- "SELECT XX.* FROM ( SELECT " +
- " \"CQ_DATABASE\" AS \"CC_DATABASE\", \"CQ_TABLE\" AS \"CC_TABLE\", \"CQ_PARTITION\" AS \"CC_PARTITION\", " +
- " \"CQ_STATE\" AS \"CC_STATE\", \"CQ_TYPE\" AS \"CC_TYPE\", \"CQ_WORKER_ID\" AS \"CC_WORKER_ID\", " +
- " \"CQ_START\" AS \"CC_START\", -1 \"CC_END\", \"CQ_RUN_AS\" AS \"CC_RUN_AS\", " +
- " \"CQ_HADOOP_JOB_ID\" AS \"CC_HADOOP_JOB_ID\", \"CQ_ID\" AS \"CC_ID\", \"CQ_ERROR_MESSAGE\" AS \"CC_ERROR_MESSAGE\", " +
- " \"CQ_ENQUEUE_TIME\" AS \"CC_ENQUEUE_TIME\", \"CQ_WORKER_VERSION\" AS \"CC_WORKER_VERSION\", " +
- " \"CQ_INITIATOR_ID\" AS \"CC_INITIATOR_ID\", \"CQ_INITIATOR_VERSION\" AS \"CC_INITIATOR_VERSION\", " +
- " \"CQ_CLEANER_START\" AS \"CC_CLEANER_START\", \"CQ_POOL_NAME\" AS \"CC_POOL_NAME\", \"CQ_TXN_ID\" AS \"CC_TXN_ID\", " +
- " \"CQ_NEXT_TXN_ID\" AS \"CC_NEXT_TXN_ID\", \"CQ_COMMIT_TIME\" AS \"CC_COMMIT_TIME\", " +
- " \"CQ_HIGHEST_WRITE_ID\" AS \"CC_HIGHEST_WRITE_ID\" " +
- "FROM " +
- " \"COMPACTION_QUEUE\" " +
- "UNION ALL " +
- "SELECT " +
- " \"CC_DATABASE\", \"CC_TABLE\", \"CC_PARTITION\", \"CC_STATE\", \"CC_TYPE\", \"CC_WORKER_ID\", " +
- " \"CC_START\", \"CC_END\", \"CC_RUN_AS\", \"CC_HADOOP_JOB_ID\", \"CC_ID\", \"CC_ERROR_MESSAGE\", " +
- " \"CC_ENQUEUE_TIME\", \"CC_WORKER_VERSION\", \"CC_INITIATOR_ID\", \"CC_INITIATOR_VERSION\", " +
- " -1 , \"CC_POOL_NAME\", \"CC_TXN_ID\", \"CC_NEXT_TXN_ID\", \"CC_COMMIT_TIME\", " +
- " \"CC_HIGHEST_WRITE_ID\"" +
- "FROM " +
- " \"COMPLETED_COMPACTIONS\" ) XX ";
+ public static final String SHOW_COMPACTION_QUERY =
+ "SELECT XX.* FROM ( SELECT " +
+ " \"CQ_DATABASE\" AS \"CC_DATABASE\", \"CQ_TABLE\" AS \"CC_TABLE\", \"CQ_PARTITION\" AS \"CC_PARTITION\", " +
+ " \"CQ_STATE\" AS \"CC_STATE\", \"CQ_TYPE\" AS \"CC_TYPE\", \"CQ_WORKER_ID\" AS \"CC_WORKER_ID\", " +
+ " \"CQ_START\" AS \"CC_START\", -1 \"CC_END\", \"CQ_RUN_AS\" AS \"CC_RUN_AS\", " +
+ " \"CQ_HADOOP_JOB_ID\" AS \"CC_HADOOP_JOB_ID\", \"CQ_ID\" AS \"CC_ID\", \"CQ_ERROR_MESSAGE\" AS \"CC_ERROR_MESSAGE\", " +
+ " \"CQ_ENQUEUE_TIME\" AS \"CC_ENQUEUE_TIME\", \"CQ_WORKER_VERSION\" AS \"CC_WORKER_VERSION\", " +
+ " \"CQ_INITIATOR_ID\" AS \"CC_INITIATOR_ID\", \"CQ_INITIATOR_VERSION\" AS \"CC_INITIATOR_VERSION\", " +
+ " \"CQ_CLEANER_START\" AS \"CC_CLEANER_START\", \"CQ_POOL_NAME\" AS \"CC_POOL_NAME\", \"CQ_TXN_ID\" AS \"CC_TXN_ID\", " +
+ " \"CQ_NEXT_TXN_ID\" AS \"CC_NEXT_TXN_ID\", \"CQ_COMMIT_TIME\" AS \"CC_COMMIT_TIME\", " +
+ " \"CQ_HIGHEST_WRITE_ID\" AS \"CC_HIGHEST_WRITE_ID\" " +
+ "FROM " +
+ " \"COMPACTION_QUEUE\" " +
+ "UNION ALL " +
+ "SELECT " +
+ " \"CC_DATABASE\" , \"CC_TABLE\", \"CC_PARTITION\", \"CC_STATE\", \"CC_TYPE\", \"CC_WORKER_ID\", " +
+ " \"CC_START\", \"CC_END\", \"CC_RUN_AS\", \"CC_HADOOP_JOB_ID\", \"CC_ID\", \"CC_ERROR_MESSAGE\", " +
+ " \"CC_ENQUEUE_TIME\", \"CC_WORKER_VERSION\", \"CC_INITIATOR_ID\", \"CC_INITIATOR_VERSION\", " +
+ " -1 , \"CC_POOL_NAME\", \"CC_TXN_ID\", \"CC_NEXT_TXN_ID\", \"CC_COMMIT_TIME\", " +
+ " \"CC_HIGHEST_WRITE_ID\"" +
+ "FROM " +
+ " \"COMPLETED_COMPACTIONS\" ) XX ";
+
+
+ public static final String SELECT_COMPACTION_QUEUE_BY_COMPID =
Review Comment:
There is no need for aliases, your are accessing the data by indexes. If the number and thype of the columns are equal, the coulmn names can differ.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,105 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error",
+ "No Such Compaction Id Available"));
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,
+ compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompactionResponseElements.put(eligibleCompactionsToAbort.get(x).id, abortCompaction(eligibleCompactionsToAbort.get(x)));
+ }
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ compactionInfo.state = TxnStore.ABORTED_STATE;
+ compactionInfo.errorMessage = "Comapction Aborted by Abort Comapction request.";
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction:Unable to update compaction record in COMPLETED_COMPACTIONS");
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction: Unable to update compaction record in COMPACTION_QUEUE");
+ } else {
+ dbConn.commit();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Success", "Successfully aborted compaction");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction:" + e.getMessage());
+ }
+ }
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database: " + e.getMessage());
+ checkRetryable(e, "abortCompaction(" + compactionInfo + ")");
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction:" + e.getMessage());
+ }
+ } catch (RetryException e) {
+ return abortCompaction(compactionInfo);
+ }
+ }
+
+ private List<CompactionInfo> findEligibleCompactionsToAbort(Map<Long,
+ AbortCompactionResponseElement> abortCompactionResponseElements, List<Long> requestedCompId) throws MetaException {
+
+ List<CompactionInfo> compactionInfoList = new ArrayList<>();
+ String queryText = TxnQueries.SELECT_COMPACTION_QUEUE_BY_COMPID + " WHERE \"CC_ID\" IN (?) " ;
+ String sqlIN = requestedCompId.stream()
+ .map(x -> String.valueOf(x))
+ .collect(Collectors.joining(",", "(", ")"));
+ queryText = queryText.replace("(?)", sqlIN);
Review Comment:
You could simply append the in clause to the end of the query, there's no need of replace here.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1083678902
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3411,6 +3415,73 @@ public void testShowCompactionOrder() throws Exception {
Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
}
+
+ @Test
+ public void testAbortCompaction() throws Exception {
+
+ d.destroy();
+ hiveConf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, "nonstrict");
+ d = new Driver(hiveConf);
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl1 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl1" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("drop table if exists T1");
+ runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+
+ //create failed compaction attempt so that compactor txn is aborted
+ HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
+ runStatementOnDriver("alter table T1 compact 'minor'");
+ TestTxnCommands2.runWorker(hiveConf);
+ // Verify compaction order
+ List<ShowCompactResponseElement> compacts =
+ txnHandler.showCompact(new ShowCompactRequest()).getCompacts();
+ Assert.assertEquals(6, compacts.size());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, compacts.get(0).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(1).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(2).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(3).getState());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, compacts.get(4).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
+
+ List<Long> compactionsToAbort = Arrays.asList(compacts.get(0).getId(),compacts.get(1).getId(),compacts.get(3).getId());
+ AbortCompactionRequest rqst= new AbortCompactionRequest();
+ rqst.setCompactionIds(compactionsToAbort);
+ AbortCompactResponse resp = txnHandler.abortCompactions(rqst);
+ Assert.assertEquals(3,resp.getAbortedcompactsSize());
+ Map<Long,AbortCompactionResponseElement> res = resp.getAbortedcompacts();
+ List<AbortCompactionResponseElement> respList = res.values().stream().collect(Collectors.toList());
+ Assert.assertEquals("Not Eligible",respList.get(0).getMessage());
+ Assert.assertEquals("Not Eligible",respList.get(1).getMessage());
+ Assert.assertEquals("Successfully aborted Compaction",respList.get(2).getMessage());
Review Comment:
fixed
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3411,6 +3415,73 @@ public void testShowCompactionOrder() throws Exception {
Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
}
+
+ @Test
+ public void testAbortCompaction() throws Exception {
+
+ d.destroy();
+ hiveConf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, "nonstrict");
+ d = new Driver(hiveConf);
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl1 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl1" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("drop table if exists T1");
+ runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+
+ //create failed compaction attempt so that compactor txn is aborted
+ HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
+ runStatementOnDriver("alter table T1 compact 'minor'");
+ TestTxnCommands2.runWorker(hiveConf);
+ // Verify compaction order
+ List<ShowCompactResponseElement> compacts =
+ txnHandler.showCompact(new ShowCompactRequest()).getCompacts();
+ Assert.assertEquals(6, compacts.size());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, compacts.get(0).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(1).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(2).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(3).getState());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, compacts.get(4).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
+
+ List<Long> compactionsToAbort = Arrays.asList(compacts.get(0).getId(),compacts.get(1).getId(),compacts.get(3).getId());
+ AbortCompactionRequest rqst= new AbortCompactionRequest();
+ rqst.setCompactionIds(compactionsToAbort);
+ AbortCompactResponse resp = txnHandler.abortCompactions(rqst);
+ Assert.assertEquals(3,resp.getAbortedcompactsSize());
+ Map<Long,AbortCompactionResponseElement> res = resp.getAbortedcompacts();
+ List<AbortCompactionResponseElement> respList = res.values().stream().collect(Collectors.toList());
+ Assert.assertEquals("Not Eligible",respList.get(0).getMessage());
+ Assert.assertEquals("Not Eligible",respList.get(1).getMessage());
+ Assert.assertEquals("Successfully aborted Compaction",respList.get(2).getMessage());
+ }
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1088849317
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3458,14 +3458,14 @@ public void testAbortCompaction() throws Exception {
runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
TestTxnCommands2.runWorker(hiveConf);
- runStatementOnDriver("drop table if exists T1");
- runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
- runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
- runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+ runStatementOnDriver("drop table if exists myT1");
+ runStatementOnDriver("create table myT1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into myT1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into myT1 values(1,4)");//makes delta_2_2 in T2
//create failed compaction attempt so that compactor txn is aborted
HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
- runStatementOnDriver("alter table T1 compact 'minor'");
+ runStatementOnDriver("alter table myT1 compact 'minor'");
Review Comment:
With T/T1 test passes locally but fails on jenkins.
http://ci.hive.apache.org/blue/organizations/jenkins/hive-precommit/detail/PR-3880/8/tests/
java.lang.RuntimeException: create table t1 (a int, b int) stored as orc TBLPROPERTIES ('serialization.null.format'='', 'transactional'='true') failed: (responseCode = 40000, errorMessage = FAILED: Execution Error, return code 40000 from org.apache.hadoop.hive.ql.ddl.DDLTask. AlreadyExistsException(message:Table hive.default.t1 already exists), hiveErrorCode = 40000, SQLState = 08S01, exception = AlreadyExistsException(message:Table hive.default.t1 already exists))
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1083679217
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
+ });
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ private void addAbortCompactionResponse(long id, String message, String status) {
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1083679630
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -247,6 +248,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
private static final int ALLOWED_REPEATED_DEADLOCKS = 10;
private static final Logger LOG = LoggerFactory.getLogger(TxnHandler.class.getName());
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1088805034
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,105 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error",
+ "No Such Compaction Id Available"));
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,
+ compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompactionResponseElements.put(eligibleCompactionsToAbort.get(x).id, abortCompaction(eligibleCompactionsToAbort.get(x)));
+ }
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ compactionInfo.state = TxnStore.ABORTED_STATE;
+ compactionInfo.errorMessage = "Comapction Aborted by Abort Comapction request.";
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction:Unable to update compaction record in COMPLETED_COMPACTIONS");
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction: Unable to update compaction record in COMPACTION_QUEUE");
+ } else {
+ dbConn.commit();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Success", "Successfully aborted compaction");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction:" + e.getMessage());
+ }
+ }
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database: " + e.getMessage());
+ checkRetryable(e, "abortCompaction(" + compactionInfo + ")");
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction:" + e.getMessage());
+ }
+ } catch (RetryException e) {
+ return abortCompaction(compactionInfo);
+ }
+ }
+
+ private List<CompactionInfo> findEligibleCompactionsToAbort(Map<Long,
+ AbortCompactionResponseElement> abortCompactionResponseElements, List<Long> requestedCompId) throws MetaException {
+
+ List<CompactionInfo> compactionInfoList = new ArrayList<>();
+ String queryText = TxnQueries.SELECT_COMPACTION_QUEUE_BY_COMPID + " WHERE \"CC_ID\" IN (?) " ;
+ String sqlIN = requestedCompId.stream()
+ .map(x -> String.valueOf(x))
+ .collect(Collectors.joining(",", "(", ")"));
+ queryText = queryText.replace("(?)", sqlIN);
Review Comment:
There was code smell warning before for dynamic formating of sql
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by GitBox <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1072024074
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
Review Comment:
As compaction related other methods are in TxnHandler, I had put it in TxnHandler. Please suggest
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ AbortCompactResponse response = new AbortCompactResponse(new ArrayList<>());
+ List<Long> requestedCompId = reqst.getCompactionIds();
+ if (requestedCompId.isEmpty()) {
+ LOG.info("Compaction ids missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("ompaction ids missing in request. No compactions to abort");
+ }
+ List<AbortCompactionResponseElement> abortCompactionResponseElementList = new ArrayList<>();
+ for (int i = 0; i < requestedCompId.size(); i++) {
+ AbortCompactionResponseElement responseEle = abortCompaction(requestedCompId.get(i));
+ abortCompactionResponseElementList.add(responseEle);
+ }
+ response.setAbortedcompacts(abortCompactionResponseElementList);
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(Long compId) throws MetaException {
+ try {
+ AbortCompactionResponseElement responseEle = new AbortCompactionResponseElement();
+ responseEle.setCompactionIds(compId);
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex)) {
+ Optional<CompactionInfo> compactionInfo = getCompactionByCompId(dbConn, compId);
+ if (compactionInfo.isPresent()) {
+ try (PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo ci = compactionInfo.get();
+ ci.errorMessage = "Compaction aborted by user";
+ ci.state = TxnStore.ABORTED_STATE;
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, ci, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] veghlaci05 commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by GitBox <gi...@apache.org>.
veghlaci05 commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1060001094
##########
ql/src/java/org/apache/hadoop/hive/ql/ddl/process/abort/compaction/AbortCompactionsOperation.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.ddl.process.abort.compaction;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.AbortCompactResponse;
+import org.apache.hadoop.hive.metastore.api.AbortCompactionRequest;
+import org.apache.hadoop.hive.metastore.api.AbortCompactionResponseElement;
+import org.apache.hadoop.hive.ql.ddl.DDLOperation;
+import org.apache.hadoop.hive.ql.ddl.DDLOperationContext;
+import org.apache.hadoop.hive.ql.ddl.ShowUtils;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+
+/**
+ * Operation process of aborting compactions.
+ */
+public class AbortCompactionsOperation extends DDLOperation<AbortCompactionsDesc> {
+ public AbortCompactionsOperation(DDLOperationContext context, AbortCompactionsDesc desc) {
+ super(context, desc);
+ }
+
+ @Override
+ public int execute() throws HiveException {
+ AbortCompactionRequest request = new AbortCompactionRequest();
+ request.setCompactionIds(desc.getCompactionIds());
+ AbortCompactResponse response = context.getDb().abortCompactions(request);
+ try (DataOutputStream os = ShowUtils.getOutputStream(new Path(desc.getResFile()), context)) {
+ writeHeader(os);
+ if (response.getAbortedcompacts() != null) {
+ for (AbortCompactionResponseElement e : response.getAbortedcompacts()) {
+ writeRow(os, e);
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("show compactions: ", e);
Review Comment:
Should be "abort compactions"
##########
standalone-metastore/metastore-common/src/main/thrift/hive_metastore.thrift:
##########
@@ -1393,6 +1393,22 @@ struct ShowCompactResponse {
1: required list<ShowCompactResponseElement> compacts,
}
+struct AbortCompactionRequest {
+ 1: required list<i64> compactionIds,
+ 2: optional string type,
+ 3: optional string poolName
+}
+
+struct AbortCompactionResponseElement {
+ 1: required i64 compactionIds,
Review Comment:
Should be compactionId
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ AbortCompactResponse response = new AbortCompactResponse(new ArrayList<>());
+ List<Long> requestedCompId = reqst.getCompactionIds();
+ if (requestedCompId.isEmpty()) {
+ LOG.info("Compaction ids missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("ompaction ids missing in request. No compactions to abort");
+ }
+ List<AbortCompactionResponseElement> abortCompactionResponseElementList = new ArrayList<>();
+ for (int i = 0; i < requestedCompId.size(); i++) {
+ AbortCompactionResponseElement responseEle = abortCompaction(requestedCompId.get(i));
+ abortCompactionResponseElementList.add(responseEle);
+ }
+ response.setAbortedcompacts(abortCompactionResponseElementList);
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(Long compId) throws MetaException {
+ try {
+ AbortCompactionResponseElement responseEle = new AbortCompactionResponseElement();
+ responseEle.setCompactionIds(compId);
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex)) {
+ Optional<CompactionInfo> compactionInfo = getCompactionByCompId(dbConn, compId);
+ if (compactionInfo.isPresent()) {
+ try (PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo ci = compactionInfo.get();
+ ci.errorMessage = "Compaction aborted by user";
+ ci.state = TxnStore.ABORTED_STATE;
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, ci, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
Review Comment:
At this point the execution should be moved to the next item. After rolling back the insert, the code still tries to run the delete command, which makes no sense. Also, the response element should be set with an appropriate status and error, before advancing to the next element.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ AbortCompactResponse response = new AbortCompactResponse(new ArrayList<>());
+ List<Long> requestedCompId = reqst.getCompactionIds();
+ if (requestedCompId.isEmpty()) {
+ LOG.info("Compaction ids missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("ompaction ids missing in request. No compactions to abort");
Review Comment:
Please fix: Compaction ids are missing
##########
standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnQueries.java:
##########
@@ -50,4 +50,20 @@ public class TxnQueries {
" \"CC_HIGHEST_WRITE_ID\"" +
"FROM " +
" \"COMPLETED_COMPACTIONS\" ) XX ";
+
+
+ public static final String SELECT_COMPACTION_QUEUE_BY_COMPID = "SELECT \"CQ_ID\", \"CQ_DATABASE\", \"CQ_TABLE\", \"CQ_PARTITION\", "
+ + "\"CQ_STATE\", \"CQ_TYPE\", \"CQ_TBLPROPERTIES\", \"CQ_WORKER_ID\", \"CQ_START\", \"CQ_RUN_AS\", "
+ + "\"CQ_HIGHEST_WRITE_ID\", \"CQ_META_INFO\", \"CQ_HADOOP_JOB_ID\", \"CQ_ERROR_MESSAGE\", "
+ + "\"CQ_ENQUEUE_TIME\", \"CQ_WORKER_VERSION\", \"CQ_INITIATOR_ID\", \"CQ_INITIATOR_VERSION\", "
+ + "\"CQ_RETRY_RETENTION\", \"CQ_NEXT_TXN_ID\", \"CQ_TXN_ID\", \"CQ_COMMIT_TIME\", \"CQ_POOL_NAME\" "
+ + "FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ? AND \"CQ_STATE\" ='i'";
+
+ public static final String INSERT_INTO_COMPLETED_COMPACTION = "INSERT INTO \"COMPLETED_COMPACTIONS\" "
Review Comment:
There is another similar query in org.apache.hadoop.hive.metastore.txn.CompactionTxnHandler.updateStatus, please use this query there as well to avoid code duplication.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ AbortCompactResponse response = new AbortCompactResponse(new ArrayList<>());
+ List<Long> requestedCompId = reqst.getCompactionIds();
+ if (requestedCompId.isEmpty()) {
+ LOG.info("Compaction ids missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("ompaction ids missing in request. No compactions to abort");
+ }
+ List<AbortCompactionResponseElement> abortCompactionResponseElementList = new ArrayList<>();
+ for (int i = 0; i < requestedCompId.size(); i++) {
+ AbortCompactionResponseElement responseEle = abortCompaction(requestedCompId.get(i));
+ abortCompactionResponseElementList.add(responseEle);
+ }
+ response.setAbortedcompacts(abortCompactionResponseElementList);
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(Long compId) throws MetaException {
+ try {
+ AbortCompactionResponseElement responseEle = new AbortCompactionResponseElement();
+ responseEle.setCompactionIds(compId);
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex)) {
+ Optional<CompactionInfo> compactionInfo = getCompactionByCompId(dbConn, compId);
+ if (compactionInfo.isPresent()) {
+ try (PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo ci = compactionInfo.get();
+ ci.errorMessage = "Compaction aborted by user";
+ ci.state = TxnStore.ABORTED_STATE;
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, ci, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
+ dbConn.rollback();
+ }
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, ci.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE <{}>");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
+ dbConn.rollback();
+ } else {
+ responseEle.setMessage("Successfully Aborted Compaction ");
+ responseEle.setStatus("Success");
+ dbConn.commit();
+ }
+ }
+ }
+ } else {
+ responseEle.setMessage("Compaction element not eligible for cancellation");
+ responseEle.setStatus("Error");
+ }
+ } catch (SQLException e) {
+ LOG.error("Failed to abort compaction request");
+ checkRetryable(e, "abortCompaction(" + compId + ")");
+ responseEle.setMessage("Error while aborting compaction");
+ responseEle.setStatus("Error");
+ }
Review Comment:
I think this method should also catch Exception in a separate block and return an appropriate list of response elements in all the cases. For example if two compactions successfully cancelled, but the third one results in a runtime exception, this method won't return anything, but throw an exception. As a result, the user will see the error of the third compaction only, without any information about the other two compactions (which were cancelled successfully).
##########
ql/src/java/org/apache/hadoop/hive/ql/ddl/process/abort/compaction/AbortCompactionsOperation.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.ddl.process.abort.compaction;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.AbortCompactResponse;
+import org.apache.hadoop.hive.metastore.api.AbortCompactionRequest;
+import org.apache.hadoop.hive.metastore.api.AbortCompactionResponseElement;
+import org.apache.hadoop.hive.ql.ddl.DDLOperation;
+import org.apache.hadoop.hive.ql.ddl.DDLOperationContext;
+import org.apache.hadoop.hive.ql.ddl.ShowUtils;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+
+/**
+ * Operation process of aborting compactions.
+ */
+public class AbortCompactionsOperation extends DDLOperation<AbortCompactionsDesc> {
+ public AbortCompactionsOperation(DDLOperationContext context, AbortCompactionsDesc desc) {
+ super(context, desc);
+ }
+
+ @Override
+ public int execute() throws HiveException {
+ AbortCompactionRequest request = new AbortCompactionRequest();
+ request.setCompactionIds(desc.getCompactionIds());
+ AbortCompactResponse response = context.getDb().abortCompactions(request);
+ try (DataOutputStream os = ShowUtils.getOutputStream(new Path(desc.getResFile()), context)) {
+ writeHeader(os);
+ if (response.getAbortedcompacts() != null) {
+ for (AbortCompactionResponseElement e : response.getAbortedcompacts()) {
+ writeRow(os, e);
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("show compactions: ", e);
+ return 1;
+ }
+ return 0;
+ }
+
+ private void writeHeader(DataOutputStream os) throws IOException {
+ os.writeBytes("CompactionId");
+ os.write(Utilities.tabCode);
+ os.writeBytes("Status");
+ os.write(Utilities.tabCode);
+ os.writeBytes("Message");
+ os.write(Utilities.newLineCode);
+
+ }
+
+ private void writeRow(DataOutputStream os, AbortCompactionResponseElement e) throws IOException {
+ os.writeBytes(Long.toString(e.getCompactionIds()));
+ os.write(Utilities.tabCode);
+ os.writeBytes(e.getStatus());
+ os.write(Utilities.tabCode);
+ os.writeBytes(e.getMessage());
+ os.write(Utilities.tabCode);
Review Comment:
Is this tab before newline required?
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ AbortCompactResponse response = new AbortCompactResponse(new ArrayList<>());
+ List<Long> requestedCompId = reqst.getCompactionIds();
+ if (requestedCompId.isEmpty()) {
+ LOG.info("Compaction ids missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("ompaction ids missing in request. No compactions to abort");
+ }
+ List<AbortCompactionResponseElement> abortCompactionResponseElementList = new ArrayList<>();
+ for (int i = 0; i < requestedCompId.size(); i++) {
+ AbortCompactionResponseElement responseEle = abortCompaction(requestedCompId.get(i));
+ abortCompactionResponseElementList.add(responseEle);
+ }
+ response.setAbortedcompacts(abortCompactionResponseElementList);
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(Long compId) throws MetaException {
+ try {
+ AbortCompactionResponseElement responseEle = new AbortCompactionResponseElement();
+ responseEle.setCompactionIds(compId);
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex)) {
+ Optional<CompactionInfo> compactionInfo = getCompactionByCompId(dbConn, compId);
+ if (compactionInfo.isPresent()) {
+ try (PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo ci = compactionInfo.get();
+ ci.errorMessage = "Compaction aborted by user";
+ ci.state = TxnStore.ABORTED_STATE;
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, ci, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
+ dbConn.rollback();
+ }
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, ci.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE <{}>");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
+ dbConn.rollback();
+ } else {
+ responseEle.setMessage("Successfully Aborted Compaction ");
+ responseEle.setStatus("Success");
+ dbConn.commit();
+ }
+ }
+ }
+ } else {
+ responseEle.setMessage("Compaction element not eligible for cancellation");
+ responseEle.setStatus("Error");
+ }
+ } catch (SQLException e) {
+ LOG.error("Failed to abort compaction request");
+ checkRetryable(e, "abortCompaction(" + compId + ")");
+ responseEle.setMessage("Error while aborting compaction");
+ responseEle.setStatus("Error");
+ }
+ return responseEle;
+ } catch (RetryException e) {
+ return abortCompaction(compId);
+ }
+
+ }
+
+ private Optional<CompactionInfo> getCompactionByCompId(Connection dbConn, Long compId) throws SQLException, MetaException {
Review Comment:
This method should take the list of ids and return them in one round. It should join on completed compactions and should not filter on status, so the error messages can be more meaningful. For example to differentiate if there is no compaction with the given id, or it is already cancelled/completed, etc.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
Review Comment:
I think this method should be in CompactionTxnHandler instead.
##########
ql/src/java/org/apache/hadoop/hive/ql/ddl/process/abort/compaction/AbortCompactionsDesc.java:
##########
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.ddl.process.abort.compaction;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.ddl.DDLDesc;
+import org.apache.hadoop.hive.ql.plan.Explain;
+
+import java.io.Serializable;
+import java.util.List;
+/**
+ * DDL task description for ABORT COMPACTIONS commands.
+ */
+@Explain(displayName = "Abort Compaction", explainLevels = { Explain.Level.USER, Explain.Level.DEFAULT, Explain.Level.EXTENDED })
+public class AbortCompactionsDesc implements DDLDesc, Serializable {
+ private static final long serialVersionUID = 1L;
+
+ private final List<Long> compactionIds;
+ private String resFile;
Review Comment:
should be final
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1083682420
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
+ });
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ private void addAbortCompactionResponse(long id, String message, String status) {
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] veghlaci05 commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by GitBox <gi...@apache.org>.
veghlaci05 commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1072059615
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
+ });
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ private void addAbortCompactionResponse(long id, String message, String status) {
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
Review Comment:
addAbortCompactionResponse() should be called here as well, stating that the compaction request could not be idnetified.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
+ });
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ private void addAbortCompactionResponse(long id, String message, String status) {
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
+ } else {
+ dbConn.commit();
+ addAbortCompactionResponse(compactionInfo.id, "Successfully aborted compaction ", "Success");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
Review Comment:
The exception message should be added to the error as a reason.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
+ });
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ private void addAbortCompactionResponse(long id, String message, String status) {
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
Review Comment:
The error message should be more specific. Like: The compaction request could not be deleted from the queue.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
+ });
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ private void addAbortCompactionResponse(long id, String message, String status) {
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
+ } else {
+ dbConn.commit();
+ addAbortCompactionResponse(compactionInfo.id, "Successfully aborted compaction ", "Success");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
+ }
+ }
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database: " + e.getMessage());
+ checkRetryable(e, "abortCompaction(" + compactionInfo + ")");
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
Review Comment:
The exception message should be added to the error as a reason.
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java:
##########
@@ -2485,6 +2486,83 @@ public void testShowCompactionFilterSortingAndLimit() throws Exception {
}
+ @Test
+ public void testAbortCompactions() throws Exception {
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ TestTxnCommands2.runInitiator(hiveConf);
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p3') compact 'MAJOR' pool 'pool0'");
+ TestTxnCommands2.runInitiator(hiveConf);
+ TestTxnCommands2.runWorker(hiveConf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(hiveConf);
+
+
+
+ runStatementOnDriver("create table mydb1.tbl2 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl2" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl2" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl3 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl3" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl3" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ SessionState.get().setCurrentDatabase("mydb1");
+
+ //testing show compaction command
+
+ List<String> r = runStatementOnDriver("SHOW COMPACTIONS SCHEMA mydb1 STATUS 'initiated'");
+ Assert.assertEquals(3,r.size());
+ List<String>compIdsToAbort = r.stream().skip(1).map(x -> x.split("\t")[0]).collect(Collectors.toList());
+ String abortCompactionCmd = "ABORT COMPACTIONS " +compIdsToAbort.get(0)+"\t"+compIdsToAbort.get(1);
+ r = runStatementOnDriver(abortCompactionCmd);
+ Assert.assertEquals(3,r.size());
+ Assert.assertEquals("CompactionId\tStatus\tMessage", r.get(0));
+ Assert.assertTrue(r.get(1).contains("Successfully Aborted Compaction"));
+ Assert.assertTrue(r.get(2).contains("Successfully Aborted Compaction"));
+
+ abortCompactionCmd = "ABORT COMPACTIONS " +compIdsToAbort.get(0)+"\t"+compIdsToAbort.get(1);
+ r = runStatementOnDriver(abortCompactionCmd);
+ Assert.assertEquals(3,r.size());
+ Assert.assertEquals("CompactionId\tStatus\tMessage", r.get(0));
+ Assert.assertTrue(r.get(1).contains("Error"));
Review Comment:
Why not doing assertion for the Message and status?
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java:
##########
@@ -2485,6 +2486,83 @@ public void testShowCompactionFilterSortingAndLimit() throws Exception {
}
+ @Test
+ public void testAbortCompactions() throws Exception {
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ TestTxnCommands2.runInitiator(hiveConf);
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p3') compact 'MAJOR' pool 'pool0'");
+ TestTxnCommands2.runInitiator(hiveConf);
+ TestTxnCommands2.runWorker(hiveConf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(hiveConf);
+
+
+
+ runStatementOnDriver("create table mydb1.tbl2 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl2" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl2" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl3 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl3" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl3" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ SessionState.get().setCurrentDatabase("mydb1");
+
+ //testing show compaction command
+
+ List<String> r = runStatementOnDriver("SHOW COMPACTIONS SCHEMA mydb1 STATUS 'initiated'");
+ Assert.assertEquals(3,r.size());
Review Comment:
You should also assert the state, so we can make sure the new statis set by the abort command.
##########
standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnQueries.java:
##########
@@ -50,4 +50,22 @@ public class TxnQueries {
" \"CC_HIGHEST_WRITE_ID\"" +
"FROM " +
" \"COMPLETED_COMPACTIONS\" ) XX ";
+
+
+ public static final String SELECT_COMPACTION_QUEUE_BY_COMPID = "SELECT \"CQ_ID\", \"CQ_DATABASE\", \"CQ_TABLE\", \"CQ_PARTITION\", "
Review Comment:
This is not used anymore
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3411,6 +3415,73 @@ public void testShowCompactionOrder() throws Exception {
Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
}
+
+ @Test
+ public void testAbortCompaction() throws Exception {
+
+ d.destroy();
+ hiveConf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, "nonstrict");
+ d = new Driver(hiveConf);
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl1 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl1" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("drop table if exists T1");
+ runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+
+ //create failed compaction attempt so that compactor txn is aborted
+ HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
+ runStatementOnDriver("alter table T1 compact 'minor'");
+ TestTxnCommands2.runWorker(hiveConf);
+ // Verify compaction order
+ List<ShowCompactResponseElement> compacts =
+ txnHandler.showCompact(new ShowCompactRequest()).getCompacts();
+ Assert.assertEquals(6, compacts.size());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, compacts.get(0).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(1).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(2).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(3).getState());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, compacts.get(4).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
+
+ List<Long> compactionsToAbort = Arrays.asList(compacts.get(0).getId(),compacts.get(1).getId(),compacts.get(3).getId());
+ AbortCompactionRequest rqst= new AbortCompactionRequest();
+ rqst.setCompactionIds(compactionsToAbort);
+ AbortCompactResponse resp = txnHandler.abortCompactions(rqst);
+ Assert.assertEquals(3,resp.getAbortedcompactsSize());
+ Map<Long,AbortCompactionResponseElement> res = resp.getAbortedcompacts();
+ List<AbortCompactionResponseElement> respList = res.values().stream().collect(Collectors.toList());
+ Assert.assertEquals("Not Eligible",respList.get(0).getMessage());
+ Assert.assertEquals("Not Eligible",respList.get(1).getMessage());
+ Assert.assertEquals("Successfully aborted Compaction",respList.get(2).getMessage());
Review Comment:
Is this the first item in compacts list, but the order is different here? As far as I know only Initiated compaction can be aborted. Would be better to check for the ids too, just to make sure it's the same.
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3411,6 +3415,73 @@ public void testShowCompactionOrder() throws Exception {
Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
}
+
+ @Test
+ public void testAbortCompaction() throws Exception {
+
+ d.destroy();
+ hiveConf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, "nonstrict");
+ d = new Driver(hiveConf);
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl1 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl1" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("drop table if exists T1");
+ runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+
+ //create failed compaction attempt so that compactor txn is aborted
+ HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
+ runStatementOnDriver("alter table T1 compact 'minor'");
+ TestTxnCommands2.runWorker(hiveConf);
+ // Verify compaction order
+ List<ShowCompactResponseElement> compacts =
+ txnHandler.showCompact(new ShowCompactRequest()).getCompacts();
+ Assert.assertEquals(6, compacts.size());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, compacts.get(0).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(1).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(2).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(3).getState());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, compacts.get(4).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
+
+ List<Long> compactionsToAbort = Arrays.asList(compacts.get(0).getId(),compacts.get(1).getId(),compacts.get(3).getId());
Review Comment:
Why not try to abort the 4th as well? Succeed compaction also should not be able to aborted.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
Review Comment:
The default error message should be sth like : Compaction request could not be found. Also, you could iterate over the ids after getting the result of findEligibleCompactionsToAbort, and create the default response element only for the missing items to avoid creating them twice.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -247,6 +248,8 @@ abstract class TxnHandler implements TxnStore, TxnStore.MutexAPI {
private static final int ALLOWED_REPEATED_DEADLOCKS = 10;
private static final Logger LOG = LoggerFactory.getLogger(TxnHandler.class.getName());
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
Review Comment:
Why did you create this field? If the same TxnHandler is shared accross multiple threads (requests) this will become a serious issue. You should not store Txn scoped state in such fields.
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java:
##########
@@ -65,6 +65,7 @@
import org.apache.hadoop.hive.metastore.api.ShowLocksResponse;
import org.apache.hadoop.hive.metastore.api.TxnInfo;
import org.apache.hadoop.hive.metastore.api.TxnState;
+import org.apache.hadoop.hive.metastore.api.AbortCompactResponse;
Review Comment:
unused import
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3411,6 +3415,73 @@ public void testShowCompactionOrder() throws Exception {
Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
}
+
+ @Test
+ public void testAbortCompaction() throws Exception {
+
+ d.destroy();
+ hiveConf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, "nonstrict");
+ d = new Driver(hiveConf);
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl1 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl1" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("drop table if exists T1");
+ runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+
+ //create failed compaction attempt so that compactor txn is aborted
+ HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
+ runStatementOnDriver("alter table T1 compact 'minor'");
+ TestTxnCommands2.runWorker(hiveConf);
+ // Verify compaction order
+ List<ShowCompactResponseElement> compacts =
+ txnHandler.showCompact(new ShowCompactRequest()).getCompacts();
+ Assert.assertEquals(6, compacts.size());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, compacts.get(0).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(1).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(2).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(3).getState());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, compacts.get(4).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
+
+ List<Long> compactionsToAbort = Arrays.asList(compacts.get(0).getId(),compacts.get(1).getId(),compacts.get(3).getId());
+ AbortCompactionRequest rqst= new AbortCompactionRequest();
+ rqst.setCompactionIds(compactionsToAbort);
+ AbortCompactResponse resp = txnHandler.abortCompactions(rqst);
+ Assert.assertEquals(3,resp.getAbortedcompactsSize());
+ Map<Long,AbortCompactionResponseElement> res = resp.getAbortedcompacts();
+ List<AbortCompactionResponseElement> respList = res.values().stream().collect(Collectors.toList());
+ Assert.assertEquals("Not Eligible",respList.get(0).getMessage());
+ Assert.assertEquals("Not Eligible",respList.get(1).getMessage());
+ Assert.assertEquals("Successfully aborted Compaction",respList.get(2).getMessage());
+ }
Review Comment:
nit: new line between methods
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1083679106
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
+ });
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ private void addAbortCompactionResponse(long id, String message, String status) {
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
+ } else {
+ dbConn.commit();
+ addAbortCompactionResponse(compactionInfo.id, "Successfully aborted compaction ", "Success");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
+ }
+ }
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database: " + e.getMessage());
+ checkRetryable(e, "abortCompaction(" + compactionInfo + ")");
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
Review Comment:
fixed
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
+ });
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ private void addAbortCompactionResponse(long id, String message, String status) {
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
+ } else {
+ dbConn.commit();
+ addAbortCompactionResponse(compactionInfo.id, "Successfully aborted compaction ", "Success");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ addAbortCompactionResponse(compactionInfo.id, "Error while aborting compaction ", "Error");
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1083679404
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ AbortCompactResponse response = new AbortCompactResponse(new ArrayList<>());
+ List<Long> requestedCompId = reqst.getCompactionIds();
+ if (requestedCompId.isEmpty()) {
+ LOG.info("Compaction ids missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("ompaction ids missing in request. No compactions to abort");
+ }
+ List<AbortCompactionResponseElement> abortCompactionResponseElementList = new ArrayList<>();
+ for (int i = 0; i < requestedCompId.size(); i++) {
+ AbortCompactionResponseElement responseEle = abortCompaction(requestedCompId.get(i));
+ abortCompactionResponseElementList.add(responseEle);
+ }
+ response.setAbortedcompacts(abortCompactionResponseElementList);
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(Long compId) throws MetaException {
+ try {
+ AbortCompactionResponseElement responseEle = new AbortCompactionResponseElement();
+ responseEle.setCompactionIds(compId);
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex)) {
+ Optional<CompactionInfo> compactionInfo = getCompactionByCompId(dbConn, compId);
+ if (compactionInfo.isPresent()) {
+ try (PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo ci = compactionInfo.get();
+ ci.errorMessage = "Compaction aborted by user";
+ ci.state = TxnStore.ABORTED_STATE;
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, ci, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
+ dbConn.rollback();
+ }
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, ci.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE <{}>");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
+ dbConn.rollback();
+ } else {
+ responseEle.setMessage("Successfully Aborted Compaction ");
+ responseEle.setStatus("Success");
+ dbConn.commit();
+ }
+ }
+ }
+ } else {
+ responseEle.setMessage("Compaction element not eligible for cancellation");
+ responseEle.setStatus("Error");
+ }
+ } catch (SQLException e) {
+ LOG.error("Failed to abort compaction request");
+ checkRetryable(e, "abortCompaction(" + compId + ")");
+ responseEle.setMessage("Error while aborting compaction");
+ responseEle.setStatus("Error");
+ }
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1083687332
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java:
##########
@@ -2485,6 +2486,83 @@ public void testShowCompactionFilterSortingAndLimit() throws Exception {
}
+ @Test
+ public void testAbortCompactions() throws Exception {
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ TestTxnCommands2.runInitiator(hiveConf);
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p3') compact 'MAJOR' pool 'pool0'");
+ TestTxnCommands2.runInitiator(hiveConf);
+ TestTxnCommands2.runWorker(hiveConf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(hiveConf);
+
+
+
+ runStatementOnDriver("create table mydb1.tbl2 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl2" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl2" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl2" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl3 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl3" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl3" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ SessionState.get().setCurrentDatabase("mydb1");
+
+ //testing show compaction command
+
+ List<String> r = runStatementOnDriver("SHOW COMPACTIONS SCHEMA mydb1 STATUS 'initiated'");
+ Assert.assertEquals(3,r.size());
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by GitBox <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1072023166
##########
standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnQueries.java:
##########
@@ -50,4 +50,20 @@ public class TxnQueries {
" \"CC_HIGHEST_WRITE_ID\"" +
"FROM " +
" \"COMPLETED_COMPACTIONS\" ) XX ";
+
+
+ public static final String SELECT_COMPACTION_QUEUE_BY_COMPID = "SELECT \"CQ_ID\", \"CQ_DATABASE\", \"CQ_TABLE\", \"CQ_PARTITION\", "
+ + "\"CQ_STATE\", \"CQ_TYPE\", \"CQ_TBLPROPERTIES\", \"CQ_WORKER_ID\", \"CQ_START\", \"CQ_RUN_AS\", "
+ + "\"CQ_HIGHEST_WRITE_ID\", \"CQ_META_INFO\", \"CQ_HADOOP_JOB_ID\", \"CQ_ERROR_MESSAGE\", "
+ + "\"CQ_ENQUEUE_TIME\", \"CQ_WORKER_VERSION\", \"CQ_INITIATOR_ID\", \"CQ_INITIATOR_VERSION\", "
+ + "\"CQ_RETRY_RETENTION\", \"CQ_NEXT_TXN_ID\", \"CQ_TXN_ID\", \"CQ_COMMIT_TIME\", \"CQ_POOL_NAME\" "
+ + "FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ? AND \"CQ_STATE\" ='i'";
+
+ public static final String INSERT_INTO_COMPLETED_COMPACTION = "INSERT INTO \"COMPLETED_COMPACTIONS\" "
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] veghlaci05 commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "veghlaci05 (via GitHub)" <gi...@apache.org>.
veghlaci05 commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1088810327
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,105 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error",
+ "No Such Compaction Id Available"));
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,
+ compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompactionResponseElements.put(eligibleCompactionsToAbort.get(x).id, abortCompaction(eligibleCompactionsToAbort.get(x)));
+ }
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ compactionInfo.state = TxnStore.ABORTED_STATE;
+ compactionInfo.errorMessage = "Comapction Aborted by Abort Comapction request.";
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction:Unable to update compaction record in COMPLETED_COMPACTIONS");
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction: Unable to update compaction record in COMPACTION_QUEUE");
+ } else {
+ dbConn.commit();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Success", "Successfully aborted compaction");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction:" + e.getMessage());
+ }
+ }
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database: " + e.getMessage());
+ checkRetryable(e, "abortCompaction(" + compactionInfo + ")");
+ return new AbortCompactionResponseElement(compactionInfo.id,
+ "Error", "Error while aborting compaction:" + e.getMessage());
+ }
+ } catch (RetryException e) {
+ return abortCompaction(compactionInfo);
+ }
+ }
+
+ private List<CompactionInfo> findEligibleCompactionsToAbort(Map<Long,
+ AbortCompactionResponseElement> abortCompactionResponseElements, List<Long> requestedCompId) throws MetaException {
+
+ List<CompactionInfo> compactionInfoList = new ArrayList<>();
+ String queryText = TxnQueries.SELECT_COMPACTION_QUEUE_BY_COMPID + " WHERE \"CC_ID\" IN (?) " ;
+ String sqlIN = requestedCompId.stream()
+ .map(x -> String.valueOf(x))
+ .collect(Collectors.joining(",", "(", ")"));
+ queryText = queryText.replace("(?)", sqlIN);
Review Comment:
Ok, if sonar prefers it this way, it's fine for me too.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1085143053
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,110 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ addAbortCompactionResponse(abortCompactionResponseElements,x, "No Such Compaction Id Available","Error");
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(abortCompactionResponseElements, eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements,
+ CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ compactionInfo.state = TxnStore.ABORTED_STATE;
+ compactionInfo.errorMessage = "Comapction Aborted by Abort Comapction request.";
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:Unable to update compaction record in COMPLETED_COMPACTIONS", "Error");
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction: Unable to update compaction record in COMPACTION_QUEUE", "Error");
+ } else {
+ dbConn.commit();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id, "Successfully aborted compaction",
+ "Success");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:"+e.getMessage(), "Error");
+ }
+ }
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database: " + e.getMessage());
+ checkRetryable(e, "abortCompaction(" + compactionInfo + ")");
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:"+ e.getMessage(), "Error" );
+ }
+ } catch (RetryException e) {
+ abortCompaction(abortCompactionResponseElements, compactionInfo);
+ }
+ }
+
+ private List<CompactionInfo> findEligibleCompactionsToAbort(Map<Long,
+ AbortCompactionResponseElement> abortCompactionResponseElements, List<Long> requestedCompId) throws MetaException {
+
+ List<CompactionInfo> compactionInfoList = new ArrayList<>();
+ String queryText = TxnQueries.SELECT_COMPACTION_QUEUE_BY_COMPID + " WHERE \"CC_ID\" IN " +
+ "(" + Joiner.on(',').join(requestedCompId) + ") ";
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ PreparedStatement pStmt = dbConn.prepareStatement(queryText)) {
+ try (ResultSet rs = pStmt.executeQuery()) {
+ while (rs.next()) {
+ if (checkIfCompactionEligibleToAbort(rs.getString(5).charAt(0))) {
+ compactionInfoList.add(CompactionInfo.loadFullFromCompactionQueue(rs));
+ } else {
+ addAbortCompactionResponse(abortCompactionResponseElements, rs.getLong(1),
+ "Error while aborting compaction as compaction is in state-" +
+ CompactionState.fromSqlConst(rs.getString(5).charAt(0)), "Error");
+ }
+ }
+ }
+ } catch (SQLException e) {
+ throw new MetaException("Unable to select from transaction database-" + StringUtils.stringifyException(e));
+ }
+ return compactionInfoList;
+ }
+
+ private boolean checkIfCompactionEligibleToAbort(char state) {
+
+ return CompactionState.INITIATED.equals(CompactionState.fromSqlConst(state));
+ }
+
+ private void addAbortCompactionResponse(Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements,
+ long id, String message, String status) {
+
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
Review Comment:
fixed
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,110 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ addAbortCompactionResponse(abortCompactionResponseElements,x, "No Such Compaction Id Available","Error");
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(abortCompactionResponseElements, eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements,
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1091558840
##########
standalone-metastore/metastore-common/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnQueries.java:
##########
@@ -18,36 +18,70 @@
package org.apache.hadoop.hive.metastore.txn;
public class TxnQueries {
- public static final String SHOW_COMPACTION_ORDERBY_CLAUSE =
- " ORDER BY CASE " +
- " WHEN \"CC_END\" > \"CC_START\" and \"CC_END\" > \"CC_COMMIT_TIME\" " +
- " THEN \"CC_END\" " +
- " WHEN \"CC_START\" > \"CC_COMMIT_TIME\" " +
- " THEN \"CC_START\" " +
- " ELSE \"CC_COMMIT_TIME\" " +
- " END desc ," +
- " \"CC_ENQUEUE_TIME\" asc";
+ public static final String SHOW_COMPACTION_ORDERBY_CLAUSE =
+ " ORDER BY CASE " +
+ " WHEN \"CC_END\" > \"CC_START\" and \"CC_END\" > \"CC_COMMIT_TIME\" " +
+ " THEN \"CC_END\" " +
+ " WHEN \"CC_START\" > \"CC_COMMIT_TIME\" " +
+ " THEN \"CC_START\" " +
+ " ELSE \"CC_COMMIT_TIME\" " +
+ " END desc ," +
+ " \"CC_ENQUEUE_TIME\" asc";
- public static final String SHOW_COMPACTION_QUERY =
- "SELECT XX.* FROM ( SELECT " +
- " \"CQ_DATABASE\" AS \"CC_DATABASE\", \"CQ_TABLE\" AS \"CC_TABLE\", \"CQ_PARTITION\" AS \"CC_PARTITION\", " +
- " \"CQ_STATE\" AS \"CC_STATE\", \"CQ_TYPE\" AS \"CC_TYPE\", \"CQ_WORKER_ID\" AS \"CC_WORKER_ID\", " +
- " \"CQ_START\" AS \"CC_START\", -1 \"CC_END\", \"CQ_RUN_AS\" AS \"CC_RUN_AS\", " +
- " \"CQ_HADOOP_JOB_ID\" AS \"CC_HADOOP_JOB_ID\", \"CQ_ID\" AS \"CC_ID\", \"CQ_ERROR_MESSAGE\" AS \"CC_ERROR_MESSAGE\", " +
- " \"CQ_ENQUEUE_TIME\" AS \"CC_ENQUEUE_TIME\", \"CQ_WORKER_VERSION\" AS \"CC_WORKER_VERSION\", " +
- " \"CQ_INITIATOR_ID\" AS \"CC_INITIATOR_ID\", \"CQ_INITIATOR_VERSION\" AS \"CC_INITIATOR_VERSION\", " +
- " \"CQ_CLEANER_START\" AS \"CC_CLEANER_START\", \"CQ_POOL_NAME\" AS \"CC_POOL_NAME\", \"CQ_TXN_ID\" AS \"CC_TXN_ID\", " +
- " \"CQ_NEXT_TXN_ID\" AS \"CC_NEXT_TXN_ID\", \"CQ_COMMIT_TIME\" AS \"CC_COMMIT_TIME\", " +
- " \"CQ_HIGHEST_WRITE_ID\" AS \"CC_HIGHEST_WRITE_ID\" " +
- "FROM " +
- " \"COMPACTION_QUEUE\" " +
- "UNION ALL " +
- "SELECT " +
- " \"CC_DATABASE\", \"CC_TABLE\", \"CC_PARTITION\", \"CC_STATE\", \"CC_TYPE\", \"CC_WORKER_ID\", " +
- " \"CC_START\", \"CC_END\", \"CC_RUN_AS\", \"CC_HADOOP_JOB_ID\", \"CC_ID\", \"CC_ERROR_MESSAGE\", " +
- " \"CC_ENQUEUE_TIME\", \"CC_WORKER_VERSION\", \"CC_INITIATOR_ID\", \"CC_INITIATOR_VERSION\", " +
- " -1 , \"CC_POOL_NAME\", \"CC_TXN_ID\", \"CC_NEXT_TXN_ID\", \"CC_COMMIT_TIME\", " +
- " \"CC_HIGHEST_WRITE_ID\"" +
- "FROM " +
- " \"COMPLETED_COMPACTIONS\" ) XX ";
+ public static final String SHOW_COMPACTION_QUERY =
+ "SELECT XX.* FROM ( SELECT " +
+ " \"CQ_DATABASE\" AS \"CC_DATABASE\", \"CQ_TABLE\" AS \"CC_TABLE\", \"CQ_PARTITION\" AS \"CC_PARTITION\", " +
+ " \"CQ_STATE\" AS \"CC_STATE\", \"CQ_TYPE\" AS \"CC_TYPE\", \"CQ_WORKER_ID\" AS \"CC_WORKER_ID\", " +
+ " \"CQ_START\" AS \"CC_START\", -1 \"CC_END\", \"CQ_RUN_AS\" AS \"CC_RUN_AS\", " +
+ " \"CQ_HADOOP_JOB_ID\" AS \"CC_HADOOP_JOB_ID\", \"CQ_ID\" AS \"CC_ID\", \"CQ_ERROR_MESSAGE\" AS \"CC_ERROR_MESSAGE\", " +
+ " \"CQ_ENQUEUE_TIME\" AS \"CC_ENQUEUE_TIME\", \"CQ_WORKER_VERSION\" AS \"CC_WORKER_VERSION\", " +
+ " \"CQ_INITIATOR_ID\" AS \"CC_INITIATOR_ID\", \"CQ_INITIATOR_VERSION\" AS \"CC_INITIATOR_VERSION\", " +
+ " \"CQ_CLEANER_START\" AS \"CC_CLEANER_START\", \"CQ_POOL_NAME\" AS \"CC_POOL_NAME\", \"CQ_TXN_ID\" AS \"CC_TXN_ID\", " +
+ " \"CQ_NEXT_TXN_ID\" AS \"CC_NEXT_TXN_ID\", \"CQ_COMMIT_TIME\" AS \"CC_COMMIT_TIME\", " +
+ " \"CQ_HIGHEST_WRITE_ID\" AS \"CC_HIGHEST_WRITE_ID\" " +
+ "FROM " +
+ " \"COMPACTION_QUEUE\" " +
+ "UNION ALL " +
+ "SELECT " +
+ " \"CC_DATABASE\" , \"CC_TABLE\", \"CC_PARTITION\", \"CC_STATE\", \"CC_TYPE\", \"CC_WORKER_ID\", " +
+ " \"CC_START\", \"CC_END\", \"CC_RUN_AS\", \"CC_HADOOP_JOB_ID\", \"CC_ID\", \"CC_ERROR_MESSAGE\", " +
+ " \"CC_ENQUEUE_TIME\", \"CC_WORKER_VERSION\", \"CC_INITIATOR_ID\", \"CC_INITIATOR_VERSION\", " +
+ " -1 , \"CC_POOL_NAME\", \"CC_TXN_ID\", \"CC_NEXT_TXN_ID\", \"CC_COMMIT_TIME\", " +
+ " \"CC_HIGHEST_WRITE_ID\"" +
+ "FROM " +
+ " \"COMPLETED_COMPACTIONS\" ) XX ";
+
+
+ public static final String SELECT_COMPACTION_QUEUE_BY_COMPID =
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by GitBox <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1072022432
##########
standalone-metastore/metastore-common/src/main/thrift/hive_metastore.thrift:
##########
@@ -1393,6 +1393,22 @@ struct ShowCompactResponse {
1: required list<ShowCompactResponseElement> compacts,
}
+struct AbortCompactionRequest {
+ 1: required list<i64> compactionIds,
+ 2: optional string type,
+ 3: optional string poolName
+}
+
+struct AbortCompactionResponseElement {
+ 1: required i64 compactionIds,
Review Comment:
fixed
##########
ql/src/java/org/apache/hadoop/hive/ql/ddl/process/abort/compaction/AbortCompactionsOperation.java:
##########
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.hive.ql.ddl.process.abort.compaction;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.api.AbortCompactResponse;
+import org.apache.hadoop.hive.metastore.api.AbortCompactionRequest;
+import org.apache.hadoop.hive.metastore.api.AbortCompactionResponseElement;
+import org.apache.hadoop.hive.ql.ddl.DDLOperation;
+import org.apache.hadoop.hive.ql.ddl.DDLOperationContext;
+import org.apache.hadoop.hive.ql.ddl.ShowUtils;
+import org.apache.hadoop.hive.ql.exec.Utilities;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+
+
+/**
+ * Operation process of aborting compactions.
+ */
+public class AbortCompactionsOperation extends DDLOperation<AbortCompactionsDesc> {
+ public AbortCompactionsOperation(DDLOperationContext context, AbortCompactionsDesc desc) {
+ super(context, desc);
+ }
+
+ @Override
+ public int execute() throws HiveException {
+ AbortCompactionRequest request = new AbortCompactionRequest();
+ request.setCompactionIds(desc.getCompactionIds());
+ AbortCompactResponse response = context.getDb().abortCompactions(request);
+ try (DataOutputStream os = ShowUtils.getOutputStream(new Path(desc.getResFile()), context)) {
+ writeHeader(os);
+ if (response.getAbortedcompacts() != null) {
+ for (AbortCompactionResponseElement e : response.getAbortedcompacts()) {
+ writeRow(os, e);
+ }
+ }
+ } catch (Exception e) {
+ LOG.warn("show compactions: ", e);
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] sonarcloud[bot] commented on pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #3880:
URL: https://github.com/apache/hive/pull/3880#issuecomment-1359439919
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=3880)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG) [1 Bug](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL) [3 Code Smells](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=3880&metric=coverage&view=list) No Coverage information
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=3880&metric=duplicated_lines_density&view=list) No Duplication information
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] veghlaci05 commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "veghlaci05 (via GitHub)" <gi...@apache.org>.
veghlaci05 commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1084128577
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,110 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ addAbortCompactionResponse(abortCompactionResponseElements,x, "No Such Compaction Id Available","Error");
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(abortCompactionResponseElements, eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements,
+ CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ compactionInfo.state = TxnStore.ABORTED_STATE;
+ compactionInfo.errorMessage = "Comapction Aborted by Abort Comapction request.";
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:Unable to update compaction record in COMPLETED_COMPACTIONS", "Error");
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction: Unable to update compaction record in COMPACTION_QUEUE", "Error");
+ } else {
+ dbConn.commit();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id, "Successfully aborted compaction",
+ "Success");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:"+e.getMessage(), "Error");
+ }
+ }
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database: " + e.getMessage());
+ checkRetryable(e, "abortCompaction(" + compactionInfo + ")");
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:"+ e.getMessage(), "Error" );
+ }
+ } catch (RetryException e) {
+ abortCompaction(abortCompactionResponseElements, compactionInfo);
+ }
+ }
+
+ private List<CompactionInfo> findEligibleCompactionsToAbort(Map<Long,
+ AbortCompactionResponseElement> abortCompactionResponseElements, List<Long> requestedCompId) throws MetaException {
+
+ List<CompactionInfo> compactionInfoList = new ArrayList<>();
+ String queryText = TxnQueries.SELECT_COMPACTION_QUEUE_BY_COMPID + " WHERE \"CC_ID\" IN " +
+ "(" + Joiner.on(',').join(requestedCompId) + ") ";
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ PreparedStatement pStmt = dbConn.prepareStatement(queryText)) {
+ try (ResultSet rs = pStmt.executeQuery()) {
+ while (rs.next()) {
+ if (checkIfCompactionEligibleToAbort(rs.getString(5).charAt(0))) {
+ compactionInfoList.add(CompactionInfo.loadFullFromCompactionQueue(rs));
+ } else {
+ addAbortCompactionResponse(abortCompactionResponseElements, rs.getLong(1),
+ "Error while aborting compaction as compaction is in state-" +
+ CompactionState.fromSqlConst(rs.getString(5).charAt(0)), "Error");
+ }
+ }
+ }
+ } catch (SQLException e) {
+ throw new MetaException("Unable to select from transaction database-" + StringUtils.stringifyException(e));
+ }
+ return compactionInfoList;
+ }
+
+ private boolean checkIfCompactionEligibleToAbort(char state) {
+
+ return CompactionState.INITIATED.equals(CompactionState.fromSqlConst(state));
+ }
Review Comment:
You could eliminate this method completely. It's a one liner and there's only one usage
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,110 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ addAbortCompactionResponse(abortCompactionResponseElements,x, "No Such Compaction Id Available","Error");
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(abortCompactionResponseElements, eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements,
+ CompactionInfo compactionInfo) throws MetaException {
+ try {
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex);
+ PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ compactionInfo.state = TxnStore.ABORTED_STATE;
+ compactionInfo.errorMessage = "Comapction Aborted by Abort Comapction request.";
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, compactionInfo, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:Unable to update compaction record in COMPLETED_COMPACTIONS", "Error");
+ } else {
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, compactionInfo.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE ");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", compactionInfo, updCount);
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction: Unable to update compaction record in COMPACTION_QUEUE", "Error");
+ } else {
+ dbConn.commit();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id, "Successfully aborted compaction",
+ "Success");
+ }
+ } catch (SQLException e) {
+ dbConn.rollback();
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:"+e.getMessage(), "Error");
+ }
+ }
+ } catch (SQLException e) {
+ LOG.error("Unable to connect to transaction database: " + e.getMessage());
+ checkRetryable(e, "abortCompaction(" + compactionInfo + ")");
+ addAbortCompactionResponse(abortCompactionResponseElements, compactionInfo.id,
+ "Error while aborting compaction:"+ e.getMessage(), "Error" );
+ }
+ } catch (RetryException e) {
+ abortCompaction(abortCompactionResponseElements, compactionInfo);
+ }
+ }
+
+ private List<CompactionInfo> findEligibleCompactionsToAbort(Map<Long,
+ AbortCompactionResponseElement> abortCompactionResponseElements, List<Long> requestedCompId) throws MetaException {
+
+ List<CompactionInfo> compactionInfoList = new ArrayList<>();
+ String queryText = TxnQueries.SELECT_COMPACTION_QUEUE_BY_COMPID + " WHERE \"CC_ID\" IN " +
+ "(" + Joiner.on(',').join(requestedCompId) + ") ";
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED);
+ PreparedStatement pStmt = dbConn.prepareStatement(queryText)) {
+ try (ResultSet rs = pStmt.executeQuery()) {
+ while (rs.next()) {
+ if (checkIfCompactionEligibleToAbort(rs.getString(5).charAt(0))) {
+ compactionInfoList.add(CompactionInfo.loadFullFromCompactionQueue(rs));
+ } else {
+ addAbortCompactionResponse(abortCompactionResponseElements, rs.getLong(1),
+ "Error while aborting compaction as compaction is in state-" +
+ CompactionState.fromSqlConst(rs.getString(5).charAt(0)), "Error");
+ }
+ }
+ }
+ } catch (SQLException e) {
+ throw new MetaException("Unable to select from transaction database-" + StringUtils.stringifyException(e));
+ }
+ return compactionInfoList;
+ }
+
+ private boolean checkIfCompactionEligibleToAbort(char state) {
+
+ return CompactionState.INITIATED.equals(CompactionState.fromSqlConst(state));
+ }
+
+ private void addAbortCompactionResponse(Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements,
+ long id, String message, String status) {
+
+ abortCompactionResponseElements.put(id, new AbortCompactionResponseElement(id, status, message));
+ }
Review Comment:
This the recent changes this method become a one-liner. You could eliminate it completely
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3418,6 +3422,100 @@ public void testShowCompactionOrder() throws Exception {
Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
}
+
+ @Test
+ public void testAbortCompaction() throws Exception {
+
+ d.destroy();
+ hiveConf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, "nonstrict");
+ d = new Driver(hiveConf);
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl1 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl1" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("drop table if exists T1");
+ runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+
+ //create failed compaction attempt so that compactor txn is aborted
+ HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
+ runStatementOnDriver("alter table T1 compact 'minor'");
+ TestTxnCommands2.runWorker(hiveConf);
+ // Verify compaction order
+ List<ShowCompactResponseElement> compacts =
+ txnHandler.showCompact(new ShowCompactRequest()).getCompacts();
+ Assert.assertEquals(6, compacts.size());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, compacts.get(0).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(1).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(2).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(3).getState());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, compacts.get(4).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
+
+ long initiatedStateCompId = compacts.get(0).getId();
+ List<Long> refusedStateCompIds = Arrays.asList(compacts.get(1).getId(),compacts.get(5).getId());
+ List<Long> compactionsToAbort = Arrays.asList(Long.parseLong("12"), compacts.get(0).getId(),
+ compacts.get(1).getId(), compacts.get(2).getId(), compacts.get(3).getId(), compacts.get(4).getId(),
+ compacts.get(5).getId());
Review Comment:
You could use a single map here where the long is the id and the value is an AbortCompactionResponseElement with the expected id, status, and message.
```
Map<Long, AbortCompactionResponseElement> map = new HashMap<Long, AbortCompactionResponseElement>() {{
put(compacts.get(0).getId(),new AbortCompactionResponseElement(compacts.get(0).getId(), "Suceess", "Successfully aborted compaction"));
...
...
}};
```
That would greatly simplify the assertions below.
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,110 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ addAbortCompactionResponse(abortCompactionResponseElements,x, "No Such Compaction Id Available","Error");
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
+ abortCompaction(abortCompactionResponseElements, eligibleCompactionsToAbort.get(x));
+ }
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public void abortCompaction(Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements,
Review Comment:
You could return with AbortCompactionResponseElement instead of passing the map
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1091558263
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] veghlaci05 merged pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "veghlaci05 (via GitHub)" <gi...@apache.org>.
veghlaci05 merged PR #3880:
URL: https://github.com/apache/hive/pull/3880
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1083678743
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java:
##########
@@ -3411,6 +3415,73 @@ public void testShowCompactionOrder() throws Exception {
Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
}
+
+ @Test
+ public void testAbortCompaction() throws Exception {
+
+ d.destroy();
+ hiveConf.setVar(HiveConf.ConfVars.DYNAMICPARTITIONINGMODE, "nonstrict");
+ d = new Driver(hiveConf);
+ //generate some compaction history
+ runStatementOnDriver("drop database if exists mydb1 cascade");
+ runStatementOnDriver("create database mydb1");
+
+ runStatementOnDriver("create table mydb1.tbl0 " + "(a int, b int) partitioned by (p string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(1,2,'p1'),(3,4,'p1'),(1,2,'p2'),(3,4,'p2'),(1,2,'p3'),(3,4,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p1') compact 'MAJOR'");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p2') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION(p='p3') compact 'MAJOR'");
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(4,5,'p1'),(6,7,'p1'),(4,5,'p2'),(6,7,'p2'),(4,5,'p3'),(6,7,'p3')");
+ TestTxnCommands2.runWorker(hiveConf);
+ TestTxnCommands2.runCleaner(hiveConf);
+ runStatementOnDriver("insert into mydb1.tbl0" + " PARTITION(p) " +
+ " values(11,12,'p1'),(13,14,'p1'),(11,12,'p2'),(13,14,'p2'),(11,12,'p3'),(13,14,'p3')");
+ runStatementOnDriver("alter table mydb1.tbl0" + " PARTITION (p='p1') compact 'MINOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("create table mydb1.tbl1 " + "(a int, b int) partitioned by (ds string) clustered by (a) into " +
+ BUCKET_COUNT + " buckets stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into mydb1.tbl1" + " PARTITION(ds) " +
+ " values(1,2,'today'),(3,4,'today'),(1,2,'tomorrow'),(3,4,'tomorrow'),(1,2,'yesterday'),(3,4,'yesterday')");
+ runStatementOnDriver("alter table mydb1.tbl1" + " PARTITION(ds='today') compact 'MAJOR'");
+ TestTxnCommands2.runWorker(hiveConf);
+
+ runStatementOnDriver("drop table if exists T1");
+ runStatementOnDriver("create table T1 (a int, b int) stored as orc TBLPROPERTIES ('transactional'='true')");
+ runStatementOnDriver("insert into T1 values(0,2)");//makes delta_1_1 in T1
+ runStatementOnDriver("insert into T1 values(1,4)");//makes delta_2_2 in T2
+
+ //create failed compaction attempt so that compactor txn is aborted
+ HiveConf.setBoolVar(hiveConf, HiveConf.ConfVars.HIVETESTMODEFAILCOMPACTION, true);
+ runStatementOnDriver("alter table T1 compact 'minor'");
+ TestTxnCommands2.runWorker(hiveConf);
+ // Verify compaction order
+ List<ShowCompactResponseElement> compacts =
+ txnHandler.showCompact(new ShowCompactRequest()).getCompacts();
+ Assert.assertEquals(6, compacts.size());
+ Assert.assertEquals(TxnStore.INITIATED_RESPONSE, compacts.get(0).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(1).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(2).getState());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, compacts.get(3).getState());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, compacts.get(4).getState());
+ Assert.assertEquals(TxnStore.REFUSED_RESPONSE, compacts.get(5).getState());
+
+ List<Long> compactionsToAbort = Arrays.asList(compacts.get(0).getId(),compacts.get(1).getId(),compacts.get(3).getId());
Review Comment:
fixed
##########
ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands.java:
##########
@@ -65,6 +65,7 @@
import org.apache.hadoop.hive.metastore.api.ShowLocksResponse;
import org.apache.hadoop.hive.metastore.api.TxnInfo;
import org.apache.hadoop.hive.metastore.api.TxnState;
+import org.apache.hadoop.hive.metastore.api.AbortCompactResponse;
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "rkirtir (via GitHub)" <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1091558532
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6245,91 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error", "Not Eligible"));
Review Comment:
fixed
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6266,4 +6271,105 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ Map<Long, AbortCompactionResponseElement> abortCompactionResponseElements = new HashMap<>();
+ AbortCompactResponse response = new AbortCompactResponse(new HashMap<>());
+ response.setAbortedcompacts(abortCompactionResponseElements);
+
+ List<Long> compactionIdsToAbort = reqst.getCompactionIds();
+ if (compactionIdsToAbort.isEmpty()) {
+ LOG.info("Compaction ids are missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("Compaction ids missing in request. No compactions to abort");
+ }
+ reqst.getCompactionIds().forEach(x -> {
+ abortCompactionResponseElements.put(x, new AbortCompactionResponseElement(x, "Error",
+ "No Such Compaction Id Available"));
+ });
+
+ List<CompactionInfo> eligibleCompactionsToAbort = findEligibleCompactionsToAbort(abortCompactionResponseElements,
+ compactionIdsToAbort);
+ for (int x = 0; x < eligibleCompactionsToAbort.size(); x++) {
Review Comment:
fixed
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] rkirtir commented on a diff in pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by GitBox <gi...@apache.org>.
rkirtir commented on code in PR #3880:
URL: https://github.com/apache/hive/pull/3880#discussion_r1072022879
##########
standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/TxnHandler.java:
##########
@@ -6242,4 +6246,86 @@ public boolean isWrapperFor(Class<?> iface) throws SQLException {
}
}
+ @Override
+ @RetrySemantics.SafeToRetry
+ public AbortCompactResponse abortCompactions(AbortCompactionRequest reqst) throws MetaException, NoSuchCompactionException {
+ AbortCompactResponse response = new AbortCompactResponse(new ArrayList<>());
+ List<Long> requestedCompId = reqst.getCompactionIds();
+ if (requestedCompId.isEmpty()) {
+ LOG.info("Compaction ids missing in request. No compactions to abort");
+ throw new NoSuchCompactionException("ompaction ids missing in request. No compactions to abort");
+ }
+ List<AbortCompactionResponseElement> abortCompactionResponseElementList = new ArrayList<>();
+ for (int i = 0; i < requestedCompId.size(); i++) {
+ AbortCompactionResponseElement responseEle = abortCompaction(requestedCompId.get(i));
+ abortCompactionResponseElementList.add(responseEle);
+ }
+ response.setAbortedcompacts(abortCompactionResponseElementList);
+ return response;
+ }
+
+ @RetrySemantics.SafeToRetry
+ public AbortCompactionResponseElement abortCompaction(Long compId) throws MetaException {
+ try {
+ AbortCompactionResponseElement responseEle = new AbortCompactionResponseElement();
+ responseEle.setCompactionIds(compId);
+ try (Connection dbConn = getDbConn(Connection.TRANSACTION_READ_COMMITTED, connPoolMutex)) {
+ Optional<CompactionInfo> compactionInfo = getCompactionByCompId(dbConn, compId);
+ if (compactionInfo.isPresent()) {
+ try (PreparedStatement pStmt = dbConn.prepareStatement(TxnQueries.INSERT_INTO_COMPLETED_COMPACTION)) {
+ CompactionInfo ci = compactionInfo.get();
+ ci.errorMessage = "Compaction aborted by user";
+ ci.state = TxnStore.ABORTED_STATE;
+ CompactionInfo.insertIntoCompletedCompactions(pStmt, ci, getDbTime(dbConn));
+ int updCount = pStmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
+ dbConn.rollback();
+ }
+ LOG.debug("Inserted {} entries into COMPLETED_COMPACTIONS", updCount);
+ try (PreparedStatement stmt = dbConn.prepareStatement("DELETE FROM \"COMPACTION_QUEUE\" WHERE \"CQ_ID\" = ?")) {
+ stmt.setLong(1, ci.id);
+ LOG.debug("Going to execute update on COMPACTION_QUEUE <{}>");
+ updCount = stmt.executeUpdate();
+ if (updCount != 1) {
+ LOG.error("Unable to update compaction record: {}. updCnt={}", ci, updCount);
+ dbConn.rollback();
+ } else {
+ responseEle.setMessage("Successfully Aborted Compaction ");
+ responseEle.setStatus("Success");
+ dbConn.commit();
+ }
+ }
+ }
+ } else {
+ responseEle.setMessage("Compaction element not eligible for cancellation");
+ responseEle.setStatus("Error");
+ }
+ } catch (SQLException e) {
+ LOG.error("Failed to abort compaction request");
+ checkRetryable(e, "abortCompaction(" + compId + ")");
+ responseEle.setMessage("Error while aborting compaction");
+ responseEle.setStatus("Error");
+ }
+ return responseEle;
+ } catch (RetryException e) {
+ return abortCompaction(compId);
+ }
+
+ }
+
+ private Optional<CompactionInfo> getCompactionByCompId(Connection dbConn, Long compId) throws SQLException, MetaException {
Review Comment:
fixed.
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] sonarcloud[bot] commented on pull request #3880: HIVE-26804 : Cancel Compactions in initiated state
Posted by "sonarcloud[bot] (via GitHub)" <gi...@apache.org>.
sonarcloud[bot] commented on PR #3880:
URL: https://github.com/apache/hive/pull/3880#issuecomment-1421381807
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=3880)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG) [0 Bugs](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL) [10 Code Smells](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=3880&metric=coverage&view=list) No Coverage information
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=3880&metric=duplicated_lines_density&view=list) No Duplication information
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org
[GitHub] [hive] sonarcloud[bot] commented on pull request #3880: [WIP] HIVE-26804 : Cancel Compactions in initiated state
Posted by GitBox <gi...@apache.org>.
sonarcloud[bot] commented on PR #3880:
URL: https://github.com/apache/hive/pull/3880#issuecomment-1360912435
Kudos, SonarCloud Quality Gate passed! [](https://sonarcloud.io/dashboard?id=apache_hive&pullRequest=3880)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG) [1 Bug](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=BUG)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY) [0 Vulnerabilities](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=VULNERABILITY)
[](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT) [](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT) [0 Security Hotspots](https://sonarcloud.io/project/security_hotspots?id=apache_hive&pullRequest=3880&resolved=false&types=SECURITY_HOTSPOT)
[](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL) [](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL) [4 Code Smells](https://sonarcloud.io/project/issues?id=apache_hive&pullRequest=3880&resolved=false&types=CODE_SMELL)
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=3880&metric=coverage&view=list) No Coverage information
[](https://sonarcloud.io/component_measures?id=apache_hive&pullRequest=3880&metric=duplicated_lines_density&view=list) No Duplication information
--
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: gitbox-unsubscribe@hive.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: gitbox-unsubscribe@hive.apache.org
For additional commands, e-mail: gitbox-help@hive.apache.org