You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by gu...@apache.org on 2013/11/14 00:03:03 UTC
svn commit: r1541755 [1/2] - in /hive/branches/tez: ./ cli/ hbase-handler/
hbase-handler/src/java/org/apache/hadoop/hive/hbase/
hbase-handler/src/test/org/apache/hadoop/hive/hbase/
hbase-handler/src/test/queries/negative/ hcatalog/ hcatalog/build-suppo...
Author: gunther
Date: Wed Nov 13 23:03:01 2013
New Revision: 1541755
URL: http://svn.apache.org/r1541755
Log:
Merge latest trunk into branch. (Gunther Hagleitner)
Added:
hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/DataInputInputStream.java
- copied unchanged from r1541725, hive/trunk/hbase-handler/src/java/org/apache/hadoop/hive/hbase/DataInputInputStream.java
hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/DataOutputOutputStream.java
- copied unchanged from r1541725, hive/trunk/hbase-handler/src/java/org/apache/hadoop/hive/hbase/DataOutputOutputStream.java
hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/PutWritable.java
- copied unchanged from r1541725, hive/trunk/hbase-handler/src/java/org/apache/hadoop/hive/hbase/PutWritable.java
hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/ResultWritable.java
- copied unchanged from r1541725, hive/trunk/hbase-handler/src/java/org/apache/hadoop/hive/hbase/ResultWritable.java
hive/branches/tez/hbase-handler/src/test/org/apache/hadoop/hive/hbase/TestPutResultWritable.java
- copied unchanged from r1541725, hive/trunk/hbase-handler/src/test/org/apache/hadoop/hive/hbase/TestPutResultWritable.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/gen-java/org/apache/hcatalog/hbase/snapshot/RevisionManagerEndpointProtos.java
- copied unchanged from r1541725, hive/trunk/hcatalog/storage-handlers/hbase/src/gen-java/org/apache/hcatalog/hbase/snapshot/RevisionManagerEndpointProtos.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RPCConverter.java
- copied unchanged from r1541725, hive/trunk/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RPCConverter.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/protobuf/
- copied from r1541725, hive/trunk/hcatalog/storage-handlers/hbase/src/protobuf/
hive/branches/tez/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestOrcSplitElimination.java
- copied unchanged from r1541725, hive/trunk/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestOrcSplitElimination.java
hive/branches/tez/ql/src/test/queries/clientpositive/join_cond_pushdown_3.q
- copied unchanged from r1541725, hive/trunk/ql/src/test/queries/clientpositive/join_cond_pushdown_3.q
hive/branches/tez/ql/src/test/queries/clientpositive/join_cond_pushdown_4.q
- copied unchanged from r1541725, hive/trunk/ql/src/test/queries/clientpositive/join_cond_pushdown_4.q
hive/branches/tez/ql/src/test/queries/clientpositive/orc_split_elimination.q
- copied unchanged from r1541725, hive/trunk/ql/src/test/queries/clientpositive/orc_split_elimination.q
hive/branches/tez/ql/src/test/results/clientpositive/join_cond_pushdown_3.q.out
- copied unchanged from r1541725, hive/trunk/ql/src/test/results/clientpositive/join_cond_pushdown_3.q.out
hive/branches/tez/ql/src/test/results/clientpositive/join_cond_pushdown_4.q.out
- copied unchanged from r1541725, hive/trunk/ql/src/test/results/clientpositive/join_cond_pushdown_4.q.out
hive/branches/tez/ql/src/test/results/clientpositive/orc_split_elimination.q.out
- copied unchanged from r1541725, hive/trunk/ql/src/test/results/clientpositive/orc_split_elimination.q.out
Removed:
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManagerProtocol.java
Modified:
hive/branches/tez/ (props changed)
hive/branches/tez/.gitignore
hive/branches/tez/cli/pom.xml
hive/branches/tez/hbase-handler/pom.xml
hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseSerDe.java
hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseStorageHandler.java
hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableInputFormat.java
hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableOutputFormat.java
hive/branches/tez/hbase-handler/src/test/org/apache/hadoop/hive/hbase/TestHBaseSerDe.java
hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop.q
hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop_hadoop20.q
hive/branches/tez/hcatalog/build-support/ant/checkstyle.xml
hive/branches/tez/hcatalog/build.properties
hive/branches/tez/hcatalog/storage-handlers/hbase/pom.xml
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBaseOutputFormat.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBulkOutputFormat.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseDirectOutputFormat.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseHCatStorageHandler.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseInputFormat.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HbaseSnapshotRecordReader.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/ImportSequenceFile.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManager.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManagerEndpoint.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManagerEndpointClient.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/TableSnapshot.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/Transaction.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/ZKBasedRevisionManager.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/test/org/apache/hcatalog/hbase/ManyMiniCluster.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/test/org/apache/hcatalog/hbase/TestHBaseBulkOutputFormat.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/test/org/apache/hcatalog/hbase/TestHBaseDirectOutputFormat.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/test/org/apache/hcatalog/hbase/TestHCatHBaseInputFormat.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/test/org/apache/hcatalog/hbase/snapshot/TestRevisionManager.java
hive/branches/tez/hcatalog/storage-handlers/hbase/src/test/org/apache/hcatalog/hbase/snapshot/TestRevisionManagerEndpoint.java
hive/branches/tez/itests/hcatalog-unit/pom.xml
hive/branches/tez/itests/qtest/pom.xml
hive/branches/tez/itests/util/pom.xml
hive/branches/tez/pom.xml
hive/branches/tez/ql/pom.xml
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/RecordReaderImpl.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/sarg/SearchArgument.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/ParseDriver.java
hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/parse/SemanticAnalyzer.java
hive/branches/tez/ql/src/test/queries/clientpositive/create_view_translate.q
hive/branches/tez/ql/src/test/results/clientpositive/create_view_translate.q.out
hive/branches/tez/ql/src/test/results/clientpositive/sample8.q.out
Propchange: hive/branches/tez/
------------------------------------------------------------------------------
Merged /hive/trunk:r1540822-1541725
Modified: hive/branches/tez/.gitignore
URL: http://svn.apache.org/viewvc/hive/branches/tez/.gitignore?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/.gitignore (original)
+++ hive/branches/tez/.gitignore Wed Nov 13 23:03:01 2013
@@ -18,3 +18,10 @@ datanucleus.log
.arc
TempStatsStore/
target/
+ql/TempStatsStore
+hcatalog/hcatalog-pig-adapter/target
+hcatalog/server-extensions/target
+hcatalog/core/target
+hcatalog/webhcat/java-client/target
+hcatalog/storage-handlers/hbase/target
+hcatalog/webhcat/svr/target
Modified: hive/branches/tez/cli/pom.xml
URL: http://svn.apache.org/viewvc/hive/branches/tez/cli/pom.xml?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/cli/pom.xml (original)
+++ hive/branches/tez/cli/pom.xml Wed Nov 13 23:03:01 2013
@@ -103,12 +103,6 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase</artifactId>
- <version>${hbase.version}</version>
- <scope>test</scope>
- </dependency>
- <dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-all</artifactId>
<version>${mockito-all.version}</version>
@@ -129,6 +123,12 @@
<version>${hadoop-20S.version}</version>
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</profile>
<profile>
@@ -146,6 +146,12 @@
<version>${hadoop-23.version}</version>
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</profile>
</profiles>
Modified: hive/branches/tez/hbase-handler/pom.xml
URL: http://svn.apache.org/viewvc/hive/branches/tez/hbase-handler/pom.xml?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hbase-handler/pom.xml (original)
+++ hive/branches/tez/hbase-handler/pom.xml Wed Nov 13 23:03:01 2013
@@ -74,17 +74,6 @@
<artifactId>commons-logging</artifactId>
<version>${commons-logging.version}</version>
</dependency>
- <dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase</artifactId>
- <version>${hbase.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.thrift</groupId>
- <artifactId>libthrift</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
<!-- test inter-project -->
<dependency>
<groupId>junit</groupId>
@@ -92,12 +81,6 @@
<version>${junit.version}</version>
<scope>test</scope>
</dependency>
- <dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase</artifactId>
- <version>${hbase.version}</version>
- <classifier>tests</classifier>
- </dependency>
</dependencies>
<profiles>
@@ -113,6 +96,58 @@
<version>${hadoop-20S.version}</version>
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop1-compat</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ <classifier>tests</classifier>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop1-compat</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</profile>
<profile>
@@ -130,6 +165,58 @@
<version>${hadoop-23.version}</version>
<optional>true</optional>
</dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop2-compat</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ <classifier>tests</classifier>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop2-compat</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</profile>
</profiles>
Modified: hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseSerDe.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseSerDe.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseSerDe.java (original)
+++ hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseSerDe.java Wed Nov 13 23:03:01 2013
@@ -28,7 +28,6 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.client.Put;
-import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hive.serde.serdeConstants;
import org.apache.hadoop.hive.serde2.AbstractSerDe;
@@ -539,11 +538,11 @@ public class HBaseSerDe extends Abstract
@Override
public Object deserialize(Writable result) throws SerDeException {
- if (!(result instanceof Result)) {
- throw new SerDeException(getClass().getName() + ": expects Result!");
+ if (!(result instanceof ResultWritable)) {
+ throw new SerDeException(getClass().getName() + ": expects ResultWritable!");
}
- cachedHBaseRow.init((Result) result, columnsMapping);
+ cachedHBaseRow.init(((ResultWritable) result).getResult(), columnsMapping);
return cachedHBaseRow;
}
@@ -555,7 +554,7 @@ public class HBaseSerDe extends Abstract
@Override
public Class<? extends Writable> getSerializedClass() {
- return Put.class;
+ return PutWritable.class;
}
@Override
@@ -605,7 +604,7 @@ public class HBaseSerDe extends Abstract
throw new SerDeException(e);
}
- return put;
+ return new PutWritable(put);
}
private byte [] serializeField(
Modified: hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseStorageHandler.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseStorageHandler.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseStorageHandler.java (original)
+++ hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HBaseStorageHandler.java Wed Nov 13 23:03:01 2013
@@ -32,7 +32,6 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.mapred.TableOutputFormat;
@@ -209,11 +208,7 @@ public class HBaseStorageHandler extends
// ensure the table is online
new HTable(hbaseConf, tableDesc.getName());
- } catch (MasterNotRunningException mnre) {
- throw new MetaException(StringUtils.stringifyException(mnre));
- } catch (IOException ie) {
- throw new MetaException(StringUtils.stringifyException(ie));
- } catch (SerDeException se) {
+ } catch (Exception se) {
throw new MetaException(StringUtils.stringifyException(se));
}
}
Modified: hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableInputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableInputFormat.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableInputFormat.java (original)
+++ hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableInputFormat.java Wed Nov 13 23:03:01 2013
@@ -35,7 +35,6 @@ import org.apache.hadoop.hbase.mapred.Ta
import org.apache.hadoop.hbase.mapreduce.TableInputFormatBase;
import org.apache.hadoop.hbase.mapreduce.TableSplit;
import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.hbase.util.Writables;
import org.apache.hadoop.hive.hbase.HBaseSerDe.ColumnMapping;
import org.apache.hadoop.hive.ql.exec.ExprNodeConstantEvaluator;
import org.apache.hadoop.hive.ql.exec.Utilities;
@@ -77,12 +76,12 @@ import org.apache.hadoop.mapreduce.lib.i
* such as column pruning and filter pushdown.
*/
public class HiveHBaseTableInputFormat extends TableInputFormatBase
- implements InputFormat<ImmutableBytesWritable, Result> {
+ implements InputFormat<ImmutableBytesWritable, ResultWritable> {
static final Log LOG = LogFactory.getLog(HiveHBaseTableInputFormat.class);
@Override
- public RecordReader<ImmutableBytesWritable, Result> getRecordReader(
+ public RecordReader<ImmutableBytesWritable, ResultWritable> getRecordReader(
InputSplit split,
JobConf jobConf,
final Reporter reporter) throws IOException {
@@ -179,7 +178,7 @@ public class HiveHBaseTableInputFormat e
final org.apache.hadoop.mapreduce.RecordReader<ImmutableBytesWritable, Result>
recordReader = createRecordReader(tableSplit, tac);
- return new RecordReader<ImmutableBytesWritable, Result>() {
+ return new RecordReader<ImmutableBytesWritable, ResultWritable>() {
@Override
public void close() throws IOException {
@@ -192,8 +191,8 @@ public class HiveHBaseTableInputFormat e
}
@Override
- public Result createValue() {
- return new Result();
+ public ResultWritable createValue() {
+ return new ResultWritable(new Result());
}
@Override
@@ -215,7 +214,7 @@ public class HiveHBaseTableInputFormat e
}
@Override
- public boolean next(ImmutableBytesWritable rowKey, Result value) throws IOException {
+ public boolean next(ImmutableBytesWritable rowKey, ResultWritable value) throws IOException {
boolean next = false;
@@ -224,7 +223,7 @@ public class HiveHBaseTableInputFormat e
if (next) {
rowKey.set(recordReader.getCurrentValue().getRow());
- Writables.copyWritable(recordReader.getCurrentValue(), value);
+ value.setResult(recordReader.getCurrentValue());
}
} catch (InterruptedException e) {
throw new IOException(e);
Modified: hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableOutputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableOutputFormat.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableOutputFormat.java (original)
+++ hive/branches/tez/hbase-handler/src/java/org/apache/hadoop/hive/hbase/HiveHBaseTableOutputFormat.java Wed Nov 13 23:03:01 2013
@@ -24,6 +24,7 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.client.Durability;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
@@ -31,6 +32,7 @@ import org.apache.hadoop.hbase.mapred.Ta
import org.apache.hadoop.hbase.mapreduce.TableOutputCommitter;
import org.apache.hadoop.hbase.mapreduce.TableOutputFormat;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.hbase.PutWritable;
import org.apache.hadoop.hive.shims.ShimLoader;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputFormat;
@@ -48,7 +50,7 @@ import org.apache.hadoop.util.Progressab
*/
public class HiveHBaseTableOutputFormat extends
TableOutputFormat<ImmutableBytesWritable> implements
- OutputFormat<ImmutableBytesWritable, Put> {
+ OutputFormat<ImmutableBytesWritable, Object> {
static final Log LOG = LogFactory.getLog(HiveHBaseTableOutputFormat.class);
public static final String HBASE_WAL_ENABLED = "hive.hbase.wal.enabled";
@@ -86,7 +88,7 @@ public class HiveHBaseTableOutputFormat
@Override
public
- org.apache.hadoop.mapred.RecordWriter<ImmutableBytesWritable, Put>
+ org.apache.hadoop.mapred.RecordWriter<ImmutableBytesWritable, Object>
getRecordWriter(
FileSystem fileSystem,
JobConf jobConf,
@@ -99,21 +101,23 @@ public class HiveHBaseTableOutputFormat
jobConf, HiveConf.ConfVars.HIVE_HBASE_WAL_ENABLED);
final HTable table = new HTable(HBaseConfiguration.create(jobConf), hbaseTableName);
table.setAutoFlush(false);
- return new MyRecordWriter(table);
+ return new MyRecordWriter(table,walEnabled);
}
@Override
- public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException,
+ public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException,
InterruptedException {
return new TableOutputCommitter();
-}
+ }
- static private class MyRecordWriter implements org.apache.hadoop.mapred.RecordWriter<ImmutableBytesWritable, Put> {
+ static private class MyRecordWriter implements org.apache.hadoop.mapred.RecordWriter<ImmutableBytesWritable, Object> {
private final HTable m_table;
+ private final boolean m_walEnabled;
- public MyRecordWriter(HTable table) {
+ public MyRecordWriter(HTable table, boolean walEnabled) {
m_table = table;
+ m_walEnabled = walEnabled;
}
public void close(Reporter reporter)
@@ -122,8 +126,21 @@ public class HiveHBaseTableOutputFormat
}
public void write(ImmutableBytesWritable key,
- Put value) throws IOException {
- m_table.put(new Put(value));
+ Object value) throws IOException {
+ Put put;
+ if (value instanceof Put){
+ put = (Put)value;
+ } else if (value instanceof PutWritable) {
+ put = new Put(((PutWritable)value).getPut());
+ } else {
+ throw new IllegalArgumentException("Illegal Argument " + (value == null ? "null" : value.getClass().getName()));
+ }
+ if(m_walEnabled) {
+ put.setDurability(Durability.SYNC_WAL);
+ } else {
+ put.setDurability(Durability.SKIP_WAL);
+ }
+ m_table.put(put);
}
}
}
Modified: hive/branches/tez/hbase-handler/src/test/org/apache/hadoop/hive/hbase/TestHBaseSerDe.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hbase-handler/src/test/org/apache/hadoop/hive/hbase/TestHBaseSerDe.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hbase-handler/src/test/org/apache/hadoop/hive/hbase/TestHBaseSerDe.java (original)
+++ hive/branches/tez/hbase-handler/src/test/org/apache/hadoop/hive/hbase/TestHBaseSerDe.java Wed Nov 13 23:03:01 2013
@@ -217,7 +217,7 @@ public class TestHBaseSerDe extends Test
assertEquals(9, fieldRefs.size());
// Deserialize
- Object row = serDe.deserialize(r);
+ Object row = serDe.deserialize(new ResultWritable(r));
for (int i = 0; i < fieldRefs.size(); i++) {
Object fieldData = oi.getStructFieldData(row, fieldRefs.get(i));
if (fieldData != null) {
@@ -226,9 +226,9 @@ public class TestHBaseSerDe extends Test
assertEquals("Field " + i, expectedFieldsData[i], fieldData);
}
// Serialize
- assertEquals(Put.class, serDe.getSerializedClass());
- Put serializedPut = (Put) serDe.serialize(row, oi);
- assertEquals("Serialized data", p.toString(), serializedPut.toString());
+ assertEquals(PutWritable.class, serDe.getSerializedClass());
+ PutWritable serializedPut = (PutWritable) serDe.serialize(row, oi);
+ assertEquals("Serialized data", p.toString(),String.valueOf(serializedPut.getPut()));
}
// No specifications default to UTF8 String storage for backwards compatibility
@@ -513,8 +513,8 @@ public class TestHBaseSerDe extends Test
// Deserialize
for (int i = 0; i < r.length; i++) {
- Object row = hbaseSerDe.deserialize(r[i]);
- Put serializedPut = (Put) hbaseSerDe.serialize(row, soi);
+ Object row = hbaseSerDe.deserialize(new ResultWritable(r[i]));
+ Put serializedPut = ((PutWritable) hbaseSerDe.serialize(row, soi)).getPut();
byte [] rowKey = serializedPut.getRow();
for (int k = 0; k < rowKey.length; k++) {
@@ -669,7 +669,7 @@ public class TestHBaseSerDe extends Test
assertEquals(9, fieldRefs.size());
// Deserialize
- Object row = hbaseSerDe.deserialize(r);
+ Object row = hbaseSerDe.deserialize(new ResultWritable(r));
for (int j = 0; j < fieldRefs.size(); j++) {
Object fieldData = soi.getStructFieldData(row, fieldRefs.get(j));
@@ -687,7 +687,7 @@ public class TestHBaseSerDe extends Test
}
// Serialize
- Put serializedPut = (Put) hbaseSerDe.serialize(row, soi);
+ Put serializedPut = ((PutWritable) hbaseSerDe.serialize(row, soi)).getPut();
assertEquals("Serialized data: ", p.toString(), serializedPut.toString());
}
@@ -775,7 +775,7 @@ public class TestHBaseSerDe extends Test
List<? extends StructField> fieldRefs = soi.getAllStructFieldRefs();
- Object row = serDe.deserialize(r);
+ Object row = serDe.deserialize(new ResultWritable(r));
int j = 0;
@@ -803,7 +803,7 @@ public class TestHBaseSerDe extends Test
SerDeUtils.getJSONString(row, soi);
// Now serialize
- Put put = (Put) serDe.serialize(row, soi);
+ Put put = ((PutWritable) serDe.serialize(row, soi)).getPut();
if (p != null) {
assertEquals("Serialized put:", p.toString(), put.toString());
Modified: hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop.q
URL: http://svn.apache.org/viewvc/hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop.q?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop.q (original)
+++ hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop.q Wed Nov 13 23:03:01 2013
@@ -16,11 +16,11 @@ STORED BY 'org.apache.hadoop.hive.hbase.
WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,cf:string")
TBLPROPERTIES ("hbase.table.name" = "hbase_table_0");
-dfs -ls ../build/ql/tmp/hbase/hbase_table_0;
+dfs -ls target/tmp/hbase/data/default/hbase_table_0;
DROP DATABASE IF EXISTS hbaseDB CASCADE;
-dfs -ls ../build/ql/tmp/hbase/hbase_table_0;
+dfs -ls target/tmp/hbase/data/default/hbase_table_0;
Modified: hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop_hadoop20.q
URL: http://svn.apache.org/viewvc/hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop_hadoop20.q?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop_hadoop20.q (original)
+++ hive/branches/tez/hbase-handler/src/test/queries/negative/cascade_dbdrop_hadoop20.q Wed Nov 13 23:03:01 2013
@@ -16,11 +16,11 @@ STORED BY 'org.apache.hadoop.hive.hbase.
WITH SERDEPROPERTIES ("hbase.columns.mapping" = ":key,cf:string")
TBLPROPERTIES ("hbase.table.name" = "hbase_table_0");
-dfs -ls target/tmp/hbase/hbase_table_0;
+dfs -ls target/tmp/hbase/data/default/hbase_table_0;
DROP DATABASE IF EXISTS hbaseDB CASCADE;
-dfs -ls target/tmp/hbase/hbase/hbase_table_0;
+dfs -ls target/tmp/hbase/data/hbase/default/hbase_table_0;
Modified: hive/branches/tez/hcatalog/build-support/ant/checkstyle.xml
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/build-support/ant/checkstyle.xml?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/build-support/ant/checkstyle.xml (original)
+++ hive/branches/tez/hcatalog/build-support/ant/checkstyle.xml Wed Nov 13 23:03:01 2013
@@ -62,6 +62,8 @@
<exclude name="RELEASE_NOTES.txt"/>
<exclude name="*.patch"/>
<exclude name="tags"/>
+ <exclude name="**/*.jar"/>
+ <exclude name="**/*.proto"/> <!-- protocol buffer def files -->
<exclude name="**/*.swp"/> <!-- vim swap files -->
<exclude name="**/*.orig"/> <!-- files created by patch -->
<exclude name="**/*.rej"/> <!-- files created by patch -->
Modified: hive/branches/tez/hcatalog/build.properties
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/build.properties?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/build.properties (original)
+++ hive/branches/tez/hcatalog/build.properties Wed Nov 13 23:03:01 2013
@@ -88,5 +88,4 @@ mvn.deploy.repo.id=apache.snapshots.http
mvn.deploy.repo.url=https://repository.apache.org/content/repositories/snapshots
maven-ant-tasks.version=2.1.3
mvn.local.repo=${user.home}/.m2/repository
-mvn.hadoop.profile=hadoop20
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/pom.xml
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/pom.xml?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/pom.xml (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/pom.xml Wed Nov 13 23:03:01 2013
@@ -56,17 +56,6 @@
<version>${guava.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase</artifactId>
- <version>${hbase.version}</version>
- <exclusions>
- <exclusion>
- <groupId>org.apache.thrift</groupId>
- <artifactId>libthrift</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
<groupId>org.apache.thrift</groupId>
<artifactId>libthrift</artifactId>
<version>${libthrift.version}</version>
@@ -90,25 +79,18 @@
<scope>test</scope>
</dependency>
<dependency>
- <groupId>org.apache.hbase</groupId>
- <artifactId>hbase</artifactId>
- <version>${hbase.version}</version>
- <classifier>tests</classifier>
- <scope>test</scope>
- <exclusions>
- <exclusion>
- <groupId>com.google.guava</groupId>
- <artifactId>guava</artifactId>
- </exclusion>
- </exclusions>
- </dependency>
- <dependency>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
<version>${zookeeper.version}</version>
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.hamcrest</groupId>
+ <artifactId>hamcrest-core</artifactId>
+ <version>1.3</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
<profiles>
@@ -123,6 +105,37 @@
<artifactId>hadoop-core</artifactId>
<version>${hadoop-20S.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop1-compat</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ <classifier>tests</classifier>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop1-compat</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ </dependency>
<!-- test -->
<dependency>
<groupId>org.apache.hadoop</groupId>
@@ -130,6 +143,27 @@
<version>${hadoop-20S.version}</version>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase.hadoop1.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</profile>
<profile>
@@ -137,13 +171,33 @@
<dependencies>
<dependency>
<groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-common</artifactId>
+ <artifactId>hadoop-mapreduce-client-core</artifactId>
<version>${hadoop-23.version}</version>
</dependency>
<dependency>
- <groupId>org.apache.hadoop</groupId>
- <artifactId>hadoop-mapreduce-client-core</artifactId>
- <version>${hadoop-23.version}</version>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-client</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop2-compat</artifactId>
+ <version>${hbase.hadoop2.version}</version>
</dependency>
<!-- test -->
<dependency>
@@ -160,6 +214,34 @@
<classifier>tests</classifier>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-common</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-server</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop-compat</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.hbase</groupId>
+ <artifactId>hbase-hadoop2-compat</artifactId>
+ <version>${hbase.hadoop2.version}</version>
+ <classifier>tests</classifier>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</profile>
</profiles>
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBaseOutputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBaseOutputFormat.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBaseOutputFormat.java (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBaseOutputFormat.java Wed Nov 13 23:03:01 2013
@@ -25,6 +25,7 @@ import java.util.Properties;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.client.Put;
+import org.apache.hadoop.hive.hbase.PutWritable;
import org.apache.hadoop.hive.ql.io.FSRecordWriter;
import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
import org.apache.hadoop.io.Writable;
@@ -37,9 +38,12 @@ import org.apache.hcatalog.common.HCatCo
import org.apache.hcatalog.common.HCatUtil;
import org.apache.hcatalog.mapreduce.OutputJobInfo;
-public class HBaseBaseOutputFormat implements OutputFormat<WritableComparable<?>, Put>,
- HiveOutputFormat<WritableComparable<?>, Put> {
-
+/**
+ * Children of this output format can be passed either Put's (typical) or
+ * PutWritable. PutWritables will come from Hive's HBase SerDe.
+ */
+public class HBaseBaseOutputFormat implements OutputFormat<WritableComparable<?>, Object>,
+ HiveOutputFormat<WritableComparable<?>, Object> {
@Override
public FSRecordWriter getHiveRecordWriter(
JobConf jc, Path finalOutPath,
@@ -51,22 +55,34 @@ public class HBaseBaseOutputFormat imple
@Override
public void checkOutputSpecs(FileSystem ignored, JobConf job) throws IOException {
- OutputFormat<WritableComparable<?>, Put> outputFormat = getOutputFormat(job);
+ OutputFormat<WritableComparable<?>, Object> outputFormat = getOutputFormat(job);
outputFormat.checkOutputSpecs(ignored, job);
}
@Override
- public RecordWriter<WritableComparable<?>, Put> getRecordWriter(FileSystem ignored,
+ public RecordWriter<WritableComparable<?>, Object> getRecordWriter(FileSystem ignored,
JobConf job, String name, Progressable progress) throws IOException {
- OutputFormat<WritableComparable<?>, Put> outputFormat = getOutputFormat(job);
+ HBaseHCatStorageHandler.setHBaseSerializers(job);
+ OutputFormat<WritableComparable<?>, Object> outputFormat = getOutputFormat(job);
return outputFormat.getRecordWriter(ignored, job, name, progress);
}
- private OutputFormat<WritableComparable<?>, Put> getOutputFormat(JobConf job)
+ protected static Put toPut(Object o) {
+ if(o != null) {
+ if(o instanceof Put) {
+ return (Put)o;
+ } else if(o instanceof PutWritable) {
+ return ((PutWritable)o).getPut();
+ }
+ }
+ throw new IllegalArgumentException("Illegal Argument " + (o == null ? "null" : o.getClass().getName()));
+ }
+
+ private OutputFormat<WritableComparable<?>, Object> getOutputFormat(JobConf job)
throws IOException {
String outputInfo = job.get(HCatConstants.HCAT_KEY_OUTPUT_INFO);
OutputJobInfo outputJobInfo = (OutputJobInfo) HCatUtil.deserialize(outputInfo);
- OutputFormat<WritableComparable<?>, Put> outputFormat = null;
+ OutputFormat<WritableComparable<?>, Object> outputFormat = null;
if (HBaseHCatStorageHandler.isBulkMode(outputJobInfo)) {
outputFormat = new HBaseBulkOutputFormat();
} else {
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBulkOutputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBulkOutputFormat.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBulkOutputFormat.java (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseBulkOutputFormat.java Wed Nov 13 23:03:01 2013
@@ -25,6 +25,7 @@ import java.util.List;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
@@ -53,10 +54,10 @@ class HBaseBulkOutputFormat extends HBas
private final static ImmutableBytesWritable EMPTY_LIST = new ImmutableBytesWritable(
new byte[0]);
- private SequenceFileOutputFormat<WritableComparable<?>, Put> baseOutputFormat;
+ private SequenceFileOutputFormat<WritableComparable<?>, Object> baseOutputFormat;
public HBaseBulkOutputFormat() {
- baseOutputFormat = new SequenceFileOutputFormat<WritableComparable<?>, Put>();
+ baseOutputFormat = new SequenceFileOutputFormat<WritableComparable<?>, Object>();
}
@Override
@@ -68,9 +69,10 @@ class HBaseBulkOutputFormat extends HBas
}
@Override
- public RecordWriter<WritableComparable<?>, Put> getRecordWriter(
+ public RecordWriter<WritableComparable<?>, Object> getRecordWriter(
FileSystem ignored, JobConf job, String name, Progressable progress)
throws IOException {
+ HBaseHCatStorageHandler.setHBaseSerializers(job);
job.setOutputKeyClass(ImmutableBytesWritable.class);
job.setOutputValueClass(Put.class);
long version = HBaseRevisionManagerUtil.getOutputRevision(job);
@@ -93,26 +95,28 @@ class HBaseBulkOutputFormat extends HBas
}
private static class HBaseBulkRecordWriter implements
- RecordWriter<WritableComparable<?>, Put> {
+ RecordWriter<WritableComparable<?>, Object> {
- private RecordWriter<WritableComparable<?>, Put> baseWriter;
+ private RecordWriter<WritableComparable<?>, Object> baseWriter;
private final Long outputVersion;
public HBaseBulkRecordWriter(
- RecordWriter<WritableComparable<?>, Put> baseWriter,
+ RecordWriter<WritableComparable<?>, Object> baseWriter,
Long outputVersion) {
this.baseWriter = baseWriter;
this.outputVersion = outputVersion;
}
@Override
- public void write(WritableComparable<?> key, Put value)
+ public void write(WritableComparable<?> key, Object value)
throws IOException {
- Put put = value;
+ Put original = toPut(value);
+ Put put = original;
if (outputVersion != null) {
- put = new Put(value.getRow(), outputVersion.longValue());
- for (List<KeyValue> row : value.getFamilyMap().values()) {
- for (KeyValue el : row) {
+ put = new Put(original.getRow(), outputVersion.longValue());
+ for (List<? extends Cell> row : original.getFamilyMap().values()) {
+ for (Cell cell : row) {
+ KeyValue el = (KeyValue)cell;
put.add(el.getFamily(), el.getQualifier(), el.getValue());
}
}
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseDirectOutputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseDirectOutputFormat.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseDirectOutputFormat.java (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseDirectOutputFormat.java Wed Nov 13 23:03:01 2013
@@ -23,6 +23,7 @@ import java.io.IOException;
import java.util.List;
import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.mapred.TableOutputFormat;
@@ -53,9 +54,10 @@ class HBaseDirectOutputFormat extends HB
}
@Override
- public RecordWriter<WritableComparable<?>, Put> getRecordWriter(FileSystem ignored,
+ public RecordWriter<WritableComparable<?>, Object> getRecordWriter(FileSystem ignored,
JobConf job, String name, Progressable progress)
throws IOException {
+ HBaseHCatStorageHandler.setHBaseSerializers(job);
long version = HBaseRevisionManagerUtil.getOutputRevision(job);
return new HBaseDirectRecordWriter(outputFormat.getRecordWriter(ignored, job, name,
progress), version);
@@ -69,26 +71,28 @@ class HBaseDirectOutputFormat extends HB
}
private static class HBaseDirectRecordWriter implements
- RecordWriter<WritableComparable<?>, Put> {
+ RecordWriter<WritableComparable<?>, Object> {
- private RecordWriter<WritableComparable<?>, Put> baseWriter;
+ private RecordWriter<WritableComparable<?>, Object> baseWriter;
private final Long outputVersion;
public HBaseDirectRecordWriter(
- RecordWriter<WritableComparable<?>, Put> baseWriter,
+ RecordWriter<WritableComparable<?>, Object> baseWriter,
Long outputVersion) {
this.baseWriter = baseWriter;
this.outputVersion = outputVersion;
}
@Override
- public void write(WritableComparable<?> key, Put value)
+ public void write(WritableComparable<?> key, Object value)
throws IOException {
- Put put = value;
+ Put original = toPut(value);
+ Put put = original;
if (outputVersion != null) {
- put = new Put(value.getRow(), outputVersion.longValue());
- for (List<KeyValue> row : value.getFamilyMap().values()) {
- for (KeyValue el : row) {
+ put = new Put(original.getRow(), outputVersion.longValue());
+ for (List<? extends Cell> row : original.getFamilyMap().values()) {
+ for (Cell cell : row) {
+ KeyValue el = (KeyValue)cell;
put.add(el.getFamily(), el.getQualifier(), el.getValue());
}
}
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseHCatStorageHandler.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseHCatStorageHandler.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseHCatStorageHandler.java (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseHCatStorageHandler.java Wed Nov 13 23:03:01 2013
@@ -33,11 +33,12 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
-import org.apache.hadoop.hbase.MasterNotRunningException;
-import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.client.HBaseAdmin;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.mapred.TableOutputFormat;
+import org.apache.hadoop.hbase.mapreduce.KeyValueSerialization;
+import org.apache.hadoop.hbase.mapreduce.MutationSerialization;
+import org.apache.hadoop.hbase.mapreduce.ResultSerialization;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
import org.apache.hadoop.hbase.mapreduce.TableMapReduceUtil;
import org.apache.hadoop.hbase.util.Bytes;
@@ -50,6 +51,7 @@ import org.apache.hadoop.hive.ql.metadat
import org.apache.hadoop.hive.ql.plan.TableDesc;
import org.apache.hadoop.hive.ql.security.authorization.HiveAuthorizationProvider;
import org.apache.hadoop.hive.serde2.SerDe;
+import org.apache.hadoop.io.serializer.WritableSerialization;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.OutputFormat;
@@ -87,6 +89,7 @@ public class HBaseHCatStorageHandler ext
public final static String DEFAULT_PREFIX = "default.";
private final static String PROPERTY_INT_OUTPUT_LOCATION = "hcat.hbase.mapreduce.intermediateOutputLocation";
+ private final static String IO_SERIALIZATIONS = "io.serializations";
private Configuration hbaseConf;
private Configuration jobConf;
@@ -135,7 +138,7 @@ public class HBaseHCatStorageHandler ext
//TODO: Remove when HCAT-308 is fixed
addOutputDependencyJars(jobConf);
jobProperties.put("tmpjars", jobConf.get("tmpjars"));
-
+ setHBaseSerializers(jobProperties);
} catch (IOException e) {
throw new IllegalStateException("Error while configuring job properties", e);
}
@@ -191,7 +194,7 @@ public class HBaseHCatStorageHandler ext
jobProperties.put(HCatConstants.HCAT_KEY_OUTPUT_INFO, HCatUtil.serialize(outputJobInfo));
addOutputDependencyJars(jobConf);
jobProperties.put("tmpjars", jobConf.get("tmpjars"));
-
+ setHBaseSerializers(jobProperties);
} catch (IOException e) {
throw new IllegalStateException("Error while configuring job properties", e);
}
@@ -339,12 +342,8 @@ public class HBaseHCatStorageHandler ext
RevisionManager rm = HBaseRevisionManagerUtil.getOpenedRevisionManager(hbaseConf);
rm.createTable(tableName, new ArrayList<String>(uniqueColumnFamilies));
- } catch (MasterNotRunningException mnre) {
- throw new MetaException(StringUtils.stringifyException(mnre));
- } catch (IOException ie) {
- throw new MetaException(StringUtils.stringifyException(ie));
- } catch (IllegalArgumentException iae) {
- throw new MetaException(StringUtils.stringifyException(iae));
+ } catch (Exception e) {
+ throw new MetaException(StringUtils.stringifyException(e));
}
}
@@ -406,10 +405,8 @@ public class HBaseHCatStorageHandler ext
admin = new HBaseAdmin(this.getConf());
}
return admin;
- } catch (MasterNotRunningException mnre) {
- throw new MetaException(StringUtils.stringifyException(mnre));
- } catch (ZooKeeperConnectionException zkce) {
- throw new MetaException(StringUtils.stringifyException(zkce));
+ } catch (Exception e) {
+ throw new MetaException(StringUtils.stringifyException(e));
}
}
@@ -544,8 +541,10 @@ public class HBaseHCatStorageHandler ext
TableMapReduceUtil.addDependencyJars(conf,
//ZK
ZooKeeper.class,
- //HBase
+ //HBase Client
HTable.class,
+ //HBase MapReduce
+ MutationSerialization.class,
//Hive
HiveException.class,
//HCatalog jar
@@ -634,4 +633,20 @@ public class HBaseHCatStorageHandler ext
return builder.toString();
}
+ static void setHBaseSerializers(Configuration configuration) {
+ configuration.setStrings(IO_SERIALIZATIONS, configuration.get(IO_SERIALIZATIONS),
+ MutationSerialization.class.getName(), ResultSerialization.class.getName(),
+ KeyValueSerialization.class.getName());
+ }
+ static void setHBaseSerializers(Map<String, String> configuration) {
+ String value = "";
+ if(configuration.containsKey(IO_SERIALIZATIONS)) {
+ value = configuration.get(IO_SERIALIZATIONS) + ",";
+ } else {
+ value = new Configuration().get(IO_SERIALIZATIONS, WritableSerialization.class.getName()) + ",";
+ }
+ value += MutationSerialization.class.getName() + "," + ResultSerialization.class.getName()
+ + "," + KeyValueSerialization.class.getName();
+ configuration.put(IO_SERIALIZATIONS, value);
+ }
}
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseInputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseInputFormat.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseInputFormat.java (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HBaseInputFormat.java Wed Nov 13 23:03:01 2013
@@ -23,11 +23,11 @@ import java.io.IOException;
import java.util.List;
import org.apache.hadoop.hbase.client.HTable;
-import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapred.TableSplit;
import org.apache.hadoop.hbase.mapreduce.TableInputFormat;
+import org.apache.hadoop.hive.hbase.ResultWritable;
import org.apache.hive.hcatalog.mapreduce.HCatMapRedUtil;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
@@ -41,7 +41,7 @@ import org.apache.hcatalog.mapreduce.Inp
/**
* This class HBaseInputFormat is a wrapper class of TableInputFormat in HBase.
*/
-class HBaseInputFormat implements InputFormat<ImmutableBytesWritable, Result> {
+class HBaseInputFormat implements InputFormat<ImmutableBytesWritable, ResultWritable> {
private final TableInputFormat inputFormat;
@@ -66,7 +66,7 @@ class HBaseInputFormat implements InputF
* org.apache.hadoop.mapreduce.TaskAttemptContext)
*/
@Override
- public RecordReader<ImmutableBytesWritable, Result> getRecordReader(
+ public RecordReader<ImmutableBytesWritable, ResultWritable> getRecordReader(
InputSplit split, JobConf job, Reporter reporter)
throws IOException {
String jobString = job.get(HCatConstants.HCAT_KEY_JOB_INFO);
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HbaseSnapshotRecordReader.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HbaseSnapshotRecordReader.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HbaseSnapshotRecordReader.java (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/HbaseSnapshotRecordReader.java Wed Nov 13 23:03:01 2013
@@ -35,8 +35,7 @@ import org.apache.hadoop.hbase.client.Re
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
-import org.apache.hadoop.io.DataInputBuffer;
-import org.apache.hadoop.io.DataOutputBuffer;
+import org.apache.hadoop.hive.hbase.ResultWritable;
import org.apache.hadoop.mapred.RecordReader;
import org.apache.hcatalog.common.HCatUtil;
import org.apache.hcatalog.hbase.snapshot.FamilyRevision;
@@ -50,7 +49,7 @@ import org.slf4j.LoggerFactory;
* The Class HbaseSnapshotRecordReader implements logic for filtering records
* based on snapshot.
*/
-class HbaseSnapshotRecordReader implements RecordReader<ImmutableBytesWritable, Result> {
+class HbaseSnapshotRecordReader implements RecordReader<ImmutableBytesWritable, ResultWritable> {
static final Logger LOG = LoggerFactory.getLogger(HbaseSnapshotRecordReader.class);
private final InputJobInfo inpJobInfo;
@@ -62,8 +61,6 @@ class HbaseSnapshotRecordReader implemen
private TableSnapshot snapshot;
private Iterator<Result> resultItr;
private Set<Long> allAbortedTransactions;
- private DataOutputBuffer valueOut = new DataOutputBuffer();
- private DataInputBuffer valueIn = new DataInputBuffer();
HbaseSnapshotRecordReader(InputJobInfo inputJobInfo, Configuration conf) throws IOException {
this.inpJobInfo = inputJobInfo;
@@ -152,8 +149,8 @@ class HbaseSnapshotRecordReader implemen
}
@Override
- public Result createValue() {
- return new Result();
+ public ResultWritable createValue() {
+ return new ResultWritable();
}
@Override
@@ -170,7 +167,7 @@ class HbaseSnapshotRecordReader implemen
}
@Override
- public boolean next(ImmutableBytesWritable key, Result value) throws IOException {
+ public boolean next(ImmutableBytesWritable key, ResultWritable value) throws IOException {
if (this.resultItr == null) {
LOG.warn("The HBase result iterator is found null. It is possible"
+ " that the record reader has already been closed.");
@@ -182,10 +179,7 @@ class HbaseSnapshotRecordReader implemen
// Update key and value. Currently no way to avoid serialization/de-serialization
// as no setters are available.
key.set(hbaseRow.getRow());
- valueOut.reset();
- hbaseRow.write(valueOut);
- valueIn.reset(valueOut.getData(), valueOut.getLength());
- value.readFields(valueIn);
+ value.setResult(hbaseRow);
return true;
}
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/ImportSequenceFile.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/ImportSequenceFile.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/ImportSequenceFile.java (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/ImportSequenceFile.java Wed Nov 13 23:03:01 2013
@@ -19,23 +19,22 @@
package org.apache.hcatalog.hbase;
-import org.apache.hadoop.filecache.DistributedCache;
-import org.apache.hadoop.fs.FileSystem;
-import org.apache.hadoop.fs.permission.FsPermission;
-import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat;
-import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
-import org.apache.hadoop.hbase.mapreduce.PutSortReducer;
-import org.apache.hadoop.hbase.mapreduce.hadoopbackport.TotalOrderPartitioner;
-
import java.io.IOException;
import java.net.URI;
+import java.net.URISyntaxException;
import java.util.Map;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.filecache.DistributedCache;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
+import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat;
+import org.apache.hadoop.hbase.mapreduce.LoadIncrementalHFiles;
+import org.apache.hadoop.hbase.mapreduce.PutSortReducer;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.JobStatus;
@@ -47,11 +46,10 @@ import org.apache.hadoop.mapreduce.lib.i
import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hive.hcatalog.mapreduce.HCatMapRedUtil;
+import org.apache.hadoop.mapreduce.lib.partition.TotalOrderPartitioner;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import static org.apache.hadoop.hbase.mapreduce.hadoopbackport.TotalOrderPartitioner.DEFAULT_PATH;
-
/**
* MapReduce job which reads a series of Puts stored in a sequence file
@@ -71,12 +69,8 @@ class ImportSequenceFile {
@Override
public void map(ImmutableBytesWritable rowKey, Put value,
Context context)
- throws IOException {
- try {
- context.write(new ImmutableBytesWritable(value.getRow()), value);
- } catch (InterruptedException e) {
- e.printStackTrace();
- }
+ throws IOException, InterruptedException {
+ context.write(new ImmutableBytesWritable(value.getRow()), value);
}
}
@@ -158,6 +152,7 @@ class ImportSequenceFile {
private static Job createSubmittableJob(Configuration conf, String tableName, Path inputDir, Path scratchDir, boolean localMode)
throws IOException {
+ HBaseHCatStorageHandler.setHBaseSerializers(conf);
Job job = new Job(conf, NAME + "_" + tableName);
job.setJarByClass(SequenceFileImporter.class);
FileInputFormat.setInputPaths(job, inputDir);
@@ -170,18 +165,35 @@ class ImportSequenceFile {
job.setMapOutputKeyClass(ImmutableBytesWritable.class);
job.setMapOutputValueClass(Put.class);
HFileOutputFormat.configureIncrementalLoad(job, table);
+ URI partitionURI;
+ try {
+ partitionURI = new URI(TotalOrderPartitioner.getPartitionFile(job.getConfiguration())
+ + "#" + TotalOrderPartitioner.DEFAULT_PATH);
+ } catch (URISyntaxException e) {
+ throw new IOException(e);
+ }
+ DistributedCache.addCacheFile(partitionURI, job.getConfiguration());
+ DistributedCache.createSymlink(job.getConfiguration());
//override OutputFormatClass with our own so we can include cleanup in the committer
job.setOutputFormatClass(ImporterOutputFormat.class);
//local mode doesn't support symbolic links so we have to manually set the actual path
if (localMode) {
String partitionFile = null;
- for (URI uri : DistributedCache.getCacheFiles(job.getConfiguration())) {
- if (DEFAULT_PATH.equals(uri.getFragment())) {
+ URI[] uris = DistributedCache.getCacheFiles(job.getConfiguration());
+ if(uris == null) {
+ throw new IllegalStateException("No cache file existed in job configuration");
+ }
+ for (URI uri : uris) {
+ if (TotalOrderPartitioner.DEFAULT_PATH.equals(uri.getFragment())) {
partitionFile = uri.toString();
break;
}
}
+ if(partitionFile == null) {
+ throw new IllegalStateException("Unable to find " +
+ TotalOrderPartitioner.DEFAULT_PATH + " in cache");
+ }
partitionFile = partitionFile.substring(0, partitionFile.lastIndexOf("#"));
job.getConfiguration().set(TotalOrderPartitioner.PARTITIONER_PATH, partitionFile.toString());
}
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManager.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManager.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManager.java (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManager.java Wed Nov 13 23:03:01 2013
@@ -93,7 +93,7 @@ public interface RevisionManager {
*/
@Deprecated
public Transaction beginWriteTransaction(String table,
- List<String> families, long keepAlive) throws IOException;
+ List<String> families, Long keepAlive) throws IOException;
/**
* Commit the write transaction.
@@ -146,7 +146,7 @@ public interface RevisionManager {
* @throws IOException
*/
@Deprecated
- public TableSnapshot createSnapshot(String tableName, long revision)
+ public TableSnapshot createSnapshot(String tableName, Long revision)
throws IOException;
/**
Modified: hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManagerEndpoint.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManagerEndpoint.java?rev=1541755&r1=1541754&r2=1541755&view=diff
==============================================================================
--- hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManagerEndpoint.java (original)
+++ hive/branches/tez/hcatalog/storage-handlers/hbase/src/java/org/apache/hcatalog/hbase/snapshot/RevisionManagerEndpoint.java Wed Nov 13 23:03:01 2013
@@ -19,14 +19,36 @@
package org.apache.hcatalog.hbase.snapshot;
import java.io.IOException;
-import java.util.List;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hbase.Coprocessor;
import org.apache.hadoop.hbase.CoprocessorEnvironment;
-import org.apache.hadoop.hbase.coprocessor.BaseEndpointCoprocessor;
+import org.apache.hadoop.hbase.coprocessor.CoprocessorService;
+import org.apache.hadoop.hbase.protobuf.ResponseConverter;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.AbortWriteTransactionRequest;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.AbortWriteTransactionResponse;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.BeginWriteTransactionRequest;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.BeginWriteTransactionResponse;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.CommitWriteTransactionRequest;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.CommitWriteTransactionResponse;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.CreateSnapshotRequest;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.CreateSnapshotResponse;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.CreateTableRequest;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.CreateTableResponse;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.DropTableRequest;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.DropTableResponse;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.GetAbortedWriteTransactionsRequest;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.GetAbortedWriteTransactionsResponse;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.KeepAliveTransactionRequest;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.KeepAliveTransactionResponse;
+import org.apache.hcatalog.hbase.snapshot.RevisionManagerEndpointProtos.RevisionManagerEndpointService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import com.google.protobuf.RpcCallback;
+import com.google.protobuf.RpcController;
+import com.google.protobuf.Service;
+
/**
* Implementation of RevisionManager as HBase RPC endpoint. This class will control the lifecycle of
* and delegate to the actual RevisionManager implementation and make it available as a service
@@ -34,21 +56,21 @@ import org.slf4j.LoggerFactory;
* In the case of {@link ZKBasedRevisionManager} now only the region servers need write access to
* manage revision data.
*/
-public class RevisionManagerEndpoint extends BaseEndpointCoprocessor implements RevisionManagerProtocol {
+public class RevisionManagerEndpoint extends RevisionManagerEndpointService implements Coprocessor, CoprocessorService {
private static final Logger LOGGER =
LoggerFactory.getLogger(RevisionManagerEndpoint.class.getName());
+ private final RPCConverter rpcConverter = new RPCConverter();
private RevisionManager rmImpl = null;
@Override
public void start(CoprocessorEnvironment env) {
- super.start(env);
try {
Configuration conf = RevisionManagerConfiguration.create(env.getConfiguration());
String className = conf.get(RMConstants.REVISION_MGR_ENDPOINT_IMPL_CLASS,
ZKBasedRevisionManager.class.getName());
- LOGGER.debug("Using Revision Manager implementation: {}", className);
+ LOGGER.info("Using Revision Manager implementation: {}", className);
rmImpl = RevisionManagerFactory.getOpenedRevisionManager(className, conf);
} catch (IOException e) {
LOGGER.error("Failed to initialize revision manager", e);
@@ -57,85 +79,140 @@ public class RevisionManagerEndpoint ext
@Override
public void stop(CoprocessorEnvironment env) {
- if (rmImpl != null) {
- try {
+ try {
+ if (rmImpl != null) {
rmImpl.close();
- } catch (IOException e) {
- LOGGER.warn("Error closing revision manager.", e);
}
+ } catch (IOException e) {
+ LOGGER.warn("Error closing revision manager.", e);
}
- super.stop(env);
- }
-
- @Override
- public void initialize(Configuration conf) {
- // do nothing, HBase controls life cycle
- }
-
- @Override
- public void open() throws IOException {
- // do nothing, HBase controls life cycle
}
@Override
- public void close() throws IOException {
- // do nothing, HBase controls life cycle
+ public Service getService() {
+ return this;
}
@Override
- public void createTable(String table, List<String> columnFamilies) throws IOException {
- rmImpl.createTable(table, columnFamilies);
- }
-
- @Override
- public void dropTable(String table) throws IOException {
- rmImpl.dropTable(table);
- }
-
- @Override
- public Transaction beginWriteTransaction(String table, List<String> families)
- throws IOException {
- return rmImpl.beginWriteTransaction(table, families);
+ public void createTable(RpcController controller,
+ CreateTableRequest request, RpcCallback<CreateTableResponse> done) {
+ if(rmImpl != null) {
+ try {
+ rmImpl.createTable(request.getTableName(), request.getColumnFamiliesList());
+ done.run(CreateTableResponse.newBuilder().build());
+ } catch(IOException e) {
+ ResponseConverter.setControllerException(controller, e);
+ }
+ }
}
@Override
- public Transaction beginWriteTransaction(String table,
- List<String> families, long keepAlive) throws IOException {
- return rmImpl.beginWriteTransaction(table, families, keepAlive);
+ public void dropTable(RpcController controller, DropTableRequest request,
+ RpcCallback<DropTableResponse> done) {
+ if(rmImpl != null) {
+ try {
+ rmImpl.dropTable(request.getTableName());
+ done.run(DropTableResponse.newBuilder().build());
+ } catch(IOException e) {
+ ResponseConverter.setControllerException(controller, e);
+ }
+ }
}
@Override
- public void commitWriteTransaction(Transaction transaction)
- throws IOException {
- rmImpl.commitWriteTransaction(transaction);
+ public void beginWriteTransaction(RpcController controller,
+ BeginWriteTransactionRequest request,
+ RpcCallback<BeginWriteTransactionResponse> done) {
+ if(rmImpl != null) {
+ try {
+ Transaction transaction;
+ if(request.hasKeepAlive()) {
+ transaction = rmImpl.beginWriteTransaction(request.getTableName(), request.getColumnFamiliesList(),
+ request.getKeepAlive());
+ } else {
+ transaction = rmImpl.beginWriteTransaction(request.getTableName(), request.getColumnFamiliesList());
+ }
+ done.run(BeginWriteTransactionResponse.newBuilder()
+ .setTransaction(rpcConverter.convertTransaction(transaction)).build());
+ } catch(IOException e) {
+ ResponseConverter.setControllerException(controller, e);
+ }
+ }
}
@Override
- public void abortWriteTransaction(Transaction transaction)
- throws IOException {
- rmImpl.abortWriteTransaction(transaction);
+ public void commitWriteTransaction(RpcController controller,
+ CommitWriteTransactionRequest request,
+ RpcCallback<CommitWriteTransactionResponse> done) {
+ if(rmImpl != null) {
+ try {
+ rmImpl.commitWriteTransaction(rpcConverter.convertTransaction(request.getTransaction()));
+ done.run(CommitWriteTransactionResponse.newBuilder().build());
+ } catch(IOException e) {
+ ResponseConverter.setControllerException(controller, e);
+ }
+ }
}
@Override
- public TableSnapshot createSnapshot(String tableName) throws IOException {
- return rmImpl.createSnapshot(tableName);
+ public void abortWriteTransaction(RpcController controller,
+ AbortWriteTransactionRequest request,
+ RpcCallback<AbortWriteTransactionResponse> done) {
+ if(rmImpl != null) {
+ try {
+ rmImpl.abortWriteTransaction(rpcConverter.convertTransaction(request.getTransaction()));
+ done.run(AbortWriteTransactionResponse.newBuilder().build());
+ } catch(IOException e) {
+ ResponseConverter.setControllerException(controller, e);
+ }
+ }
}
@Override
- public TableSnapshot createSnapshot(String tableName, long revision)
- throws IOException {
- return rmImpl.createSnapshot(tableName, revision);
+ public void getAbortedWriteTransactions(RpcController controller,
+ GetAbortedWriteTransactionsRequest request,
+ RpcCallback<GetAbortedWriteTransactionsResponse> done) {
+ if(rmImpl != null) {
+ try {
+ rmImpl.getAbortedWriteTransactions(request.getTableName(), request.getColumnFamily());
+ done.run(GetAbortedWriteTransactionsResponse.newBuilder().build());
+ } catch(IOException e) {
+ ResponseConverter.setControllerException(controller, e);
+ }
+ }
}
@Override
- public void keepAlive(Transaction transaction) throws IOException {
- rmImpl.keepAlive(transaction);
+ public void createSnapshot(RpcController controller,
+ CreateSnapshotRequest request, RpcCallback<CreateSnapshotResponse> done) {
+ if(rmImpl != null) {
+ try {
+ TableSnapshot snapshot;
+ if(request.hasRevision()) {
+ snapshot = rmImpl.createSnapshot(request.getTableName(), request.getRevision());
+ } else {
+ snapshot = rmImpl.createSnapshot(request.getTableName());
+ }
+ done.run(CreateSnapshotResponse.newBuilder()
+ .setTableSnapshot(rpcConverter.convertTableSnapshot(snapshot)).build());
+ } catch(IOException e) {
+ ResponseConverter.setControllerException(controller, e);
+ }
+ }
}
@Override
- public List<FamilyRevision> getAbortedWriteTransactions(String table,
- String columnFamily) throws IOException {
- return rmImpl.getAbortedWriteTransactions(table, columnFamily);
+ public void keepAliveTransaction(RpcController controller,
+ KeepAliveTransactionRequest request,
+ RpcCallback<KeepAliveTransactionResponse> done) {
+ if(rmImpl != null) {
+ try {
+ rmImpl.keepAlive(rpcConverter.convertTransaction(request.getTransaction()));
+ done.run(KeepAliveTransactionResponse.newBuilder().build());
+ } catch(IOException e) {
+ ResponseConverter.setControllerException(controller, e);
+ }
+ }
}
}