You are viewing a plain text version of this content. The canonical link for it is here.
Posted to hcatalog-commits@incubator.apache.org by to...@apache.org on 2012/02/15 03:53:52 UTC
svn commit: r1244334 [2/4] - in /incubator/hcatalog/trunk: ./
src/java/org/apache/hadoop/ src/java/org/apache/hadoop/mapred/
src/java/org/apache/hcatalog/cli/SemanticAnalysis/
src/java/org/apache/hcatalog/common/ src/java/org/apache/hcatalog/mapred/
sr...
Added: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatEximOutputCommitter.java.broken
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatEximOutputCommitter.java.broken?rev=1244334&view=auto
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatEximOutputCommitter.java.broken (added)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatEximOutputCommitter.java.broken Wed Feb 15 03:53:50 2012
@@ -0,0 +1,165 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hcatalog.mapreduce;
+
+import java.io.IOException;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.metastore.Warehouse;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.parse.EximUtil;
+import org.apache.hadoop.hive.ql.parse.SemanticException;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.JobStatus;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hcatalog.common.ErrorType;
+import org.apache.hcatalog.common.HCatException;
+
+public class HCatEximOutputCommitter extends OutputCommitter {
+
+ private static final Log LOG = LogFactory.getLog(HCatEximOutputCommitter.class);
+
+ private final OutputCommitter baseCommitter;
+
+ public HCatEximOutputCommitter(JobContext context, OutputCommitter baseCommitter) {
+ this.baseCommitter = baseCommitter;
+ }
+
+ @Override
+ public void abortTask(TaskAttemptContext context) throws IOException {
+ baseCommitter.abortTask(context);
+ }
+
+ @Override
+ public void commitTask(TaskAttemptContext context) throws IOException {
+ baseCommitter.commitTask(context);
+ }
+
+ @Override
+ public boolean needsTaskCommit(TaskAttemptContext context) throws IOException {
+ return baseCommitter.needsTaskCommit(context);
+ }
+
+ @Override
+ public void setupJob(JobContext context) throws IOException {
+ if( baseCommitter != null ) {
+ baseCommitter.setupJob(context);
+ }
+ }
+
+ @Override
+ public void setupTask(TaskAttemptContext context) throws IOException {
+ baseCommitter.setupTask(context);
+ }
+
+ @Override
+ public void abortJob(JobContext jobContext, JobStatus.State state) throws IOException {
+ if(baseCommitter != null) {
+ baseCommitter.abortJob(jobContext, state);
+ }
+ OutputJobInfo jobInfo = HCatOutputFormat.getJobInfo(jobContext);
+
+ Path src = new Path(jobInfo.getLocation());
+ FileSystem fs = src.getFileSystem(jobContext.getConfiguration());
+ fs.delete(src, true);
+ }
+
+ @Override
+ public void commitJob(JobContext jobContext) throws IOException {
+ if(baseCommitter != null) {
+ baseCommitter.commitJob(jobContext);
+ }
+ }
+
+ @Override
+ public void cleanupJob(JobContext jobContext) throws IOException {
+ LOG.info("HCatEximOutputCommitter.cleanup invoked; m.o.d : " +
+ jobContext.getConfiguration().get("mapred.output.dir"));
+ if (baseCommitter != null) {
+ LOG.info("baseCommitter.class = " + baseCommitter.getClass().getName());
+ baseCommitter.cleanupJob(jobContext);
+ }
+
+ OutputJobInfo jobInfo = HCatBaseOutputFormat.getJobInfo(jobContext);
+ Configuration conf = jobContext.getConfiguration();
+ FileSystem fs;
+ try {
+ fs = FileSystem.get(new URI(jobInfo.getTableInfo().getTable().getSd().getLocation()), conf);
+ } catch (URISyntaxException e) {
+ throw new IOException(e);
+ }
+ doCleanup(jobInfo, fs);
+ }
+
+ private static void doCleanup(OutputJobInfo jobInfo, FileSystem fs) throws IOException,
+ HCatException {
+ try {
+ Table ttable = jobInfo.getTableInfo().getTable();
+ org.apache.hadoop.hive.ql.metadata.Table table = new org.apache.hadoop.hive.ql.metadata.Table(
+ ttable);
+ StorageDescriptor tblSD = ttable.getSd();
+ Path tblPath = new Path(tblSD.getLocation());
+ Path path = new Path(tblPath, "_metadata");
+ List<Partition> tpartitions = null;
+ try {
+ Map.Entry<org.apache.hadoop.hive.metastore.api.Table, List<Partition>> rv = EximUtil
+ .readMetaData(fs, path);
+ tpartitions = rv.getValue();
+ } catch (IOException e) {
+ }
+ List<org.apache.hadoop.hive.ql.metadata.Partition> partitions =
+ new ArrayList<org.apache.hadoop.hive.ql.metadata.Partition>();
+ if (tpartitions != null) {
+ for (Partition tpartition : tpartitions) {
+ partitions.add(new org.apache.hadoop.hive.ql.metadata.Partition(table, tpartition));
+ }
+ }
+ if (!table.getPartitionKeys().isEmpty()) {
+ Map<String, String> partitionValues = jobInfo.getPartitionValues();
+ org.apache.hadoop.hive.ql.metadata.Partition partition =
+ new org.apache.hadoop.hive.ql.metadata.Partition(table,
+ partitionValues,
+ new Path(tblPath, Warehouse.makePartPath(partitionValues)));
+ partition.getTPartition().setParameters(table.getParameters());
+ partitions.add(partition);
+ }
+ EximUtil.createExportDump(fs, path, (table), partitions);
+ } catch (SemanticException e) {
+ throw new HCatException(ErrorType.ERROR_PUBLISHING_PARTITION, e);
+ } catch (HiveException e) {
+ throw new HCatException(ErrorType.ERROR_PUBLISHING_PARTITION, e);
+ } catch (MetaException e) {
+ throw new HCatException(ErrorType.ERROR_PUBLISHING_PARTITION, e);
+ }
+ }
+}
Added: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatEximOutputFormat.java.broken
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatEximOutputFormat.java.broken?rev=1244334&view=auto
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatEximOutputFormat.java.broken (added)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatEximOutputFormat.java.broken Wed Feb 15 03:53:50 2012
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hcatalog.mapreduce;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+import java.util.TreeMap;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.Warehouse;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.SerDeInfo;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.io.RCFileInputFormat;
+import org.apache.hadoop.hive.ql.io.RCFileOutputFormat;
+import org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.JobContext;
+import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.RecordWriter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hcatalog.common.ErrorType;
+import org.apache.hcatalog.common.HCatConstants;
+import org.apache.hcatalog.common.HCatException;
+import org.apache.hcatalog.common.HCatUtil;
+import org.apache.hcatalog.data.HCatRecord;
+import org.apache.hcatalog.data.schema.HCatFieldSchema;
+import org.apache.hcatalog.data.schema.HCatSchema;
+import org.apache.hcatalog.data.schema.HCatSchemaUtils;
+import org.apache.hcatalog.rcfile.RCFileInputDriver;
+import org.apache.hcatalog.rcfile.RCFileOutputDriver;
+
+/**
+ * The OutputFormat to use to write data to HCat without a hcat server. This can then
+ * be imported into a hcat instance, or used with a HCatEximInputFormat. As in
+ * HCatOutputFormat, the key value is ignored and
+ * and should be given as null. The value is the HCatRecord to write.
+ */
+public class HCatEximOutputFormat extends HCatBaseOutputFormat {
+
+ private static final Log LOG = LogFactory.getLog(HCatEximOutputFormat.class);
+
+ /**
+ * Get the record writer for the job. Uses the Table's default OutputStorageDriver
+ * to get the record writer.
+ *
+ * @param context
+ * the information about the current task.
+ * @return a RecordWriter to write the output for the job.
+ * @throws IOException
+ */
+ @Override
+ public RecordWriter<WritableComparable<?>, HCatRecord>
+ getRecordWriter(TaskAttemptContext context
+ ) throws IOException, InterruptedException {
+ return getOutputFormat(context).getRecordWriter(context);
+ }
+
+ /**
+ * Get the output committer for this output format. This is responsible
+ * for ensuring the output is committed correctly.
+ * @param context the task context
+ * @return an output committer
+ * @throws IOException
+ * @throws InterruptedException
+ */
+ @Override
+ public OutputCommitter getOutputCommitter(TaskAttemptContext context) throws IOException, InterruptedException {
+ return new HCatEximOutputCommitter(context,((OutputCommitterContainer)getOutputFormat(context).getOutputCommitter(context)).getBaseOutputCommitter());
+ }
+
+ /**
+ * Check for validity of the output-specification for the job.
+ * @param context information about the job
+ * @throws IOException when output should not be attempted
+ */
+ @Override
+ public void checkOutputSpecs(JobContext context
+ ) throws IOException, InterruptedException {
+ ((OutputFormatContainer)getOutputFormat(context)).getBaseOutputFormat().checkOutputSpecs(context);
+ }
+
+ public static void setOutput(Job job, String dbname, String tablename, String location,
+ HCatSchema partitionSchema, List<String> partitionValues, HCatSchema columnSchema) throws HCatException {
+ setOutput(job, dbname, tablename, location, partitionSchema, partitionValues, columnSchema,
+ RCFileInputDriver.class.getName(),
+ RCFileOutputDriver.class.getName(),
+ RCFileInputFormat.class.getName(),
+ RCFileOutputFormat.class.getName(),
+ ColumnarSerDe.class.getName());
+ }
+
+ @SuppressWarnings("unchecked")
+ public static void setOutput(Job job, String dbname, String tablename, String location,
+ HCatSchema partitionSchema,
+ List<String> partitionValues,
+ HCatSchema columnSchema,
+ String isdname, String osdname,
+ String ifname, String ofname,
+ String serializationLib) throws HCatException {
+ Map<String, String> partSpec = new TreeMap<String, String>();
+ List<HCatFieldSchema> partKeys = null;
+ if (partitionSchema != null) {
+ partKeys = partitionSchema.getFields();
+ if (partKeys.size() != partitionValues.size()) {
+ throw new IllegalArgumentException("Partition key size differs from partition value size");
+ }
+ for (int i = 0; i < partKeys.size(); ++i) {
+ HCatFieldSchema partKey = partKeys.get(i);
+ if (partKey.getType() != HCatFieldSchema.Type.STRING) {
+ throw new IllegalArgumentException("Partition key type string is only supported");
+ }
+ partSpec.put(partKey.getName(), partitionValues.get(i));
+ }
+ }
+ StorerInfo storerInfo = new StorerInfo(isdname, osdname, new Properties());
+ OutputJobInfo outputJobInfo = OutputJobInfo.create(dbname,tablename,partSpec,null,null);
+ org.apache.hadoop.hive.ql.metadata.Table tbl = new
+ org.apache.hadoop.hive.ql.metadata.Table(dbname, tablename);
+ Table table = tbl.getTTable();
+ table.getParameters().put(HCatConstants.HCAT_ISD_CLASS, isdname);
+ table.getParameters().put(HCatConstants.HCAT_OSD_CLASS, osdname);
+ try {
+ String partname = null;
+ if ((partKeys != null) && !partKeys.isEmpty()) {
+ List<FieldSchema> partSchema = HCatSchemaUtils.getFieldSchemas(partKeys);
+ table.setPartitionKeys(partSchema);
+ partname = Warehouse.makePartName(partSchema, partitionValues);
+ } else {
+ partname = "data";
+ }
+ StorageDescriptor sd = table.getSd();
+ sd.setLocation(location);
+ String dataLocation = location + "/" + partname;
+ outputJobInfo.setTableInfo(new HCatTableInfo(dbname,tablename,columnSchema,null,storerInfo,table));
+ outputJobInfo.setOutputSchema(columnSchema);
+ outputJobInfo.setLocation(dataLocation);
+ setPartDetails(outputJobInfo, columnSchema, partSpec);
+ sd.setCols(HCatUtil.getFieldSchemaList(outputJobInfo.getOutputSchema().getFields()));
+ sd.setInputFormat(ifname);
+ sd.setOutputFormat(ofname);
+ SerDeInfo serdeInfo = sd.getSerdeInfo();
+ serdeInfo.setSerializationLib(serializationLib);
+ Configuration conf = job.getConfiguration();
+ conf.set(HCatConstants.HCAT_KEY_OUTPUT_INFO, HCatUtil.serialize(outputJobInfo));
+ } catch (IOException e) {
+ throw new HCatException(ErrorType.ERROR_SET_OUTPUT, e);
+ } catch (MetaException e) {
+ throw new HCatException(ErrorType.ERROR_SET_OUTPUT, e);
+ }
+ }
+}
Modified: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatOutputFormat.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatOutputFormat.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatOutputFormat.java (original)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatOutputFormat.java Wed Feb 15 03:53:50 2012
@@ -73,7 +73,7 @@ public class HCatOutputFormat extends HC
try {
Configuration conf = job.getConfiguration();
- client = createHiveClient(outputJobInfo.getServerUri(), conf);
+ client = createHiveClient(null, conf);
Table table = client.getTable(outputJobInfo.getDatabaseName(), outputJobInfo.getTableName());
if (table.getPartitionKeysSize() == 0 ){
@@ -95,10 +95,8 @@ public class HCatOutputFormat extends HC
}
}
- if (
- (outputJobInfo.getPartitionValues() == null)
- || (outputJobInfo.getPartitionValues().size() < table.getPartitionKeysSize())
- ){
+ if ((outputJobInfo.getPartitionValues() == null)
+ || (outputJobInfo.getPartitionValues().size() < table.getPartitionKeysSize())){
// dynamic partition usecase - partition values were null, or not all were specified
// need to figure out which keys are not specified.
List<String> dynamicPartitioningKeys = new ArrayList<String>();
@@ -131,31 +129,24 @@ public class HCatOutputFormat extends HC
StorageDescriptor tblSD = table.getSd();
HCatSchema tableSchema = HCatUtil.extractSchemaFromStorageDescriptor(tblSD);
- StorerInfo storerInfo = InitializeInput.extractStorerInfo(tblSD,table.getParameters());
+ StorerInfo storerInfo = InternalUtil.extractStorerInfo(tblSD,table.getParameters());
List<String> partitionCols = new ArrayList<String>();
for(FieldSchema schema : table.getPartitionKeys()) {
partitionCols.add(schema.getName());
}
- Class<? extends HCatOutputStorageDriver> driverClass =
- (Class<? extends HCatOutputStorageDriver>) Class.forName(storerInfo.getOutputSDClass());
- HCatOutputStorageDriver driver = driverClass.newInstance();
-
- String tblLocation = tblSD.getLocation();
- String location = driver.getOutputLocation(job,
- tblLocation, partitionCols,
- outputJobInfo.getPartitionValues(),conf.get(HCatConstants.HCAT_DYNAMIC_PTN_JOBID));
+ HCatStorageHandler storageHandler = HCatUtil.getStorageHandler(job.getConfiguration(), storerInfo);
//Serialize the output info into the configuration
outputJobInfo.setTableInfo(HCatTableInfo.valueOf(table));
outputJobInfo.setOutputSchema(tableSchema);
- outputJobInfo.setLocation(location);
outputJobInfo.setHarRequested(harRequested);
outputJobInfo.setMaximumDynamicPartitions(maxDynamicPartitions);
- conf.set(HCatConstants.HCAT_KEY_OUTPUT_INFO, HCatUtil.serialize(outputJobInfo));
- Path tblPath = new Path(tblLocation);
+ HCatUtil.configureOutputStorageHandler(storageHandler,job,outputJobInfo);
+
+ Path tblPath = new Path(table.getSd().getLocation());
/* Set the umask in conf such that files/dirs get created with table-dir
* permissions. Following three assumptions are made:
@@ -231,6 +222,7 @@ public class HCatOutputFormat extends HC
return getOutputFormat(context).getOutputCommitter(context);
}
+ //TODO remove url component, everything should be encapsulated in HiveConf
static HiveMetaStoreClient createHiveClient(String url, Configuration conf) throws IOException, MetaException {
HiveConf hiveConf = getHiveConf(url, conf);
// HCatUtil.logHiveConf(LOG, hiveConf);
Modified: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatOutputStorageDriver.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatOutputStorageDriver.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatOutputStorageDriver.java (original)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatOutputStorageDriver.java Wed Feb 15 03:53:50 2012
@@ -174,8 +174,10 @@ public abstract class HCatOutputStorageD
* @param outputFormat format the returned container will contain
* @return
*/
+
+ //TODO broken this entire class will disappear anyway
OutputFormatContainer getOutputFormatContainer(OutputFormat outputFormat) {
- return new DefaultOutputFormatContainer(outputFormat);
+ return new DefaultOutputFormatContainer(null);
}
}
Added: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatStorageHandler.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatStorageHandler.java?rev=1244334&view=auto
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatStorageHandler.java (added)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatStorageHandler.java Wed Feb 15 03:53:50 2012
@@ -0,0 +1,117 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hcatalog.mapreduce;
+
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.ql.metadata.HiveException;
+import org.apache.hadoop.hive.ql.metadata.HiveStorageHandler;
+import org.apache.hadoop.hive.ql.plan.TableDesc;
+import org.apache.hadoop.hive.ql.security.authorization.HiveAuthorizationProvider;
+import org.apache.hadoop.mapred.OutputFormat;
+
+/**
+ * The abstract Class HCatStorageHandler would server as the base class for all
+ * the storage handlers required for non-native tables in HCatalog.
+ */
+public abstract class HCatStorageHandler implements HiveStorageHandler {
+
+ //TODO move this to HiveStorageHandler
+ /**
+ * This method is called to allow the StorageHandlers the chance
+ * to populate the JobContext.getConfiguration() with properties that
+ * maybe be needed by the handler's bundled artifacts (ie InputFormat, SerDe, etc).
+ * Key value pairs passed into jobProperties is guaranteed to be set in the job's
+ * configuration object. User's can retrieve "context" information from tableDesc.
+ * User's should avoid mutating tableDesc and only make changes in jobProperties.
+ * This method is expected to be idempotent such that a job called with the
+ * same tableDesc values should return the same key-value pairs in jobProperties.
+ * Any external state set by this method should remain the same if this method is
+ * called again. It is up to the user to determine how best guarantee this invariant.
+ *
+ * This method in particular is to create a configuration for input.
+ * @param tableDesc
+ * @param jobProperties
+ */
+ public abstract void configureInputJobProperties(TableDesc tableDesc, Map<String, String> jobProperties);
+
+ //TODO move this to HiveStorageHandler
+ /**
+ * This method is called to allow the StorageHandlers the chance
+ * to populate the JobContext.getConfiguration() with properties that
+ * maybe be needed by the handler's bundled artifacts (ie InputFormat, SerDe, etc).
+ * Key value pairs passed into jobProperties is guaranteed to be set in the job's
+ * configuration object. User's can retrieve "context" information from tableDesc.
+ * User's should avoid mutating tableDesc and only make changes in jobProperties.
+ * This method is expected to be idempotent such that a job called with the
+ * same tableDesc values should return the same key-value pairs in jobProperties.
+ * Any external state set by this method should remain the same if this method is
+ * called again. It is up to the user to determine how best guarantee this invariant.
+ *
+ * This method in particular is to create a configuration for output.
+ * @param tableDesc
+ * @param jobProperties
+ */
+ public abstract void configureOutputJobProperties(TableDesc tableDesc, Map<String, String> jobProperties);
+
+ /**
+ *
+ *
+ * @return authorization provider
+ * @throws HiveException
+ */
+ public abstract HiveAuthorizationProvider getAuthorizationProvider()
+ throws HiveException;
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.hadoop.hive.ql.metadata.HiveStorageHandler#
+ * configureTableJobProperties(org.apache.hadoop.hive.ql.plan.TableDesc,
+ * java.util.Map)
+ */
+ @Override
+ @Deprecated
+ public final void configureTableJobProperties(TableDesc tableDesc,
+ Map<String, String> jobProperties) {
+ }
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.hadoop.conf.Configurable#getConf()
+ */
+ @Override
+ public abstract Configuration getConf();
+
+ /*
+ * (non-Javadoc)
+ *
+ * @see org.apache.hadoop.conf.Configurable#setConf(org.apache.hadoop.conf.
+ * Configuration)
+ */
+ @Override
+ public abstract void setConf(Configuration conf);
+
+ OutputFormatContainer getOutputFormatContainer(OutputFormat outputFormat) {
+ return new DefaultOutputFormatContainer(outputFormat);
+ }
+
+}
Modified: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatTableInfo.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatTableInfo.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatTableInfo.java (original)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/HCatTableInfo.java Wed Feb 15 03:53:50 2012
@@ -117,6 +117,10 @@ public class HCatTableInfo implements Se
return storerInfo;
}
+ public String getTableLocation() {
+ return table.getSd().getLocation();
+ }
+
/**
* minimize dependency on hive classes so this is package private
* this should eventually no longer be used
@@ -134,7 +138,7 @@ public class HCatTableInfo implements Se
*/
static HCatTableInfo valueOf(Table table) throws IOException {
HCatSchema dataColumns = HCatUtil.extractSchemaFromStorageDescriptor(table.getSd());
- StorerInfo storerInfo = InitializeInput.extractStorerInfo(table.getSd(), table.getParameters());
+ StorerInfo storerInfo = InternalUtil.extractStorerInfo(table.getSd(), table.getParameters());
HCatSchema partitionColumns = HCatUtil.getPartitionColumns(table);
return new HCatTableInfo(table.getDbName(),
table.getTableName(),
Added: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/InternalUtil.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/InternalUtil.java?rev=1244334&view=auto
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/InternalUtil.java (added)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/InternalUtil.java Wed Feb 15 03:53:50 2012
@@ -0,0 +1,147 @@
+package org.apache.hcatalog.mapreduce;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hive.metastore.MetaStoreUtils;
+import org.apache.hadoop.hive.metastore.api.FieldSchema;
+import org.apache.hadoop.hive.metastore.api.StorageDescriptor;
+import org.apache.hadoop.hive.serde2.SerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ListObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.MapObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.primitive.PrimitiveObjectInspectorFactory;
+import org.apache.hadoop.hive.serde2.typeinfo.ListTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.MapTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hcatalog.common.HCatConstants;
+import org.apache.hcatalog.common.HCatUtil;
+import org.apache.hcatalog.data.schema.HCatFieldSchema;
+import org.apache.hcatalog.data.schema.HCatSchema;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.Properties;
+
+class InternalUtil {
+
+ static StorerInfo extractStorerInfo(StorageDescriptor sd, Map<String, String> properties) throws IOException {
+ String inputSDClass, outputSDClass;
+
+ if (properties.containsKey(HCatConstants.HCAT_ISD_CLASS)){
+ inputSDClass = properties.get(HCatConstants.HCAT_ISD_CLASS);
+ }else{
+ // attempt to default to RCFile if the storage descriptor says it's an RCFile
+ if ((sd.getInputFormat() != null) && (sd.getInputFormat().equals(HCatConstants.HIVE_RCFILE_IF_CLASS))){
+ inputSDClass = HCatConstants.HCAT_RCFILE_ISD_CLASS;
+ }else{
+ throw new IOException("No input storage driver classname found for table, cannot write partition");
+ }
+ }
+
+ Properties hcatProperties = new Properties();
+ for (String key : properties.keySet()){
+ hcatProperties.put(key, properties.get(key));
+ }
+
+ return new StorerInfo(inputSDClass, null,
+ sd.getInputFormat(), sd.getOutputFormat(), sd.getSerdeInfo().getSerializationLib(),
+ properties.get(org.apache.hadoop.hive.metastore.api.Constants.META_TABLE_STORAGE),
+ hcatProperties);
+ }
+
+ static StructObjectInspector createStructObjectInspector(HCatSchema outputSchema) throws IOException {
+
+ if(outputSchema == null ) {
+ throw new IOException("Invalid output schema specified");
+ }
+
+ List<ObjectInspector> fieldInspectors = new ArrayList<ObjectInspector>();
+ List<String> fieldNames = new ArrayList<String>();
+
+ for(HCatFieldSchema hcatFieldSchema : outputSchema.getFields()) {
+ TypeInfo type = TypeInfoUtils.getTypeInfoFromTypeString(hcatFieldSchema.getTypeString());
+
+ fieldNames.add(hcatFieldSchema.getName());
+ fieldInspectors.add(getObjectInspector(type));
+ }
+
+ StructObjectInspector structInspector = ObjectInspectorFactory.
+ getStandardStructObjectInspector(fieldNames, fieldInspectors);
+ return structInspector;
+ }
+
+ private static ObjectInspector getObjectInspector(TypeInfo type) throws IOException {
+
+ switch(type.getCategory()) {
+
+ case PRIMITIVE :
+ PrimitiveTypeInfo primitiveType = (PrimitiveTypeInfo) type;
+ return PrimitiveObjectInspectorFactory.
+ getPrimitiveJavaObjectInspector(primitiveType.getPrimitiveCategory());
+
+ case MAP :
+ MapTypeInfo mapType = (MapTypeInfo) type;
+ MapObjectInspector mapInspector = ObjectInspectorFactory.getStandardMapObjectInspector(
+ getObjectInspector(mapType.getMapKeyTypeInfo()), getObjectInspector(mapType.getMapValueTypeInfo()));
+ return mapInspector;
+
+ case LIST :
+ ListTypeInfo listType = (ListTypeInfo) type;
+ ListObjectInspector listInspector = ObjectInspectorFactory.getStandardListObjectInspector(
+ getObjectInspector(listType.getListElementTypeInfo()));
+ return listInspector;
+
+ case STRUCT :
+ StructTypeInfo structType = (StructTypeInfo) type;
+ List<TypeInfo> fieldTypes = structType.getAllStructFieldTypeInfos();
+
+ List<ObjectInspector> fieldInspectors = new ArrayList<ObjectInspector>();
+ for(TypeInfo fieldType : fieldTypes) {
+ fieldInspectors.add(getObjectInspector(fieldType));
+ }
+
+ StructObjectInspector structInspector = ObjectInspectorFactory.getStandardStructObjectInspector(
+ structType.getAllStructFieldNames(), fieldInspectors);
+ return structInspector;
+
+ default :
+ throw new IOException("Unknown field schema type");
+ }
+ }
+
+ //TODO this has to find a better home, it's also hardcoded as default in hive would be nice
+ // if the default was decided by the serde
+ static void initializeOutputSerDe(SerDe serDe, Configuration conf, OutputJobInfo jobInfo) throws SerDeException {
+ Properties props = new Properties();
+ List<FieldSchema> fields = HCatUtil.getFieldSchemaList(jobInfo.getOutputSchema().getFields());
+ props.setProperty(org.apache.hadoop.hive.serde.Constants.LIST_COLUMNS,
+ MetaStoreUtils.getColumnNamesFromFieldSchema(fields));
+ props.setProperty(org.apache.hadoop.hive.serde.Constants.LIST_COLUMN_TYPES,
+ MetaStoreUtils.getColumnTypesFromFieldSchema(fields));
+
+ // setting these props to match LazySimpleSerde
+ props.setProperty(org.apache.hadoop.hive.serde.Constants.SERIALIZATION_NULL_FORMAT, "\\N");
+ props.setProperty(org.apache.hadoop.hive.serde.Constants.SERIALIZATION_FORMAT, "1");
+
+ //add props from params set in table schema
+ props.putAll(jobInfo.getTableInfo().getStorerInfo().getProperties());
+ //add props from job properties
+ props.putAll(jobInfo.getProperties());
+
+ serDe.initialize(conf,props);
+ }
+
+ static Reporter createReporter(TaskAttemptContext context) {
+ return new ProgressReporter(context);
+ }
+
+}
Modified: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputCommitterContainer.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputCommitterContainer.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputCommitterContainer.java (original)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputCommitterContainer.java Wed Feb 15 03:53:50 2012
@@ -26,13 +26,13 @@ import org.apache.hadoop.mapreduce.Outpu
* See {@link OutputFormatContainer} for more information about containers.
*/
abstract class OutputCommitterContainer extends OutputCommitter {
- private final OutputCommitter committer;
+ private final org.apache.hadoop.mapred.OutputCommitter committer;
/**
* @param context current JobContext
* @param committer OutputCommitter that this instance will contain
*/
- public OutputCommitterContainer(JobContext context, OutputCommitter committer) {
+ public OutputCommitterContainer(JobContext context, org.apache.hadoop.mapred.OutputCommitter committer) {
this.committer = committer;
}
Modified: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputFormatContainer.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputFormatContainer.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputFormatContainer.java (original)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputFormatContainer.java Wed Feb 15 03:53:50 2012
@@ -33,19 +33,19 @@ import org.apache.hcatalog.data.HCatReco
* such as partitioning isn't supported.
*/
abstract class OutputFormatContainer extends OutputFormat<WritableComparable<?>, HCatRecord> {
- private OutputFormat<? super WritableComparable<?>, ? super Writable> of;
+ private org.apache.hadoop.mapred.OutputFormat<? super WritableComparable<?>, ? super Writable> of;
/**
* @param of OutputFormat this instance will contain
*/
- public OutputFormatContainer(OutputFormat<? super WritableComparable<?>,? super Writable> of) {
+ public OutputFormatContainer(org.apache.hadoop.mapred.OutputFormat<? super WritableComparable<?>,? super Writable> of) {
this.of = of;
}
/**
* @return underlying OutputFormat
*/
- public OutputFormat getBaseOutputFormat() {
+ public org.apache.hadoop.mapred.OutputFormat getBaseOutputFormat() {
return of;
}
Modified: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputJobInfo.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputJobInfo.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputJobInfo.java (original)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/OutputJobInfo.java Wed Feb 15 03:53:50 2012
@@ -52,15 +52,6 @@ public class OutputJobInfo implements Se
/** The partition values to publish to, if used for output*/
private Map<String, String> partitionValues;
- /** The Metadata server uri */
- private final String serverUri;
-
- /** If the hcat server is configured to work with hadoop security, this
- * variable will hold the principal name of the server - this will be used
- * in the authentication to the hcat server using kerberos
- */
- private final String serverKerberosPrincipal;
-
private List<Integer> posOfPartCols;
private List<Integer> posOfDynPartCols;
@@ -79,8 +70,6 @@ public class OutputJobInfo implements Se
* @param databaseName the db name
* @param tableName the table name
* @param partitionValues The partition values to publish to, can be null or empty Map to
- * @param serverUri the Metadata server uri
- * @param serverKerberosPrincipal If the hcat server is configured to
* work with hadoop security, the kerberos principal name of the server - else null
* The principal name should be of the form:
* <servicename>/_HOST@<realm> like "hcat/_HOST@myrealm.com"
@@ -90,25 +79,17 @@ public class OutputJobInfo implements Se
*/
public static OutputJobInfo create(String databaseName,
String tableName,
- Map<String, String> partitionValues,
- String serverUri,
- String serverKerberosPrincipal) {
+ Map<String, String> partitionValues) {
return new OutputJobInfo(databaseName,
tableName,
- partitionValues,
- serverUri,
- serverKerberosPrincipal);
+ partitionValues);
}
private OutputJobInfo(String databaseName,
String tableName,
- Map<String, String> partitionValues,
- String serverUri,
- String serverKerberosPrincipal) {
+ Map<String, String> partitionValues) {
this.databaseName = (databaseName == null) ? MetaStoreUtils.DEFAULT_DATABASE_NAME : databaseName;
this.tableName = tableName;
- this.serverUri = serverUri;
- this.serverKerberosPrincipal = serverKerberosPrincipal;
this.partitionValues = partitionValues;
this.properties = new Properties();
}
@@ -180,7 +161,7 @@ public class OutputJobInfo implements Se
/**
* @param location location to write to
*/
- void setLocation(String location) {
+ public void setLocation(String location) {
this.location = location;
}
/**
@@ -200,20 +181,6 @@ public class OutputJobInfo implements Se
}
/**
- * @return metastore thrift server URI
- */
- public String getServerUri() {
- return serverUri;
- }
-
- /**
- * @return the serverKerberosPrincipal
- */
- public String getServerKerberosPrincipal() {
- return serverKerberosPrincipal;
- }
-
- /**
* set the tablInfo instance
* this should be the same instance
* determined by this object's DatabaseName and TableName
Added: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/ProgressReporter.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/ProgressReporter.java?rev=1244334&view=auto
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/ProgressReporter.java (added)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/ProgressReporter.java Wed Feb 15 03:53:50 2012
@@ -0,0 +1,55 @@
+package org.apache.hcatalog.mapreduce;
+
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
+import org.apache.hadoop.mapreduce.TaskInputOutputContext;
+import org.apache.hadoop.util.Progressable;
+
+class ProgressReporter implements Reporter {
+
+ private Progressable progressable;
+
+ public ProgressReporter(TaskAttemptContext context) {
+ this(context instanceof TaskInputOutputContext?
+ (TaskInputOutputContext)context:
+ Reporter.NULL);
+ }
+
+ public ProgressReporter(Progressable progressable) {
+ this.progressable = progressable;
+ }
+
+ @Override
+ public void setStatus(String status) {
+ }
+
+ @Override
+ public Counters.Counter getCounter(Enum<?> name) {
+ return Reporter.NULL.getCounter(name);
+ }
+
+ @Override
+ public Counters.Counter getCounter(String group, String name) {
+ return Reporter.NULL.getCounter(group,name);
+ }
+
+ @Override
+ public void incrCounter(Enum<?> key, long amount) {
+ }
+
+ @Override
+ public void incrCounter(String group, String counter, long amount) {
+ }
+
+ @Override
+ public InputSplit getInputSplit() throws UnsupportedOperationException {
+ return Reporter.NULL.getInputSplit();
+ }
+
+ @Override
+ public void progress() {
+ progressable.progress();
+ }
+}
Modified: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/RecordWriterContainer.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/RecordWriterContainer.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/RecordWriterContainer.java (original)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/RecordWriterContainer.java Wed Feb 15 03:53:50 2012
@@ -31,21 +31,21 @@ import org.apache.hcatalog.data.HCatReco
*/
abstract class RecordWriterContainer extends RecordWriter<WritableComparable<?>, HCatRecord> {
- private final RecordWriter<? super WritableComparable<?>, ? super Writable> baseRecordWriter;
+ private final org.apache.hadoop.mapred.RecordWriter<? super WritableComparable<?>, ? super Writable> baseRecordWriter;
/**
* @param context current JobContext
* @param baseRecordWriter RecordWriter that this instance will contain
*/
public RecordWriterContainer(TaskAttemptContext context,
- RecordWriter<? super WritableComparable<?>, ? super Writable> baseRecordWriter) {
+ org.apache.hadoop.mapred.RecordWriter<? super WritableComparable<?>, ? super Writable> baseRecordWriter) {
this.baseRecordWriter = baseRecordWriter;
}
/**
* @return underlying RecordWriter
*/
- public RecordWriter getBaseRecordWriter() {
+ public org.apache.hadoop.mapred.RecordWriter getBaseRecordWriter() {
return baseRecordWriter;
}
Modified: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/StorerInfo.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/StorerInfo.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/StorerInfo.java (original)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/mapreduce/StorerInfo.java Wed Feb 15 03:53:50 2012
@@ -25,23 +25,34 @@ public class StorerInfo implements Seria
/** The serialization version */
private static final long serialVersionUID = 1L;
+ //TODO remove this
/** The name of the input storage driver class */
private String inputSDClass;
+ //TODO remove this
/** The name of the output storage driver class */
private String outputSDClass;
/** The properties for the storage driver */
private Properties properties;
+ private String ofClass;
+ private String ifClass;
+
+ private String serdeClass;
+
+ private String storageHandlerClass;
+
+
+ //TODO remove this
/**
* Initialize the storage driver
* @param inputSDClass
* @param outputSDClass
* @param properties
*/
- StorerInfo(String inputSDClass, String outputSDClass, Properties properties) {
+ public StorerInfo(String inputSDClass, String outputSDClass, Properties properties) {
super();
this.inputSDClass = inputSDClass;
this.outputSDClass = outputSDClass;
@@ -49,17 +60,27 @@ public class StorerInfo implements Seria
}
/**
- * @return the inputSDClass
+ * Initialize the storage driver
+ * @param inputSDClass
+ * @param outputSDClass
+ * @param properties
*/
- public String getInputSDClass() {
- return inputSDClass;
+ public StorerInfo(String inputSDClass, String outputSDClass, String ifClass, String ofClass, String serdeClass, String storageHandlerClass, Properties properties) {
+ super();
+ this.inputSDClass = inputSDClass;
+ this.outputSDClass = outputSDClass;
+ this.ifClass =ifClass;
+ this.ofClass = ofClass;
+ this.serdeClass = serdeClass;
+ this.storageHandlerClass = storageHandlerClass;
+ this.properties = properties;
}
/**
- * @param inputSDClass the inputSDClass to set
+ * @return the inputSDClass
*/
- public void setInputSDClass(String inputSDClass) {
- this.inputSDClass = inputSDClass;
+ public String getInputSDClass() {
+ return inputSDClass;
}
/**
@@ -69,11 +90,24 @@ public class StorerInfo implements Seria
return outputSDClass;
}
- /**
- * @param outputSDClass the outputSDClass to set
- */
- public void setOutputSDClass(String outputSDClass) {
- this.outputSDClass = outputSDClass;
+public String getIfClass() {
+ return ifClass;
+ }
+
+ public void setIfClass(String ifClass) {
+ this.ifClass = ifClass;
+ }
+
+ public String getOfClass() {
+ return ofClass;
+ }
+
+ public String getSerdeClass() {
+ return serdeClass;
+ }
+
+ public String getStorageHandlerClass() {
+ return storageHandlerClass;
}
/**
Added: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatEximLoader.java.broken
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatEximLoader.java.broken?rev=1244334&view=auto
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatEximLoader.java.broken (added)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatEximLoader.java.broken Wed Feb 15 03:53:50 2012
@@ -0,0 +1,127 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hcatalog.pig;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hcatalog.data.schema.HCatFieldSchema;
+import org.apache.hcatalog.data.schema.HCatSchema;
+import org.apache.hcatalog.mapreduce.HCatBaseInputFormat;
+import org.apache.hcatalog.mapreduce.HCatEximInputFormat;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.pig.Expression;
+import org.apache.pig.LoadFunc;
+import org.apache.pig.ResourceSchema;
+import org.apache.pig.impl.util.UDFContext;
+
+/**
+ * Pig {@link LoadFunc} to read data/metadata from hcatalog exported location
+ */
+
+public class HCatEximLoader extends HCatBaseLoader {
+
+ private static final Log LOG = LogFactory.getLog(HCatEximLoader.class);
+
+ private HCatSchema tableSchema;
+ private HCatSchema partitionSchema;
+ private HCatEximInputFormat inputFormat;
+
+ public HCatEximLoader() {
+ LOG.debug("HCatEximLoader ctored");
+ }
+
+ @Override
+ public ResourceSchema getSchema(String location, Job job) throws IOException {
+ LOG.debug("getSchema with location :" + location);
+ if (tableSchema == null) {
+ List<HCatSchema> rv = HCatEximInputFormat.setInput(job, location, null);
+ tableSchema = rv.get(0);
+ partitionSchema = rv.get(1);
+ }
+ LOG.debug("getSchema got schema :" + tableSchema.toString());
+ List<HCatFieldSchema> colsPlusPartKeys = new ArrayList<HCatFieldSchema>();
+ colsPlusPartKeys.addAll(tableSchema.getFields());
+ colsPlusPartKeys.addAll(partitionSchema.getFields());
+ outputSchema = new HCatSchema(colsPlusPartKeys);
+ return PigHCatUtil.getResourceSchema(outputSchema);
+ }
+
+ @Override
+ public String[] getPartitionKeys(String location, Job job) throws IOException {
+ LOG.warn("getPartitionKeys with location :" + location);
+ /*
+ if (tableSchema == null) {
+ List<HCatSchema> rv = HCatEximInputFormat.setInput(job, location, null);
+ tableSchema = rv.get(0);
+ partitionSchema = rv.get(1);
+ }
+ return partitionSchema.getFieldNames().toArray(new String[0]);
+ */
+ return null;
+ }
+
+ @Override
+ public void setPartitionFilter(Expression partitionFilter) throws IOException {
+ LOG.debug("setPartitionFilter with filter :" + partitionFilter.toString());
+ }
+
+ @Override
+ public void setLocation(String location, Job job) throws IOException {
+ LOG.debug("setLocation with location :" + location);
+ List<HCatSchema> rv = HCatEximInputFormat.setInput(job, location, null);
+ tableSchema = rv.get(0);
+ partitionSchema = rv.get(1);
+ List<HCatFieldSchema> colsPlusPartKeys = new ArrayList<HCatFieldSchema>();
+ colsPlusPartKeys.addAll(tableSchema.getFields());
+ colsPlusPartKeys.addAll(partitionSchema.getFields());
+ outputSchema = new HCatSchema(colsPlusPartKeys);
+ UDFContext udfContext = UDFContext.getUDFContext();
+ Properties props = udfContext.getUDFProperties(this.getClass(),
+ new String[] {signature});
+ RequiredFieldList requiredFieldsInfo =
+ (RequiredFieldList) props.get(PRUNE_PROJECTION_INFO);
+ if (requiredFieldsInfo != null) {
+ ArrayList<HCatFieldSchema> fcols = new ArrayList<HCatFieldSchema>();
+ for (RequiredField rf : requiredFieldsInfo.getFields()) {
+ fcols.add(tableSchema.getFields().get(rf.getIndex()));
+ }
+ outputSchema = new HCatSchema(fcols);
+ try {
+ HCatBaseInputFormat.setOutputSchema(job, outputSchema);
+ } catch (Exception e) {
+ throw new IOException(e);
+ }
+ }
+ }
+
+
+ @Override
+ public InputFormat getInputFormat() throws IOException {
+ if (inputFormat == null) {
+ inputFormat = new HCatEximInputFormat();
+ }
+ return inputFormat;
+ }
+
+}
Added: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatEximStorer.java.broken
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatEximStorer.java.broken?rev=1244334&view=auto
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatEximStorer.java.broken (added)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatEximStorer.java.broken Wed Feb 15 03:53:50 2012
@@ -0,0 +1,151 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hcatalog.pig;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Properties;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hcatalog.common.HCatConstants;
+import org.apache.hcatalog.common.HCatException;
+import org.apache.hcatalog.common.HCatUtil;
+import org.apache.hcatalog.data.schema.HCatFieldSchema;
+import org.apache.hcatalog.data.schema.HCatSchema;
+import org.apache.hcatalog.mapreduce.HCatEximOutputCommitter;
+import org.apache.hcatalog.mapreduce.HCatEximOutputFormat;
+import org.apache.hadoop.hive.metastore.MetaStoreUtils;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.pig.ResourceSchema;
+import org.apache.pig.impl.logicalLayer.FrontendException;
+import org.apache.pig.impl.logicalLayer.schema.Schema;
+import org.apache.pig.impl.util.ObjectSerializer;
+import org.apache.pig.impl.util.UDFContext;
+
+/**
+ * HCatEximStorer.
+ *
+ */
+
+public class HCatEximStorer extends HCatBaseStorer {
+
+ private static final Log LOG = LogFactory.getLog(HCatEximStorer.class);
+
+ private final String outputLocation;
+
+ public HCatEximStorer(String outputLocation) throws Exception {
+ this(outputLocation, null, null);
+ }
+
+ public HCatEximStorer(String outputLocation, String partitionSpec) throws Exception {
+ this(outputLocation, partitionSpec, null);
+ }
+
+ public HCatEximStorer(String outputLocation, String partitionSpec, String schema)
+ throws Exception {
+ super(partitionSpec, schema);
+ this.outputLocation = outputLocation;
+ LOG.debug("HCatEximStorer called");
+ }
+
+ @Override
+ public OutputFormat getOutputFormat() throws IOException {
+ LOG.debug("getOutputFormat called");
+ return new HCatEximOutputFormat();
+ }
+
+ @Override
+ public void setStoreLocation(String location, Job job) throws IOException {
+ LOG.debug("setStoreLocation called with :" + location);
+ String[] userStr = location.split("\\.");
+ String dbname = MetaStoreUtils.DEFAULT_DATABASE_NAME;
+ String tablename = null;
+ if (userStr.length == 2) {
+ dbname = userStr[0];
+ tablename = userStr[1];
+ } else {
+ tablename = userStr[0];
+ }
+ Properties p = UDFContext.getUDFContext()
+ .getUDFProperties(this.getClass(), new String[] {sign});
+ Configuration config = job.getConfiguration();
+ if (!HCatUtil.checkJobContextIfRunningFromBackend(job)) {
+ Schema schema = (Schema) ObjectSerializer.deserialize(p.getProperty(PIG_SCHEMA));
+ if (schema != null) {
+ pigSchema = schema;
+ }
+ if (pigSchema == null) {
+ throw new FrontendException("Schema for data cannot be determined.",
+ PigHCatUtil.PIG_EXCEPTION_CODE);
+ }
+ HCatSchema hcatTblSchema = new HCatSchema(new ArrayList<HCatFieldSchema>());
+ try {
+ doSchemaValidations(pigSchema, hcatTblSchema);
+ } catch (HCatException he) {
+ throw new FrontendException(he.getMessage(), PigHCatUtil.PIG_EXCEPTION_CODE, he);
+ }
+
+ List<HCatFieldSchema> hcatFields = new ArrayList<HCatFieldSchema>();
+ List<String> partVals = new ArrayList<String>();
+ for (String key : partitionKeys) {
+ hcatFields.add(new HCatFieldSchema(key, HCatFieldSchema.Type.STRING, ""));
+ partVals.add(partitions.get(key));
+ }
+
+ HCatSchema outputSchema = convertPigSchemaToHCatSchema(pigSchema,
+ hcatTblSchema);
+ LOG.debug("Pig Schema '" + pigSchema.toString() + "' was converted to HCatSchema '"
+ + outputSchema);
+ HCatEximOutputFormat.setOutput(job,
+ dbname, tablename,
+ outputLocation,
+ new HCatSchema(hcatFields),
+ partVals,
+ outputSchema);
+ p.setProperty(COMPUTED_OUTPUT_SCHEMA, ObjectSerializer.serialize(outputSchema));
+ p.setProperty(HCatConstants.HCAT_KEY_OUTPUT_INFO,
+ config.get(HCatConstants.HCAT_KEY_OUTPUT_INFO));
+ if (config.get(HCatConstants.HCAT_KEY_HIVE_CONF) != null) {
+ p.setProperty(HCatConstants.HCAT_KEY_HIVE_CONF,
+ config.get(HCatConstants.HCAT_KEY_HIVE_CONF));
+ }
+ } else {
+ config.set(HCatConstants.HCAT_KEY_OUTPUT_INFO,
+ p.getProperty(HCatConstants.HCAT_KEY_OUTPUT_INFO));
+ if (p.getProperty(HCatConstants.HCAT_KEY_HIVE_CONF) != null) {
+ config.set(HCatConstants.HCAT_KEY_HIVE_CONF,
+ p.getProperty(HCatConstants.HCAT_KEY_HIVE_CONF));
+ }
+ }
+ }
+
+ @Override
+ public void storeSchema(ResourceSchema schema, String arg1, Job job) throws IOException {
+ if( job.getConfiguration().get("mapred.job.tracker", "").equalsIgnoreCase("local") ) {
+ //In local mode, mapreduce will not call OutputCommitter.cleanupJob.
+ //Calling it from here so that the partition publish happens.
+ //This call needs to be removed after MAPREDUCE-1447 is fixed.
+ new HCatEximOutputCommitter(job,null).cleanupJob(job);
+ }
+ }
+}
Modified: incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatStorer.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatStorer.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatStorer.java (original)
+++ incubator/hcatalog/trunk/src/java/org/apache/hcatalog/pig/HCatStorer.java Wed Feb 15 03:53:50 2012
@@ -80,15 +80,11 @@ public class HCatStorer extends HCatBase
if(userStr.length == 2) {
outputJobInfo = OutputJobInfo.create(userStr[0],
userStr[1],
- partitions,
- PigHCatUtil.getHCatServerUri(job),
- PigHCatUtil.getHCatServerPrincipal(job));
+ partitions);
} else if(userStr.length == 1) {
outputJobInfo = OutputJobInfo.create(null,
userStr[0],
- partitions,
- PigHCatUtil.getHCatServerUri(job),
- PigHCatUtil.getHCatServerPrincipal(job));
+ partitions);
} else {
throw new FrontendException("location "+location+" is invalid. It must be of the form [db.]table", PigHCatUtil.PIG_EXCEPTION_CODE);
}
Modified: incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/GroupByAge.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/GroupByAge.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/GroupByAge.java (original)
+++ incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/GroupByAge.java Wed Feb 15 03:53:50 2012
@@ -118,7 +118,7 @@ public class GroupByAge extends Configur
job.setOutputKeyClass(WritableComparable.class);
job.setOutputValueClass(DefaultHCatRecord.class);
HCatOutputFormat.setOutput(job, OutputJobInfo.create(dbName,
- outputTableName, null, serverUri, principalID));
+ outputTableName, null));
HCatSchema s = HCatOutputFormat.getTableSchema(job);
System.err.println("INFO: output schema explicitly set for writing:"
+ s);
Modified: incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/HBaseReadWrite.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/HBaseReadWrite.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/HBaseReadWrite.java (original)
+++ incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/HBaseReadWrite.java Wed Feb 15 03:53:50 2012
@@ -159,7 +159,7 @@ public class HBaseReadWrite extends Conf
job.setOutputKeyClass(WritableComparable.class);
job.setOutputValueClass(DefaultHCatRecord.class);
HCatOutputFormat.setOutput(job, OutputJobInfo.create(dbName,
- tableName, null, serverUri, principalID));
+ tableName, null));
boolean succ = job.waitForCompletion(true);
Modified: incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/ReadWrite.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/ReadWrite.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/ReadWrite.java (original)
+++ incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/ReadWrite.java Wed Feb 15 03:53:50 2012
@@ -95,7 +95,7 @@ public class ReadWrite extends Configure
job.setOutputKeyClass(Text.class);
job.setOutputValueClass(DefaultHCatRecord.class);
HCatOutputFormat.setOutput(job, OutputJobInfo.create(dbName,
- outputTableName, null, serverUri, principalID));
+ outputTableName, null));
HCatSchema s = HCatInputFormat.getTableSchema(job);
System.err.println("INFO: output schema explicitly set for writing:"
+ s);
Modified: incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreComplex.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreComplex.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreComplex.java (original)
+++ incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreComplex.java Wed Feb 15 03:53:50 2012
@@ -106,7 +106,7 @@ public class StoreComplex {
dbName, tableName, null, serverUri, principalID));
// initialize HCatOutputFormat
HCatOutputFormat.setOutput(job, OutputJobInfo.create(
- dbName, outputTableName, outputPartitionKvps, serverUri, principalID));
+ dbName, outputTableName, outputPartitionKvps));
HCatSchema s = HCatInputFormat.getTableSchema(job);
Modified: incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreDemo.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreDemo.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreDemo.java (original)
+++ incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreDemo.java Wed Feb 15 03:53:50 2012
@@ -117,7 +117,7 @@ public class StoreDemo {
dbName, tableName, null, serverUri, principalID));
// initialize HCatOutputFormat
HCatOutputFormat.setOutput(job, OutputJobInfo.create(
- dbName, outputTableName, outputPartitionKvps, serverUri, principalID));
+ dbName, outputTableName, outputPartitionKvps));
// test with and without specifying schema randomly
HCatSchema s = HCatInputFormat.getTableSchema(job);
System.err.println("INFO: output schema explicitly set for writing:" + s);
Modified: incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreNumbers.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreNumbers.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreNumbers.java (original)
+++ incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/StoreNumbers.java Wed Feb 15 03:53:50 2012
@@ -181,7 +181,7 @@ public class StoreNumbers {
dbName, tableName, null, serverUri, principalID));
// initialize HCatOutputFormat
HCatOutputFormat.setOutput(job, OutputJobInfo.create(
- dbName, outputTableName, outputPartitionKvps, serverUri, principalID));
+ dbName, outputTableName, outputPartitionKvps));
// test with and without specifying schema randomly
HCatSchema s = HCatInputFormat.getTableSchema(job);
if(writeToNonPartPigTable) {
Modified: incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteJson.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteJson.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteJson.java (original)
+++ incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteJson.java Wed Feb 15 03:53:50 2012
@@ -103,7 +103,7 @@ public class WriteJson extends Configure
job.setOutputValueClass(DefaultHCatRecord.class);
job.setNumReduceTasks(0);
HCatOutputFormat.setOutput(job, OutputJobInfo.create(dbName,
- outputTableName, null, serverUri, principalID));
+ outputTableName, null));
HCatSchema s = HCatInputFormat.getTableSchema(job);
System.err.println("INFO: output schema explicitly set for writing:"
+ s);
Modified: incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteRC.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteRC.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteRC.java (original)
+++ incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteRC.java Wed Feb 15 03:53:50 2012
@@ -105,7 +105,7 @@ public class WriteRC extends Configured
job.setOutputValueClass(DefaultHCatRecord.class);
job.setNumReduceTasks(0);
HCatOutputFormat.setOutput(job, OutputJobInfo.create(dbName,
- outputTableName, null, serverUri, principalID));
+ outputTableName, null));
HCatSchema s = HCatInputFormat.getTableSchema(job);
System.err.println("INFO: output schema explicitly set for writing:"
+ s);
Modified: incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteText.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteText.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteText.java (original)
+++ incubator/hcatalog/trunk/src/test/e2e/hcatalog/udfs/java/org/apache/hcatalog/utils/WriteText.java Wed Feb 15 03:53:50 2012
@@ -114,7 +114,7 @@ public class WriteText extends Configure
job.setOutputValueClass(DefaultHCatRecord.class);
job.setNumReduceTasks(0);
HCatOutputFormat.setOutput(job, OutputJobInfo.create(dbName,
- outputTableName, null, serverUri, principalID));
+ outputTableName, null));
HCatSchema s = HCatInputFormat.getTableSchema(job);
System.err.println("INFO: output schema explicitly set for writing:"
+ s);
Modified: incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/DummyStorageHandler.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/DummyStorageHandler.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/DummyStorageHandler.java (original)
+++ incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/DummyStorageHandler.java Wed Feb 15 03:53:50 2012
@@ -17,13 +17,17 @@
*/
package org.apache.hcatalog.cli;
+import java.io.IOException;
import java.util.List;
import java.util.Map;
+import java.util.Properties;
+
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.metastore.HiveMetaHook;
import org.apache.hadoop.hive.metastore.api.Database;
-import org.apache.hadoop.hive.metastore.api.MetaException;
-import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
import org.apache.hadoop.hive.ql.metadata.AuthorizationException;
import org.apache.hadoop.hive.ql.metadata.HiveException;
import org.apache.hadoop.hive.ql.metadata.Partition;
@@ -33,9 +37,18 @@ import org.apache.hadoop.hive.ql.securit
import org.apache.hadoop.hive.ql.security.authorization.Privilege;
import org.apache.hadoop.hive.serde2.SerDe;
import org.apache.hadoop.hive.serde2.columnar.ColumnarSerDe;
-import org.apache.hcatalog.mapreduce.HCatInputStorageDriver;
-import org.apache.hcatalog.mapreduce.HCatOutputStorageDriver;
-import org.apache.hcatalog.storagehandler.HCatStorageHandler;
+import org.apache.hadoop.io.Writable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.InputSplit;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapred.OutputFormat;
+import org.apache.hadoop.mapred.RecordReader;
+import org.apache.hadoop.mapred.RecordWriter;
+import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.util.Progressable;
+import org.apache.hcatalog.data.HCatRecord;
+import org.apache.hcatalog.mapreduce.HCatStorageHandler;
class DummyStorageHandler extends HCatStorageHandler {
@@ -49,13 +62,13 @@ class DummyStorageHandler extends HCatSt
}
@Override
- public void configureTableJobProperties(TableDesc arg0,
- Map<String, String> arg1) {
+ public Class<? extends InputFormat> getInputFormatClass() {
+ return DummyInputFormat.class;
}
@Override
- public HiveMetaHook getMetaHook() {
- return this;
+ public Class<? extends OutputFormat> getOutputFormatClass() {
+ return DummyOutputFormat.class;
}
@Override
@@ -64,39 +77,16 @@ class DummyStorageHandler extends HCatSt
}
@Override
- public void preCreateTable(Table table) throws MetaException {
- }
-
- @Override
- public void rollbackCreateTable(Table table) throws MetaException {
- }
-
- @Override
- public void commitCreateTable(Table table) throws MetaException {
- }
-
- @Override
- public void preDropTable(Table table) throws MetaException {
- }
-
- @Override
- public void rollbackDropTable(Table table) throws MetaException {
-
- }
-
- @Override
- public void commitDropTable(Table table, boolean deleteData)
- throws MetaException {
+ public HiveMetaHook getMetaHook() {
+ return null;
}
@Override
- public Class<? extends HCatInputStorageDriver> getInputStorageDriver() {
- return HCatInputStorageDriver.class;
+ public void configureInputJobProperties(TableDesc tableDesc, Map<String, String> jobProperties) {
}
@Override
- public Class<? extends HCatOutputStorageDriver> getOutputStorageDriver() {
- return HCatOutputStorageDriver.class;
+ public void configureOutputJobProperties(TableDesc tableDesc, Map<String, String> jobProperties) {
}
@Override
@@ -210,6 +200,89 @@ class DummyStorageHandler extends HCatSt
}
+ /**
+ * The Class DummyInputFormat is a dummy implementation of the old hadoop
+ * mapred.InputFormat required by HiveStorageHandler.
+ */
+ class DummyInputFormat implements
+ InputFormat<WritableComparable, HCatRecord> {
+
+ /*
+ * @see
+ * org.apache.hadoop.mapred.InputFormat#getRecordReader(org.apache.hadoop
+ * .mapred.InputSplit, org.apache.hadoop.mapred.JobConf,
+ * org.apache.hadoop.mapred.Reporter)
+ */
+ @Override
+ public RecordReader<WritableComparable, HCatRecord> getRecordReader(
+ InputSplit split, JobConf jobconf, Reporter reporter)
+ throws IOException {
+ throw new IOException("This operation is not supported.");
+ }
+
+ /*
+ * @see
+ * org.apache.hadoop.mapred.InputFormat#getSplits(org.apache.hadoop.
+ * mapred .JobConf, int)
+ */
+ @Override
+ public InputSplit[] getSplits(JobConf jobconf, int number)
+ throws IOException {
+ throw new IOException("This operation is not supported.");
+ }
+ }
+
+ /**
+ * The Class DummyOutputFormat is a dummy implementation of the old hadoop
+ * mapred.OutputFormat and HiveOutputFormat required by HiveStorageHandler.
+ */
+ class DummyOutputFormat implements
+ OutputFormat<WritableComparable<?>, HCatRecord>,
+ HiveOutputFormat<WritableComparable<?>, HCatRecord> {
+
+ /*
+ * @see
+ * org.apache.hadoop.mapred.OutputFormat#checkOutputSpecs(org.apache
+ * .hadoop .fs.FileSystem, org.apache.hadoop.mapred.JobConf)
+ */
+ @Override
+ public void checkOutputSpecs(FileSystem fs, JobConf jobconf)
+ throws IOException {
+ throw new IOException("This operation is not supported.");
+
+ }
+
+ /*
+ * @see
+ * org.apache.hadoop.mapred.OutputFormat#getRecordWriter(org.apache.
+ * hadoop .fs.FileSystem, org.apache.hadoop.mapred.JobConf,
+ * java.lang.String, org.apache.hadoop.util.Progressable)
+ */
+ @Override
+ public RecordWriter<WritableComparable<?>, HCatRecord> getRecordWriter(
+ FileSystem fs, JobConf jobconf, String str,
+ Progressable progress) throws IOException {
+ throw new IOException("This operation is not supported.");
+ }
+
+ /*
+ * @see
+ * org.apache.hadoop.hive.ql.io.HiveOutputFormat#getHiveRecordWriter(org
+ * .apache.hadoop.mapred.JobConf, org.apache.hadoop.fs.Path,
+ * java.lang.Class, boolean, java.util.Properties,
+ * org.apache.hadoop.util.Progressable)
+ */
+ @Override
+ public org.apache.hadoop.hive.ql.exec.FileSinkOperator.RecordWriter getHiveRecordWriter(
+ JobConf jc, Path finalOutPath,
+ Class<? extends Writable> valueClass, boolean isCompressed,
+ Properties tableProperties, Progressable progress)
+ throws IOException {
+ throw new IOException("This operation is not supported.");
+ }
+
+ }
+
}
Modified: incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/TestPermsGrp.java
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/TestPermsGrp.java?rev=1244334&r1=1244333&r2=1244334&view=diff
==============================================================================
--- incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/TestPermsGrp.java (original)
+++ incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/TestPermsGrp.java Wed Feb 15 03:53:50 2012
@@ -64,7 +64,7 @@ public class TestPermsGrp extends TestCa
@Override
public void run() {
try {
- HiveMetaStore.main(new String[]{msPort});
+ HiveMetaStore.main(new String[]{"-v","-p",msPort});
} catch(Throwable t) {
System.err.println("Exiting. Got exception from metastore: " + t.getMessage());
}
Added: incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/TestStorageHandlerProperties.java.broken
URL: http://svn.apache.org/viewvc/incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/TestStorageHandlerProperties.java.broken?rev=1244334&view=auto
==============================================================================
--- incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/TestStorageHandlerProperties.java.broken (added)
+++ incubator/hcatalog/trunk/src/test/org/apache/hcatalog/cli/TestStorageHandlerProperties.java.broken Wed Feb 15 03:53:50 2012
@@ -0,0 +1,84 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hcatalog.cli;
+
+import static org.junit.Assert.assertEquals;
+
+import org.apache.hadoop.hive.cli.CliSessionState;
+import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
+import org.apache.hadoop.hive.metastore.HiveMetaStoreClient;
+import org.apache.hadoop.hive.metastore.MetaStoreUtils;
+import org.apache.hadoop.hive.metastore.api.MetaException;
+import org.apache.hadoop.hive.metastore.api.NoSuchObjectException;
+import org.apache.hadoop.hive.metastore.api.Table;
+import org.apache.hadoop.hive.ql.CommandNeedRetryException;
+import org.apache.hadoop.hive.ql.Driver;
+import org.apache.hadoop.hive.ql.processors.CommandProcessorResponse;
+import org.apache.hadoop.hive.ql.session.SessionState;
+import org.apache.hcatalog.cli.SemanticAnalysis.HCatSemanticAnalyzer;
+import org.apache.hcatalog.common.HCatConstants;
+import org.apache.thrift.TException;
+
+import junit.framework.TestCase;
+
+public class TestStorageHandlerProperties extends TestCase {
+
+ private Driver hcatDriver;
+ private Driver hiveDriver;
+ private HiveMetaStoreClient msc;
+
+ protected void setUp() throws Exception {
+ HiveConf hcatConf = new HiveConf(this.getClass());
+ hcatConf.set(ConfVars.PREEXECHOOKS.varname, "");
+ hcatConf.set(ConfVars.POSTEXECHOOKS.varname, "");
+ hcatConf.set(ConfVars.HIVE_SUPPORT_CONCURRENCY.varname, "false");
+
+ HiveConf hiveConf = new HiveConf(hcatConf,this.getClass());
+ hiveDriver = new Driver(hiveConf);
+
+ hcatConf.set(ConfVars.SEMANTIC_ANALYZER_HOOK.varname, HCatSemanticAnalyzer.class.getName());
+ hcatDriver = new Driver(hcatConf);
+
+ msc = new HiveMetaStoreClient(hcatConf);
+ SessionState.start(new CliSessionState(hcatConf));
+ }
+
+ public void testTableProperties() throws CommandNeedRetryException, MetaException ,TException, NoSuchObjectException{
+ hcatDriver.run("drop table test_table");
+ CommandProcessorResponse response = hcatDriver
+ .run("create table test_table(key int, value string) STORED BY " +
+ "'org.apache.hcatalog.cli.DummyStorageHandler' ");
+
+ assertEquals(0, response.getResponseCode());
+ Table tbl = msc.getTable(MetaStoreUtils.DEFAULT_DATABASE_NAME, "test_table");
+ DummyStorageHandler dsh = new DummyStorageHandler();
+ assertTrue(tbl.getParameters().containsKey(HCatConstants.HCAT_ISD_CLASS));
+ assertTrue(tbl.getParameters().containsKey(HCatConstants.HCAT_OSD_CLASS));
+ assertEquals(tbl.getParameters().get(HCatConstants.HCAT_ISD_CLASS), dsh.getInputStorageDriver().getName());
+ assertEquals(tbl.getParameters().get(HCatConstants.HCAT_OSD_CLASS), dsh.getOutputStorageDriver().getName());
+ }
+
+ /* @throws java.lang.Exception
+ * @see junit.framework.TestCase#tearDown()
+ */
+ protected void tearDown() throws Exception {
+ super.tearDown();
+ }
+
+}