HIVE-15899 Make CTAS with acid target table and insert into acid_tbl select ... union all ... work (Eugene Koifman, reviewed by Ashutosh Chauhan)
Project: http://git-wip-us.apache.org/repos/asf/hive/repo Commit: http://git-wip-us.apache.org/repos/asf/hive/commit/02d359db Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/02d359db Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/02d359db Branch: refs/heads/master Commit: 02d359db235cc646928ff2b7f5b1fe2d88c7ad12 Parents: 5608300 Author: Eugene Koifman <ekoif...@hortonworks.com> Authored: Wed Sep 20 10:46:21 2017 -0700 Committer: Eugene Koifman <ekoif...@hortonworks.com> Committed: Wed Sep 20 10:46:21 2017 -0700 ---------------------------------------------------------------------- .../apache/hadoop/hive/ql/TestAcidOnTez.java | 329 +++++++++++++++---- .../java/org/apache/hadoop/hive/ql/Context.java | 3 + .../java/org/apache/hadoop/hive/ql/Driver.java | 6 +- .../apache/hadoop/hive/ql/exec/MoveTask.java | 13 +- .../org/apache/hadoop/hive/ql/io/AcidUtils.java | 12 + .../apache/hadoop/hive/ql/metadata/Hive.java | 43 ++- .../ql/optimizer/QueryPlanPostProcessor.java | 166 ++++++++++ .../optimizer/unionproc/UnionProcFactory.java | 3 +- .../hadoop/hive/ql/parse/SemanticAnalyzer.java | 28 +- .../apache/hadoop/hive/ql/plan/LoadDesc.java | 22 +- .../hadoop/hive/ql/plan/LoadFileDesc.java | 20 +- .../hadoop/hive/ql/plan/LoadTableDesc.java | 26 +- .../hadoop/hive/ql/plan/TezEdgeProperty.java | 10 +- .../apache/hadoop/hive/ql/TestTxnNoBuckets.java | 189 +++++++++-- .../clientpositive/autoColumnStats_4.q.out | 1 + .../clientpositive/llap/acid_no_buckets.q.out | 8 + .../llap/dynamic_semijoin_reduction_3.q.out | 7 + .../llap/dynpart_sort_optimization_acid.q.out | 12 + .../results/clientpositive/llap/sqlmerge.q.out | 4 + 19 files changed, 751 insertions(+), 151 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java ---------------------------------------------------------------------- diff --git a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java index d0b5cf6..8b4b21f 100644 --- a/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java +++ b/itests/hive-unit/src/test/java/org/apache/hadoop/hive/ql/TestAcidOnTez.java @@ -33,6 +33,7 @@ import org.apache.hadoop.hive.metastore.api.ShowCompactResponse; import org.apache.hadoop.hive.metastore.txn.TxnDbUtil; import org.apache.hadoop.hive.metastore.txn.TxnStore; import org.apache.hadoop.hive.metastore.txn.TxnUtils; +import org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator; import org.apache.hadoop.hive.ql.io.BucketCodec; import org.apache.hadoop.hive.ql.io.HiveInputFormat; import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse; @@ -88,7 +89,6 @@ public class TestAcidOnTez { hiveConf = new HiveConf(this.getClass()); hiveConf.set(HiveConf.ConfVars.PREEXECHOOKS.varname, ""); hiveConf.set(HiveConf.ConfVars.POSTEXECHOOKS.varname, ""); - hiveConf.set(HiveConf.ConfVars.HIVE_SUPPORT_CONCURRENCY.varname, "false"); hiveConf.set(HiveConf.ConfVars.METASTOREWAREHOUSE.varname, TEST_WAREHOUSE_DIR); hiveConf.setVar(HiveConf.ConfVars.HIVEMAPREDMODE, "nonstrict"); hiveConf.setVar(HiveConf.ConfVars.HIVEINPUTFORMAT, HiveInputFormat.class.getName()); @@ -179,14 +179,14 @@ public class TestAcidOnTez { runStatementOnDriver("create table " + Table.NONACIDNONBUCKET + " stored as ORC TBLPROPERTIES('transactional'='false') as " + "select a, b from " + Table.ACIDTBL + " where a <= 5 union all select a, b from " + Table.NONACIDORCTBL + " where a >= 5", confForTez); - List<String> rs = runStatementOnDriver("select a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by a, b, INPUT__FILE__NAME"); + List<String> rs = runStatementOnDriver("select a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by a, b, INPUT__FILE__NAME", confForTez); String expected0[][] = { - {"1\t2", "/1/000000_0"}, - {"3\t4", "/1/000000_0"}, - {"5\t6", "/1/000000_0"}, - {"5\t6", "/2/000000_0"}, - {"7\t8", "/2/000000_0"}, - {"9\t10", "/2/000000_0"}, + {"1\t2", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"3\t4", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"5\t6", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"5\t6", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, + {"7\t8", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, + {"9\t10", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, }; Assert.assertEquals("Unexpected row count after ctas", expected0.length, rs.size()); //verify data and layout @@ -195,9 +195,9 @@ public class TestAcidOnTez { Assert.assertTrue("Actual line(file) " + i + " bc: " + rs.get(i), rs.get(i).endsWith(expected0[i][1])); } //make the table ACID - runStatementOnDriver("alter table " + Table.NONACIDNONBUCKET + " SET TBLPROPERTIES ('transactional'='true')"); + runStatementOnDriver("alter table " + Table.NONACIDNONBUCKET + " SET TBLPROPERTIES ('transactional'='true')", confForTez); - rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by ROW__ID"); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by ROW__ID", confForTez); LOG.warn("after ctas:"); for (String s : rs) { LOG.warn(s); @@ -206,12 +206,12 @@ public class TestAcidOnTez { /* * Expected result 0th entry i the RecordIdentifier + data. 1st entry file before compact*/ String expected[][] = { - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":0}\t1\t2", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":1}\t3\t4", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":2}\t5\t6", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":3}\t9\t10", "/2/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":4}\t7\t8", "/2/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":5}\t5\t6", "/2/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":0}\t1\t2", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":1}\t3\t4", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":2}\t5\t6", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":3}\t9\t10", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":4}\t7\t8", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":5}\t5\t6", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, }; Assert.assertEquals("Unexpected row count after ctas", expected.length, rs.size()); //verify data and layout @@ -220,17 +220,17 @@ public class TestAcidOnTez { Assert.assertTrue("Actual line(file) " + i + " bc: " + rs.get(i), rs.get(i).endsWith(expected[i][1])); } //perform some Update/Delete - runStatementOnDriver("update " + Table.NONACIDNONBUCKET + " set a = 70, b = 80 where a = 7"); - runStatementOnDriver("delete from " + Table.NONACIDNONBUCKET + " where a = 5"); - rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by ROW__ID"); + runStatementOnDriver("update " + Table.NONACIDNONBUCKET + " set a = 70, b = 80 where a = 7", confForTez); + runStatementOnDriver("delete from " + Table.NONACIDNONBUCKET + " where a = 5", confForTez); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by ROW__ID", confForTez); LOG.warn("after update/delete:"); for (String s : rs) { LOG.warn(s); } String[][] expected2 = { - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":0}\t1\t2", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":1}\t3\t4", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":3}\t9\t10", "/2/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":0}\t1\t2", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":1}\t3\t4", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":3}\t9\t10", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, {"{\"transactionid\":21,\"bucketid\":536870912,\"rowid\":0}\t70\t80", "delta_0000021_0000021_0000/bucket_00000"} }; Assert.assertEquals("Unexpected row count after update", expected2.length, rs.size()); @@ -255,9 +255,9 @@ public class TestAcidOnTez { Assert.assertNull("at " + i + " " + expectedDelDelta[i] + " not found on disk", expectedDelDelta[i]); } //run Minor compaction - runStatementOnDriver("alter table " + Table.NONACIDNONBUCKET + " compact 'minor'"); + runStatementOnDriver("alter table " + Table.NONACIDNONBUCKET + " compact 'minor'", confForTez); TestTxnCommands2.runWorker(hiveConf); - rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by ROW__ID"); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by ROW__ID", confForTez); LOG.warn("after compact minor:"); for (String s : rs) { LOG.warn(s); @@ -285,9 +285,9 @@ public class TestAcidOnTez { Assert.assertNull("at " + i + " " + expectedDelDelta2[i] + " not found on disk", expectedDelDelta2[i]); } //run Major compaction - runStatementOnDriver("alter table " + Table.NONACIDNONBUCKET + " compact 'major'"); + runStatementOnDriver("alter table " + Table.NONACIDNONBUCKET + " compact 'major'", confForTez); TestTxnCommands2.runWorker(hiveConf); - rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by ROW__ID"); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.NONACIDNONBUCKET + " order by ROW__ID", confForTez); LOG.warn("after compact major:"); for (String s : rs) { LOG.warn(s); @@ -306,8 +306,12 @@ public class TestAcidOnTez { * How to do this? CTAS is the only way to create data files which are not immediate children * of the partition dir. CTAS/Union/Tez doesn't support partition tables. The only way is to copy * data files in directly. + * + * Actually Insert Into ... select ... union all ... with + * HIVE_OPTIMIZE_UNION_REMOVE (and HIVEFETCHTASKCONVERSION="none"?) will create subdirs + * but if writing to non acid table there is a merge task on MR (but not on Tez) */ - @Ignore("HIVE-17214") + @Ignore("HIVE-17214")//this consistently works locally but never in ptest.... @Test public void testNonStandardConversion02() throws Exception { HiveConf confForTez = new HiveConf(hiveConf); // make a clone of existing hive conf @@ -320,13 +324,13 @@ public class TestAcidOnTez { "union all select a, b from " + Table.ACIDTBL + " where a = 5", confForTez); List<String> rs = runStatementOnDriver("select a, b, INPUT__FILE__NAME from " + - Table.NONACIDNONBUCKET + " order by a, b"); + Table.NONACIDNONBUCKET + " order by a, b", confForTez); String expected0[][] = { - {"1\t2", "/1/000000_0"}, - {"3\t4", "/1/000000_0"}, - {"5\t6", "/3/000000_0"}, - {"7\t8", "/2/000000_0"}, - {"9\t10", "/2/000000_0"}, + {"1\t2", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"3\t4", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"5\t6", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "3/000000_0"}, + {"7\t8", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, + {"9\t10", AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, }; Assert.assertEquals("Unexpected row count after ctas", expected0.length, rs.size()); //verify data and layout @@ -338,7 +342,7 @@ public class TestAcidOnTez { FileStatus[] status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + (Table.NONACIDNONBUCKET).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); //ensure there is partition dir - runStatementOnDriver("insert into " + Table.NONACIDPART + " partition (p=1) values (100,110)"); + runStatementOnDriver("insert into " + Table.NONACIDPART + " partition (p=1) values (100,110)", confForTez); //creates more files in that partition for(FileStatus stat : status) { int limit = 5; @@ -357,29 +361,29 @@ public class TestAcidOnTez { nonacidpart/ âââ p=1 âââ 000000_0 - âââ 1 + âââ HIVE_UNION_SUBDIR__1 â  âââ 000000_0 - âââ 2 + âââ HIVE_UNION_SUBDIR_2 â  âââ 000000_0 - âââ 3 + âââ HIVE_UNION_SUBDIR_3 âââ 000000_0 4 directories, 4 files **/ //make the table ACID - runStatementOnDriver("alter table " + Table.NONACIDPART + " SET TBLPROPERTIES ('transactional'='true')"); - rs = runStatementOnDriver("select ROW__ID, a, b, p, INPUT__FILE__NAME from " + Table.NONACIDPART + " order by ROW__ID"); + runStatementOnDriver("alter table " + Table.NONACIDPART + " SET TBLPROPERTIES ('transactional'='true')", confForTez); + rs = runStatementOnDriver("select ROW__ID, a, b, p, INPUT__FILE__NAME from " + Table.NONACIDPART + " order by ROW__ID", confForTez); LOG.warn("after acid conversion:"); for (String s : rs) { LOG.warn(s); } String[][] expected = { {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":0}\t100\t110\t1", "nonacidpart/p=1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":1}\t1\t2\t1", "nonacidpart/p=1/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":2}\t3\t4\t1", "nonacidpart/p=1/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":3}\t9\t10\t1", "nonacidpart/p=1/2/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":4}\t7\t8\t1", "nonacidpart/p=1/2/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":5}\t5\t6\t1", "nonacidpart/p=1/3/000000_0"} + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":1}\t1\t2\t1", "nonacidpart/p=1/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":2}\t3\t4\t1", "nonacidpart/p=1/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":3}\t9\t10\t1", "nonacidpart/p=1/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":4}\t7\t8\t1", "nonacidpart/p=1/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":5}\t5\t6\t1", "nonacidpart/p=1/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "3/000000_0"} }; Assert.assertEquals("Wrong row count", expected.length, rs.size()); //verify data and layout @@ -389,9 +393,9 @@ public class TestAcidOnTez { } //run Major compaction - runStatementOnDriver("alter table " + Table.NONACIDPART + " partition (p=1) compact 'major'"); + runStatementOnDriver("alter table " + Table.NONACIDPART + " partition (p=1) compact 'major'", confForTez); TestTxnCommands2.runWorker(hiveConf); - rs = runStatementOnDriver("select ROW__ID, a, b, p, INPUT__FILE__NAME from " + Table.NONACIDPART + " order by ROW__ID"); + rs = runStatementOnDriver("select ROW__ID, a, b, p, INPUT__FILE__NAME from " + Table.NONACIDPART + " order by ROW__ID", confForTez); LOG.warn("after major compaction:"); for (String s : rs) { LOG.warn(s); @@ -406,33 +410,42 @@ public class TestAcidOnTez { } /** * CTAS + Tez + Union creates a non-standard layout in table dir - * Each leg of the union places data into a subdir of the table/partition. Subdirs are named 1/, 2/, etc - * The way this currently works is that CTAS creates an Acid table but the insert statement writes - * the data in non-acid layout. Then on read, it's treated like an non-acid to acid conversion. - * Longer term CTAS should create acid layout from the get-go. + * Each leg of the union places data into a subdir of the table/partition. + * Subdirs are named HIVE_UNION_SUBDIR_1/, HIVE_UNION_SUBDIR_2/, etc + * For Acid tables the writer for each dir must have a different statementId ensured by + * {@link org.apache.hadoop.hive.ql.optimizer.QueryPlanPostProcessor}. + * {@link org.apache.hadoop.hive.ql.metadata.Hive#moveAcidFiles(FileSystem, FileStatus[], Path, List)} drops the union subdirs + * since each delta file has a unique name. */ @Test public void testCtasTezUnion() throws Exception { HiveConf confForTez = new HiveConf(hiveConf); // make a clone of existing hive conf + confForTez.setBoolVar(HiveConf.ConfVars.HIVE_EXPLAIN_USER, false); setupTez(confForTez); //CTAS with ACID target table + List<String> rs0 = runStatementOnDriver("explain create table " + Table.ACIDNOBUCKET + " stored as ORC TBLPROPERTIES('transactional'='true') as " + + "select a, b from " + Table.ACIDTBL + " where a <= 5 union all select a, b from " + Table.NONACIDORCTBL + " where a >= 5", confForTez); + LOG.warn("explain ctas:");//TezEdgeProperty.EdgeType + for (String s : rs0) { + LOG.warn(s); + } runStatementOnDriver("create table " + Table.ACIDNOBUCKET + " stored as ORC TBLPROPERTIES('transactional'='true') as " + "select a, b from " + Table.ACIDTBL + " where a <= 5 union all select a, b from " + Table.NONACIDORCTBL + " where a >= 5", confForTez); - List<String> rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDNOBUCKET + " order by ROW__ID"); + List<String> rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDNOBUCKET + " order by ROW__ID", confForTez); LOG.warn("after ctas:"); for (String s : rs) { LOG.warn(s); } Assert.assertEquals(0, BucketCodec.determineVersion(536870912).decodeWriterId(536870912)); /* - * Expected result 0th entry i the RecordIdentifier + data. 1st entry file before compact*/ + * Expected result 0th entry is the RecordIdentifier + data. 1st entry file before compact*/ String expected[][] = { - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":0}\t1\t2", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":1}\t3\t4", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":2}\t5\t6", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":3}\t9\t10", "/2/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":4}\t7\t8", "/2/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":5}\t5\t6", "/2/000000_0"}, + {"{\"transactionid\":18,\"bucketid\":536870913,\"rowid\":0}\t1\t2", "/delta_0000018_0000018_0001/bucket_00000"}, + {"{\"transactionid\":18,\"bucketid\":536870913,\"rowid\":1}\t3\t4", "/delta_0000018_0000018_0001/bucket_00000"}, + {"{\"transactionid\":18,\"bucketid\":536870913,\"rowid\":2}\t5\t6", "/delta_0000018_0000018_0001/bucket_00000"}, + {"{\"transactionid\":18,\"bucketid\":536870914,\"rowid\":0}\t9\t10", "/delta_0000018_0000018_0002/bucket_00000"}, + {"{\"transactionid\":18,\"bucketid\":536870914,\"rowid\":1}\t7\t8", "/delta_0000018_0000018_0002/bucket_00000"}, + {"{\"transactionid\":18,\"bucketid\":536870914,\"rowid\":2}\t5\t6", "/delta_0000018_0000018_0002/bucket_00000"}, }; Assert.assertEquals("Unexpected row count after ctas", expected.length, rs.size()); //verify data and layout @@ -441,18 +454,18 @@ public class TestAcidOnTez { Assert.assertTrue("Actual line(file) " + i + " bc: " + rs.get(i), rs.get(i).endsWith(expected[i][1])); } //perform some Update/Delete - runStatementOnDriver("update " + Table.ACIDNOBUCKET + " set a = 70, b = 80 where a = 7"); - runStatementOnDriver("delete from " + Table.ACIDNOBUCKET + " where a = 5"); - rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDNOBUCKET + " order by ROW__ID"); + runStatementOnDriver("update " + Table.ACIDNOBUCKET + " set a = 70, b = 80 where a = 7", confForTez); + runStatementOnDriver("delete from " + Table.ACIDNOBUCKET + " where a = 5", confForTez); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDNOBUCKET + " order by ROW__ID", confForTez); LOG.warn("after update/delete:"); for (String s : rs) { LOG.warn(s); } String[][] expected2 = { - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":0}\t1\t2", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":1}\t3\t4", "/1/000000_0"}, - {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":3}\t9\t10", "/2/000000_0"}, - {"{\"transactionid\":19,\"bucketid\":536870912,\"rowid\":0}\t70\t80", "delta_0000019_0000019_0000/bucket_00000"} + {"{\"transactionid\":18,\"bucketid\":536870913,\"rowid\":0}\t1\t2", "/delta_0000018_0000018_0001/bucket_00000"}, + {"{\"transactionid\":18,\"bucketid\":536870913,\"rowid\":1}\t3\t4", "/delta_0000018_0000018_0001/bucket_00000"}, + {"{\"transactionid\":18,\"bucketid\":536870914,\"rowid\":0}\t9\t10", "/delta_0000018_0000018_0002/bucket_00000"}, + {"{\"transactionid\":20,\"bucketid\":536870912,\"rowid\":0}\t70\t80", "delta_0000020_0000020_0000/bucket_00000"} }; Assert.assertEquals("Unexpected row count after update", expected2.length, rs.size()); //verify data and layout @@ -464,7 +477,7 @@ public class TestAcidOnTez { FileSystem fs = FileSystem.get(hiveConf); FileStatus[] status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + (Table.ACIDNOBUCKET).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); - String[] expectedDelDelta = {"delete_delta_0000019_0000019_0000", "delete_delta_0000020_0000020_0000"}; + String[] expectedDelDelta = {"delete_delta_0000020_0000020_0000", "delete_delta_0000021_0000021_0000"}; for(FileStatus stat : status) { for(int i = 0; i < expectedDelDelta.length; i++) { if(expectedDelDelta[i] != null && stat.getPath().toString().endsWith(expectedDelDelta[i])) { @@ -476,9 +489,9 @@ public class TestAcidOnTez { Assert.assertNull("at " + i + " " + expectedDelDelta[i] + " not found on disk", expectedDelDelta[i]); } //run Minor compaction - runStatementOnDriver("alter table " + Table.ACIDNOBUCKET + " compact 'minor'"); + runStatementOnDriver("alter table " + Table.ACIDNOBUCKET + " compact 'minor'", confForTez); TestTxnCommands2.runWorker(hiveConf); - rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDNOBUCKET + " order by ROW__ID"); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDNOBUCKET + " order by ROW__ID", confForTez); LOG.warn("after compact minor:"); for (String s : rs) { LOG.warn(s); @@ -493,7 +506,7 @@ public class TestAcidOnTez { //check we have right delete delta files after minor compaction status = fs.listStatus(new Path(TEST_WAREHOUSE_DIR + "/" + (Table.ACIDNOBUCKET).toString().toLowerCase()), FileUtils.STAGING_DIR_PATH_FILTER); - String[] expectedDelDelta2 = { "delete_delta_0000019_0000019_0000", "delete_delta_0000020_0000020_0000", "delete_delta_0000019_0000020"}; + String[] expectedDelDelta2 = { "delete_delta_0000020_0000020_0000", "delete_delta_0000021_0000021_0000", "delete_delta_0000018_0000021"}; for(FileStatus stat : status) { for(int i = 0; i < expectedDelDelta2.length; i++) { if(expectedDelDelta2[i] != null && stat.getPath().toString().endsWith(expectedDelDelta2[i])) { @@ -506,9 +519,9 @@ public class TestAcidOnTez { Assert.assertNull("at " + i + " " + expectedDelDelta2[i] + " not found on disk", expectedDelDelta2[i]); } //run Major compaction - runStatementOnDriver("alter table " + Table.ACIDNOBUCKET + " compact 'major'"); + runStatementOnDriver("alter table " + Table.ACIDNOBUCKET + " compact 'major'", confForTez); TestTxnCommands2.runWorker(hiveConf); - rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDNOBUCKET + " order by ROW__ID"); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from " + Table.ACIDNOBUCKET + " order by ROW__ID", confForTez); LOG.warn("after compact major:"); for (String s : rs) { LOG.warn(s); @@ -517,7 +530,178 @@ public class TestAcidOnTez { for(int i = 0; i < expected2.length; i++) { Assert.assertTrue("Actual line " + i + " bc: " + rs.get(i), rs.get(i).startsWith(expected2[i][0])); //everything is now in base/ - Assert.assertTrue("Actual line(file) " + i + " bc: " + rs.get(i), rs.get(i).endsWith("base_0000020/bucket_00000")); + Assert.assertTrue("Actual line(file) " + i + " bc: " + rs.get(i), rs.get(i).endsWith("base_0000021/bucket_00000")); + } + } + /** + * 1. Insert into regular unbucketed table from Union all - union is removed and data is placed in + * subdirs of target table. + * 2. convert to acid table and check data + * 3. compact and check data + * Compare with {@link #testAcidInsertWithRemoveUnion()} where T is transactional=true + */ + @Test + public void testInsertWithRemoveUnion() throws Exception { + int[][] values = {{1,2},{3,4},{5,6},{7,8},{9,10}}; + HiveConf confForTez = new HiveConf(hiveConf); // make a clone of existing hive conf + setupTez(confForTez); + runStatementOnDriver("drop table if exists T", confForTez); + runStatementOnDriver("create table T (a int, b int) stored as ORC TBLPROPERTIES ('transactional'='false')", confForTez); + /* +ekoifman:apache-hive-3.0.0-SNAPSHOT-bin ekoifman$ tree ~/dev/hiverwgit/itests/hive-unit/target/tmp/org.apache.hadoop.hive.ql.TestAcidOnTez-1505502329802/warehouse/t/.hive-staging_hive_2017-09-15_12-07-33_224_7717909516029836949-1/ +/Users/ekoifman/dev/hiverwgit/itests/hive-unit/target/tmp/org.apache.hadoop.hive.ql.TestAcidOnTez-1505502329802/warehouse/t/.hive-staging_hive_2017-09-15_12-07-33_224_7717909516029836949-1/ +âââ -ext-10000 + âââ HIVE_UNION_SUBDIR_1 + â  âââ 000000_0 + âââ HIVE_UNION_SUBDIR_2 + â  âââ 000000_0 + âââ HIVE_UNION_SUBDIR_3 + âââ 000000_0 + +4 directories, 3 files + */ + runStatementOnDriver("insert into T(a,b) select a, b from " + Table.ACIDTBL + " where a between 1 and 3 group by a, b union all select a, b from " + Table.ACIDTBL + " where a between 5 and 7 union all select a, b from " + Table.ACIDTBL + " where a >= 9", confForTez); + List<String> rs = runStatementOnDriver("select a, b, INPUT__FILE__NAME from T order by a, b, INPUT__FILE__NAME", confForTez); + LOG.warn(testName.getMethodName() + ": before converting to acid"); + for(String s : rs) { + LOG.warn(s); + } + String[][] expected = { + {"1\t2","warehouse/t/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"3\t4","warehouse/t/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "1/000000_0"}, + {"5\t6","warehouse/t/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, + {"7\t8","warehouse/t/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "2/000000_0"}, + {"9\t10","warehouse/t/" + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "3/000000_0"} + }; + Assert.assertEquals("Unexpected row count after conversion", expected.length, rs.size()); + for(int i = 0; i < expected.length; i++) { + Assert.assertTrue("Actual line " + i + " bc: " + rs.get(i), rs.get(i).startsWith(expected[i][0])); + Assert.assertTrue("Actual line(file) " + i + " bc: " + rs.get(i), rs.get(i).endsWith(expected[i][1])); + } + //make the table ACID + runStatementOnDriver("alter table T SET TBLPROPERTIES ('transactional'='true')", confForTez); + rs = runStatementOnDriver("select a,b from T order by a, b", confForTez); + Assert.assertEquals("After to Acid conversion", TestTxnCommands2.stringifyValues(values), rs); + + //run Major compaction + runStatementOnDriver("alter table T compact 'major'", confForTez); + TestTxnCommands2.runWorker(hiveConf); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from T order by ROW__ID", confForTez); + LOG.warn(testName.getMethodName() + ": after compact major of T:"); + for (String s : rs) { + LOG.warn(s); + } + String[][] expected2 = { + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":0}\t1\t2", "warehouse/t/base_-9223372036854775808/bucket_00000"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":1}\t3\t4", "warehouse/t/base_-9223372036854775808/bucket_00000"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":2}\t5\t6", "warehouse/t/base_-9223372036854775808/bucket_00000"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":3}\t7\t8", "warehouse/t/base_-9223372036854775808/bucket_00000"}, + {"{\"transactionid\":0,\"bucketid\":536870912,\"rowid\":4}\t9\t10", "warehouse/t/base_-9223372036854775808/bucket_00000"} + }; + Assert.assertEquals("Unexpected row count after major compact", expected2.length, rs.size()); + for(int i = 0; i < expected2.length; i++) { + Assert.assertTrue("Actual line " + i + " ac: " + rs.get(i), rs.get(i).startsWith(expected2[i][0])); + Assert.assertTrue("Actual line(file) " + i + " ac: " + rs.get(i), rs.get(i).endsWith(expected2[i][1])); + } + } + /** + * 1. Insert into unbucketed acid table from Union all - union is removed and data is placed in + * subdirs of target table. + * 2. convert to acid table and check data + * 3. compact and check data + * Compare with {@link #testInsertWithRemoveUnion()} where T is transactional=false + */ + @Test + public void testAcidInsertWithRemoveUnion() throws Exception { + HiveConf confForTez = new HiveConf(hiveConf); // make a clone of existing hive conf + setupTez(confForTez); + int[][] values = {{1,2},{3,4},{5,6},{7,8},{9,10}}; + runStatementOnDriver("drop table if exists T", confForTez); + runStatementOnDriver("create table T (a int, b int) stored as ORC TBLPROPERTIES ('transactional'='true')", confForTez); + /*On Tez, below (T is transactional), we get the following layout +ekoifman:apache-hive-3.0.0-SNAPSHOT-bin ekoifman$ tree ~/dev/hiverwgit/itests/hive-unit/target/tmp/org.apache.hadoop.hive.ql.TestAcidOnTez-1505500035574/warehouse/t/.hive-staging_hive_2017-09-15_11-28-33_960_9111484239090506828-1/ +/Users/ekoifman/dev/hiverwgit/itests/hive-unit/target/tmp/org.apache.hadoop.hive.ql.TestAcidOnTez-1505500035574/warehouse/t/.hive-staging_hive_2017-09-15_11-28-33_960_9111484239090506828-1/ +âââ -ext-10000 + âââ HIVE_UNION_SUBDIR_1 + â  âââ 000000_0 + â  âââ _orc_acid_version + â  âââ delta_0000019_0000019_0001 + â  âââ bucket_00000 + âââ HIVE_UNION_SUBDIR_2 + â  âââ 000000_0 + â  âââ _orc_acid_version + â  âââ delta_0000019_0000019_0002 + â  âââ bucket_00000 + âââ HIVE_UNION_SUBDIR_3 + âââ 000000_0 + âââ _orc_acid_version + âââ delta_0000019_0000019_0003 + âââ bucket_00000 + +10 directories, 6 files */ + runStatementOnDriver("insert into T(a,b) select a, b from " + Table.ACIDTBL + " where a between 1 and 3 union all select a, b from " + Table.ACIDTBL + " where a between 5 and 7 union all select a, b from " + Table.ACIDTBL + " where a >= 9", confForTez); + List<String> rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from T order by a, b", confForTez); + LOG.warn(testName.getMethodName() + ": reading acid table T"); + for(String s : rs) { + LOG.warn(s); + } + + String[][] expected2 = { + {"{\"transactionid\":19,\"bucketid\":536870913,\"rowid\":0}\t1\t2", "warehouse/t/delta_0000019_0000019_0001/bucket_00000"}, + {"{\"transactionid\":19,\"bucketid\":536870913,\"rowid\":1}\t3\t4", "warehouse/t/delta_0000019_0000019_0001/bucket_00000"}, + {"{\"transactionid\":19,\"bucketid\":536870914,\"rowid\":0}\t5\t6", "warehouse/t/delta_0000019_0000019_0002/bucket_00000"}, + {"{\"transactionid\":19,\"bucketid\":536870914,\"rowid\":1}\t7\t8", "warehouse/t/delta_0000019_0000019_0002/bucket_00000"}, + {"{\"transactionid\":19,\"bucketid\":536870915,\"rowid\":0}\t9\t10", "warehouse/t/delta_0000019_0000019_0003/bucket_00000"} + }; + Assert.assertEquals("Unexpected row count", expected2.length, rs.size()); + for(int i = 0; i < expected2.length; i++) { + Assert.assertTrue("Actual line " + i + " ac: " + rs.get(i), rs.get(i).startsWith(expected2[i][0])); + Assert.assertTrue("Actual line(file) " + i + " ac: " + rs.get(i), rs.get(i).endsWith(expected2[i][1])); + } + } + @Test + public void testBucketedAcidInsertWithRemoveUnion() throws Exception { + HiveConf confForTez = new HiveConf(hiveConf); // make a clone of existing hive conf + setupTez(confForTez); + int[][] values = {{1,2},{2,4},{5,6},{6,8},{9,10}}; + runStatementOnDriver("delete from " + Table.ACIDTBL, confForTez); + runStatementOnDriver("insert into " + Table.ACIDTBL + TestTxnCommands2.makeValuesClause(values));//make sure both buckets are not empty + runStatementOnDriver("drop table if exists T", confForTez); + /* + With bucketed target table Union All is not removed + + ekoifman:apache-hive-3.0.0-SNAPSHOT-bin ekoifman$ tree ~/dev/hiverwgit/itests/hive-unit/target/tmp/org.apache.hadoop.hive.ql.TestAcidOnTez-1505510130462/warehouse/t/.hive-staging_hive_2017-09-15_14-16-32_422_4626314315862498838-1/ +/Users/ekoifman/dev/hiverwgit/itests/hive-unit/target/tmp/org.apache.hadoop.hive.ql.TestAcidOnTez-1505510130462/warehouse/t/.hive-staging_hive_2017-09-15_14-16-32_422_4626314315862498838-1/ +âââ -ext-10000 + âââ 000000_0 + â  âââ _orc_acid_version + â  âââ delta_0000021_0000021_0000 + â  âââ bucket_00000 + âââ 000001_0 + âââ _orc_acid_version + âââ delta_0000021_0000021_0000 + âââ bucket_00001 + +5 directories, 4 files +*/ + runStatementOnDriver("create table T (a int, b int) clustered by (a) into 2 buckets stored as ORC TBLPROPERTIES ('transactional'='true')", confForTez); + runStatementOnDriver("insert into T(a,b) select a, b from " + Table.ACIDTBL + " where a between 1 and 3 union all select a, b from " + Table.ACIDTBL + " where a between 5 and 7 union all select a, b from " + Table.ACIDTBL + " where a >= 9", confForTez); + List<String> rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from T order by a, b", confForTez); + LOG.warn(testName.getMethodName() + ": reading bucketed acid table T"); + for(String s : rs) { + LOG.warn(s); + } + String[][] expected2 = { + {"{\"transactionid\":21,\"bucketid\":536936448,\"rowid\":0}\t1\t2", "warehouse/t/delta_0000021_0000021_0000/bucket_00001"}, + {"{\"transactionid\":21,\"bucketid\":536870912,\"rowid\":0}\t2\t4", "warehouse/t/delta_0000021_0000021_0000/bucket_00000"}, + {"{\"transactionid\":21,\"bucketid\":536936448,\"rowid\":1}\t5\t6", "warehouse/t/delta_0000021_0000021_0000/bucket_00001"}, + {"{\"transactionid\":21,\"bucketid\":536870912,\"rowid\":1}\t6\t8", "warehouse/t/delta_0000021_0000021_0000/bucket_00000"}, + {"{\"transactionid\":21,\"bucketid\":536936448,\"rowid\":2}\t9\t10", "warehouse/t/delta_0000021_0000021_0000/bucket_00001"} + }; + Assert.assertEquals("Unexpected row count", expected2.length, rs.size()); + for(int i = 0; i < expected2.length; i++) { + Assert.assertTrue("Actual line " + i + " ac: " + rs.get(i), rs.get(i).startsWith(expected2[i][0])); + Assert.assertTrue("Actual line(file) " + i + " ac: " + rs.get(i), rs.get(i).endsWith(expected2[i][1])); } } // Ideally test like this should be a qfile test. However, the explain output from qfile is always @@ -613,6 +797,7 @@ public class TestAcidOnTez { private List<String> runStatementOnDriver(String stmt, HiveConf conf) throws Exception { Driver driver = new Driver(conf); + driver.setMaxRows(10000); CommandProcessorResponse cpr = driver.run(stmt); if(cpr.getResponseCode() != 0) { throw new RuntimeException(stmt + " failed: " + cpr); http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/Context.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/Context.java b/ql/src/java/org/apache/hadoop/hive/ql/Context.java index ed091ae..cedb486 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/Context.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/Context.java @@ -1001,4 +1001,7 @@ public class Context { public void setIsUpdateDeleteMerge(boolean isUpdate) { this.isUpdateDeleteMerge = isUpdate; } + public String getExecutionId() { + return executionId; + } } http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/Driver.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java index 4e7c80f..a3bbf00 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java @@ -1199,7 +1199,11 @@ public class Driver implements CommandProcessor { } // Set the transaction id in all of the acid file sinks if (haveAcidWrite()) { - for (FileSinkDesc desc : plan.getAcidSinks()) { + List<FileSinkDesc> acidSinks = new ArrayList<>(plan.getAcidSinks()); + //sorting makes tests easier to write since file names and ROW__IDs depend on statementId + //so this makes (file name -> data) mapping stable + acidSinks.sort((FileSinkDesc fsd1, FileSinkDesc fsd2) -> fsd1.getDirName().compareTo(fsd2.getDirName())); + for (FileSinkDesc desc : acidSinks) { desc.setTransactionId(txnMgr.getCurrentTxnId()); //it's possible to have > 1 FileSink writing to the same table/partition //e.g. Merge stmt, multi-insert stmt when mixing DP and SP writes http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java index cde2805..a1e4f96 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/MoveTask.java @@ -269,9 +269,18 @@ public class MoveTask extends Task<MoveWork> implements Serializable { if (lfd != null) { Path targetPath = lfd.getTargetDir(); Path sourcePath = lfd.getSourcePath(); - moveFile(sourcePath, targetPath, lfd.getIsDfsDir()); + if(lfd.getWriteType() == AcidUtils.Operation.INSERT) { + //'targetPath' is table root of un-partitioned table/partition + //'sourcePath' result of 'select ...' part of CTAS statement + assert lfd.getIsDfsDir(); + FileSystem srcFs = sourcePath.getFileSystem(conf); + List<Path> newFiles = new ArrayList<>(); + Hive.moveAcidFiles(srcFs, srcFs.globStatus(sourcePath), targetPath, newFiles); + } + else { + moveFile(sourcePath, targetPath, lfd.getIsDfsDir()); + } } - // Multi-file load is for dynamic partitions when some partitions do not // need to merge and they can simply be moved to the target directory. LoadMultiFilesDesc lmfd = work.getLoadMultiFilesWork(); http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java b/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java index fa0ba63..c64bc8c 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/io/AcidUtils.java @@ -41,6 +41,8 @@ import org.apache.hadoop.hive.metastore.TransactionalValidationListener; import org.apache.hadoop.hive.ql.ErrorMsg; import org.apache.hadoop.hive.ql.io.orc.OrcRecordUpdater; import org.apache.hadoop.hive.ql.metadata.Table; +import org.apache.hadoop.hive.ql.plan.CreateTableDesc; +import org.apache.hadoop.hive.ql.plan.TableDesc; import org.apache.hadoop.hive.shims.HadoopShims; import org.apache.hadoop.hive.shims.HadoopShims.HdfsFileStatusWithId; import org.apache.hadoop.hive.shims.ShimLoader; @@ -1142,6 +1144,16 @@ public class AcidUtils { return tableIsTransactional != null && tableIsTransactional.equalsIgnoreCase("true"); } + public static boolean isAcidTable(CreateTableDesc table) { + if (table == null || table.getTblProps() == null) { + return false; + } + String tableIsTransactional = table.getTblProps().get(hive_metastoreConstants.TABLE_IS_TRANSACTIONAL); + if (tableIsTransactional == null) { + tableIsTransactional = table.getTblProps().get(hive_metastoreConstants.TABLE_IS_TRANSACTIONAL.toUpperCase()); + } + return tableIsTransactional != null && tableIsTransactional.equalsIgnoreCase("true"); + } /** * Sets the acidOperationalProperties in the configuration object argument. http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java index 9f98b69..0f59917 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java @@ -130,6 +130,7 @@ import org.apache.hadoop.hive.metastore.api.StorageDescriptor; import org.apache.hadoop.hive.metastore.api.UniqueConstraintsRequest; import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants; import org.apache.hadoop.hive.ql.ErrorMsg; +import org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator; import org.apache.hadoop.hive.ql.exec.FunctionRegistry; import org.apache.hadoop.hive.ql.exec.FunctionTask; import org.apache.hadoop.hive.ql.exec.FunctionUtils; @@ -3469,12 +3470,14 @@ private void constructOneLBLocationMap(FileStatus fSta, } } - private static void moveAcidFiles(FileSystem fs, FileStatus[] stats, Path dst, + public static void moveAcidFiles(FileSystem fs, FileStatus[] stats, Path dst, List<Path> newFiles) throws HiveException { // The layout for ACID files is table|partname/base|delta|delete_delta/bucket - // We will always only be writing delta files. In the buckets created by FileSinkOperator - // it will look like bucket/delta|delete_delta/bucket. So we need to move that into - // the above structure. For the first mover there will be no delta directory, + // We will always only be writing delta files ( except IOW which writes base_X/ ). + // In the buckets created by FileSinkOperator + // it will look like original_bucket/delta|delete_delta/bucket + // (e.g. .../-ext-10004/000000_0/delta_0000014_0000014_0000/bucket_00000). So we need to + // move that into the above structure. For the first mover there will be no delta directory, // so we can move the whole directory. // For everyone else we will need to just move the buckets under the existing delta // directory. @@ -3489,6 +3492,36 @@ private void constructOneLBLocationMap(FileStatus fSta, FileStatus[] origBucketStats = null; try { origBucketStats = fs.listStatus(srcPath, AcidUtils.originalBucketFilter); + if(origBucketStats == null || origBucketStats.length == 0) { + /** + check if we are dealing with data with non-standard layout. For example a write + produced by a (optimized) Union All query + which looks like + âââ -ext-10000 + âââ HIVE_UNION_SUBDIR_1 + â  âââ 000000_0 + â  âââ _orc_acid_version + â  âââ delta_0000019_0000019_0001 + â  âââ bucket_00000 + âââ HIVE_UNION_SUBDIR_2 + â  âââ 000000_0 + â  âââ _orc_acid_version + â  âââ delta_0000019_0000019_0002 + â  âââ bucket_00000 + The assumption is that we either have all data in subdirs or root of srcPath + but not both. + For Union case, we expect delta dirs to have unique names which is assured by + {@link org.apache.hadoop.hive.ql.optimizer.QueryPlanPostProcessor} + */ + FileStatus[] unionSubdirs = fs.globStatus(new Path(srcPath, + AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + "[0-9]*")); + List<FileStatus> buckets = new ArrayList<>(); + for(FileStatus unionSubdir : unionSubdirs) { + Collections.addAll(buckets, + fs.listStatus(unionSubdir.getPath(), AcidUtils.originalBucketFilter)); + } + origBucketStats = buckets.toArray(new FileStatus[buckets.size()]); + } } catch (IOException e) { String msg = "Unable to look for bucket files in src path " + srcPath.toUri().toString(); LOG.error(msg); @@ -3502,7 +3535,7 @@ private void constructOneLBLocationMap(FileStatus fSta, fs, dst, origBucketPath, createdDeltaDirs, newFiles); moveAcidFiles(AcidUtils.DELETE_DELTA_PREFIX, AcidUtils.deleteEventDeltaDirFilter, fs, dst,origBucketPath, createdDeltaDirs, newFiles); - moveAcidFiles(AcidUtils.BASE_PREFIX, AcidUtils.baseFileFilter, + moveAcidFiles(AcidUtils.BASE_PREFIX, AcidUtils.baseFileFilter,//for Insert Overwrite fs, dst, origBucketPath, createdDeltaDirs, newFiles); } } http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/optimizer/QueryPlanPostProcessor.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/QueryPlanPostProcessor.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/QueryPlanPostProcessor.java new file mode 100644 index 0000000..b5bc386 --- /dev/null +++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/QueryPlanPostProcessor.java @@ -0,0 +1,166 @@ +package org.apache.hadoop.hive.ql.optimizer; + +import org.apache.hadoop.hive.ql.exec.FileSinkOperator; +import org.apache.hadoop.hive.ql.exec.Operator; +import org.apache.hadoop.hive.ql.exec.OperatorUtils; +import org.apache.hadoop.hive.ql.exec.Task; +import org.apache.hadoop.hive.ql.exec.repl.ReplStateLogWork; +import org.apache.hadoop.hive.ql.exec.repl.bootstrap.ReplLoadWork; +import org.apache.hadoop.hive.ql.index.IndexMetadataChangeWork; +import org.apache.hadoop.hive.ql.io.AcidUtils; +import org.apache.hadoop.hive.ql.parse.GenTezProcContext; +import org.apache.hadoop.hive.ql.parse.GenTezWork; +import org.apache.hadoop.hive.ql.parse.spark.GenSparkWork; +import org.apache.hadoop.hive.ql.plan.ArchiveWork; +import org.apache.hadoop.hive.ql.plan.BaseWork; +import org.apache.hadoop.hive.ql.plan.ColumnStatsUpdateWork; +import org.apache.hadoop.hive.ql.plan.ColumnStatsWork; +import org.apache.hadoop.hive.ql.plan.ConditionalWork; +import org.apache.hadoop.hive.ql.plan.CopyWork; +import org.apache.hadoop.hive.ql.plan.DDLWork; +import org.apache.hadoop.hive.ql.plan.DependencyCollectionWork; +import org.apache.hadoop.hive.ql.plan.ExplainSQRewriteWork; +import org.apache.hadoop.hive.ql.plan.ExplainWork; +import org.apache.hadoop.hive.ql.plan.FetchWork; +import org.apache.hadoop.hive.ql.plan.FileSinkDesc; +import org.apache.hadoop.hive.ql.plan.FunctionWork; +import org.apache.hadoop.hive.ql.plan.MapredLocalWork; +import org.apache.hadoop.hive.ql.plan.MapredWork; +import org.apache.hadoop.hive.ql.plan.MoveWork; +import org.apache.hadoop.hive.ql.plan.SparkWork; +import org.apache.hadoop.hive.ql.plan.StatsNoJobWork; +import org.apache.hadoop.hive.ql.plan.StatsWork; +import org.apache.hadoop.hive.ql.plan.TezWork; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import java.util.List; +import java.util.Set; + +/** + * Finds Acid FileSinkDesc objects which can be created in the physical (disconnected) plan, e.g. + * {@link org.apache.hadoop.hive.ql.parse.GenTezUtils#removeUnionOperators(GenTezProcContext, BaseWork, int)} + * so that statementId can be properly assigned to ensure unique ROW__IDs + * {@link org.apache.hadoop.hive.ql.optimizer.unionproc.UnionProcFactory} is another example where + * Union All optimizations create new FileSinkDescS + */ +public class QueryPlanPostProcessor { + private static final Logger LOG = LoggerFactory.getLogger(QueryPlanPostProcessor.class); + + public QueryPlanPostProcessor(List<Task<?>> rootTasks, Set<FileSinkDesc> acidSinks, String executionId) { + for(Task<?> t : rootTasks) { + //Work + Object work = t.getWork(); + if(work instanceof TezWork) { + for(BaseWork bw : ((TezWork)work).getAllWorkUnsorted()) { + collectFileSinkDescs(bw.getAllLeafOperators(), acidSinks); + } + } + else if(work instanceof BaseWork) { + collectFileSinkDescs(((BaseWork)work).getAllLeafOperators(), acidSinks); + } + else if(work instanceof MapredWork) { + MapredWork w = (MapredWork)work; + if(w.getMapWork() != null) { + collectFileSinkDescs(w.getMapWork().getAllLeafOperators(), acidSinks); + } + if(w.getReduceWork() != null) { + collectFileSinkDescs(w.getReduceWork().getAllLeafOperators(), acidSinks); + } + } + else if(work instanceof SparkWork) { + for(BaseWork bw : ((SparkWork)work).getRoots()) { + collectFileSinkDescs(bw.getAllLeafOperators(), acidSinks); + } + } + else if(work instanceof MapredLocalWork) { + //I don't think this can have any FileSinkOperatorS - more future proofing + Set<FileSinkOperator> fileSinkOperatorSet = OperatorUtils.findOperators(((MapredLocalWork)work).getAliasToWork().values(), FileSinkOperator.class); + for(FileSinkOperator fsop : fileSinkOperatorSet) { + collectFileSinkDescs(fsop, acidSinks); + } + } + else if(work instanceof ExplainWork) { + new QueryPlanPostProcessor(((ExplainWork)work).getRootTasks(), acidSinks, executionId); + } + /* + ekoifman:~ ekoifman$ cd dev/hiverwgit/ql/src/java/org/apache/ +ekoifman:apache ekoifman$ find . -name *Work.java +./hadoop/hive/ql/exec/repl/bootstrap/ReplLoadWork.java +./hadoop/hive/ql/exec/repl/ReplDumpWork.java +./hadoop/hive/ql/exec/repl/ReplStateLogWork.java +./hadoop/hive/ql/index/IndexMetadataChangeWork.java +./hadoop/hive/ql/io/merge/MergeFileWork.java - extends MapWork +./hadoop/hive/ql/io/rcfile/stats/PartialScanWork.java - extends MapWork +./hadoop/hive/ql/io/rcfile/truncate/ColumnTruncateWork.java - extends MapWork +./hadoop/hive/ql/parse/GenTezWork.java +./hadoop/hive/ql/parse/spark/GenSparkWork.java +./hadoop/hive/ql/plan/ArchiveWork.java +./hadoop/hive/ql/plan/BaseWork.java +./hadoop/hive/ql/plan/ColumnStatsUpdateWork.java +./hadoop/hive/ql/plan/ColumnStatsWork.java +./hadoop/hive/ql/plan/ConditionalWork.java +./hadoop/hive/ql/plan/CopyWork.java +./hadoop/hive/ql/plan/DDLWork.java +./hadoop/hive/ql/plan/DependencyCollectionWork.java +./hadoop/hive/ql/plan/ExplainSQRewriteWork.java +./hadoop/hive/ql/plan/ExplainWork.java +./hadoop/hive/ql/plan/FetchWork.java +./hadoop/hive/ql/plan/FunctionWork.java +./hadoop/hive/ql/plan/MapredLocalWork.java +./hadoop/hive/ql/plan/MapredWork.java +./hadoop/hive/ql/plan/MapWork.java - extends BaseWork +./hadoop/hive/ql/plan/MergeJoinWork.java - extends BaseWork +./hadoop/hive/ql/plan/MoveWork.java +./hadoop/hive/ql/plan/ReduceWork.java +./hadoop/hive/ql/plan/ReplCopyWork.java - extends CopyWork +./hadoop/hive/ql/plan/SparkWork.java +./hadoop/hive/ql/plan/StatsNoJobWork.java +./hadoop/hive/ql/plan/StatsWork.java +./hadoop/hive/ql/plan/TezWork.java +./hadoop/hive/ql/plan/UnionWork.java - extends BaseWork + */ + else if(work instanceof ReplLoadWork || + work instanceof ReplStateLogWork || + work instanceof IndexMetadataChangeWork || + work instanceof GenTezWork || + work instanceof GenSparkWork || + work instanceof ArchiveWork || + work instanceof ColumnStatsUpdateWork || + work instanceof ColumnStatsWork || + work instanceof ConditionalWork || + work instanceof CopyWork || + work instanceof DDLWork || + work instanceof DependencyCollectionWork || + work instanceof ExplainSQRewriteWork || + work instanceof FetchWork || + work instanceof FunctionWork || + work instanceof MoveWork || + work instanceof StatsNoJobWork || + work instanceof StatsWork) { + LOG.debug("Found " + work.getClass().getName() + " - no FileSinkOperation can be present. executionId=" + executionId); + } + else { + //if here, someone must have added new Work object - should it be walked to find FileSinks? + throw new IllegalArgumentException("Unexpected Work object: " + work.getClass() + " executionId=" + executionId); + } + } + } + private void collectFileSinkDescs(Operator<?> leaf, Set<FileSinkDesc> acidSinks) { + if(leaf instanceof FileSinkOperator) { + FileSinkDesc fsd = ((FileSinkOperator) leaf).getConf(); + if(fsd.getWriteType() != AcidUtils.Operation.NOT_ACID) { + if(acidSinks.add(fsd)) { + if(LOG.isDebugEnabled()) { + LOG.debug("Found Acid Sink: " + fsd.getDirName()); + } + } + } + } + } + private void collectFileSinkDescs(Set<Operator<?>> leaves, Set<FileSinkDesc> acidSinks) { + for(Operator<?> leaf : leaves) { + collectFileSinkDescs(leaf, acidSinks); + } + } +} http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java index 2a7f3d4..2fbadd3 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/optimizer/unionproc/UnionProcFactory.java @@ -24,6 +24,7 @@ import java.util.Map; import java.util.Stack; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.ql.exec.AbstractFileMergeOperator; import org.apache.hadoop.hive.ql.exec.FileSinkOperator; import org.apache.hadoop.hive.ql.exec.Operator; import org.apache.hadoop.hive.ql.exec.OperatorFactory; @@ -219,7 +220,7 @@ public final class UnionProcFactory { for (Operator<? extends OperatorDesc> parent : parents) { FileSinkDesc fileSinkDesc = (FileSinkDesc) fileSinkOp.getConf().clone(); - fileSinkDesc.setDirName(new Path(parentDirName, parent.getIdentifier())); + fileSinkDesc.setDirName(new Path(parentDirName, AbstractFileMergeOperator.UNION_SUDBIR_PREFIX + parent.getIdentifier())); fileSinkDesc.setLinkedFileSink(true); fileSinkDesc.setParentDir(parentDirName); parent.setChildOperators(null); http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java index 7bdf39e..fee5a79 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java @@ -131,6 +131,7 @@ import org.apache.hadoop.hive.ql.metadata.SessionHiveMetaStoreClient; import org.apache.hadoop.hive.ql.metadata.Table; import org.apache.hadoop.hive.ql.metadata.VirtualColumn; import org.apache.hadoop.hive.ql.optimizer.Optimizer; +import org.apache.hadoop.hive.ql.optimizer.QueryPlanPostProcessor; import org.apache.hadoop.hive.ql.optimizer.Transform; import org.apache.hadoop.hive.ql.optimizer.calcite.CalciteSemanticException; import org.apache.hadoop.hive.ql.optimizer.calcite.CalciteSemanticException.UnsupportedFeature; @@ -7129,6 +7130,8 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer { viewDesc.setSchema(new ArrayList<FieldSchema>(field_schemas)); } + destTableIsAcid = tblDesc != null && AcidUtils.isAcidTable(tblDesc); + boolean isDestTempFile = true; if (!ctx.isMRTmpFileURI(dest_path.toUri().toString())) { idToTableNameMap.put(String.valueOf(destTableId), dest_path.toUri().toString()); @@ -7139,8 +7142,7 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer { boolean isDfsDir = (dest_type.intValue() == QBMetaData.DEST_DFS_FILE); loadFileWork.add(new LoadFileDesc(tblDesc, viewDesc, queryTmpdir, dest_path, isDfsDir, cols, - colTypes)); - + colTypes, destTableIsAcid ? Operation.INSERT : Operation.NOT_ACID)); if (tblDesc == null) { if (viewDesc != null) { table_desc = PlanUtils.getTableDesc(viewDesc, cols, colTypes); @@ -7248,10 +7250,22 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer { (deleting(dest) ? AcidUtils.Operation.DELETE : AcidUtils.Operation.INSERT); fileSinkDesc.setWriteType(wt); - String destTableFullName = dest_tab.getCompleteName().replace('@', '.'); - Map<String, ASTNode> iowMap = qb.getParseInfo().getInsertOverwriteTables(); - if (iowMap.containsKey(destTableFullName)) { - fileSinkDesc.setInsertOverwrite(true); + switch (dest_type) { + case QBMetaData.DEST_PARTITION: + //fall through + case QBMetaData.DEST_TABLE: + //INSERT [OVERWRITE] path + String destTableFullName = dest_tab.getCompleteName().replace('@', '.'); + Map<String, ASTNode> iowMap = qb.getParseInfo().getInsertOverwriteTables(); + if (iowMap.containsKey(destTableFullName)) { + fileSinkDesc.setInsertOverwrite(true); + } + break; + case QBMetaData.DEST_DFS_FILE: + //CTAS path + break; + default: + throw new IllegalStateException("Unexpected dest_type=" + dest_tab); } acidFileSinks.add(fileSinkDesc); } @@ -11497,6 +11511,8 @@ public class SemanticAnalyzer extends BaseSemanticAnalyzer { compiler.compile(pCtx, rootTasks, inputs, outputs); fetchTask = pCtx.getFetchTask(); } + //find all Acid FileSinkOperatorS + QueryPlanPostProcessor qp = new QueryPlanPostProcessor(rootTasks, acidFileSinks, ctx.getExecutionId()); LOG.info("Completed plan generation"); // 10. put accessed columns to readEntity http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadDesc.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadDesc.java index d708df3..45d4fb0 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadDesc.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadDesc.java @@ -21,6 +21,7 @@ package org.apache.hadoop.hive.ql.plan; import java.io.Serializable; import org.apache.hadoop.fs.Path; +import org.apache.hadoop.hive.ql.io.AcidUtils; import org.apache.hadoop.hive.ql.plan.Explain.Level; /** @@ -30,12 +31,15 @@ import org.apache.hadoop.hive.ql.plan.Explain.Level; public class LoadDesc implements Serializable { private static final long serialVersionUID = 1L; private Path sourcePath; + /** + * Need to remember whether this is an acid compliant operation, and if so whether it is an + * insert, update, or delete. + */ + private final AcidUtils.Operation writeType; - public LoadDesc() { - } - - public LoadDesc(final Path sourcePath) { + public LoadDesc(final Path sourcePath, AcidUtils.Operation writeType) { this.sourcePath = sourcePath; + this.writeType = writeType; } @Explain(displayName = "source", explainLevels = { Level.EXTENDED }) @@ -46,4 +50,14 @@ public class LoadDesc implements Serializable { public void setSourcePath(Path sourcePath) { this.sourcePath = sourcePath; } + + public AcidUtils.Operation getWriteType() { + return writeType; + } + + @Explain(displayName = "Write Type") + public String getWriteTypeString() { + //if acid write, add to plan output, else don't bother + return getWriteType() == AcidUtils.Operation.NOT_ACID ? null : getWriteType().toString(); + } } http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadFileDesc.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadFileDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadFileDesc.java index 03202fb..0292af5 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadFileDesc.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadFileDesc.java @@ -21,7 +21,7 @@ package org.apache.hadoop.hive.ql.plan; import java.io.Serializable; import org.apache.hadoop.fs.Path; -import org.apache.hadoop.hive.ql.exec.PTFUtils; +import org.apache.hadoop.hive.ql.io.AcidUtils; /** * LoadFileDesc. @@ -36,11 +36,8 @@ public class LoadFileDesc extends LoadDesc implements Serializable { private String columnTypes; private String destinationCreateTable; - public LoadFileDesc() { - } - public LoadFileDesc(final LoadFileDesc o) { - super(o.getSourcePath()); + super(o.getSourcePath(), o.getWriteType()); this.targetDir = o.targetDir; this.isDfsDir = o.isDfsDir; @@ -51,8 +48,8 @@ public class LoadFileDesc extends LoadDesc implements Serializable { public LoadFileDesc(final CreateTableDesc createTableDesc, final CreateViewDesc createViewDesc, final Path sourcePath, final Path targetDir, final boolean isDfsDir, - final String columns, final String columnTypes) { - this(sourcePath, targetDir, isDfsDir, columns, columnTypes); + final String columns, final String columnTypes, AcidUtils.Operation writeType) { + this(sourcePath, targetDir, isDfsDir, columns, columnTypes, writeType); if (createTableDesc != null && createTableDesc.getDatabaseName() != null && createTableDesc.getTableName() != null) { destinationCreateTable = (createTableDesc.getTableName().contains(".") ? "" : createTableDesc @@ -66,9 +63,14 @@ public class LoadFileDesc extends LoadDesc implements Serializable { } public LoadFileDesc(final Path sourcePath, final Path targetDir, - final boolean isDfsDir, final String columns, final String columnTypes) { + final boolean isDfsDir, final String columns, final String columnTypes) { + this(sourcePath, targetDir, isDfsDir, columns, columnTypes, AcidUtils.Operation.NOT_ACID); + } + private LoadFileDesc(final Path sourcePath, final Path targetDir, + final boolean isDfsDir, final String columns, + final String columnTypes, AcidUtils.Operation writeType) { - super(sourcePath); + super(sourcePath, writeType); this.targetDir = targetDir; this.isDfsDir = isDfsDir; this.columns = columns; http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java index aa77850..ab0a92a 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/LoadTableDesc.java @@ -30,30 +30,25 @@ import org.apache.hadoop.hive.ql.plan.Explain.Level; * LoadTableDesc. * */ -public class LoadTableDesc extends org.apache.hadoop.hive.ql.plan.LoadDesc - implements Serializable { +public class LoadTableDesc extends LoadDesc implements Serializable { private static final long serialVersionUID = 1L; private boolean replace; private DynamicPartitionCtx dpCtx; private ListBucketingCtx lbCtx; private boolean inheritTableSpecs = true; //For partitions, flag controlling whether the current //table specs are to be used - // Need to remember whether this is an acid compliant operation, and if so whether it is an - // insert, update, or delete. - private AcidUtils.Operation writeType; // TODO: the below seems like they should just be combined into partitionDesc private org.apache.hadoop.hive.ql.plan.TableDesc table; private Map<String, String> partitionSpec; // NOTE: this partitionSpec has to be ordered map public LoadTableDesc(final LoadTableDesc o) { - super(o.getSourcePath()); + super(o.getSourcePath(), o.getWriteType()); this.replace = o.replace; this.dpCtx = o.dpCtx; this.lbCtx = o.lbCtx; this.inheritTableSpecs = o.inheritTableSpecs; - this.writeType = o.writeType; this.table = o.table; this.partitionSpec = o.partitionSpec; } @@ -63,8 +58,8 @@ public class LoadTableDesc extends org.apache.hadoop.hive.ql.plan.LoadDesc final Map<String, String> partitionSpec, final boolean replace, final AcidUtils.Operation writeType) { - super(sourcePath); - init(table, partitionSpec, replace, writeType); + super(sourcePath, writeType); + init(table, partitionSpec, replace); } /** @@ -104,24 +99,22 @@ public class LoadTableDesc extends org.apache.hadoop.hive.ql.plan.LoadDesc final org.apache.hadoop.hive.ql.plan.TableDesc table, final DynamicPartitionCtx dpCtx, final AcidUtils.Operation writeType) { - super(sourcePath); + super(sourcePath, writeType); this.dpCtx = dpCtx; if (dpCtx != null && dpCtx.getPartSpec() != null && partitionSpec == null) { - init(table, dpCtx.getPartSpec(), true, writeType); + init(table, dpCtx.getPartSpec(), true); } else { - init(table, new LinkedHashMap<String, String>(), true, writeType); + init(table, new LinkedHashMap<String, String>(), true); } } private void init( final org.apache.hadoop.hive.ql.plan.TableDesc table, final Map<String, String> partitionSpec, - final boolean replace, - AcidUtils.Operation writeType) { + final boolean replace) { this.table = table; this.partitionSpec = partitionSpec; this.replace = replace; - this.writeType = writeType; } @Explain(displayName = "table", explainLevels = { Level.USER, Level.DEFAULT, Level.EXTENDED }) @@ -181,7 +174,4 @@ public class LoadTableDesc extends org.apache.hadoop.hive.ql.plan.LoadDesc this.lbCtx = lbCtx; } - public AcidUtils.Operation getWriteType() { - return writeType; - } } http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/java/org/apache/hadoop/hive/ql/plan/TezEdgeProperty.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/plan/TezEdgeProperty.java b/ql/src/java/org/apache/hadoop/hive/ql/plan/TezEdgeProperty.java index b695f0f..bbed9be 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/plan/TezEdgeProperty.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/plan/TezEdgeProperty.java @@ -22,12 +22,12 @@ import org.apache.hadoop.hive.conf.HiveConf; public class TezEdgeProperty { - public enum EdgeType { - SIMPLE_EDGE, + public enum EdgeType {//todo: HIVE-15549 + SIMPLE_EDGE,//SORT_PARTITION_EDGE BROADCAST_EDGE, - CONTAINS, - CUSTOM_EDGE, - CUSTOM_SIMPLE_EDGE, + CONTAINS,//used for union (all?) + CUSTOM_EDGE,//CO_PARTITION_EDGE + CUSTOM_SIMPLE_EDGE,//PARTITION_EDGE ONE_TO_ONE_EDGE } http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/test/org/apache/hadoop/hive/ql/TestTxnNoBuckets.java ---------------------------------------------------------------------- diff --git a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnNoBuckets.java b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnNoBuckets.java index 1f0c269..880329d 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnNoBuckets.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnNoBuckets.java @@ -5,7 +5,10 @@ import org.apache.hadoop.hive.ql.io.BucketCodec; import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse; import org.junit.Assert; import org.junit.Before; +import org.junit.Ignore; +import org.junit.Rule; import org.junit.Test; +import org.junit.rules.TestName; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -20,6 +23,8 @@ public class TestTxnNoBuckets extends TxnCommandsBaseForTests { File.separator + TestTxnNoBuckets.class.getCanonicalName() + "-" + System.currentTimeMillis() ).getPath().replaceAll("\\\\", "/"); + @Rule + public TestName testName = new TestName(); @Override String getTestDataDir() { return TEST_DATA_DIR; @@ -65,17 +70,7 @@ public class TestTxnNoBuckets extends TxnCommandsBaseForTests { Assert.assertTrue(rs.get(2), rs.get(2).endsWith("nobuckets/delta_0000019_0000019_0000/bucket_00001")); Assert.assertTrue(rs.get(3), rs.get(3).startsWith("{\"transactionid\":19,\"bucketid\":536936448,\"rowid\":1}\t2\t2\t2\t")); Assert.assertTrue(rs.get(3), rs.get(3).endsWith("nobuckets/delta_0000019_0000019_0000/bucket_00001")); - /*todo: WTF? - RS for update seems to spray randomly... is that OK? maybe as long as all resultant files have different names... will they? - Assuming we name them based on taskId, we should create bucketX and bucketY. - we delete events can be written to bucketX file it could be useful for filter delete for a split by file name since the insert - events seem to be written to a proper bucketX file. In fact this may reduce the number of changes elsewhere like compactor... maybe - But this limits the parallelism - what is worse, you don't know what the parallelism should be until you have a list of all the - input files since bucket count is no longer a metadata property. Also, with late Update split, the file name has already been determined - from taskId so the Insert part won't end up matching the bucketX property necessarily. - With early Update split, the Insert can still be an insert - i.e. go to appropriate bucketX. But deletes will still go wherever (random shuffle) - unless you know all the bucketX files to be read - may not be worth the trouble. - * 2nd: something in FS fails. ArrayIndexOutOfBoundsException: 1 at FileSinkOperator.process(FileSinkOperator.java:779)*/ + runStatementOnDriver("update nobuckets set c3 = 17 where c3 in(0,1)"); rs = runStatementOnDriver("select ROW__ID, c1, c2, c3, INPUT__FILE__NAME from nobuckets order by INPUT__FILE__NAME, ROW__ID"); LOG.warn("after update"); @@ -152,15 +147,6 @@ public class TestTxnNoBuckets extends TxnCommandsBaseForTests { } /** - * all of these pass but don't do exactly the right thing - * files land as if it's not an acid table "warehouse/myctas4/000000_0" - * even though in {@link org.apache.hadoop.hive.metastore.TransactionalValidationListener} fires - * and sees it as transactional table - * look for QB.isCTAS() and CreateTableDesc() in SemanticAnalyzer - * - * On read, these files are treated like non acid to acid conversion - * - * see HIVE-15899 * See CTAS tests in TestAcidOnTez */ @Test @@ -169,30 +155,177 @@ public class TestTxnNoBuckets extends TxnCommandsBaseForTests { runStatementOnDriver("insert into " + Table.NONACIDORCTBL + makeValuesClause(values)); runStatementOnDriver("create table myctas stored as ORC TBLPROPERTIES ('transactional" + "'='true', 'transactional_properties'='default') as select a, b from " + Table.NONACIDORCTBL); - List<String> rs = runStatementOnDriver("select * from myctas order by a, b"); - Assert.assertEquals(stringifyValues(values), rs); + List<String> rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from myctas order by ROW__ID"); + String expected[][] = { + {"{\"transactionid\":14,\"bucketid\":536870912,\"rowid\":0}\t3\t4", "warehouse/myctas/delta_0000014_0000014_0000/bucket_00000"}, + {"{\"transactionid\":14,\"bucketid\":536870912,\"rowid\":1}\t1\t2", "warehouse/myctas/delta_0000014_0000014_0000/bucket_00000"}, + }; + checkExpected(rs, expected, "Unexpected row count after ctas from non acid table"); runStatementOnDriver("insert into " + Table.ACIDTBL + makeValuesClause(values)); runStatementOnDriver("create table myctas2 stored as ORC TBLPROPERTIES ('transactional" + "'='true', 'transactional_properties'='default') as select a, b from " + Table.ACIDTBL); - rs = runStatementOnDriver("select * from myctas2 order by a, b"); - Assert.assertEquals(stringifyValues(values), rs); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from myctas2 order by ROW__ID"); + String expected2[][] = { + {"{\"transactionid\":17,\"bucketid\":536870912,\"rowid\":0}\t3\t4", "warehouse/myctas2/delta_0000017_0000017_0000/bucket_00000"}, + {"{\"transactionid\":17,\"bucketid\":536870912,\"rowid\":1}\t1\t2", "warehouse/myctas2/delta_0000017_0000017_0000/bucket_00000"}, + }; + checkExpected(rs, expected2, "Unexpected row count after ctas from acid table"); runStatementOnDriver("create table myctas3 stored as ORC TBLPROPERTIES ('transactional" + "'='true', 'transactional_properties'='default') as select a, b from " + Table.NONACIDORCTBL + " union all select a, b from " + Table.ACIDTBL); - rs = runStatementOnDriver("select * from myctas3 order by a, b"); - Assert.assertEquals(stringifyValues(new int[][] {{1,2},{1,2},{3,4},{3,4}}), rs); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from myctas3 order by ROW__ID"); + String expected3[][] = { + {"{\"transactionid\":19,\"bucketid\":536870912,\"rowid\":0}\t3\t4", "warehouse/myctas3/delta_0000019_0000019_0000/bucket_00000"}, + {"{\"transactionid\":19,\"bucketid\":536870912,\"rowid\":1}\t1\t2", "warehouse/myctas3/delta_0000019_0000019_0000/bucket_00000"}, + {"{\"transactionid\":19,\"bucketid\":536936448,\"rowid\":0}\t3\t4", "warehouse/myctas3/delta_0000019_0000019_0000/bucket_00001"}, + {"{\"transactionid\":19,\"bucketid\":536936448,\"rowid\":1}\t1\t2", "warehouse/myctas3/delta_0000019_0000019_0000/bucket_00001"}, + }; + checkExpected(rs, expected3, "Unexpected row count after ctas from union all query"); runStatementOnDriver("create table myctas4 stored as ORC TBLPROPERTIES ('transactional" + "'='true', 'transactional_properties'='default') as select a, b from " + Table.NONACIDORCTBL + " union distinct select a, b from " + Table.ACIDTBL); - rs = runStatementOnDriver("select * from myctas4 order by a, b"); + rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from myctas4 order by ROW__ID"); + String expected4[][] = { + {"{\"transactionid\":21,\"bucketid\":536870912,\"rowid\":0}\t1\t2", "/delta_0000021_0000021_0000/bucket_00000"}, + {"{\"transactionid\":21,\"bucketid\":536870912,\"rowid\":1}\t3\t4", "/delta_0000021_0000021_0000/bucket_00000"}, + }; + checkExpected(rs, expected4, "Unexpected row count after ctas from union distinct query"); + } + /** + * Insert into unbucketed acid table from union all query + * Union All is flattend so nested subdirs are created and acid move drops them since + * delta dirs have unique names + */ + @Test + public void testInsertToAcidWithUnionRemove() throws Exception { + hiveConf.setBoolVar(HiveConf.ConfVars.HIVE_OPTIMIZE_UNION_REMOVE, true); + hiveConf.setVar(HiveConf.ConfVars.HIVEFETCHTASKCONVERSION, "none"); + d.close(); + d = new Driver(hiveConf); + int[][] values = {{1,2},{3,4},{5,6},{7,8},{9,10}}; + runStatementOnDriver("insert into " + TxnCommandsBaseForTests.Table.ACIDTBL + makeValuesClause(values));//HIVE-17138: this creates 1 delta_0000013_0000013_0000/bucket_00001 + runStatementOnDriver("drop table if exists T"); + runStatementOnDriver("create table T (a int, b int) stored as ORC TBLPROPERTIES ('transactional'='true')"); + /* + So Union All removal kicks in and we get 3 subdirs in staging. +ekoifman:apache-hive-3.0.0-SNAPSHOT-bin ekoifman$ tree /Users/ekoifman/dev/hiverwgit/ql/target/tmp/org.apache.hadoop.hive.ql.TestTxnNoBuckets-1505516390532/warehouse/t/.hive-staging_hive_2017-09-15_16-05-06_895_1123322677843388168-1/ +âââ -ext-10000 + âââ HIVE_UNION_SUBDIR_19 + â  âââ 000000_0 + â  âââ _orc_acid_version + â  âââ delta_0000016_0000016_0001 + âââ HIVE_UNION_SUBDIR_20 + â  âââ 000000_0 + â  âââ _orc_acid_version + â  âââ delta_0000016_0000016_0002 + âââ HIVE_UNION_SUBDIR_21 + âââ 000000_0 + âââ _orc_acid_version + âââ delta_0000016_0000016_0003*/ + runStatementOnDriver("insert into T(a,b) select a, b from " + TxnCommandsBaseForTests.Table.ACIDTBL + " where a between 1 and 3 group by a, b union all select a, b from " + TxnCommandsBaseForTests.Table.ACIDTBL + " where a between 5 and 7 union all select a, b from " + TxnCommandsBaseForTests.Table.ACIDTBL + " where a >= 9"); + + List<String> rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from T order by ROW__ID"); + + String expected[][] = { + {"{\"transactionid\":16,\"bucketid\":536870913,\"rowid\":0}\t1\t2", "/delta_0000016_0000016_0001/bucket_00000"}, + {"{\"transactionid\":16,\"bucketid\":536870913,\"rowid\":1}\t3\t4", "/delta_0000016_0000016_0001/bucket_00000"}, + {"{\"transactionid\":16,\"bucketid\":536870914,\"rowid\":0}\t7\t8", "/delta_0000016_0000016_0002/bucket_00000"}, + {"{\"transactionid\":16,\"bucketid\":536870914,\"rowid\":1}\t5\t6", "/delta_0000016_0000016_0002/bucket_00000"}, + {"{\"transactionid\":16,\"bucketid\":536870915,\"rowid\":0}\t9\t10", "/delta_0000016_0000016_0003/bucket_00000"}, + }; + checkExpected(rs, expected, "Unexpected row count after ctas"); + } + private void checkExpected(List<String> rs, String[][] expected, String msg) { + LOG.warn(testName.getMethodName() + ": read data(" + msg + "): "); + for(String s : rs) { + LOG.warn(s); + } + Assert.assertEquals( testName.getMethodName() + ": " + msg, expected.length, rs.size()); + //verify data and layout + for(int i = 0; i < expected.length; i++) { + Assert.assertTrue("Actual line " + i + " bc: " + rs.get(i), rs.get(i).startsWith(expected[i][0])); + Assert.assertTrue("Actual line(file) " + i + " bc: " + rs.get(i), rs.get(i).endsWith(expected[i][1])); + } + } + /** + * The idea here is to create a non acid table that was written by multiple writers, i.e. + * unbucketed table that has 000000_0 & 000001_0, for example. Unfortunately this doesn't work + * due to 'merge' logic - see comments in the method + */ + @Ignore + @Test + public void testToAcidConversionMultiBucket() throws Exception { + int[][] values = {{1,2},{2,4},{5,6},{6,8},{9,10}}; + runStatementOnDriver("insert into " + Table.ACIDTBL + makeValuesClause(values)); + runStatementOnDriver("drop table if exists T"); + runStatementOnDriver("create table T (a int, b int) stored as ORC TBLPROPERTIES ('transactional'='false')"); + /*T non-acid + non bucketd - 3 writers are created and then followed by merge to create a single output file + though how the data from union is split between writers is a mystery + (bucketed tables don't do merge) + Processing data file file:/Users/ekoifman/dev/hiverwgit/ql/target/tmp/org.apache.hadoop.hive.ql.TestTxnNoBuckets-1505317179157/warehouse/t/.hive-staging_hive_2017-09-13_08-40-30_275_8623609103176711840-1/-ext-10000/000000_0 [length: 515] +{"a":6,"b":8} +{"a":9,"b":10} +{"a":5,"b":6} +{"a":1,"b":2} +{"a":2,"b":4} +________________________________________________________________________________________________________________________ + +Processing data file file:/Users/ekoifman/dev/hiverwgit/ql/target/tmp/org.apache.hadoop.hive.ql.TestTxnNoBuckets-1505317179157/warehouse/t/.hive-staging_hive_2017-09-13_08-40-30_275_8623609103176711840-1/-ext-10003/000000_0 [length: 242] +{"a":6,"b":8} +________________________________________________________________________________________________________________________ + +Processing data file file:/Users/ekoifman/dev/hiverwgit/ql/target/tmp/org.apache.hadoop.hive.ql.TestTxnNoBuckets-1505317179157/warehouse/t/.hive-staging_hive_2017-09-13_08-40-30_275_8623609103176711840-1/-ext-10003/000001_0 [length: 244] +{"a":9,"b":10} +{"a":5,"b":6} +________________________________________________________________________________________________________________________ + +Processing data file file:/Users/ekoifman/dev/hiverwgit/ql/target/tmp/org.apache.hadoop.hive.ql.TestTxnNoBuckets-1505317179157/warehouse/t/.hive-staging_hive_2017-09-13_08-40-30_275_8623609103176711840-1/-ext-10003/000002_0 [length: 242] +{"a":1,"b":2} +{"a":2,"b":4} + */ + runStatementOnDriver("insert into T(a,b) select a, b from " + Table.ACIDTBL + " where a between 1 and 3 group by a, b union all select a, b from " + Table.ACIDTBL + " where a between 5 and 7 union all select a, b from " + Table.ACIDTBL + " where a >= 9"); + List<String> rs = runStatementOnDriver("select a, b, INPUT__FILE__NAME from T order by a, b, INPUT__FILE__NAME"); + LOG.warn("before converting to acid"); + for(String s : rs) { + LOG.warn(s); + } + } + @Test + public void testInsertFromUnion() throws Exception { + int[][] values = {{1,2},{2,4},{5,6},{6,8},{9,10}}; + runStatementOnDriver("insert into " + Table.NONACIDNONBUCKET + makeValuesClause(values)); + runStatementOnDriver("drop table if exists T"); + runStatementOnDriver("create table T (a int, b int) stored as ORC TBLPROPERTIES ('transactional'='true')"); + runStatementOnDriver("insert into T(a,b) select a, b from " + Table.NONACIDNONBUCKET + " where a between 1 and 3 group by a, b union all select a, b from " + Table.NONACIDNONBUCKET + " where a between 5 and 7 union all select a, b from " + Table.NONACIDNONBUCKET + " where a >= 9"); + List<String> rs = runStatementOnDriver("select ROW__ID, a, b, INPUT__FILE__NAME from T order by a, b, INPUT__FILE__NAME"); + LOG.warn("before converting to acid"); + for(String s : rs) { + LOG.warn(s); + } + /* + The number of writers seems to be based on number of MR jobs for the src query. todo check number of FileSinks + warehouse/t/.hive-staging_hive_2017-09-13_08-59-28_141_6304543600372946004-1/-ext-10000/000000_0/delta_0000016_0000016_0000/bucket_00000 [length: 648] + {"operation":0,"originalTransaction":16,"bucket":536870912,"rowId":0,"currentTransaction":16,"row":{"_col0":1,"_col1":2}} + {"operation":0,"originalTransaction":16,"bucket":536870912,"rowId":1,"currentTransaction":16,"row":{"_col0":2,"_col1":4}} + ________________________________________________________________________________________________________________________ + warehouse/t/.hive-staging_hive_2017-09-13_08-59-28_141_6304543600372946004-1/-ext-10000/000001_0/delta_0000016_0000016_0000/bucket_00001 [length: 658] + {"operation":0,"originalTransaction":16,"bucket":536936448,"rowId":0,"currentTransaction":16,"row":{"_col0":5,"_col1":6}} + {"operation":0,"originalTransaction":16,"bucket":536936448,"rowId":1,"currentTransaction":16,"row":{"_col0":6,"_col1":8}} + {"operation":0,"originalTransaction":16,"bucket":536936448,"rowId":2,"currentTransaction":16,"row":{"_col0":9,"_col1":10}} + */ + rs = runStatementOnDriver("select a, b from T order by a, b"); Assert.assertEquals(stringifyValues(values), rs); + rs = runStatementOnDriver("select ROW__ID from T group by ROW__ID having count(*) > 1"); + if(rs.size() > 0) { + Assert.assertEquals("Duplicate ROW__IDs: " + rs.get(0), 0, rs.size()); + } } /** * see HIVE-16177 - * See also {@link TestTxnCommands2#testNonAcidToAcidConversion02()} todo need test with > 1 bucket file + * See also {@link TestTxnCommands2#testNonAcidToAcidConversion02()} */ @Test public void testToAcidConversion02() throws Exception { http://git-wip-us.apache.org/repos/asf/hive/blob/02d359db/ql/src/test/results/clientpositive/autoColumnStats_4.q.out ---------------------------------------------------------------------- diff --git a/ql/src/test/results/clientpositive/autoColumnStats_4.q.out b/ql/src/test/results/clientpositive/autoColumnStats_4.q.out index a0581f8..7a5f8b4 100644 --- a/ql/src/test/results/clientpositive/autoColumnStats_4.q.out +++ b/ql/src/test/results/clientpositive/autoColumnStats_4.q.out @@ -136,6 +136,7 @@ STAGE PLANS: output format: org.apache.hadoop.hive.ql.io.orc.OrcOutputFormat serde: org.apache.hadoop.hive.ql.io.orc.OrcSerde name: default.acid_dtt + Write Type: INSERT Stage: Stage-3 Stats-Aggr Operator
