Repository: hive Updated Branches: refs/heads/master f44bd6f63 -> a1d6b2de0
HIVE-14241 Acid clashes with ConfVars.HIVEFETCHTASKCONVERSION <> none (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/a1d6b2de Tree: http://git-wip-us.apache.org/repos/asf/hive/tree/a1d6b2de Diff: http://git-wip-us.apache.org/repos/asf/hive/diff/a1d6b2de Branch: refs/heads/master Commit: a1d6b2de0f9781782abb6b3533315f51c2440a22 Parents: f44bd6f Author: Eugene Koifman <ekoif...@hortonworks.com> Authored: Fri Jul 15 11:23:08 2016 -0700 Committer: Eugene Koifman <ekoif...@hortonworks.com> Committed: Fri Jul 15 11:23:08 2016 -0700 ---------------------------------------------------------------------- .../apache/hive/hcatalog/streaming/TestStreaming.java | 2 ++ ql/src/java/org/apache/hadoop/hive/ql/Driver.java | 8 ++++++++ .../org/apache/hadoop/hive/ql/exec/FetchOperator.java | 4 ++++ .../java/org/apache/hadoop/hive/ql/exec/FetchTask.java | 4 +++- .../test/org/apache/hadoop/hive/ql/TestTxnCommands2.java | 11 +++++++++++ 5 files changed, 28 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/hive/blob/a1d6b2de/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java ---------------------------------------------------------------------- diff --git a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java index dedfe3f..7da5365 100644 --- a/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java +++ b/hcatalog/streaming/src/test/org/apache/hive/hcatalog/streaming/TestStreaming.java @@ -790,6 +790,8 @@ public class TestStreaming { , txnBatch.getCurrentTransactionState()); connection.close(); + List<String> rs = queryTable(driver, "select * from " + dbName + "." + tblName); + Assert.assertEquals(1, rs.size()); } @Test http://git-wip-us.apache.org/repos/asf/hive/blob/a1d6b2de/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 5fe3525..f27f938 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/Driver.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/Driver.java @@ -904,6 +904,14 @@ public class Driver implements CommandProcessor { ValidTxnList txns = txnMgr.getValidTxns(); String txnStr = txns.toString(); conf.set(ValidTxnList.VALID_TXNS_KEY, txnStr); + if(plan.getFetchTask() != null) { + /** + * This is needed for {@link HiveConf.ConfVars.HIVEFETCHTASKCONVERSION} optimization which + * initializes JobConf in FetchOperator before recordValidTxns() but this has to be done + * after locks are acquired to avoid race conditions in ACID. + */ + plan.getFetchTask().setValidTxnList(txnStr); + } LOG.debug("Encoding valid txns info " + txnStr + " txnid:" + txnMgr.getCurrentTxnId()); } http://git-wip-us.apache.org/repos/asf/hive/blob/a1d6b2de/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java index d8ac6ae..601ad08 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchOperator.java @@ -34,6 +34,7 @@ import org.apache.hadoop.fs.FileStatus; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.hadoop.hive.common.FileUtils; +import org.apache.hadoop.hive.common.ValidTxnList; import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.exec.mr.ExecMapperContext; import org.apache.hadoop.hive.ql.io.AcidUtils; @@ -145,6 +146,9 @@ public class FetchOperator implements Serializable { initialize(); } + public void setValidTxnList(String txnStr) { + job.set(ValidTxnList.VALID_TXNS_KEY, txnStr); + } private void initialize() throws HiveException { if (isStatReader) { outputOI = work.getStatRowOI(); http://git-wip-us.apache.org/repos/asf/hive/blob/a1d6b2de/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchTask.java ---------------------------------------------------------------------- diff --git a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchTask.java b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchTask.java index dff1815..db6848a 100644 --- a/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchTask.java +++ b/ql/src/java/org/apache/hadoop/hive/ql/exec/FetchTask.java @@ -24,7 +24,6 @@ import java.util.List; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import org.apache.hadoop.hive.conf.HiveConf; import org.apache.hadoop.hive.ql.CommandNeedRetryException; import org.apache.hadoop.hive.ql.CompilationOpContext; import org.apache.hadoop.hive.ql.DriverContext; @@ -59,6 +58,9 @@ public class FetchTask extends Task<FetchWork> implements Serializable { super(); } + public void setValidTxnList(String txnStr) { + fetch.setValidTxnList(txnStr); + } @Override public void initialize(QueryState queryState, QueryPlan queryPlan, DriverContext ctx, CompilationOpContext opContext) { http://git-wip-us.apache.org/repos/asf/hive/blob/a1d6b2de/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java ---------------------------------------------------------------------- diff --git a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java index 245a3bc..3fec3f9 100644 --- a/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java +++ b/ql/src/test/org/apache/hadoop/hive/ql/TestTxnCommands2.java @@ -714,6 +714,17 @@ public class TestTxnCommands2 { } @Test + public void testSimpleRead() throws Exception { + hiveConf.setVar(HiveConf.ConfVars.HIVEFETCHTASKCONVERSION, "more"); + int[][] tableData = {{1,2},{3,3}}; + runStatementOnDriver("insert into " + Table.ACIDTBL + " " + makeValuesClause(tableData)); + int[][] tableData2 = {{5,3}}; + runStatementOnDriver("insert into " + Table.ACIDTBL + " " + makeValuesClause(tableData2)); + hiveConf.set(ValidTxnList.VALID_TXNS_KEY, "0:"); + List<String> rs = runStatementOnDriver("select * from " + Table.ACIDTBL); + Assert.assertEquals("Missing data", 3, rs.size()); + } + @Test public void testUpdateMixedCase() throws Exception { int[][] tableData = {{1,2},{3,3},{5,3}}; runStatementOnDriver("insert into " + Table.ACIDTBL + "(a,b) " + makeValuesClause(tableData));