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 2020/10/02 13:45:03 UTC
[GitHub] [hive] deniskuzZ opened a new pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
deniskuzZ opened a new pull request #1548:
URL: https://github.com/apache/hive/pull/1548
<!--
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?
Below changes are only with respect to branch-3.1.
Design: taken from https://issues.apache.org/jira/secure/attachment/12954375/Aborted%20Txn%20w_Direct%20Write.pdf
**Overview:**
1. add a dummy row to TXN_COMPONENTS with operation type 'p' in enqueueLockWithRetry, which will be removed in addDynamicPartition
2. If anytime txn is aborted, this dummy entry will be block initiator to remove this txnId from TXNS
3. Initiator will add a row in COMPACTION_QUEUE (with type 'p') for the above aborted txn with the state as READY_FOR_CLEANING, at a time there will be a single entry of this type for a table in COMPACTION_QUEUE.
4. Cleaner will directly pickup above request, and process it via new cleanAborted code path(scan all partitions and remove aborted dirs), once successful cleaner will remove dummy row from TXN_COMPONENTS
**Cleaner Design:**
- We are keeping cleaner single thread, and this new type of cleanup will be handled similar to any regular cleanup
**Aborted dirs cleanup:**
- In p-type cleanup, cleaner will iterate over all the partitions and remove all delta/base dirs with given aborted writeId list
- added cleanup of aborted base/delta in the worker also
**TXN_COMPONENTS cleanup:**
- If successful, p-type entry will be removed from TXN_COMPONENTS during addDynamicPartitions
- If aborted, cleaner will clean in markCleaned after successful processing of p-type cleanup
**TXNS cleanup:**
- No change, will be cleaned up by the initiator
<!--
Please clarify what changes you are proposing. The purpose of this section is to outline the changes and how this PR fixes the issue.
If possible, please consider writing useful notes for better and faster reviews in your PR. See the examples below.
1. If you refactor some codes with changing classes, showing the class hierarchy will help reviewers.
2. If you fix some SQL features, you can provide some references of other DBMSes.
3. If there is design documentation, please add the link.
4. If there is a discussion in the mailing list, please add the link.
-->
### Why are the changes needed?
To fix above mentioned issue.
<!--
Please clarify why the changes are needed. For instance,
1. If you propose a new API, clarify the use case for a new API.
2. If you fix a bug, you can clarify why it is a bug.
-->
### Does this PR introduce _any_ user-facing change?
No
<!--
Note that it means *any* user-facing change including all aspects such as the documentation fix.
If yes, please clarify the previous behavior and the change this PR proposes - provide the console output, description, screenshot and/or a reproducable example to show the behavior difference if possible.
If possible, please also clarify if this is a user-facing change compared to the released Hive versions or within the unreleased branches such as master.
If no, write 'No'.
-->
### How was this patch tested?
unit-tests added
<!--
If tests were added, say they were added here. Please make sure to add some test cases that check the changes thoroughly including negative and positive cases if possible.
If it was tested in a way different from regular unit tests, please clarify how you tested step by step, ideally copy and paste-able, so that other reviewers can test and check, and descendants can verify in the future.
If tests were not added, please describe why they were not added and/or why it was difficult to add.
-->
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501757123
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+ }
+
+ @Test
+ public void testCleanAbortCompactSeveralTables() throws Exception {
+ String dbName = "default";
+ String tblName1 = "cws1";
+ String tblName2 = "cws2";
+
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName1, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName2, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,1".getBytes());
+ connection2.write("2,2".getBytes());
+ connection2.abortTransaction();
+
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ FileSystem fs = FileSystem.get(conf);
+ Table table1 = msClient.getTable(dbName, tblName1);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table1.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+ Table table2 = msClient.getTable(dbName, tblName2);
+ stat = fs.listStatus(new Path(table2.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table1.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCorrectlyCleaned() throws Exception {
+ // Test that at commit the tables are cleaned properly
+ String dbName = "default";
+ String tblName = "cws";
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 1, count);
+
+ connection.commitTransaction();
+
+ // After commit the row should have been deleted
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+ }
+
+ @Test
+ public void testCleanAbortAndMinorCompact() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+ connection.abortTransaction();
+
+ executeStatementOnDriver("insert into " + tblName + " partition (a) values (1, '1')", driver);
+ executeStatementOnDriver("delete from " + tblName + " where b=1", driver);
+
+ conf.setIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_DELTA_NUM_THRESHOLD, 0);
+ runInitiator(conf);
+ runWorker(conf);
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+ // Cleaning should happen in threads concurrently for the minor compaction and the clean abort one.
+ runCleaner(conf);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ }
+
+ private HiveStreamingConnection prepareTableAndConnection(String dbName, String tblName, int batchSize) throws Exception {
Review comment:
added
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499404466
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
+ return true;
+ } else if (name.startsWith(baseDir(wId)) && !name.contains("=")) {
+ return true;
+ }
+ }
+ return false;
+ };
+ List<FileStatus> deleted = new ArrayList<>();
+ deleteDeltaDirectoriesAux(rootPartition, fs, filter, deleted);
+ return deleted;
+ }
+
+ private static void deleteDeltaDirectoriesAux(Path root, FileSystem fs, PathFilter filter, List<FileStatus> deleted)
+ throws IOException {
+ RemoteIterator<FileStatus> it = listIterator(fs, root, null);
+
+ while (it.hasNext()) {
+ FileStatus fStatus = it.next();
+ if (fStatus.isDirectory()) {
+ if (filter.accept(fStatus.getPath())) {
+ fs.delete(fStatus.getPath(), true);
+ deleted.add(fStatus);
+ } else {
+ deleteDeltaDirectoriesAux(fStatus.getPath(), fs, filter, deleted);
+ if (isDirectoryEmpty(fs, fStatus.getPath())) {
Review comment:
Are we doing this to delete newly created partitions if there are no other writes? Is this ok, what if we found a valid empty partition that is registered in the HMS? We should not delete that. I think this can be skipped all together, the empty partition dir will not bother anybody
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501732794
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
Previously if you had aborted txn above threshold this would generate a "normal" compaction that would clean up everything. However now if you have one dynpart aborted the type will be CLEAN_ABORTED that will only clean the writeids belonging to p-type records and leave everything else. This will delay the normal cleaning. I am not sure that is a problem or not.
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501737923
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -414,77 +436,56 @@ public void markCleaned(CompactionInfo info) throws MetaException {
* aborted TXN_COMPONENTS above tc_writeid (and consequently about aborted txns).
* See {@link ql.txn.compactor.Cleaner.removeFiles()}
*/
- s = "SELECT DISTINCT \"TXN_ID\" FROM \"TXNS\", \"TXN_COMPONENTS\" WHERE \"TXN_ID\" = \"TC_TXNID\" "
- + "AND \"TXN_STATE\" = " + TxnStatus.ABORTED + " AND \"TC_DATABASE\" = ? AND \"TC_TABLE\" = ?";
- if (info.highestWriteId != 0) s += " AND \"TC_WRITEID\" <= ?";
- if (info.partName != null) s += " AND \"TC_PARTITION\" = ?";
+ List<String> queries = new ArrayList<>();
+ Iterator<Long> writeIdsIter = null;
+ List<Integer> counts = null;
- pStmt = dbConn.prepareStatement(s);
- paramCount = 1;
- pStmt.setString(paramCount++, info.dbname);
- pStmt.setString(paramCount++, info.tableName);
- if(info.highestWriteId != 0) {
- pStmt.setLong(paramCount++, info.highestWriteId);
+ s = "DELETE FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\" IN (" +
+ " SELECT \"TXN_ID\" FROM \"TXNS\" WHERE \"TXN_STATE\" = " + TxnStatus.ABORTED + ") " +
Review comment:
cool
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499620518
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
+ return true;
+ } else if (name.startsWith(baseDir(wId)) && !name.contains("=")) {
+ return true;
+ }
+ }
+ return false;
+ };
+ List<FileStatus> deleted = new ArrayList<>();
+ deleteDeltaDirectoriesAux(rootPartition, fs, filter, deleted);
+ return deleted;
+ }
+
+ private static void deleteDeltaDirectoriesAux(Path root, FileSystem fs, PathFilter filter, List<FileStatus> deleted)
Review comment:
getHdfsDirSnapshots does the same recursive listing, isn't it?
```
RemoteIterator<LocatedFileStatus> itr = fs.listFiles(path, true);
while (itr.hasNext()) {
```
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499827644
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
##########
@@ -237,6 +237,7 @@ void run(HiveConf conf, String jobName, Table t, Partition p, StorageDescriptor
}
JobConf job = createBaseJobConf(conf, jobName, t, sd, writeIds, ci);
+ QueryCompactor.Util.removeAbortedDirsForAcidTable(conf, dir);
Review comment:
removed
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499623864
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
Why would it read the aborted data as valid if txn is in aborted state?
----------------------------------------------------------------
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.
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] deniskuzZ commented on pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on pull request #1548:
URL: https://github.com/apache/hive/pull/1548#issuecomment-705545322
looks like master is broken right now
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499754423
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
changed, also excluded base directory from listing
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501713463
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
Review comment:
I think this assert is quit misleading. I might be wrong but the recursive listing skips empty directories, and actually this new version of cleaning will keep the partition directories (it should) and only delete the delta dirs and files.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499639831
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
+ return true;
+ } else if (name.startsWith(baseDir(wId)) && !name.contains("=")) {
+ return true;
+ }
+ }
+ return false;
+ };
+ List<FileStatus> deleted = new ArrayList<>();
+ deleteDeltaDirectoriesAux(rootPartition, fs, filter, deleted);
+ return deleted;
+ }
+
+ private static void deleteDeltaDirectoriesAux(Path root, FileSystem fs, PathFilter filter, List<FileStatus> deleted)
+ throws IOException {
+ RemoteIterator<FileStatus> it = listIterator(fs, root, null);
+
+ while (it.hasNext()) {
+ FileStatus fStatus = it.next();
+ if (fStatus.isDirectory()) {
+ if (filter.accept(fStatus.getPath())) {
+ fs.delete(fStatus.getPath(), true);
+ deleted.add(fStatus);
+ } else {
+ deleteDeltaDirectoriesAux(fStatus.getPath(), fs, filter, deleted);
+ if (isDirectoryEmpty(fs, fStatus.getPath())) {
Review comment:
agree, that would simplify re-use of getHdfsDirSnapshots
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501731802
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+ }
+
+ @Test
+ public void testCleanAbortCompactSeveralTables() throws Exception {
+ String dbName = "default";
+ String tblName1 = "cws1";
+ String tblName2 = "cws2";
+
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName1, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName2, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,1".getBytes());
+ connection2.write("2,2".getBytes());
+ connection2.abortTransaction();
+
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ FileSystem fs = FileSystem.get(conf);
+ Table table1 = msClient.getTable(dbName, tblName1);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table1.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+ Table table2 = msClient.getTable(dbName, tblName2);
+ stat = fs.listStatus(new Path(table2.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table1.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCorrectlyCleaned() throws Exception {
+ // Test that at commit the tables are cleaned properly
+ String dbName = "default";
+ String tblName = "cws";
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
Review comment:
fixed
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
Review comment:
yes, recursive listing skips empty directories - that's was done intentionally. changed the comment
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499754423
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
changed, included delete_delta as well
----------------------------------------------------------------
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.
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] vpnvishv commented on pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
vpnvishv commented on pull request #1548:
URL: https://github.com/apache/hive/pull/1548#issuecomment-704423354
@deniskuzZ Yes.. Sorry I missed those while porting changes from my private branch to open source Hive-3.
I think you have already covered it now. Still if you want to refer I have Updated Hive-3 PR with those tests, https://github.com/apache/hive/pull/1415/commits/4596b50e0f4e00c363a744bad13a8c750330d9a0
----------------------------------------------------------------
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.
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] deniskuzZ commented on pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on pull request #1548:
URL: https://github.com/apache/hive/pull/1548#issuecomment-702804902
@vpnvishv , could you please check.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501766620
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -232,6 +240,51 @@ public Object run() throws Exception {
private static String idWatermark(CompactionInfo ci) {
return " id=" + ci.id;
}
+
+ private void cleanAborted(CompactionInfo ci) throws MetaException {
+ if (ci.writeIds == null || ci.writeIds.size() == 0) {
+ LOG.warn("Attempted cleaning aborted transaction with empty writeId list");
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499475796
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
I might be mistaken here, but does this mean, that if we have many "normal" aborted txn and 1 aborted dynpart txn, we will not initiate a normal compaction until the dynpart stuff is not cleaned up? Is this ok, shouldn't we doing both?
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499754423
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
changed, included delete_delta as well
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed. @vpnvishv is that correct?
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed. @vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
1. In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed. @vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
2. was related to the following issue based on Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() design:
"Suppose you have p-type clean on table T that is running (i.e. has the Write lock) and you have 30 different partition clean requests (in T). The 30 per partition cleans will get blocked but they will tie up every thread in the pool while they are blocked, right? If so, no other clean (on any other table) will actually make progress until the p-type on T is done."
I think, it's not valid now.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
1. In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed:
https://github.com/apache/hive/pull/1548/files#diff-9cf3ae764b7a33b568a984d695aff837R328
@vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
2. was related to the following issue based on Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() design:
"Suppose you have p-type clean on table T that is running (i.e. has the Write lock) and you have 30 different partition clean requests (in T). The 30 per partition cleans will get blocked but they will tie up every thread in the pool while they are blocked, right? If so, no other clean (on any other table) will actually make progress until the p-type on T is done."
I think, it's not valid now.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
1. In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed:
https://github.com/apache/hive/pull/1548/files#diff-9cf3ae764b7a33b568a984d695aff837R328
@vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
2. was related to the following issue based on Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() design:
"Suppose you have p-type clean on table T that is running (i.e. has the Write lock) and you have 30 different partition clean requests (in T). The 30 per partition cleans will get blocked but they will tie up every thread in the pool while they are blocked, right? If so, no other clean (on any other table) will actually make progress until the p-type on T is done."
Yes, it's still the case that we'll have to wait for all tasks to complete and if there is one slow task, we won't be able to submit new tasks. However not sure if it's a critical issue. I think, we can address it in a separate jira.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
1. In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed:
https://github.com/apache/hive/pull/1548/files#diff-9cf3ae764b7a33b568a984d695aff837R328
@vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
2. was related to the following issue based on Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() design:
"Suppose you have p-type clean on table T that is running (i.e. has the Write lock) and you have 30 different partition clean requests (in T). The 30 per partition cleans will get blocked but they will tie up every thread in the pool while they are blocked, right? If so, no other clean (on any other table) will actually make progress until the p-type on T is done."
Yes, it's still the case that we'll have to wait for all tasks to complete and if there is one long-running task, we won't be able to submit new ones. However not sure if it's a critical issue. I think, we can address it in a separate jira.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -232,6 +240,51 @@ public Object run() throws Exception {
private static String idWatermark(CompactionInfo ci) {
return " id=" + ci.id;
}
+
+ private void cleanAborted(CompactionInfo ci) throws MetaException {
+ if (ci.writeIds == null || ci.writeIds.size() == 0) {
+ LOG.warn("Attempted cleaning aborted transaction with empty writeId list");
Review comment:
yep, good catch!
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+ }
+
+ @Test
+ public void testCleanAbortCompactSeveralTables() throws Exception {
+ String dbName = "default";
+ String tblName1 = "cws1";
+ String tblName2 = "cws2";
+
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName1, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName2, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,1".getBytes());
+ connection2.write("2,2".getBytes());
+ connection2.abortTransaction();
+
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ FileSystem fs = FileSystem.get(conf);
+ Table table1 = msClient.getTable(dbName, tblName1);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table1.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+ Table table2 = msClient.getTable(dbName, tblName2);
+ stat = fs.listStatus(new Path(table2.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table1.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCorrectlyCleaned() throws Exception {
+ // Test that at commit the tables are cleaned properly
+ String dbName = "default";
+ String tblName = "cws";
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
Review comment:
fixed
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
Review comment:
yes, recursive listing skips empty directories - that's was done intentionally. changed the comment
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+ }
+
+ @Test
+ public void testCleanAbortCompactSeveralTables() throws Exception {
+ String dbName = "default";
+ String tblName1 = "cws1";
+ String tblName2 = "cws2";
+
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName1, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName2, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,1".getBytes());
+ connection2.write("2,2".getBytes());
+ connection2.abortTransaction();
+
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ FileSystem fs = FileSystem.get(conf);
+ Table table1 = msClient.getTable(dbName, tblName1);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table1.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+ Table table2 = msClient.getTable(dbName, tblName2);
+ stat = fs.listStatus(new Path(table2.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table1.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCorrectlyCleaned() throws Exception {
+ // Test that at commit the tables are cleaned properly
+ String dbName = "default";
+ String tblName = "cws";
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 1, count);
+
+ connection.commitTransaction();
+
+ // After commit the row should have been deleted
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+ }
+
+ @Test
+ public void testCleanAbortAndMinorCompact() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+ connection.abortTransaction();
+
+ executeStatementOnDriver("insert into " + tblName + " partition (a) values (1, '1')", driver);
+ executeStatementOnDriver("delete from " + tblName + " where b=1", driver);
+
+ conf.setIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_DELTA_NUM_THRESHOLD, 0);
+ runInitiator(conf);
+ runWorker(conf);
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+ // Cleaning should happen in threads concurrently for the minor compaction and the clean abort one.
+ runCleaner(conf);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ }
+
+ private HiveStreamingConnection prepareTableAndConnection(String dbName, String tblName, int batchSize) throws Exception {
Review comment:
added
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -232,6 +240,51 @@ public Object run() throws Exception {
private static String idWatermark(CompactionInfo ci) {
return " id=" + ci.id;
}
+
+ private void cleanAborted(CompactionInfo ci) throws MetaException {
+ if (ci.writeIds == null || ci.writeIds.size() == 0) {
+ LOG.warn("Attempted cleaning aborted transaction with empty writeId list");
Review comment:
fixed
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
I still don't follow. Aborted txn check is done per db/table/partition, so if you have db1/tbl1/{p1-p100}/type=NOT_DP and db1/tbl1/null/type=DP - that should generate 2 entries in potential compactions.
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
oh, sorry, I only considered time based threshold
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
oh, sorry, I only considered time based threshold for DYN_PART
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
I still don't follow. Aborted txn check is done per db/table/partition, so if you have db1/tbl1/p1/type=NOT_DP and db1/tbl1/null/type=DP - that should generate 2 entries in potential compactions.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499747334
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
why is that? aborted dynPart is just a special case that would be handled separately (IS_DP=1).
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501646861
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed. @vpnvishv is that correct?
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501723853
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -232,6 +240,51 @@ public Object run() throws Exception {
private static String idWatermark(CompactionInfo ci) {
return " id=" + ci.id;
}
+
+ private void cleanAborted(CompactionInfo ci) throws MetaException {
+ if (ci.writeIds == null || ci.writeIds.size() == 0) {
+ LOG.warn("Attempted cleaning aborted transaction with empty writeId list");
Review comment:
Shouldn't you mark the compaction failed or cleaned?
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499544579
##########
File path: ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
##########
@@ -2128,6 +2128,395 @@ public void testCleanerForTxnToWriteId() throws Exception {
0, TxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));
}
+ @Test
+ public void testMmTableAbortWithCompaction() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}};
+ verifyDeltaDirAndResult(1, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // Verify query result
+ int [][] resultData2 = new int[][] {{1,2}, {5,6}};
+
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData2);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
Review comment:
fixed, turned off StatsOptimizer
##########
File path: ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
##########
@@ -2128,6 +2128,395 @@ public void testCleanerForTxnToWriteId() throws Exception {
0, TxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));
}
+ @Test
+ public void testMmTableAbortWithCompaction() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}};
+ verifyDeltaDirAndResult(1, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // Verify query result
+ int [][] resultData2 = new int[][] {{1,2}, {5,6}};
+
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData2);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 4. Perform a MINOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'");
+ // The worker should remove the subdir for aborted transaction
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData2);
+ verifyBaseDirAndResult(0, Table.MMTBL.toString(), "", resultData2);
+ // 5. Run Cleaner. Shouldn't impact anything.
+ runCleaner(hiveConf);
+ // 6. Run initiator remove aborted entry from TXNS table
+ runInitiator(hiveConf);
+
+ // Verify query result
+ List<String> rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData2), rs);
+
+ int [][] resultData3 = new int[][] {{1,2}, {5,6}, {7,8}};
+ // 7. add few more rows
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(7,8)");
+ // 8. add one more aborted delta
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(9,10)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+
+ // 9. Perform a MAJOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MAJOR'");
+ verifyDeltaDirAndResult(4, Table.MMTBL.toString(), "", resultData3);
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+ runCleaner(hiveConf);
+ verifyDeltaDirAndResult(0, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+ runInitiator(hiveConf);
+ verifyDeltaDirAndResult(0, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+
+ // Verify query result
+ rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData3), rs);
+ }
+ @Test
+ public void testMmTableAbortWithCompactionNoCleanup() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}, {5,6}};
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + " values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData1);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 3. Perform a MINOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'");
+ // The worker should remove the subdir for aborted transaction
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+ verifyBaseDirAndResult(0, Table.MMTBL.toString(), "", resultData1);
+ // Verify query result
+ List<String> rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData1), rs);
+
+ int [][] resultData3 = new int[][] {{1,2}, {5,6}, {7,8}};
+ // 4. add few more rows
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(7,8)");
+ // 5. add one more aborted delta
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(9,10)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ verifyDeltaDirAndResult(4, Table.MMTBL.toString(), "", resultData3);
+
+ // 6. Perform a MAJOR compaction, expectation is it should remove aborted delta dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MAJOR'");
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+
+ // 7. Run one more Major compaction this should not have any affect
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MAJOR'");
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+
+ runCleaner(hiveConf);
+
+ // Verify query result
+ rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData3), rs);
+ }
+
+ @Test
+ public void testFullACIDAbortWithMinorMajorCompaction() throws Exception {
+ // 1. Insert some rows into acid table
+ runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(1,2)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}};
+ verifyDeltaDirAndResult(1, Table.ACIDTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.ACIDTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 2 delta directories.
+ verifyDeltaDirAndResult(2, Table.ACIDTBL.toString(), "", resultData1);
+
+ r1 = runStatementOnDriver("select count(*) from " + Table.ACIDTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // Verify query result
+ int [][] resultData2 = new int[][] {{1,2}, {5,6}};
+ // 3. insert few more rows in acid table
+ runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(5,6)");
+ verifyDeltaDirAndResult(3, Table.ACIDTBL.toString(), "", resultData2);
+ r1 = runStatementOnDriver("select count(*) from " + Table.ACIDTBL);
+ Assert.assertEquals("2", r1.get(0));
Review comment:
fixed, turned of StatsOptimizer
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501884660
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
oh, sorry, I only considered time based threshold for DYN_PART
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499545127
##########
File path: ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
##########
@@ -2128,6 +2128,395 @@ public void testCleanerForTxnToWriteId() throws Exception {
0, TxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));
}
+ @Test
+ public void testMmTableAbortWithCompaction() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}};
+ verifyDeltaDirAndResult(1, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // Verify query result
+ int [][] resultData2 = new int[][] {{1,2}, {5,6}};
+
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData2);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 4. Perform a MINOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'");
+ // The worker should remove the subdir for aborted transaction
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData2);
+ verifyBaseDirAndResult(0, Table.MMTBL.toString(), "", resultData2);
+ // 5. Run Cleaner. Shouldn't impact anything.
+ runCleaner(hiveConf);
+ // 6. Run initiator remove aborted entry from TXNS table
+ runInitiator(hiveConf);
+
+ // Verify query result
+ List<String> rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData2), rs);
+
+ int [][] resultData3 = new int[][] {{1,2}, {5,6}, {7,8}};
+ // 7. add few more rows
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(7,8)");
+ // 8. add one more aborted delta
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(9,10)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+
+ // 9. Perform a MAJOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MAJOR'");
+ verifyDeltaDirAndResult(4, Table.MMTBL.toString(), "", resultData3);
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+ runCleaner(hiveConf);
+ verifyDeltaDirAndResult(0, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+ runInitiator(hiveConf);
+ verifyDeltaDirAndResult(0, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+
+ // Verify query result
+ rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData3), rs);
+ }
+ @Test
+ public void testMmTableAbortWithCompactionNoCleanup() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}, {5,6}};
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + " values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData1);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 3. Perform a MINOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'");
+ // The worker should remove the subdir for aborted transaction
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499623864
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
Why would it read the aborted data as valid if txn is in still in aborted state?
----------------------------------------------------------------
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.
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] vpnvishv commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
vpnvishv commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499137459
##########
File path: ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
##########
@@ -2128,6 +2128,395 @@ public void testCleanerForTxnToWriteId() throws Exception {
0, TxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));
}
+ @Test
+ public void testMmTableAbortWithCompaction() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}};
+ verifyDeltaDirAndResult(1, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // Verify query result
+ int [][] resultData2 = new int[][] {{1,2}, {5,6}};
+
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData2);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 4. Perform a MINOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'");
+ // The worker should remove the subdir for aborted transaction
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData2);
+ verifyBaseDirAndResult(0, Table.MMTBL.toString(), "", resultData2);
+ // 5. Run Cleaner. Shouldn't impact anything.
+ runCleaner(hiveConf);
+ // 6. Run initiator remove aborted entry from TXNS table
+ runInitiator(hiveConf);
+
+ // Verify query result
+ List<String> rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData2), rs);
+
+ int [][] resultData3 = new int[][] {{1,2}, {5,6}, {7,8}};
+ // 7. add few more rows
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(7,8)");
+ // 8. add one more aborted delta
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(9,10)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+
+ // 9. Perform a MAJOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MAJOR'");
+ verifyDeltaDirAndResult(4, Table.MMTBL.toString(), "", resultData3);
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+ runCleaner(hiveConf);
+ verifyDeltaDirAndResult(0, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+ runInitiator(hiveConf);
+ verifyDeltaDirAndResult(0, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+
+ // Verify query result
+ rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData3), rs);
+ }
+ @Test
+ public void testMmTableAbortWithCompactionNoCleanup() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}, {5,6}};
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + " values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData1);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 3. Perform a MINOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'");
+ // The worker should remove the subdir for aborted transaction
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
Review comment:
Here we need some correction in expected values, as these tests were written wrt to branch-3.1 in which we don't support minor compaction for MM tables.
##########
File path: ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
##########
@@ -2128,6 +2128,395 @@ public void testCleanerForTxnToWriteId() throws Exception {
0, TxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));
}
+ @Test
+ public void testMmTableAbortWithCompaction() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}};
+ verifyDeltaDirAndResult(1, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // Verify query result
+ int [][] resultData2 = new int[][] {{1,2}, {5,6}};
+
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData2);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 4. Perform a MINOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'");
+ // The worker should remove the subdir for aborted transaction
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData2);
+ verifyBaseDirAndResult(0, Table.MMTBL.toString(), "", resultData2);
+ // 5. Run Cleaner. Shouldn't impact anything.
+ runCleaner(hiveConf);
+ // 6. Run initiator remove aborted entry from TXNS table
+ runInitiator(hiveConf);
+
+ // Verify query result
+ List<String> rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData2), rs);
+
+ int [][] resultData3 = new int[][] {{1,2}, {5,6}, {7,8}};
+ // 7. add few more rows
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(7,8)");
+ // 8. add one more aborted delta
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(9,10)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+
+ // 9. Perform a MAJOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MAJOR'");
+ verifyDeltaDirAndResult(4, Table.MMTBL.toString(), "", resultData3);
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+ runCleaner(hiveConf);
+ verifyDeltaDirAndResult(0, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+ runInitiator(hiveConf);
+ verifyDeltaDirAndResult(0, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+
+ // Verify query result
+ rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData3), rs);
+ }
+ @Test
+ public void testMmTableAbortWithCompactionNoCleanup() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}, {5,6}};
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + " values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData1);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
+
+ // 3. Perform a MINOR compaction, expectation is it should remove aborted base dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MINOR'");
+ // The worker should remove the subdir for aborted transaction
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+ verifyBaseDirAndResult(0, Table.MMTBL.toString(), "", resultData1);
+ // Verify query result
+ List<String> rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData1), rs);
+
+ int [][] resultData3 = new int[][] {{1,2}, {5,6}, {7,8}};
+ // 4. add few more rows
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(7,8)");
+ // 5. add one more aborted delta
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(9,10)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ verifyDeltaDirAndResult(4, Table.MMTBL.toString(), "", resultData3);
+
+ // 6. Perform a MAJOR compaction, expectation is it should remove aborted delta dir
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MAJOR'");
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+
+ // 7. Run one more Major compaction this should not have any affect
+ runStatementOnDriver("alter table "+ Table.MMTBL + " compact 'MAJOR'");
+ runWorker(hiveConf);
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData3);
+ verifyBaseDirAndResult(1, Table.MMTBL.toString(), "", resultData3);
+
+ runCleaner(hiveConf);
+
+ // Verify query result
+ rs = runStatementOnDriver("select a,b from " + Table.MMTBL + " order by a");
+ Assert.assertEquals(stringifyValues(resultData3), rs);
+ }
+
+ @Test
+ public void testFullACIDAbortWithMinorMajorCompaction() throws Exception {
+ // 1. Insert some rows into acid table
+ runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(1,2)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}};
+ verifyDeltaDirAndResult(1, Table.ACIDTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.ACIDTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 2 delta directories.
+ verifyDeltaDirAndResult(2, Table.ACIDTBL.toString(), "", resultData1);
+
+ r1 = runStatementOnDriver("select count(*) from " + Table.ACIDTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // Verify query result
+ int [][] resultData2 = new int[][] {{1,2}, {5,6}};
+ // 3. insert few more rows in acid table
+ runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) values(5,6)");
+ verifyDeltaDirAndResult(3, Table.ACIDTBL.toString(), "", resultData2);
+ r1 = runStatementOnDriver("select count(*) from " + Table.ACIDTBL);
+ Assert.assertEquals("2", r1.get(0));
Review comment:
Again, here expected value looks correct to me and this looks like a genuine issue. Is this due to some incorrect stats, any idea?
##########
File path: ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
##########
@@ -2128,6 +2128,395 @@ public void testCleanerForTxnToWriteId() throws Exception {
0, TxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));
}
+ @Test
+ public void testMmTableAbortWithCompaction() throws Exception {
+ // 1. Insert some rows into MM table
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(1,2)");
+ // There should be 1 delta directory
+ int [][] resultData1 = new int[][] {{1,2}};
+ verifyDeltaDirAndResult(1, Table.MMTBL.toString(), "", resultData1);
+ List<String> r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // 2. Let a transaction be aborted
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, true);
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(3,4)");
+ hiveConf.setBoolVar(HiveConf.ConfVars.HIVETESTMODEROLLBACKTXN, false);
+ // There should be 1 delta and 1 base directory. The base one is the aborted one.
+ verifyDeltaDirAndResult(2, Table.MMTBL.toString(), "", resultData1);
+
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("1", r1.get(0));
+
+ // Verify query result
+ int [][] resultData2 = new int[][] {{1,2}, {5,6}};
+
+ runStatementOnDriver("insert into " + Table.MMTBL + "(a,b) values(5,6)");
+ verifyDeltaDirAndResult(3, Table.MMTBL.toString(), "", resultData2);
+ r1 = runStatementOnDriver("select count(*) from " + Table.MMTBL);
+ Assert.assertEquals("2", r1.get(0));
Review comment:
Here expected value looks correct to me and this looks like a genuine issue. Is this due to some incorrect stats, any idea?
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r508809944
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
##########
@@ -589,4 +593,9 @@ private void checkInterrupt() throws InterruptedException {
throw new InterruptedException("Compaction execution is interrupted");
}
}
-}
+
+ private static boolean isDynPartAbort(Table t, CompactionInfo ci) {
Review comment:
could be, do you know if there is some helper class I could move isDynPartAbort method?
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r508805163
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -414,76 +403,30 @@ public void markCleaned(CompactionInfo info) throws MetaException {
* aborted TXN_COMPONENTS above tc_writeid (and consequently about aborted txns).
* See {@link ql.txn.compactor.Cleaner.removeFiles()}
*/
- s = "SELECT DISTINCT \"TXN_ID\" FROM \"TXNS\", \"TXN_COMPONENTS\" WHERE \"TXN_ID\" = \"TC_TXNID\" "
- + "AND \"TXN_STATE\" = " + TxnStatus.ABORTED + " AND \"TC_DATABASE\" = ? AND \"TC_TABLE\" = ?";
- if (info.highestWriteId != 0) s += " AND \"TC_WRITEID\" <= ?";
- if (info.partName != null) s += " AND \"TC_PARTITION\" = ?";
-
+ s = "DELETE FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\" IN (" +
Review comment:
this is an optimization that makes everything in 1 db request instead of 2 (select + delete)
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501883427
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
I still don't follow. Aborted txn check is done per db/table/partition, so if you have db1/tbl1/{p1-p100}/type=NOT_DP and db1/tbl1/null/type=DP - that should generate 2 entries in potential compactions.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r508809944
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
##########
@@ -589,4 +593,9 @@ private void checkInterrupt() throws InterruptedException {
throw new InterruptedException("Compaction execution is interrupted");
}
}
-}
+
+ private static boolean isDynPartAbort(Table t, CompactionInfo ci) {
Review comment:
those are actually 2 diff methods the only common part is the check for isDynPart. Also there is no CompactionUtils only CompactorUtil, that contains thread factory stuff.
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499671848
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
You are right, I got confused, the p entry will solve this.
----------------------------------------------------------------
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.
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] deniskuzZ edited a comment on pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ edited a comment on pull request #1548:
URL: https://github.com/apache/hive/pull/1548#issuecomment-704456432
> @deniskuzZ Yes.. Sorry I missed those while porting changes from my private branch to open source [Hive-3](https://issues.apache.org/jira/browse/HIVE-3).
> I think you have already covered it now. Still if you want to refer I have Updated [Hive-3](https://issues.apache.org/jira/browse/HIVE-3) PR with those tests, [4596b50](https://github.com/apache/hive/commit/4596b50e0f4e00c363a744bad13a8c750330d9a0)
Thanks @vpnvishv, I'll check them. Also noticed that new test property is not actually configured + delete_delta isn't cleaned up. Fixed in master patch.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r508804039
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -400,11 +389,11 @@ public void markCleaned(CompactionInfo info) throws MetaException {
pStmt.setString(paramCount++, info.partName);
}
if(info.highestWriteId != 0) {
- pStmt.setLong(paramCount++, info.highestWriteId);
+ pStmt.setLong(paramCount, info.highestWriteId);
Review comment:
redundant post increment
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -134,9 +132,6 @@ public CompactionTxnHandler() {
response.add(info);
}
}
-
- LOG.debug("Going to rollback");
- dbConn.rollback();
Review comment:
no idea :)
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499457494
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
##########
@@ -237,6 +237,7 @@ void run(HiveConf conf, String jobName, Table t, Partition p, StorageDescriptor
}
JobConf job = createBaseJobConf(conf, jobName, t, sd, writeIds, ci);
+ QueryCompactor.Util.removeAbortedDirsForAcidTable(conf, dir);
Review comment:
@vpnvishv Why do we do this here? I understand we can, but why don't we let the Cleaner to delete the files? This just makes the compactor slower. Do we have a functionality reason for this?
After this change it will run in CompactorMR and in MMQueryCompactors, but not in normal QueryCompactors?
----------------------------------------------------------------
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.
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] klcopp commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
klcopp commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r508641433
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Worker.java
##########
@@ -589,4 +593,9 @@ private void checkInterrupt() throws InterruptedException {
throw new InterruptedException("Compaction execution is interrupted");
}
}
-}
+
+ private static boolean isDynPartAbort(Table t, CompactionInfo ci) {
Review comment:
This can be consolidated with most of isDynPartIngest in CompactionUtils
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -400,11 +389,11 @@ public void markCleaned(CompactionInfo info) throws MetaException {
pStmt.setString(paramCount++, info.partName);
}
if(info.highestWriteId != 0) {
- pStmt.setLong(paramCount++, info.highestWriteId);
+ pStmt.setLong(paramCount, info.highestWriteId);
Review comment:
Why was this changed?
##########
File path: ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java
##########
@@ -2128,24 +2129,601 @@ public void testCleanerForTxnToWriteId() throws Exception {
0, TxnDbUtil.countQueryAgent(hiveConf, "select count(*) from TXN_TO_WRITE_ID"));
}
- private void verifyDirAndResult(int expectedDeltas) throws Exception {
- FileSystem fs = FileSystem.get(hiveConf);
- // Verify the content of subdirs
- FileStatus[] status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" +
- (Table.MMTBL).toString().toLowerCase()), FileUtils.HIDDEN_FILES_PATH_FILTER);
+ @Test
+ public void testMmTableAbortWithCompaction() throws Exception {
Review comment:
FYI MM tests are usually in TestTxnCommandsForMmTable.java but I don't really care about this
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -134,9 +132,6 @@ public CompactionTxnHandler() {
response.add(info);
}
}
-
- LOG.debug("Going to rollback");
- dbConn.rollback();
Review comment:
Any ideas about why this was here? Just curious
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -414,76 +403,30 @@ public void markCleaned(CompactionInfo info) throws MetaException {
* aborted TXN_COMPONENTS above tc_writeid (and consequently about aborted txns).
* See {@link ql.txn.compactor.Cleaner.removeFiles()}
*/
- s = "SELECT DISTINCT \"TXN_ID\" FROM \"TXNS\", \"TXN_COMPONENTS\" WHERE \"TXN_ID\" = \"TC_TXNID\" "
- + "AND \"TXN_STATE\" = " + TxnStatus.ABORTED + " AND \"TC_DATABASE\" = ? AND \"TC_TABLE\" = ?";
- if (info.highestWriteId != 0) s += " AND \"TC_WRITEID\" <= ?";
- if (info.partName != null) s += " AND \"TC_PARTITION\" = ?";
-
+ s = "DELETE FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\" IN (" +
Review comment:
This is just refactoring right? LGTM but can you make sure @pvary sees this as well?
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501884660
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
oh, sorry, I only considered time based threshold
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499405728
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
+ return true;
+ } else if (name.startsWith(baseDir(wId)) && !name.contains("=")) {
+ return true;
+ }
+ }
+ return false;
+ };
+ List<FileStatus> deleted = new ArrayList<>();
+ deleteDeltaDirectoriesAux(rootPartition, fs, filter, deleted);
+ return deleted;
+ }
+
+ private static void deleteDeltaDirectoriesAux(Path root, FileSystem fs, PathFilter filter, List<FileStatus> deleted)
+ throws IOException {
+ RemoteIterator<FileStatus> it = listIterator(fs, root, null);
+
+ while (it.hasNext()) {
+ FileStatus fStatus = it.next();
+ if (fStatus.isDirectory()) {
+ if (filter.accept(fStatus.getPath())) {
+ fs.delete(fStatus.getPath(), true);
+ deleted.add(fStatus);
+ } else {
+ deleteDeltaDirectoriesAux(fStatus.getPath(), fs, filter, deleted);
+ if (isDirectoryEmpty(fs, fStatus.getPath())) {
+ fs.delete(fStatus.getPath(), false);
+ deleted.add(fStatus);
+ }
+ }
+ }
+ }
+ }
+
+ private static boolean isDirectoryEmpty(FileSystem fs, Path path) throws IOException {
+ RemoteIterator<FileStatus> it = listIterator(fs, path, null);
+ return !it.hasNext();
+ }
+
+ private static RemoteIterator<FileStatus> listIterator(FileSystem fs, Path path, PathFilter filter)
+ throws IOException {
+ try {
+ return new ToFileStatusIterator(SHIMS.listLocatedHdfsStatusIterator(fs, path, filter));
+ } catch (Throwable t) {
Review comment:
This should be similar to tryListLocatedHdfsStatus don't catch all Throwable. And maybe add all this to the HdfsUtils class
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501715311
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+ }
+
+ @Test
+ public void testCleanAbortCompactSeveralTables() throws Exception {
+ String dbName = "default";
+ String tblName1 = "cws1";
+ String tblName2 = "cws2";
+
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName1, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName2, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,1".getBytes());
+ connection2.write("2,2".getBytes());
+ connection2.abortTransaction();
+
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ FileSystem fs = FileSystem.get(conf);
+ Table table1 = msClient.getTable(dbName, tblName1);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table1.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+ Table table2 = msClient.getTable(dbName, tblName2);
+ stat = fs.listStatus(new Path(table2.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table1.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCorrectlyCleaned() throws Exception {
+ // Test that at commit the tables are cleaned properly
+ String dbName = "default";
+ String tblName = "cws";
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
Review comment:
this comment is copied I guess
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501731697
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -232,6 +240,51 @@ public Object run() throws Exception {
private static String idWatermark(CompactionInfo ci) {
return " id=" + ci.id;
}
+
+ private void cleanAborted(CompactionInfo ci) throws MetaException {
+ if (ci.writeIds == null || ci.writeIds.size() == 0) {
+ LOG.warn("Attempted cleaning aborted transaction with empty writeId list");
Review comment:
yep, good catch!
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499449636
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
Two questions here:
1. In the original Jira there was discussion about not allowing concurrent cleanings of the same stuff (partition / table). Should we worry about this?
2. The slow cleanAborted will clog the executor service, we should do something about this, either in this patch, or follow up something like https://issues.apache.org/jira/browse/HIVE-21150 immediately after this.
----------------------------------------------------------------
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.
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] deniskuzZ commented on pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on pull request #1548:
URL: https://github.com/apache/hive/pull/1548#issuecomment-705545322
looks like master is broken right now
----------------------------------------------------------------
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.
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] deniskuzZ commented on pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on pull request #1548:
URL: https://github.com/apache/hive/pull/1548#issuecomment-704246954
@vpnvishv I can't find any tests that would actually cover dynamic part abort. HIVETESTMODEFAILLOADDYNAMICPARTITION is not used anywhere. Am I missing something 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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499402826
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
I am wondering are we covering all the use cases here? Is it possible that this dynamic part query was writing to an existing partition with existing older writes and a compaction was running before we managed to delete the aborted delta? I think in this case sadly, we still going to read the aborted data as valid. Could you add a test case to check if it is indeed a problem or not? (I do not have an idea for a solution...)
----------------------------------------------------------------
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.
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] vpnvishv commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
vpnvishv commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499649919
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/CompactorMR.java
##########
@@ -237,6 +237,7 @@ void run(HiveConf conf, String jobName, Table t, Partition p, StorageDescriptor
}
JobConf job = createBaseJobConf(conf, jobName, t, sd, writeIds, ci);
+ QueryCompactor.Util.removeAbortedDirsForAcidTable(conf, dir);
Review comment:
@pvargacl You are right, this is not required, as now compactor run in a transaction and the cleaner has validTxnList with aborted bits set. This we have added wrt to Hive-3, in which cleaner doesn't have aborted bits set, as we create validWriteIdList for cleaner based on the highestWriteId.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499754946
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
+ return true;
+ } else if (name.startsWith(baseDir(wId)) && !name.contains("=")) {
+ return true;
+ }
+ }
+ return false;
+ };
+ List<FileStatus> deleted = new ArrayList<>();
+ deleteDeltaDirectoriesAux(rootPartition, fs, filter, deleted);
+ return deleted;
+ }
+
+ private static void deleteDeltaDirectoriesAux(Path root, FileSystem fs, PathFilter filter, List<FileStatus> deleted)
+ throws IOException {
+ RemoteIterator<FileStatus> it = listIterator(fs, root, null);
+
+ while (it.hasNext()) {
+ FileStatus fStatus = it.next();
+ if (fStatus.isDirectory()) {
+ if (filter.accept(fStatus.getPath())) {
+ fs.delete(fStatus.getPath(), true);
+ deleted.add(fStatus);
+ } else {
+ deleteDeltaDirectoriesAux(fStatus.getPath(), fs, filter, deleted);
+ if (isDirectoryEmpty(fs, fStatus.getPath())) {
+ fs.delete(fStatus.getPath(), false);
+ deleted.add(fStatus);
+ }
+ }
+ }
+ }
+ }
+
+ private static boolean isDirectoryEmpty(FileSystem fs, Path path) throws IOException {
+ RemoteIterator<FileStatus> it = listIterator(fs, path, null);
+ return !it.hasNext();
+ }
+
+ private static RemoteIterator<FileStatus> listIterator(FileSystem fs, Path path, PathFilter filter)
+ throws IOException {
+ try {
+ return new ToFileStatusIterator(SHIMS.listLocatedHdfsStatusIterator(fs, path, filter));
+ } catch (Throwable t) {
Review comment:
removed it
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r508807499
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -414,76 +403,30 @@ public void markCleaned(CompactionInfo info) throws MetaException {
* aborted TXN_COMPONENTS above tc_writeid (and consequently about aborted txns).
* See {@link ql.txn.compactor.Cleaner.removeFiles()}
*/
- s = "SELECT DISTINCT \"TXN_ID\" FROM \"TXNS\", \"TXN_COMPONENTS\" WHERE \"TXN_ID\" = \"TC_TXNID\" "
- + "AND \"TXN_STATE\" = " + TxnStatus.ABORTED + " AND \"TC_DATABASE\" = ? AND \"TC_TABLE\" = ?";
- if (info.highestWriteId != 0) s += " AND \"TC_WRITEID\" <= ?";
- if (info.partName != null) s += " AND \"TC_PARTITION\" = ?";
-
+ s = "DELETE FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\" IN (" +
Review comment:
@pvary, could you please take a quick look? thanks!
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501723043
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
I agree, it can be addressed in a follow up Jira.
----------------------------------------------------------------
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.
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] vpnvishv commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
vpnvishv commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499655306
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
@pvargacl Sorry I may be missing something here, but with this change, how can compactor read the data of an aborted delta. It should be in the aborted list right, due to this dummy p type entry in TXN_COMPONENTS?
----------------------------------------------------------------
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.
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] deniskuzZ commented on pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on pull request #1548:
URL: https://github.com/apache/hive/pull/1548#issuecomment-704456432
> @deniskuzZ Yes.. Sorry I missed those while porting changes from my private branch to open source [Hive-3](https://issues.apache.org/jira/browse/HIVE-3).
> I think you have already covered it now. Still if you want to refer I have Updated [Hive-3](https://issues.apache.org/jira/browse/HIVE-3) PR with those tests, [4596b50](https://github.com/apache/hive/commit/4596b50e0f4e00c363a744bad13a8c750330d9a0)
Thanks @vpnvishv, I'll check them. Also noticed that new test property is not actually configured. Fixed in master patch.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499749748
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
+ return true;
+ } else if (name.startsWith(baseDir(wId)) && !name.contains("=")) {
+ return true;
+ }
+ }
+ return false;
+ };
+ List<FileStatus> deleted = new ArrayList<>();
+ deleteDeltaDirectoriesAux(rootPartition, fs, filter, deleted);
+ return deleted;
+ }
+
+ private static void deleteDeltaDirectoriesAux(Path root, FileSystem fs, PathFilter filter, List<FileStatus> deleted)
+ throws IOException {
+ RemoteIterator<FileStatus> it = listIterator(fs, root, null);
+
+ while (it.hasNext()) {
+ FileStatus fStatus = it.next();
+ if (fStatus.isDirectory()) {
+ if (filter.accept(fStatus.getPath())) {
+ fs.delete(fStatus.getPath(), true);
+ deleted.add(fStatus);
+ } else {
+ deleteDeltaDirectoriesAux(fStatus.getPath(), fs, filter, deleted);
+ if (isDirectoryEmpty(fs, fStatus.getPath())) {
Review comment:
+ partitions are not removed in HMS
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499405728
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
+ return true;
+ } else if (name.startsWith(baseDir(wId)) && !name.contains("=")) {
+ return true;
+ }
+ }
+ return false;
+ };
+ List<FileStatus> deleted = new ArrayList<>();
+ deleteDeltaDirectoriesAux(rootPartition, fs, filter, deleted);
+ return deleted;
+ }
+
+ private static void deleteDeltaDirectoriesAux(Path root, FileSystem fs, PathFilter filter, List<FileStatus> deleted)
+ throws IOException {
+ RemoteIterator<FileStatus> it = listIterator(fs, root, null);
+
+ while (it.hasNext()) {
+ FileStatus fStatus = it.next();
+ if (fStatus.isDirectory()) {
+ if (filter.accept(fStatus.getPath())) {
+ fs.delete(fStatus.getPath(), true);
+ deleted.add(fStatus);
+ } else {
+ deleteDeltaDirectoriesAux(fStatus.getPath(), fs, filter, deleted);
+ if (isDirectoryEmpty(fs, fStatus.getPath())) {
+ fs.delete(fStatus.getPath(), false);
+ deleted.add(fStatus);
+ }
+ }
+ }
+ }
+ }
+
+ private static boolean isDirectoryEmpty(FileSystem fs, Path path) throws IOException {
+ RemoteIterator<FileStatus> it = listIterator(fs, path, null);
+ return !it.hasNext();
+ }
+
+ private static RemoteIterator<FileStatus> listIterator(FileSystem fs, Path path, PathFilter filter)
+ throws IOException {
+ try {
+ return new ToFileStatusIterator(SHIMS.listLocatedHdfsStatusIterator(fs, path, filter));
+ } catch (Throwable t) {
Review comment:
This should be similar to tryListLocatedHdfsStatus don't catch all Throwable
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501646861
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
1. In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed:
https://github.com/apache/hive/pull/1548/files#diff-9cf3ae764b7a33b568a984d695aff837R328
@vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
2. was related to the following issue based on Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() design:
"Suppose you have p-type clean on table T that is running (i.e. has the Write lock) and you have 30 different partition clean requests (in T). The 30 per partition cleans will get blocked but they will tie up every thread in the pool while they are blocked, right? If so, no other clean (on any other table) will actually make progress until the p-type on T is done."
I think, it's not valid now.
----------------------------------------------------------------
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.
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] deniskuzZ merged pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ merged pull request #1548:
URL: https://github.com/apache/hive/pull/1548
----------------------------------------------------------------
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.
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] vpnvishv commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
vpnvishv commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499656397
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
I was also wondering the same, as this code was there in the earlier patches so I have just kept it. We can remove this.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499754142
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
+ return true;
+ } else if (name.startsWith(baseDir(wId)) && !name.contains("=")) {
+ return true;
+ }
+ }
+ return false;
+ };
+ List<FileStatus> deleted = new ArrayList<>();
+ deleteDeltaDirectoriesAux(rootPartition, fs, filter, deleted);
+ return deleted;
+ }
+
+ private static void deleteDeltaDirectoriesAux(Path root, FileSystem fs, PathFilter filter, List<FileStatus> deleted)
Review comment:
changed to use getHdfsDirSnapshots,
@pvargacl do you know. if i should access cached data somehow?
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501646861
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
1. In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed:
https://github.com/apache/hive/pull/1548/files#diff-9cf3ae764b7a33b568a984d695aff837R328
@vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
2. was related to the following issue based on Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() design:
"Suppose you have p-type clean on table T that is running (i.e. has the Write lock) and you have 30 different partition clean requests (in T). The 30 per partition cleans will get blocked but they will tie up every thread in the pool while they are blocked, right? If so, no other clean (on any other table) will actually make progress until the p-type on T is done."
Yes, it's still the case that we'll have to wait for all tasks to complete and if there is one long-running task, we won't be able to submit new ones. However not sure if it's a critical issue. I think, we can address it in a separate jira.
----------------------------------------------------------------
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.
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] deniskuzZ commented on pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on pull request #1548:
URL: https://github.com/apache/hive/pull/1548#issuecomment-703585092
> @deniskuzZ Overall change LGTM.
> Looked into the test failures, one of test requires change in expected values wrt master branch. Other two looks genuine failures to me. Please check the inline comments.
@vpnvishv, thank you for the review!
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501646861
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed. @vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501646861
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
1. In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed. @vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
2. was related to the following issue based on Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() design:
"Suppose you have p-type clean on table T that is running (i.e. has the Write lock) and you have 30 different partition clean requests (in T). The 30 per partition cleans will get blocked but they will tie up every thread in the pool while they are blocked, right? If so, no other clean (on any other table) will actually make progress until the p-type on T is done."
I think, it's not valid now.
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501718248
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+ }
+
+ @Test
+ public void testCleanAbortCompactSeveralTables() throws Exception {
+ String dbName = "default";
+ String tblName1 = "cws1";
+ String tblName2 = "cws2";
+
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName1, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName2, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,1".getBytes());
+ connection2.write("2,2".getBytes());
+ connection2.abortTransaction();
+
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ FileSystem fs = FileSystem.get(conf);
+ Table table1 = msClient.getTable(dbName, tblName1);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table1.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+ Table table2 = msClient.getTable(dbName, tblName2);
+ stat = fs.listStatus(new Path(table2.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table1.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCorrectlyCleaned() throws Exception {
+ // Test that at commit the tables are cleaned properly
+ String dbName = "default";
+ String tblName = "cws";
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 1, count);
+
+ connection.commitTransaction();
+
+ // After commit the row should have been deleted
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+ }
+
+ @Test
+ public void testCleanAbortAndMinorCompact() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+ connection.abortTransaction();
+
+ executeStatementOnDriver("insert into " + tblName + " partition (a) values (1, '1')", driver);
+ executeStatementOnDriver("delete from " + tblName + " where b=1", driver);
+
+ conf.setIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_DELTA_NUM_THRESHOLD, 0);
+ runInitiator(conf);
+ runWorker(conf);
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+ // Cleaning should happen in threads concurrently for the minor compaction and the clean abort one.
+ runCleaner(conf);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ }
+
+ private HiveStreamingConnection prepareTableAndConnection(String dbName, String tblName, int batchSize) throws Exception {
Review comment:
Could you add two more test with batch size 2. first batch writes to p1 and p2 and commits, second batch writes to p2 and p3 and aborts. And one with the aborted / committed order changed.
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501883427
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
I still don't follow. Aborted txn check is done per db/table/partition, so if you have db1/tbl1/p1/type=NOT_DP and db1/tbl1/null/type=DP - that should generate 2 entries in potential compactions.
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499388356
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
+ return true;
+ } else if (name.startsWith(baseDir(wId)) && !name.contains("=")) {
+ return true;
+ }
+ }
+ return false;
+ };
+ List<FileStatus> deleted = new ArrayList<>();
+ deleteDeltaDirectoriesAux(rootPartition, fs, filter, deleted);
+ return deleted;
+ }
+
+ private static void deleteDeltaDirectoriesAux(Path root, FileSystem fs, PathFilter filter, List<FileStatus> deleted)
Review comment:
This is going issue many filesystem listing on a table with many partitions, that is going to be very slow on S3. I think you should consider changing this logic to be similar to getHdfsDirSnapshots that would do 1 recursive listing, iterate all the files and collect the deltas that needs to be deleted and delete them at the end (possible concurrently)
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501713463
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
Review comment:
I think this assert is quit misleading. I might be wrong but the recursive listing skips empty directories, and actually this new version of cleaning will keep the partition directories (it should) and only delete the delta dirs and files.
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+ }
+
+ @Test
+ public void testCleanAbortCompactSeveralTables() throws Exception {
+ String dbName = "default";
+ String tblName1 = "cws1";
+ String tblName2 = "cws2";
+
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName1, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName2, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,1".getBytes());
+ connection2.write("2,2".getBytes());
+ connection2.abortTransaction();
+
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ FileSystem fs = FileSystem.get(conf);
+ Table table1 = msClient.getTable(dbName, tblName1);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table1.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+ Table table2 = msClient.getTable(dbName, tblName2);
+ stat = fs.listStatus(new Path(table2.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table1.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCorrectlyCleaned() throws Exception {
+ // Test that at commit the tables are cleaned properly
+ String dbName = "default";
+ String tblName = "cws";
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
Review comment:
this comment is copied I guess
##########
File path: itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/txn/compactor/TestCompactor.java
##########
@@ -853,6 +857,273 @@ public void majorCompactAfterAbort() throws Exception {
Lists.newArrayList(5, 6), 1);
}
+ @Test
+ public void testCleanAbortCompactAfterAbortTwoPartitions() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection1 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableTwoPartitionsAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCompactAfterAbort() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ // Create three folders with two different transactions
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,3".getBytes());
+ connection2.write("2,3".getBytes());
+ connection2.write("3,3".getBytes());
+ connection2.abortTransaction();
+
+ assertAndCompactCleanAbort(dbName, tblName);
+
+ connection1.close();
+ connection2.close();
+ }
+
+ private void assertAndCompactCleanAbort(String dbName, String tblName) throws Exception {
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ TxnStore txnHandler = TxnUtils.getTxnStore(conf);
+ Table table = msClient.getTable(dbName, tblName);
+ FileSystem fs = FileSystem.get(conf);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table.getSd().getLocation()));
+ if (3 != stat.length) {
+ Assert.fail("Expecting three directories corresponding to three partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 1, count);
+
+ ShowCompactResponse rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.CLEANING_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ rsp = txnHandler.showCompact(new ShowCompactRequest());
+ Assert.assertEquals(1, rsp.getCompacts().size());
+ Assert.assertEquals(TxnStore.SUCCEEDED_RESPONSE, rsp.getCompacts().get(0).getState());
+ Assert.assertEquals("cws", rsp.getCompacts().get(0).getTablename());
+ Assert.assertEquals(CompactionType.CLEAN_ABORTED,
+ rsp.getCompacts().get(0).getType());
+ }
+
+ @Test
+ public void testCleanAbortCompactSeveralTables() throws Exception {
+ String dbName = "default";
+ String tblName1 = "cws1";
+ String tblName2 = "cws2";
+
+ HiveStreamingConnection connection1 = prepareTableAndConnection(dbName, tblName1, 1);
+ HiveStreamingConnection connection2 = prepareTableAndConnection(dbName, tblName2, 1);
+
+ connection1.beginTransaction();
+ connection1.write("1,1".getBytes());
+ connection1.write("2,2".getBytes());
+ connection1.abortTransaction();
+
+ connection2.beginTransaction();
+ connection2.write("1,1".getBytes());
+ connection2.write("2,2".getBytes());
+ connection2.abortTransaction();
+
+ IMetaStoreClient msClient = new HiveMetaStoreClient(conf);
+ FileSystem fs = FileSystem.get(conf);
+ Table table1 = msClient.getTable(dbName, tblName1);
+ FileStatus[] stat =
+ fs.listStatus(new Path(table1.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+ Table table2 = msClient.getTable(dbName, tblName2);
+ stat = fs.listStatus(new Path(table2.getSd().getLocation()));
+ if (2 != stat.length) {
+ Assert.fail("Expecting two directories corresponding to two partitions, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 2, count);
+
+ runInitiator(conf);
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE where CQ_TYPE='p'");
+ // Only one job is added to the queue per table. This job corresponds to all the entries for a particular table
+ // with rows in TXN_COMPONENTS
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+
+ runCleaner(conf);
+
+ // After the cleaner runs TXN_COMPONENTS and COMPACTION_QUEUE should have zero rows, also the folders should have been deleted.
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ RemoteIterator it =
+ fs.listFiles(new Path(table1.getSd().getLocation()), true);
+ if (it.hasNext()) {
+ Assert.fail("Expecting compaction to have cleaned the directories, FileStatus[] stat " + Arrays.toString(stat));
+ }
+
+ connection1.close();
+ connection2.close();
+ }
+
+ @Test
+ public void testCleanAbortCorrectlyCleaned() throws Exception {
+ // Test that at commit the tables are cleaned properly
+ String dbName = "default";
+ String tblName = "cws";
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 1, count);
+
+ connection.commitTransaction();
+
+ // After commit the row should have been deleted
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS where TC_OPERATION_TYPE='p'");
+ // We should have two rows corresponding to the two aborted transactions
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+ }
+
+ @Test
+ public void testCleanAbortAndMinorCompact() throws Exception {
+ String dbName = "default";
+ String tblName = "cws";
+
+ HiveStreamingConnection connection = prepareTableAndConnection(dbName, tblName, 1);
+
+ connection.beginTransaction();
+ connection.write("1,1".getBytes());
+ connection.write("2,2".getBytes());
+ connection.abortTransaction();
+
+ executeStatementOnDriver("insert into " + tblName + " partition (a) values (1, '1')", driver);
+ executeStatementOnDriver("delete from " + tblName + " where b=1", driver);
+
+ conf.setIntVar(HiveConf.ConfVars.HIVE_COMPACTOR_DELTA_NUM_THRESHOLD, 0);
+ runInitiator(conf);
+ runWorker(conf);
+
+ int count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 2, count);
+ // Cleaning should happen in threads concurrently for the minor compaction and the clean abort one.
+ runCleaner(conf);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from COMPACTION_QUEUE");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from COMPACTION_QUEUE"), 0, count);
+
+ count = TxnDbUtil.countQueryAgent(conf, "select count(*) from TXN_COMPONENTS");
+ Assert.assertEquals(TxnDbUtil.queryToString(conf, "select * from TXN_COMPONENTS"), 0, count);
+
+ }
+
+ private HiveStreamingConnection prepareTableAndConnection(String dbName, String tblName, int batchSize) throws Exception {
Review comment:
Could you add two more test with batch size 2. first batch writes to p1 and p2 and commits, second batch writes to p2 and p3 and aborts. And one with the aborted / committed order changed.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
I agree, it can be addressed in a follow up Jira.
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -232,6 +240,51 @@ public Object run() throws Exception {
private static String idWatermark(CompactionInfo ci) {
return " id=" + ci.id;
}
+
+ private void cleanAborted(CompactionInfo ci) throws MetaException {
+ if (ci.writeIds == null || ci.writeIds.size() == 0) {
+ LOG.warn("Attempted cleaning aborted transaction with empty writeId list");
Review comment:
Shouldn't you mark the compaction failed or cleaned?
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -107,11 +107,12 @@ public CompactionTxnHandler() {
// Check for aborted txns: number of aborted txns past threshold and age of aborted txns
// past time threshold
boolean checkAbortedTimeThreshold = abortedTimeThreshold >= 0;
- final String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\","
- + "MIN(\"TXN_STARTED\"), COUNT(*)"
+ String sCheckAborted = "SELECT \"TC_DATABASE\", \"TC_TABLE\", \"TC_PARTITION\", "
+ + "MIN(\"TXN_STARTED\"), COUNT(*), "
+ + "MAX(CASE WHEN \"TC_OPERATION_TYPE\" = " + OperationType.DYNPART + " THEN 1 ELSE 0 END) AS \"IS_DP\" "
Review comment:
Previously if you had aborted txn above threshold this would generate a "normal" compaction that would clean up everything. However now if you have one dynpart aborted the type will be CLEAN_ABORTED that will only clean the writeids belonging to p-type records and leave everything else. This will delay the normal cleaning. I am not sure that is a problem or not.
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -414,77 +436,56 @@ public void markCleaned(CompactionInfo info) throws MetaException {
* aborted TXN_COMPONENTS above tc_writeid (and consequently about aborted txns).
* See {@link ql.txn.compactor.Cleaner.removeFiles()}
*/
- s = "SELECT DISTINCT \"TXN_ID\" FROM \"TXNS\", \"TXN_COMPONENTS\" WHERE \"TXN_ID\" = \"TC_TXNID\" "
- + "AND \"TXN_STATE\" = " + TxnStatus.ABORTED + " AND \"TC_DATABASE\" = ? AND \"TC_TABLE\" = ?";
- if (info.highestWriteId != 0) s += " AND \"TC_WRITEID\" <= ?";
- if (info.partName != null) s += " AND \"TC_PARTITION\" = ?";
+ List<String> queries = new ArrayList<>();
+ Iterator<Long> writeIdsIter = null;
+ List<Integer> counts = null;
- pStmt = dbConn.prepareStatement(s);
- paramCount = 1;
- pStmt.setString(paramCount++, info.dbname);
- pStmt.setString(paramCount++, info.tableName);
- if(info.highestWriteId != 0) {
- pStmt.setLong(paramCount++, info.highestWriteId);
+ s = "DELETE FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\" IN (" +
+ " SELECT \"TXN_ID\" FROM \"TXNS\" WHERE \"TXN_STATE\" = " + TxnStatus.ABORTED + ") " +
Review comment:
cool
----------------------------------------------------------------
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.
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] deniskuzZ commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
deniskuzZ commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r501646861
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/txn/compactor/Cleaner.java
##########
@@ -97,9 +100,9 @@ public void run() {
long minOpenTxnId = txnHandler.findMinOpenTxnIdForCleaner();
LOG.info("Cleaning based on min open txn id: " + minOpenTxnId);
List<CompletableFuture> cleanerList = new ArrayList<>();
- for(CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
+ for (CompactionInfo compactionInfo : txnHandler.findReadyToClean()) {
cleanerList.add(CompletableFuture.runAsync(CompactorUtil.ThrowingRunnable.unchecked(() ->
- clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
+ clean(compactionInfo, minOpenTxnId)), cleanerExecutor));
Review comment:
1. In original patch Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() was used to prevent a concurrent p-clean (where the whole table will be scanned). I think, that is resolved by grouping p-cleans and recording list of writeIds that needs to be removed:
https://github.com/apache/hive/pull/1548/files#diff-9cf3ae764b7a33b568a984d695aff837R328
@vpnvishv is that correct? Also we do not allow concurrent Cleaners, their execution is mutexed.
2. was related to the following issue based on Map<String, NonReentrantReadWriteLock> tableLock = new ConcurrentHashMap<>() design:
"Suppose you have p-type clean on table T that is running (i.e. has the Write lock) and you have 30 different partition clean requests (in T). The 30 per partition cleans will get blocked but they will tie up every thread in the pool while they are blocked, right? If so, no other clean (on any other table) will actually make progress until the p-type on T is done."
Yes, it's still the case that we'll have to wait for all tasks to complete and if there is one slow task, we won't be able to submit new tasks. However not sure if it's a critical issue. I think, we can address it in a separate jira.
----------------------------------------------------------------
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.
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] pvargacl commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
pvargacl commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r499399709
##########
File path: ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java
##########
@@ -2839,6 +2848,87 @@ public static void setNonTransactional(Map<String, String> tblProps) {
tblProps.remove(hive_metastoreConstants.TABLE_TRANSACTIONAL_PROPERTIES);
}
+ /**
+ * Look for delta directories matching the list of writeIds and deletes them.
+ * @param rootPartition root partition to look for the delta directories
+ * @param conf configuration
+ * @param writeIds list of writeIds to look for in the delta directories
+ * @return list of deleted directories.
+ * @throws IOException
+ */
+ public static List<FileStatus> deleteDeltaDirectories(Path rootPartition, Configuration conf, Set<Long> writeIds)
+ throws IOException {
+ FileSystem fs = rootPartition.getFileSystem(conf);
+
+ PathFilter filter = (p) -> {
+ String name = p.getName();
+ for (Long wId : writeIds) {
+ if (name.startsWith(deltaSubdir(wId, wId)) && !name.contains("=")) {
Review comment:
Why the contains "=", are we checking for a partition where the user named the column exactly like a valid delta dir? I don't think we should support that
----------------------------------------------------------------
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.
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] klcopp commented on a change in pull request #1548: HIVE-21052: Make sure transactions get cleaned if they are aborted be…
Posted by GitBox <gi...@apache.org>.
klcopp commented on a change in pull request #1548:
URL: https://github.com/apache/hive/pull/1548#discussion_r509054430
##########
File path: standalone-metastore/metastore-server/src/main/java/org/apache/hadoop/hive/metastore/txn/CompactionTxnHandler.java
##########
@@ -414,76 +403,30 @@ public void markCleaned(CompactionInfo info) throws MetaException {
* aborted TXN_COMPONENTS above tc_writeid (and consequently about aborted txns).
* See {@link ql.txn.compactor.Cleaner.removeFiles()}
*/
- s = "SELECT DISTINCT \"TXN_ID\" FROM \"TXNS\", \"TXN_COMPONENTS\" WHERE \"TXN_ID\" = \"TC_TXNID\" "
- + "AND \"TXN_STATE\" = " + TxnStatus.ABORTED + " AND \"TC_DATABASE\" = ? AND \"TC_TABLE\" = ?";
- if (info.highestWriteId != 0) s += " AND \"TC_WRITEID\" <= ?";
- if (info.partName != null) s += " AND \"TC_PARTITION\" = ?";
-
+ s = "DELETE FROM \"TXN_COMPONENTS\" WHERE \"TC_TXNID\" IN (" +
Review comment:
never mind, LGTM
----------------------------------------------------------------
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.
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