You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by pr...@apache.org on 2015/04/08 01:01:18 UTC
svn commit: r1671970 [1/2] - in /hive/branches/llap: ./
common/src/java/org/apache/hadoop/hive/common/
common/src/java/org/apache/hadoop/hive/conf/
hcatalog/core/src/main/java/org/apache/hive/hcatalog/mapreduce/
metastore/src/java/org/apache/hadoop/hiv...
Author: prasanthj
Date: Tue Apr 7 23:01:18 2015
New Revision: 1671970
URL: http://svn.apache.org/r1671970
Log:
Merge from trunk to llap (Prasanth Jayachandran)
Modified:
hive/branches/llap/ (props changed)
hive/branches/llap/common/src/java/org/apache/hadoop/hive/common/FileUtils.java
hive/branches/llap/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
hive/branches/llap/hcatalog/core/src/main/java/org/apache/hive/hcatalog/mapreduce/FileOutputCommitterContainer.java
hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileTezProcessor.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcRawRecordMerger.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSplit.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/StorageBasedAuthorizationProvider.java
hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/plugin/sqlstd/SQLAuthorizationUtils.java
hive/branches/llap/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java
hive/branches/llap/ql/src/test/queries/clientpositive/groupby3_map.q
hive/branches/llap/ql/src/test/results/clientpositive/groupby3_map.q.out
hive/branches/llap/ql/src/test/results/clientpositive/spark/groupby3_map.q.out
hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/ColumnProjectionUtils.java
hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/avro/AvroDeserializer.java
hive/branches/llap/serde/src/test/org/apache/hadoop/hive/serde2/avro/TestAvroDeserializer.java
hive/branches/llap/serde/src/test/org/apache/hadoop/hive/serde2/avro/TestAvroObjectInspectorGenerator.java
hive/branches/llap/serde/src/test/org/apache/hadoop/hive/serde2/avro/Utils.java
Propchange: hive/branches/llap/
------------------------------------------------------------------------------
--- svn:mergeinfo (original)
+++ svn:mergeinfo Tue Apr 7 23:01:18 2015
@@ -4,4 +4,4 @@
/hive/branches/spark:1608589-1660298
/hive/branches/tez:1494760-1622766
/hive/branches/vectorization:1466908-1527856
-/hive/trunk:1624170-1671667
+/hive/trunk:1624170-1671942
Modified: hive/branches/llap/common/src/java/org/apache/hadoop/hive/common/FileUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/common/src/java/org/apache/hadoop/hive/common/FileUtils.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/common/src/java/org/apache/hadoop/hive/common/FileUtils.java (original)
+++ hive/branches/llap/common/src/java/org/apache/hadoop/hive/common/FileUtils.java Tue Apr 7 23:01:18 2015
@@ -335,16 +335,16 @@ public final class FileUtils {
* @param fs
* file system
* @param path
- * @return the argument path if it exists or a parent path exists. Returns
- * NULL root is only parent that exists
+ * @return FileStatus for argument path if it exists or the first ancestor in the path that exists
* @throws IOException
*/
- public static Path getPathOrParentThatExists(FileSystem fs, Path path) throws IOException {
- if (!fs.exists(path)) {
- Path parentPath = path.getParent();
- return getPathOrParentThatExists(fs, parentPath);
+ public static FileStatus getPathOrParentThatExists(FileSystem fs, Path path) throws IOException {
+ FileStatus stat = FileUtils.getFileStatusOrNull(fs, path);
+ if (stat != null) {
+ return stat;
}
- return path;
+ Path parentPath = path.getParent();
+ return getPathOrParentThatExists(fs, parentPath);
}
/**
@@ -743,4 +743,20 @@ public final class FileUtils {
}
+ /**
+ * Attempts to get file status. This method differs from the FileSystem API in that it returns
+ * null instead of throwing FileNotFoundException if the path does not exist.
+ *
+ * @param fs file system to check
+ * @param path file system path to check
+ * @return FileStatus for path or null if path does not exist
+ * @throws IOException if there is an I/O error
+ */
+ public static FileStatus getFileStatusOrNull(FileSystem fs, Path path) throws IOException {
+ try {
+ return fs.getFileStatus(path);
+ } catch (FileNotFoundException e) {
+ return null;
+ }
+ }
}
Modified: hive/branches/llap/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java (original)
+++ hive/branches/llap/common/src/java/org/apache/hadoop/hive/conf/HiveConf.java Tue Apr 7 23:01:18 2015
@@ -948,6 +948,13 @@ public class HiveConf extends Configurat
"Define the compression strategy to use while writing data. \n" +
"This changes the compression level of higher level compression codec (like ZLIB)."),
+ HIVE_ORC_SPLIT_STRATEGY("hive.exec.orc.split.strategy", "HYBRID", new StringSet("HYBRID", "BI", "ETL"),
+ "This is not a user level config. BI strategy is used when the requirement is to spend less time in split generation" +
+ " as opposed to query execution (split generation does not read or cache file footers)." +
+ " ETL strategy is used when spending little more time in split generation is acceptable" +
+ " (split generation reads and caches file footers). HYBRID chooses between the above strategies" +
+ " based on heuristics."),
+
HIVE_ORC_INCLUDE_FILE_FOOTER_IN_SPLITS("hive.orc.splits.include.file.footer", false,
"If turned on splits generated by orc will include metadata about the stripes in the file. This\n" +
"data is read remotely (from the client or HS2 machine) and sent to all the tasks."),
Modified: hive/branches/llap/hcatalog/core/src/main/java/org/apache/hive/hcatalog/mapreduce/FileOutputCommitterContainer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/hcatalog/core/src/main/java/org/apache/hive/hcatalog/mapreduce/FileOutputCommitterContainer.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/hcatalog/core/src/main/java/org/apache/hive/hcatalog/mapreduce/FileOutputCommitterContainer.java (original)
+++ hive/branches/llap/hcatalog/core/src/main/java/org/apache/hive/hcatalog/mapreduce/FileOutputCommitterContainer.java Tue Apr 7 23:01:18 2015
@@ -510,8 +510,9 @@ class FileOutputCommitterContainer exten
}
final Path finalOutputPath = getFinalPath(fs, file, srcDir, destDir, immutable);
+ FileStatus fileStatus = fs.getFileStatus(file);
- if (fs.isFile(file)) {
+ if (fileStatus.isFile()) {
if (dryRun){
if (immutable){
// Dryrun checks are meaningless for mutable table - we should always succeed
@@ -541,7 +542,7 @@ class FileOutputCommitterContainer exten
}
}
}
- } else if(fs.getFileStatus(file).isDir()) {
+ } else if (fileStatus.isDirectory()) {
FileStatus[] children = fs.listStatus(file);
FileStatus firstChild = null;
Modified: hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java (original)
+++ hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/MetaStoreDirectSql.java Tue Apr 7 23:01:18 2015
@@ -852,6 +852,7 @@ class MetaStoreDirectSql {
func.apply(entry.getValue(), fields);
fields = null;
}
+ Deadline.checkTimeout();
}
int rv = list.size();
query.closeAll();
@@ -1152,6 +1153,7 @@ class MetaStoreDirectSql {
list.size());
for (Object[] row : list) {
colStats.add(prepareCSObj(row, 0));
+ Deadline.checkTimeout();
}
query.closeAll();
return colStats;
@@ -1193,6 +1195,7 @@ class MetaStoreDirectSql {
} else {
extraColumnNameTypeParts.put(colName, new String[] { colType, String.valueOf(count) });
}
+ Deadline.checkTimeout();
}
query.closeAll();
// Extrapolation is not needed for columns noExtraColumnNames
@@ -1212,6 +1215,7 @@ class MetaStoreDirectSql {
list = ensureList(qResult);
for (Object[] row : list) {
colStats.add(prepareCSObj(row, 0));
+ Deadline.checkTimeout();
}
end = doTrace ? System.nanoTime() : 0;
timingTrace(doTrace, queryText, start, end);
@@ -1251,6 +1255,7 @@ class MetaStoreDirectSql {
indexToObject.put(sumIndex[ind - 1], row[ind]);
}
sumMap.put((String) row[0], indexToObject);
+ Deadline.checkTimeout();
}
end = doTrace ? System.nanoTime() : 0;
timingTrace(doTrace, queryText, start, end);
@@ -1318,6 +1323,7 @@ class MetaStoreDirectSql {
}
}
colStats.add(prepareCSObj(row, 0));
+ Deadline.checkTimeout();
}
}
return colStats;
@@ -1389,6 +1395,7 @@ class MetaStoreDirectSql {
}
lastPartName = partName;
from = i;
+ Deadline.checkTimeout();
}
timingTrace(doTrace, queryText, start, queryTime);
@@ -1416,6 +1423,7 @@ class MetaStoreDirectSql {
csd.setLastAnalyzed(extractSqlLong(laObj));
}
csos.add(prepareCSObj(row, offset));
+ Deadline.checkTimeout();
}
result.setStatsObj(csos);
return result;
Modified: hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java (original)
+++ hive/branches/llap/metastore/src/java/org/apache/hadoop/hive/metastore/ObjectStore.java Tue Apr 7 23:01:18 2015
@@ -1817,6 +1817,7 @@ public class ObjectStore implements RawS
}
for (MPartition mp : src) {
dest.add(convertToPart(mp));
+ Deadline.checkTimeout();
}
return dest;
}
@@ -1826,6 +1827,7 @@ public class ObjectStore implements RawS
List<Partition> parts = new ArrayList<Partition>(mparts.size());
for (MPartition mp : mparts) {
parts.add(convertToPart(dbName, tblName, mp));
+ Deadline.checkTimeout();
}
return parts;
}
@@ -6049,6 +6051,7 @@ public class ObjectStore implements RawS
desc.setLastAnalyzed(mStat.getLastAnalyzed());
}
statObjs.add(StatObjectConverter.getTableColumnStatisticsObj(mStat));
+ Deadline.checkTimeout();
}
return new ColumnStatistics(desc, statObjs);
}
@@ -6097,6 +6100,7 @@ public class ObjectStore implements RawS
}
curList.add(StatObjectConverter.getPartitionColumnStatisticsObj(mStatsObj));
lastPartName = partName;
+ Deadline.checkTimeout();
}
return result;
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/DagUtils.java Tue Apr 7 23:01:18 2015
@@ -45,6 +45,7 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.common.FileUtils;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.ql.Context;
@@ -945,8 +946,9 @@ public class DagUtils {
throws IOException {
FileSystem destFS = dest.getFileSystem(conf);
FileSystem sourceFS = src.getFileSystem(conf);
- if (destFS.exists(dest)) {
- return (sourceFS.getFileStatus(src).getLen() == destFS.getFileStatus(dest).getLen());
+ FileStatus destStatus = FileUtils.getFileStatusOrNull(destFS, dest);
+ if (destStatus != null) {
+ return (sourceFS.getFileStatus(src).getLen() == destStatus.getLen());
}
return false;
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MapRecordProcessor.java Tue Apr 7 23:01:18 2015
@@ -27,6 +27,7 @@ import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
+import java.util.TreeMap;
import java.util.concurrent.Callable;
import org.apache.commons.logging.Log;
@@ -87,7 +88,11 @@ public class MapRecordProcessor extends
List<String> cacheKeys;
ObjectCache cache;
- public MapRecordProcessor(final JobConf jconf) throws Exception {
+ private static Map<Integer, DummyStoreOperator> connectOps =
+ new TreeMap<Integer, DummyStoreOperator>();
+
+ public MapRecordProcessor(final JobConf jconf, final ProcessorContext context) throws Exception {
+ super(jconf, context);
if (LlapIoProxy.isDaemon()) { // do not cache plan
cache = new org.apache.hadoop.hive.ql.exec.mr.ObjectCache();
} else {
@@ -99,10 +104,10 @@ public class MapRecordProcessor extends
}
@Override
- void init(final JobConf jconf, ProcessorContext processorContext, MRTaskReporter mrReporter,
+ void init(MRTaskReporter mrReporter,
Map<String, LogicalInput> inputs, Map<String, LogicalOutput> outputs) throws Exception {
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.TEZ_INIT_OPERATORS);
- super.init(jconf, processorContext, mrReporter, inputs, outputs);
+ super.init(mrReporter, inputs, outputs);
String queryId = HiveConf.getVar(jconf, HiveConf.ConfVars.HIVEQUERYID);
String key = queryId + processorContext.getTaskVertexName() + MAP_PLAN_KEY;
@@ -129,14 +134,14 @@ public class MapRecordProcessor extends
key = queryId + processorContext.getTaskVertexName() + prefix;
cacheKeys.add(key);
- mergeWorkList.add(
- (MapWork) cache.retrieve(key,
- new Callable<Object>() {
- @Override
- public Object call() {
- return Utilities.getMergeWork(jconf, prefix);
- }
- }));
+ mergeWorkList.add(
+ (MapWork) cache.retrieve(key,
+ new Callable<Object>() {
+ @Override
+ public Object call() {
+ return Utilities.getMergeWork(jconf, prefix);
+ }
+ }));
}
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileRecordProcessor.java Tue Apr 7 23:01:18 2015
@@ -64,12 +64,16 @@ public class MergeFileRecordProcessor ex
private final Object[] row = new Object[2];
ObjectCache cache;
+ public MergeFileRecordProcessor(final JobConf jconf, final ProcessorContext context) {
+ super(jconf, context);
+ }
+
@Override
- void init(final JobConf jconf, ProcessorContext processorContext,
+ void init(
MRTaskReporter mrReporter, Map<String, LogicalInput> inputs,
Map<String, LogicalOutput> outputs) throws Exception {
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.TEZ_INIT_OPERATORS);
- super.init(jconf, processorContext, mrReporter, inputs, outputs);
+ super.init(mrReporter, inputs, outputs);
execContext = new ExecMapperContext(jconf);
//Update JobConf using MRInput, info like filename comes via this
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileTezProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileTezProcessor.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileTezProcessor.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/MergeFileTezProcessor.java Tue Apr 7 23:01:18 2015
@@ -38,7 +38,7 @@ public class MergeFileTezProcessor exten
@Override
public void run(Map<String, LogicalInput> inputs,
Map<String, LogicalOutput> outputs) throws Exception {
- rproc = new MergeFileRecordProcessor();
+ rproc = new MergeFileRecordProcessor(jobConf, getContext());
initializeAndRunProcessor(inputs, outputs);
}
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/RecordProcessor.java Tue Apr 7 23:01:18 2015
@@ -50,11 +50,11 @@ import java.util.concurrent.Callable;
*/
public abstract class RecordProcessor {
- protected JobConf jconf;
+ protected final JobConf jconf;
protected Map<String, LogicalInput> inputs;
protected Map<String, LogicalOutput> outputs;
protected Map<String, OutputCollector> outMap;
- protected ProcessorContext processorContext;
+ protected final ProcessorContext processorContext;
public static final Log l4j = LogFactory.getLog(RecordProcessor.class);
@@ -67,22 +67,23 @@ public abstract class RecordProcessor {
protected PerfLogger perfLogger = PerfLogger.getPerfLogger();
protected String CLASS_NAME = RecordProcessor.class.getName();
+ public RecordProcessor(JobConf jConf, ProcessorContext processorContext) {
+ this.jconf = jConf;
+ this.processorContext = processorContext;
+ }
+
/**
* Common initialization code for RecordProcessors
- * @param jconf
- * @param processorContext the {@link ProcessorContext}
* @param mrReporter
* @param inputs map of Input names to {@link LogicalInput}s
* @param outputs map of Output names to {@link LogicalOutput}s
* @throws Exception
*/
- void init(JobConf jconf, ProcessorContext processorContext, MRTaskReporter mrReporter,
+ void init(MRTaskReporter mrReporter,
Map<String, LogicalInput> inputs, Map<String, LogicalOutput> outputs) throws Exception {
- this.jconf = jconf;
this.reporter = mrReporter;
this.inputs = inputs;
this.outputs = outputs;
- this.processorContext = processorContext;
isLogInfoEnabled = l4j.isInfoEnabled();
isLogTraceEnabled = l4j.isTraceEnabled();
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/ReduceRecordProcessor.java Tue Apr 7 23:01:18 2015
@@ -84,7 +84,8 @@ public class ReduceRecordProcessor exte
private boolean abort;
- public ReduceRecordProcessor(final JobConf jconf) throws Exception {
+ public ReduceRecordProcessor(final JobConf jconf, final ProcessorContext context) throws Exception {
+ super(jconf, context);
ObjectCache cache;
@@ -108,11 +109,11 @@ public class ReduceRecordProcessor exte
}
@Override
- void init(JobConf jconf, ProcessorContext processorContext,
+ void init(
MRTaskReporter mrReporter, Map<String, LogicalInput> inputs,
Map<String, LogicalOutput> outputs) throws Exception {
perfLogger.PerfLogBegin(CLASS_NAME, PerfLogger.TEZ_INIT_OPERATORS);
- super.init(jconf, processorContext, mrReporter, inputs, outputs);
+ super.init(mrReporter, inputs, outputs);
MapredContext.init(false, new JobConf(jconf));
List<LogicalInput> shuffleInputs = getShuffleInputs(inputs);
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/exec/tez/TezProcessor.java Tue Apr 7 23:01:18 2015
@@ -129,9 +129,9 @@ public class TezProcessor extends Abstra
LOG.info("Running task: " + getContext().getUniqueIdentifier());
if (isMap) {
- rproc = new MapRecordProcessor(jobConf);
+ rproc = new MapRecordProcessor(jobConf, getContext());
} else {
- rproc = new ReduceRecordProcessor(jobConf);
+ rproc = new ReduceRecordProcessor(jobConf, getContext());
}
initializeAndRunProcessor(inputs, outputs);
@@ -144,7 +144,7 @@ public class TezProcessor extends Abstra
try {
MRTaskReporter mrReporter = new MRTaskReporter(getContext());
- rproc.init(jobConf, getContext(), mrReporter, inputs, outputs);
+ rproc.init(mrReporter, inputs, outputs);
rproc.run();
//done - output does not need to be committed as hive does not use outputcommitter
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java Tue Apr 7 23:01:18 2015
@@ -21,14 +21,16 @@ package org.apache.hadoop.hive.ql.io.orc
import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
+import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.TreeMap;
+import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
-import java.util.concurrent.TimeUnit;
+import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.commons.logging.Log;
@@ -38,15 +40,12 @@ import org.apache.hadoop.fs.BlockLocatio
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.common.ValidReadTxnList;
+import org.apache.hadoop.hive.common.ValidTxnList;
import org.apache.hadoop.hive.conf.HiveConf;
import org.apache.hadoop.hive.conf.HiveConf.ConfVars;
import org.apache.hadoop.hive.metastore.api.hive_metastoreConstants;
import org.apache.hadoop.hive.ql.exec.Utilities;
-import org.apache.hadoop.hive.ql.exec.vector.ColumnVector;
-import org.apache.hadoop.hive.ql.exec.vector.DoubleColumnVector;
-import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedInputFormatInterface;
import org.apache.hadoop.hive.ql.io.AcidInputFormat;
import org.apache.hadoop.hive.ql.io.AcidOutputFormat;
@@ -56,13 +55,11 @@ import org.apache.hadoop.hive.ql.io.Inpu
import org.apache.hadoop.hive.ql.io.LlapWrappableInputFormatInterface;
import org.apache.hadoop.hive.ql.io.RecordIdentifier;
import org.apache.hadoop.hive.ql.io.StatsProvidingRecordReader;
-import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat.Context;
import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument.TruthValue;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
import org.apache.hadoop.hive.ql.log.PerfLogger;
-import org.apache.hadoop.hive.ql.plan.TableScanDesc;
import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
import org.apache.hadoop.hive.serde2.SerDeStats;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
@@ -75,14 +72,13 @@ import org.apache.hadoop.io.NullWritable
import org.apache.hadoop.mapred.FileSplit;
import org.apache.hadoop.mapred.InputFormat;
import org.apache.hadoop.mapred.InputSplit;
-import org.apache.hadoop.mapred.InvalidInputException;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.StringUtils;
-import com.google.common.annotations.VisibleForTesting;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
+import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
/**
* A MapReduce/Hive input format for ORC files.
@@ -111,7 +107,14 @@ public class OrcInputFormat implements
InputFormatChecker, VectorizedInputFormatInterface, LlapWrappableInputFormatInterface,
AcidInputFormat<NullWritable, OrcStruct>, CombineHiveInputFormat.AvoidSplitCombination {
+ static enum SplitStrategyKind{
+ HYBRID,
+ BI,
+ ETL
+ }
+
private static final Log LOG = LogFactory.getLog(OrcInputFormat.class);
+ private static boolean isDebugEnabled = LOG.isDebugEnabled();
static final HadoopShims SHIMS = ShimLoader.getHadoopShims();
static final String MIN_SPLIT_SIZE =
SHIMS.getHadoopConfNames().get("MAPREDMINSPLITSIZE");
@@ -270,7 +273,6 @@ public class OrcInputFormat implements
/**
* Take the configuration and figure out which columns we need to include.
- * @param options the options to update
* @param types the types for the file
* @param conf the configuration
* @param isOriginal is the file in the original format?
@@ -371,30 +373,28 @@ public class OrcInputFormat implements
static class Context {
private final Configuration conf;
private static Cache<Path, FileInfo> footerCache;
- private final ExecutorService threadPool;
- private final List<OrcSplit> splits =
- new ArrayList<OrcSplit>(10000);
+ private static ExecutorService threadPool = null;
private final int numBuckets;
- private final List<Throwable> errors = new ArrayList<Throwable>();
private final long maxSize;
private final long minSize;
private final boolean footerInSplits;
private final boolean cacheStripeDetails;
private final AtomicInteger cacheHitCounter = new AtomicInteger(0);
private final AtomicInteger numFilesCounter = new AtomicInteger(0);
- private Throwable fatalError = null;
private ValidTxnList transactionList;
-
- /**
- * A count of the number of threads that may create more work for the
- * thread pool.
- */
- private int schedulers = 0;
+ private SplitStrategyKind splitStrategyKind;
Context(Configuration conf) {
this.conf = conf;
minSize = conf.getLong(MIN_SPLIT_SIZE, DEFAULT_MIN_SPLIT_SIZE);
maxSize = conf.getLong(MAX_SPLIT_SIZE, DEFAULT_MAX_SPLIT_SIZE);
+ String ss = conf.get(ConfVars.HIVE_ORC_SPLIT_STRATEGY.varname);
+ if (ss == null || ss.equals(SplitStrategyKind.HYBRID.name())) {
+ splitStrategyKind = SplitStrategyKind.HYBRID;
+ } else {
+ LOG.info("Enforcing " + ss + " ORC split strategy");
+ splitStrategyKind = SplitStrategyKind.valueOf(ss);
+ }
footerInSplits = HiveConf.getBoolVar(conf,
ConfVars.HIVE_ORC_INCLUDE_FILE_FOOTER_IN_SPLITS);
numBuckets =
@@ -407,11 +407,13 @@ public class OrcInputFormat implements
cacheStripeDetails = (cacheStripeDetailsSize > 0);
- threadPool = Executors.newFixedThreadPool(numThreads,
- new ThreadFactoryBuilder().setDaemon(true)
- .setNameFormat("ORC_GET_SPLITS #%d").build());
-
synchronized (Context.class) {
+ if (threadPool == null) {
+ threadPool = Executors.newFixedThreadPool(numThreads,
+ new ThreadFactoryBuilder().setDaemon(true)
+ .setNameFormat("ORC_GET_SPLITS #%d").build());
+ }
+
if (footerCache == null && cacheStripeDetails) {
footerCache = CacheBuilder.newBuilder().concurrencyLevel(numThreads)
.initialCapacity(cacheStripeDetailsSize).softValues().build();
@@ -421,87 +423,205 @@ public class OrcInputFormat implements
Long.MAX_VALUE + ":");
transactionList = new ValidReadTxnList(value);
}
+ }
- int getSchedulers() {
- return schedulers;
- }
+ interface SplitStrategy<T> {
+ List<T> getSplits() throws IOException;
+ }
- /**
- * Get the Nth split.
- * @param index if index >= 0, count from the front, otherwise count from
- * the back.
- * @return the Nth file split
- */
- OrcSplit getResult(int index) {
- if (index >= 0) {
- return splits.get(index);
- } else {
- return splits.get(splits.size() + index);
- }
+ static final class SplitInfo extends ACIDSplitStrategy {
+ private final Context context;
+ private final FileSystem fs;
+ private final HdfsFileStatusWithId file;
+ private final FileInfo fileInfo;
+ private final boolean isOriginal;
+ private final List<Long> deltas;
+ private final boolean hasBase;
+
+ SplitInfo(Context context, FileSystem fs,
+ HdfsFileStatusWithId file, FileInfo fileInfo,
+ boolean isOriginal,
+ List<Long> deltas,
+ boolean hasBase, Path dir, boolean[] covered) throws IOException {
+ super(dir, context.numBuckets, deltas, covered);
+ this.context = context;
+ this.fs = fs;
+ this.file = file;
+ this.fileInfo = fileInfo;
+ this.isOriginal = isOriginal;
+ this.deltas = deltas;
+ this.hasBase = hasBase;
}
+ }
+
+ /**
+ * ETL strategy is used when spending little more time in split generation is acceptable
+ * (split generation reads and caches file footers).
+ */
+ static final class ETLSplitStrategy implements SplitStrategy<SplitInfo> {
+ Context context;
+ FileSystem fs;
+ List<HdfsFileStatusWithId> files;
+ boolean isOriginal;
+ List<Long> deltas;
+ Path dir;
+ boolean[] covered;
- List<Throwable> getErrors() {
- return errors;
+ public ETLSplitStrategy(Context context, FileSystem fs, Path dir, List<HdfsFileStatusWithId> children,
+ boolean isOriginal, List<Long> deltas, boolean[] covered) {
+ this.context = context;
+ this.dir = dir;
+ this.fs = fs;
+ this.files = children;
+ this.isOriginal = isOriginal;
+ this.deltas = deltas;
+ this.covered = covered;
}
- /**
- * Add a unit of work.
- * @param runnable the object to run
- */
- synchronized void schedule(Runnable runnable) {
- if (fatalError == null) {
- if (runnable instanceof FileGenerator ||
- runnable instanceof SplitGenerator) {
- schedulers += 1;
+ private FileInfo verifyCachedFileInfo(HdfsFileStatusWithId fileId) {
+ context.numFilesCounter.incrementAndGet();
+ FileStatus file = fileId.getFileStatus();
+ FileInfo fileInfo = Context.footerCache.getIfPresent(file.getPath());
+ if (fileInfo != null) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Info cached for path: " + file.getPath());
+ }
+ if (fileInfo.modificationTime == file.getModificationTime() &&
+ fileInfo.size == file.getLen()) {
+ // Cached copy is valid
+ context.cacheHitCounter.incrementAndGet();
+ return fileInfo;
+ } else {
+ // Invalidate
+ Context.footerCache.invalidate(file.getPath());
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Meta-Info for : " + file.getPath() +
+ " changed. CachedModificationTime: "
+ + fileInfo.modificationTime + ", CurrentModificationTime: "
+ + file.getModificationTime()
+ + ", CachedLength: " + fileInfo.size + ", CurrentLength: " +
+ file.getLen());
+ }
}
- threadPool.execute(runnable);
} else {
- throw new RuntimeException("serious problem", fatalError);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Info not cached for path: " + file.getPath());
+ }
}
+ return null;
}
- /**
- * Mark a worker that may generate more work as done.
- */
- synchronized void decrementSchedulers() {
- schedulers -= 1;
- if (schedulers == 0) {
- notify();
+ @Override
+ public List<SplitInfo> getSplits() throws IOException {
+ List<SplitInfo> result = Lists.newArrayList();
+ for (HdfsFileStatusWithId file : files) {
+ FileInfo info = null;
+ if (context.cacheStripeDetails) {
+ info = verifyCachedFileInfo(file);
+ }
+ // ignore files of 0 length
+ if (file.getFileStatus().getLen() > 0) {
+ result.add(new SplitInfo(context, fs, file, info, isOriginal, deltas, true, dir, covered));
+ }
}
+ return result;
}
- synchronized void notifyOnNonIOException(Throwable th) {
- fatalError = th;
- notify();
+ @Override
+ public String toString() {
+ return ETLSplitStrategy.class.getSimpleName() + " strategy for " + dir;
}
+ }
- /**
- * Wait until all of the tasks are done. It waits until all of the
- * threads that may create more work are done and then shuts down the
- * thread pool and waits for the final threads to finish.
- */
- synchronized void waitForTasks() {
- try {
- while (schedulers != 0) {
- wait();
- if (fatalError != null) {
- threadPool.shutdownNow();
- throw new RuntimeException("serious problem", fatalError);
+ /**
+ * BI strategy is used when the requirement is to spend less time in split generation
+ * as opposed to query execution (split generation does not read or cache file footers).
+ */
+ static final class BISplitStrategy extends ACIDSplitStrategy {
+ List<HdfsFileStatusWithId> fileStatuses;
+ boolean isOriginal;
+ List<Long> deltas;
+ FileSystem fs;
+ Context context;
+ Path dir;
+
+ public BISplitStrategy(Context context, FileSystem fs,
+ Path dir, List<HdfsFileStatusWithId> fileStatuses, boolean isOriginal,
+ List<Long> deltas, boolean[] covered) {
+ super(dir, context.numBuckets, deltas, covered);
+ this.context = context;
+ this.fileStatuses = fileStatuses;
+ this.isOriginal = isOriginal;
+ this.deltas = deltas;
+ this.fs = fs;
+ this.dir = dir;
+ }
+
+ @Override
+ public List<OrcSplit> getSplits() throws IOException {
+ List<OrcSplit> splits = Lists.newArrayList();
+ for (HdfsFileStatusWithId child : fileStatuses) {
+ String[] hosts = SHIMS.getLocationsWithOffset(fs, child.getFileStatus()).firstEntry().getValue()
+ .getHosts();
+ OrcSplit orcSplit = new OrcSplit(child.getFileStatus().getPath(), null, 0, child.getFileStatus().getLen(), hosts,
+ null, isOriginal, true, deltas, -1);
+ splits.add(orcSplit);
+ }
+
+ // add uncovered ACID delta splits
+ splits.addAll(super.getSplits());
+ return splits;
+ }
+
+ @Override
+ public String toString() {
+ return BISplitStrategy.class.getSimpleName() + " strategy for " + dir;
+ }
+ }
+
+ /**
+ * ACID split strategy is used when there is no base directory (when transactions are enabled).
+ */
+ static class ACIDSplitStrategy implements SplitStrategy<OrcSplit> {
+ Path dir;
+ List<Long> deltas;
+ boolean[] covered;
+ int numBuckets;
+
+ public ACIDSplitStrategy(Path dir, int numBuckets, List<Long> deltas, boolean[] covered) {
+ this.dir = dir;
+ this.numBuckets = numBuckets;
+ this.deltas = deltas;
+ this.covered = covered;
+ }
+
+ @Override
+ public List<OrcSplit> getSplits() throws IOException {
+ // Generate a split for any buckets that weren't covered.
+ // This happens in the case where a bucket just has deltas and no
+ // base.
+ List<OrcSplit> splits = Lists.newArrayList();
+ if (!deltas.isEmpty()) {
+ for (int b = 0; b < numBuckets; ++b) {
+ if (!covered[b]) {
+ splits.add(new OrcSplit(dir, null, b, 0, new String[0], null, false, false, deltas, -1));
}
}
- threadPool.shutdown();
- threadPool.awaitTermination(Long.MAX_VALUE, TimeUnit.DAYS);
- } catch (InterruptedException ie) {
- throw new IllegalStateException("interrupted", ie);
}
+ return splits;
+ }
+
+ @Override
+ public String toString() {
+ return ACIDSplitStrategy.class.getSimpleName() + " strategy for " + dir;
}
}
/**
* Given a directory, get the list of files and blocks in those files.
- * A thread is used for each directory.
+ * To parallelize file generator use "mapreduce.input.fileinputformat.list-status.num-threads"
*/
- static final class FileGenerator implements Runnable {
+ static final class FileGenerator implements Callable<SplitStrategy> {
private final Context context;
private final FileSystem fs;
private final Path dir;
@@ -514,83 +634,69 @@ public class OrcInputFormat implements
this.useFileIds = useFileIds;
}
- private void scheduleSplits(HdfsFileStatusWithId child,
- boolean isOriginal,
- boolean hasBase,
- List<Long> deltas) throws IOException{
- FileInfo info = null;
- if (context.cacheStripeDetails) {
- info = verifyCachedFileInfo(child.getFileStatus());
- }
- new SplitGenerator(context, fs, child, info, isOriginal, deltas,
- hasBase).schedule();
- }
-
- /**
- * For each path, get the list of files and blocks that they consist of.
- */
@Override
- public void run() {
- try {
- AcidUtils.Directory dirInfo = AcidUtils.getAcidState(dir,
- context.conf, context.transactionList, useFileIds);
- List<Long> deltas =
- AcidUtils.serializeDeltas(dirInfo.getCurrentDirectories());
- Path base = dirInfo.getBaseDirectory();
- List<HdfsFileStatusWithId> original = dirInfo.getOriginalFiles();
-
- boolean[] covered = new boolean[context.numBuckets];
- boolean isOriginal = base == null;
-
- // if we have a base to work from
- if (base != null || !original.isEmpty()) {
-
- // find the base files (original or new style)
- List<HdfsFileStatusWithId> children = original;
- if (base != null) {
- children = findBaseFiles(base, useFileIds);
- }
-
- // for each child, schedule splits and mark off the bucket
- for (HdfsFileStatusWithId child : children) {
- AcidOutputFormat.Options opts = AcidUtils.parseBaseBucketFilename
- (child.getFileStatus().getPath(), context.conf);
- scheduleSplits(child, isOriginal, true, deltas);
- int b = opts.getBucket();
- // If the bucket is in the valid range, mark it as covered.
- // I wish Hive actually enforced bucketing all of the time.
- if (b >= 0 && b < covered.length) {
- covered[b] = true;
- }
+ public SplitStrategy call() throws IOException {
+ final SplitStrategy splitStrategy;
+ AcidUtils.Directory dirInfo = AcidUtils.getAcidState(dir,
+ context.conf, context.transactionList);
+ List<Long> deltas = AcidUtils.serializeDeltas(dirInfo.getCurrentDirectories());
+ Path base = dirInfo.getBaseDirectory();
+ List<HdfsFileStatusWithId> original = dirInfo.getOriginalFiles();
+ boolean[] covered = new boolean[context.numBuckets];
+ boolean isOriginal = base == null;
+
+ // if we have a base to work from
+ if (base != null || !original.isEmpty()) {
+
+ // find the base files (original or new style)
+ List<HdfsFileStatusWithId> children = original;
+ if (base != null) {
+ children = findBaseFiles(base, useFileIds);
+ }
+
+ long totalFileSize = 0;
+ for (HdfsFileStatusWithId child : children) {
+ totalFileSize += child.getFileStatus().getLen();
+ AcidOutputFormat.Options opts = AcidUtils.parseBaseBucketFilename
+ (child.getFileStatus().getPath(), context.conf);
+ int b = opts.getBucket();
+ // If the bucket is in the valid range, mark it as covered.
+ // I wish Hive actually enforced bucketing all of the time.
+ if (b >= 0 && b < covered.length) {
+ covered[b] = true;
}
}
- // Generate a split for any buckets that weren't covered.
- // This happens in the case where a bucket just has deltas and no
- // base.
- if (!deltas.isEmpty()) {
- for (int b = 0; b < context.numBuckets; ++b) {
- if (!covered[b]) {
- synchronized (context.splits) {
- context.splits.add(new OrcSplit(dir, null, b, 0, new String[0], null,
- false, false, deltas));
- }
+ int numFiles = children.size();
+ long avgFileSize = totalFileSize / numFiles;
+ switch(context.splitStrategyKind) {
+ case BI:
+ // BI strategy requested through config
+ splitStrategy = new BISplitStrategy(context, fs, dir, children, isOriginal,
+ deltas, covered);
+ break;
+ case ETL:
+ // ETL strategy requested through config
+ splitStrategy = new ETLSplitStrategy(context, fs, dir, children, isOriginal,
+ deltas, covered);
+ break;
+ default:
+ // HYBRID strategy
+ if (avgFileSize > context.maxSize) {
+ splitStrategy = new ETLSplitStrategy(context, fs, dir, children, isOriginal, deltas,
+ covered);
+ } else {
+ splitStrategy = new BISplitStrategy(context, fs, dir, children, isOriginal, deltas,
+ covered);
}
- }
- }
- } catch (Throwable th) {
- if (!(th instanceof IOException)) {
- LOG.error("Unexpected Exception", th);
+ break;
}
- synchronized (context.errors) {
- context.errors.add(th);
- }
- if (!(th instanceof IOException)) {
- context.notifyOnNonIOException(th);
- }
- } finally {
- context.decrementSchedulers();
+ } else {
+ // no base, only deltas
+ splitStrategy = new ACIDSplitStrategy(dir, context.numBuckets, deltas, covered);
}
+
+ return splitStrategy;
}
private List<HdfsFileStatusWithId> findBaseFiles(
@@ -609,49 +715,16 @@ public class OrcInputFormat implements
}
return result;
}
-
- private FileInfo verifyCachedFileInfo(FileStatus file) {
- context.numFilesCounter.incrementAndGet();
- FileInfo fileInfo = Context.footerCache.getIfPresent(file.getPath());
- if (fileInfo != null) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Info cached for path: " + file.getPath());
- }
- if (fileInfo.modificationTime == file.getModificationTime() &&
- fileInfo.size == file.getLen()) {
- // Cached copy is valid
- context.cacheHitCounter.incrementAndGet();
- return fileInfo;
- } else {
- // Invalidate
- Context.footerCache.invalidate(file.getPath());
- if (LOG.isDebugEnabled()) {
- LOG.debug("Meta-Info for : " + file.getPath() +
- " changed. CachedModificationTime: "
- + fileInfo.modificationTime + ", CurrentModificationTime: "
- + file.getModificationTime()
- + ", CachedLength: " + fileInfo.size + ", CurrentLength: " +
- file.getLen());
- }
- }
- } else {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Info not cached for path: " + file.getPath());
- }
- }
- return null;
- }
}
/**
* Split the stripes of a given file into input splits.
* A thread is used for each file.
*/
- static final class SplitGenerator implements Runnable {
+ static final class SplitGenerator implements Callable<List<OrcSplit>> {
private final Context context;
private final FileSystem fs;
private final HdfsFileStatusWithId fileWithId;
- private final FileStatus file;
private final long blockSize;
private final TreeMap<Long, BlockLocation> locations;
private final FileInfo fileInfo;
@@ -663,54 +736,30 @@ public class OrcInputFormat implements
private final List<Long> deltas;
private final boolean hasBase;
private OrcFile.WriterVersion writerVersion;
+ private long projColsUncompressedSize;
+ private List<OrcSplit> deltaSplits;
- SplitGenerator(Context context, FileSystem fs,
- HdfsFileStatusWithId fileWithId, FileInfo fileInfo,
- boolean isOriginal,
- List<Long> deltas,
- boolean hasBase) throws IOException {
- this.context = context;
- this.fs = fs;
- this.fileWithId = fileWithId;
- this.file = this.fileWithId.getFileStatus();
- this.blockSize = this.file.getBlockSize();
- this.fileInfo = fileInfo;
- locations = SHIMS.getLocationsWithOffset(fs, this.file);
- this.isOriginal = isOriginal;
- this.deltas = deltas;
- this.hasBase = hasBase;
- }
-
- @VisibleForTesting
- SplitGenerator(Context context, FileSystem fs,
- FileStatus file, FileInfo fileInfo,
- boolean isOriginal,
- List<Long> deltas,
- boolean hasBase) throws IOException {
- this(context, fs, AcidUtils.createOriginalObj(null, file),
- fileInfo, isOriginal, deltas, hasBase);
+ public SplitGenerator(SplitInfo splitInfo) throws IOException {
+ this.context = splitInfo.context;
+ this.fs = splitInfo.fs;
+ this.fileWithId = splitInfo.file;
+ this.blockSize = fileWithId.getFileStatus().getBlockSize();
+ this.fileInfo = splitInfo.fileInfo;
+ locations = SHIMS.getLocationsWithOffset(fs, fileWithId.getFileStatus());
+ this.isOriginal = splitInfo.isOriginal;
+ this.deltas = splitInfo.deltas;
+ this.hasBase = splitInfo.hasBase;
+ this.projColsUncompressedSize = -1;
+ this.deltaSplits = splitInfo.getSplits();
}
Path getPath() {
return fileWithId.getFileStatus().getPath();
}
- void schedule() throws IOException {
- if(locations.size() == 1 && file.getLen() < context.maxSize) {
- String[] hosts = locations.firstEntry().getValue().getHosts();
- synchronized (context.splits) {
- context.splits.add(new OrcSplit(file.getPath(), fileWithId.getFileId(), 0,
- file.getLen(), hosts, fileMetaInfo, isOriginal, hasBase, deltas));
- }
- } else {
- // if it requires a compute task
- context.schedule(this);
- }
- }
-
@Override
public String toString() {
- return "splitter(" + file.getPath() + ")";
+ return "splitter(" + fileWithId.getFileStatus().getPath() + ")";
}
/**
@@ -742,7 +791,7 @@ public class OrcInputFormat implements
* @param fileMetaInfo file metadata from footer and postscript
* @throws IOException
*/
- void createSplit(long offset, long length,
+ OrcSplit createSplit(long offset, long length,
FileMetaInfo fileMetaInfo) throws IOException {
String[] hosts;
Map.Entry<Long, BlockLocation> startEntry = locations.floorEntry(offset);
@@ -774,7 +823,7 @@ public class OrcInputFormat implements
maxSize = Math.max(maxSize, val.get());
}
} else {
- throw new IOException("File " + file.getPath().toString() +
+ throw new IOException("File " + fileWithId.getFileStatus().getPath().toString() +
" should have had overlap on block starting at " + block.getOffset());
}
}
@@ -796,10 +845,8 @@ public class OrcInputFormat implements
hosts = new String[hostList.size()];
hostList.toArray(hosts);
}
- synchronized (context.splits) {
- context.splits.add(new OrcSplit(file.getPath(), fileWithId.getFileId(), offset, length,
- hosts, fileMetaInfo, isOriginal, hasBase, deltas));
- }
+ return new OrcSplit(fileWithId.getFileStatus().getPath(), null, offset, length, hosts, fileMetaInfo,
+ isOriginal, hasBase, deltas, projColsUncompressedSize);
}
/**
@@ -807,145 +854,123 @@ public class OrcInputFormat implements
* block size and the configured minimum and maximum sizes.
*/
@Override
- public void run() {
- try {
- populateAndCacheStripeDetails();
-
- // figure out which stripes we need to read
- boolean[] includeStripe = null;
- // we can't eliminate stripes if there are deltas because the
- // deltas may change the rows making them match the predicate.
- if (deltas.isEmpty()) {
- Reader.Options options = new Reader.Options();
- options.include(genIncludedColumns(types, context.conf, isOriginal));
- setSearchArgument(options, types, context.conf, isOriginal);
- // only do split pruning if HIVE-8732 has been fixed in the writer
- if (options.getSearchArgument() != null &&
- writerVersion != OrcFile.WriterVersion.ORIGINAL) {
- SearchArgument sarg = options.getSearchArgument();
- List<PredicateLeaf> sargLeaves = sarg.getLeaves();
- int[] filterColumns = RecordReaderImpl.mapSargColumns(sargLeaves,
- options.getColumnNames(), getRootColumn(isOriginal));
-
- if (stripeStats != null) {
- // eliminate stripes that doesn't satisfy the predicate condition
- includeStripe = new boolean[stripes.size()];
- for(int i=0; i < stripes.size(); ++i) {
- includeStripe[i] = (i >= stripeStats.size()) ||
- isStripeSatisfyPredicate(stripeStats.get(i), sarg,
- filterColumns);
- if (LOG.isDebugEnabled() && !includeStripe[i]) {
- LOG.debug("Eliminating ORC stripe-" + i + " of file '" +
- file.getPath() + "' as it did not satisfy " +
- "predicate condition.");
- }
+ public List<OrcSplit> call() throws IOException {
+ populateAndCacheStripeDetails();
+ List<OrcSplit> splits = Lists.newArrayList();
+
+ // figure out which stripes we need to read
+ boolean[] includeStripe = null;
+ // we can't eliminate stripes if there are deltas because the
+ // deltas may change the rows making them match the predicate.
+ if (deltas.isEmpty()) {
+ Reader.Options options = new Reader.Options();
+ options.include(genIncludedColumns(types, context.conf, isOriginal));
+ setSearchArgument(options, types, context.conf, isOriginal);
+ // only do split pruning if HIVE-8732 has been fixed in the writer
+ if (options.getSearchArgument() != null &&
+ writerVersion != OrcFile.WriterVersion.ORIGINAL) {
+ SearchArgument sarg = options.getSearchArgument();
+ List<PredicateLeaf> sargLeaves = sarg.getLeaves();
+ int[] filterColumns = RecordReaderImpl.mapSargColumns(sargLeaves,
+ options.getColumnNames(), getRootColumn(isOriginal));
+
+ if (stripeStats != null) {
+ // eliminate stripes that doesn't satisfy the predicate condition
+ includeStripe = new boolean[stripes.size()];
+ for (int i = 0; i < stripes.size(); ++i) {
+ includeStripe[i] = (i >= stripeStats.size()) ||
+ isStripeSatisfyPredicate(stripeStats.get(i), sarg,
+ filterColumns);
+ if (LOG.isDebugEnabled() && !includeStripe[i]) {
+ LOG.debug("Eliminating ORC stripe-" + i + " of file '" +
+ fileWithId.getFileStatus().getPath() + "' as it did not satisfy " +
+ "predicate condition.");
}
}
}
}
+ }
- // if we didn't have predicate pushdown, read everything
- if (includeStripe == null) {
- includeStripe = new boolean[stripes.size()];
- Arrays.fill(includeStripe, true);
- }
-
- long currentOffset = -1;
- long currentLength = 0;
- int idx = -1;
- for(StripeInformation stripe: stripes) {
- idx++;
-
- if (!includeStripe[idx]) {
- // create split for the previous unfinished stripe
- if (currentOffset != -1) {
- createSplit(currentOffset, currentLength, fileMetaInfo);
- currentOffset = -1;
- }
- continue;
- }
-
- // if we are working on a stripe, over the min stripe size, and
- // crossed a block boundary, cut the input split here.
- if (currentOffset != -1 && currentLength > context.minSize &&
- (currentOffset / blockSize != stripe.getOffset() / blockSize)) {
- createSplit(currentOffset, currentLength, fileMetaInfo);
- currentOffset = -1;
- }
- // if we aren't building a split, start a new one.
- if (currentOffset == -1) {
- currentOffset = stripe.getOffset();
- currentLength = stripe.getLength();
- } else {
- currentLength =
- (stripe.getOffset() + stripe.getLength()) - currentOffset;
- }
- if (currentLength >= context.maxSize) {
- createSplit(currentOffset, currentLength, fileMetaInfo);
+ // if we didn't have predicate pushdown, read everything
+ if (includeStripe == null) {
+ includeStripe = new boolean[stripes.size()];
+ Arrays.fill(includeStripe, true);
+ }
+
+ long currentOffset = -1;
+ long currentLength = 0;
+ int idx = -1;
+ for (StripeInformation stripe : stripes) {
+ idx++;
+
+ if (!includeStripe[idx]) {
+ // create split for the previous unfinished stripe
+ if (currentOffset != -1) {
+ splits.add(createSplit(currentOffset, currentLength, fileMetaInfo));
currentOffset = -1;
}
+ continue;
}
- if (currentOffset != -1) {
- createSplit(currentOffset, currentLength, fileMetaInfo);
- }
- } catch (Throwable th) {
- if (!(th instanceof IOException)) {
- LOG.error("Unexpected Exception", th);
- }
- synchronized (context.errors) {
- context.errors.add(th);
+
+ // if we are working on a stripe, over the min stripe size, and
+ // crossed a block boundary, cut the input split here.
+ if (currentOffset != -1 && currentLength > context.minSize &&
+ (currentOffset / blockSize != stripe.getOffset() / blockSize)) {
+ splits.add(createSplit(currentOffset, currentLength, fileMetaInfo));
+ currentOffset = -1;
+ }
+ // if we aren't building a split, start a new one.
+ if (currentOffset == -1) {
+ currentOffset = stripe.getOffset();
+ currentLength = stripe.getLength();
+ } else {
+ currentLength =
+ (stripe.getOffset() + stripe.getLength()) - currentOffset;
}
- if (!(th instanceof IOException)) {
- context.notifyOnNonIOException(th);
+ if (currentLength >= context.maxSize) {
+ splits.add(createSplit(currentOffset, currentLength, fileMetaInfo));
+ currentOffset = -1;
}
- } finally {
- context.decrementSchedulers();
}
+ if (currentOffset != -1) {
+ splits.add(createSplit(currentOffset, currentLength, fileMetaInfo));
+ }
+
+ // add uncovered ACID delta splits
+ splits.addAll(deltaSplits);
+ return splits;
}
- private void populateAndCacheStripeDetails() {
- try {
- Reader orcReader;
- if (fileInfo != null) {
- stripes = fileInfo.stripeInfos;
- fileMetaInfo = fileInfo.fileMetaInfo;
- stripeStats = fileInfo.stripeStats;
- types = fileInfo.types;
- writerVersion = fileInfo.writerVersion;
- // For multiple runs, in case sendSplitsInFooter changes
- if (fileMetaInfo == null && context.footerInSplits) {
- orcReader = OrcFile.createReader(file.getPath(),
- OrcFile.readerOptions(context.conf).filesystem(fs));
- fileInfo.fileMetaInfo = ((ReaderImpl) orcReader).getFileMetaInfo();
- fileInfo.stripeStats = orcReader.getStripeStatistics();
- fileInfo.types = orcReader.getTypes();
- fileInfo.writerVersion = orcReader.getWriterVersion();
- }
- } else {
- orcReader = OrcFile.createReader(file.getPath(),
- OrcFile.readerOptions(context.conf).filesystem(fs));
- stripes = orcReader.getStripes();
- stripeStats = orcReader.getStripeStatistics();
- types = orcReader.getTypes();
- writerVersion = orcReader.getWriterVersion();
- fileMetaInfo = context.footerInSplits ?
- ((ReaderImpl) orcReader).getFileMetaInfo() : null;
- if (context.cacheStripeDetails) {
- // Populate into cache.
- Context.footerCache.put(file.getPath(),
- new FileInfo(file.getModificationTime(), file.getLen(), stripes,
- stripeStats, types, fileMetaInfo, writerVersion));
- }
- }
- } catch (Throwable th) {
- if (!(th instanceof IOException)) {
- LOG.error("Unexpected Exception", th);
- }
- synchronized (context.errors) {
- context.errors.add(th);
+ private void populateAndCacheStripeDetails() throws IOException {
+ Reader orcReader = OrcFile.createReader(fileWithId.getFileStatus().getPath(),
+ OrcFile.readerOptions(context.conf).filesystem(fs));
+ List<String> projCols = Arrays.asList(ColumnProjectionUtils.getReadColumnNames(context.conf));
+ projColsUncompressedSize = orcReader.getRawDataSizeOfColumns(projCols);
+ if (fileInfo != null) {
+ stripes = fileInfo.stripeInfos;
+ stripeStats = fileInfo.stripeStats;
+ fileMetaInfo = fileInfo.fileMetaInfo;
+ types = fileInfo.types;
+ writerVersion = fileInfo.writerVersion;
+ // For multiple runs, in case sendSplitsInFooter changes
+ if (fileMetaInfo == null && context.footerInSplits) {
+ fileInfo.fileMetaInfo = ((ReaderImpl) orcReader).getFileMetaInfo();
+ fileInfo.types = orcReader.getTypes();
+ fileInfo.writerVersion = orcReader.getWriterVersion();
}
- if (!(th instanceof IOException)) {
- context.notifyOnNonIOException(th);
+ } else {
+ stripes = orcReader.getStripes();
+ types = orcReader.getTypes();
+ writerVersion = orcReader.getWriterVersion();
+ stripeStats = orcReader.getStripeStatistics();
+ fileMetaInfo = context.footerInSplits ?
+ ((ReaderImpl) orcReader).getFileMetaInfo() : null;
+ if (context.cacheStripeDetails) {
+ // Populate into cache.
+ Context.footerCache.put(fileWithId.getFileStatus().getPath(),
+ new FileInfo(fileWithId.getFileStatus().getModificationTime(),
+ fileWithId.getFileStatus().getLen(), stripes,
+ stripeStats, types, fileMetaInfo, writerVersion));
}
}
}
@@ -978,29 +1003,64 @@ public class OrcInputFormat implements
// use threads to resolve directories into splits
Context context = new Context(conf);
boolean useFileIds = HiveConf.getBoolVar(conf, ConfVars.HIVE_ORC_INCLUDE_FILE_ID_IN_SPLITS);
+ List<OrcSplit> splits = Lists.newArrayList();
+ List<Future<?>> pathFutures = Lists.newArrayList();
+ List<Future<?>> splitFutures = Lists.newArrayList();
+
+ // multi-threaded file statuses and split strategy
for (Path dir : getInputPaths(conf)) {
FileSystem fs = dir.getFileSystem(conf);
- context.schedule(new FileGenerator(context, fs, dir, useFileIds));
+ FileGenerator fileGenerator = new FileGenerator(context, fs, dir, useFileIds);
+ pathFutures.add(context.threadPool.submit(fileGenerator));
}
- context.waitForTasks();
- // deal with exceptions
- if (!context.errors.isEmpty()) {
- List<IOException> errors =
- new ArrayList<IOException>(context.errors.size());
- for(Throwable th: context.errors) {
- if (th instanceof IOException) {
- errors.add((IOException) th);
+
+ // complete path futures and schedule split generation
+ try {
+ for (Future<?> pathFuture : pathFutures) {
+ SplitStrategy splitStrategy = (SplitStrategy) pathFuture.get();
+
+ if (isDebugEnabled) {
+ LOG.debug(splitStrategy);
+ }
+
+ if (splitStrategy instanceof ETLSplitStrategy) {
+ List<SplitInfo> splitInfos = splitStrategy.getSplits();
+ for (SplitInfo splitInfo : splitInfos) {
+ splitFutures.add(context.threadPool.submit(new SplitGenerator(splitInfo)));
+ }
} else {
- throw new RuntimeException("serious problem", th);
+ splits.addAll(splitStrategy.getSplits());
}
}
- throw new InvalidInputException(errors);
+
+ // complete split futures
+ for (Future<?> splitFuture : splitFutures) {
+ splits.addAll((Collection<? extends OrcSplit>) splitFuture.get());
+ }
+ } catch (Exception e) {
+ cancelFutures(pathFutures);
+ cancelFutures(splitFutures);
+ throw new RuntimeException("serious problem", e);
}
+
if (context.cacheStripeDetails) {
LOG.info("FooterCacheHitRatio: " + context.cacheHitCounter.get() + "/"
+ context.numFilesCounter.get());
}
- return context.splits;
+
+ if (isDebugEnabled) {
+ for (OrcSplit split : splits) {
+ LOG.debug(split + " projected_columns_uncompressed_size: "
+ + split.getProjectedColumnsUncompressedSize());
+ }
+ }
+ return splits;
+ }
+
+ private static void cancelFutures(List<Future<?>> futures) {
+ for (Future future : futures) {
+ future.cancel(true);
+ }
}
@Override
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcRawRecordMerger.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcRawRecordMerger.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcRawRecordMerger.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcRawRecordMerger.java Tue Apr 7 23:01:18 2015
@@ -460,7 +460,7 @@ public class OrcRawRecordMerger implemen
Path deltaFile = AcidUtils.createBucketFile(delta, bucket);
FileSystem fs = deltaFile.getFileSystem(conf);
long length = getLastFlushLength(fs, deltaFile);
- if (fs.exists(deltaFile) && length != -1) {
+ if (length != -1 && fs.exists(deltaFile)) {
Reader deltaReader = OrcFile.createReader(deltaFile,
OrcFile.readerOptions(conf).maxLength(length));
ReaderPair deltaPair = new ReaderPair(key, deltaReader, bucket, minKey,
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSplit.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSplit.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSplit.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSplit.java Tue Apr 7 23:01:18 2015
@@ -48,6 +48,7 @@ public class OrcSplit extends FileSplit
private final List<Long> deltas = new ArrayList<Long>();
private OrcFile.WriterVersion writerVersion;
private transient Long fileId;
+ private long projColsUncompressedSize;
static final int HAS_FILEID_FLAG = 8;
static final int BASE_FLAG = 4;
@@ -63,7 +64,7 @@ public class OrcSplit extends FileSplit
public OrcSplit(Path path, Long fileId, long offset, long length, String[] hosts,
FileMetaInfo fileMetaInfo, boolean isOriginal, boolean hasBase,
- List<Long> deltas) {
+ List<Long> deltas, long projectedDataSize) {
super(path, offset, length, hosts);
// We could avoid serializing file ID and just replace the path with inode-based path.
// However, that breaks bunch of stuff because Hive later looks up things by split path.
@@ -73,6 +74,7 @@ public class OrcSplit extends FileSplit
this.isOriginal = isOriginal;
this.hasBase = hasBase;
this.deltas.addAll(deltas);
+ this.projColsUncompressedSize = projectedDataSize;
}
@Override
@@ -170,4 +172,9 @@ public class OrcSplit extends FileSplit
public Long getFileId() {
return fileId;
}
+
+ public long getProjectedColumnsUncompressedSize() {
+ return projColsUncompressedSize;
+ }
+
}
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/metadata/Hive.java Tue Apr 7 23:01:18 2015
@@ -2416,7 +2416,7 @@ private void constructOneLBLocationMap(F
List<List<Path[]>> result = new ArrayList<List<Path[]>>();
try {
- FileStatus destStatus = !replace && fs.exists(destf) ? fs.getFileStatus(destf) : null;
+ FileStatus destStatus = !replace ? FileUtils.getFileStatusOrNull(fs, destf) : null;
if (destStatus != null && !destStatus.isDir()) {
throw new HiveException("checkPaths: destination " + destf
+ " should be a directory");
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/StorageBasedAuthorizationProvider.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/StorageBasedAuthorizationProvider.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/StorageBasedAuthorizationProvider.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/StorageBasedAuthorizationProvider.java Tue Apr 7 23:01:18 2015
@@ -361,19 +361,22 @@ public class StorageBasedAuthorizationPr
final FileSystem fs = path.getFileSystem(conf);
- if (fs.exists(path)) {
- checkPermissions(fs, path, actions, authenticator.getUserName());
+ FileStatus pathStatus = FileUtils.getFileStatusOrNull(fs, path);
+ if (pathStatus != null) {
+ checkPermissions(fs, pathStatus, actions, authenticator.getUserName());
} else if (path.getParent() != null) {
// find the ancestor which exists to check its permissions
Path par = path.getParent();
+ FileStatus parStatus = null;
while (par != null) {
- if (fs.exists(par)) {
+ parStatus = FileUtils.getFileStatusOrNull(fs, par);
+ if (parStatus != null) {
break;
}
par = par.getParent();
}
- checkPermissions(fs, par, actions, authenticator.getUserName());
+ checkPermissions(fs, parStatus, actions, authenticator.getUserName());
}
}
@@ -382,18 +385,20 @@ public class StorageBasedAuthorizationPr
* does not exists, it returns.
*/
@SuppressWarnings("deprecation")
- protected static void checkPermissions(final FileSystem fs, final Path path,
+ protected static void checkPermissions(final FileSystem fs, final FileStatus stat,
final EnumSet<FsAction> actions, String user) throws IOException,
AccessControlException, HiveException {
- try {
- FileStatus stat = fs.getFileStatus(path);
- for (FsAction action : actions) {
- FileUtils.checkFileAccessWithImpersonation(fs, stat, action, user);
- }
- } catch (FileNotFoundException fnfe) {
+ if (stat == null) {
// File named by path doesn't exist; nothing to validate.
return;
+ }
+ FsAction checkActions = FsAction.NONE;
+ for (FsAction action : actions) {
+ checkActions = checkActions.or(action);
+ }
+ try {
+ FileUtils.checkFileAccessWithImpersonation(fs, stat, checkActions, user);
} catch (org.apache.hadoop.fs.permission.AccessControlException ace) {
// Older hadoop version will throw this @deprecated Exception.
throw accessControlException(ace);
Modified: hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/plugin/sqlstd/SQLAuthorizationUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/plugin/sqlstd/SQLAuthorizationUtils.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/plugin/sqlstd/SQLAuthorizationUtils.java (original)
+++ hive/branches/llap/ql/src/java/org/apache/hadoop/hive/ql/security/authorization/plugin/sqlstd/SQLAuthorizationUtils.java Tue Apr 7 23:01:18 2015
@@ -385,8 +385,7 @@ public class SQLAuthorizationUtils {
FileSystem fs;
try {
fs = FileSystem.get(filePath.toUri(), conf);
- Path path = FileUtils.getPathOrParentThatExists(fs, filePath);
- FileStatus fileStatus = fs.getFileStatus(path);
+ FileStatus fileStatus = FileUtils.getPathOrParentThatExists(fs, filePath);
if (FileUtils.isOwnerOfFileHierarchy(fs, fileStatus, userName)) {
availPrivs.addPrivilege(SQLPrivTypeGrant.OWNER_PRIV);
}
Modified: hive/branches/llap/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java (original)
+++ hive/branches/llap/ql/src/test/org/apache/hadoop/hive/ql/io/orc/TestInputOutputFormat.java Tue Apr 7 23:01:18 2015
@@ -62,10 +62,12 @@ import org.apache.hadoop.hive.ql.exec.ve
import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector;
import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch;
import org.apache.hadoop.hive.ql.io.AcidOutputFormat;
+import org.apache.hadoop.hive.ql.io.AcidUtils;
import org.apache.hadoop.hive.ql.io.CombineHiveInputFormat;
import org.apache.hadoop.hive.ql.io.HiveInputFormat;
import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
import org.apache.hadoop.hive.ql.io.InputFormatChecker;
+import org.apache.hadoop.hive.ql.io.orc.OrcInputFormat.SplitStrategy;
import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.hive.ql.io.sarg.SearchArgumentFactory;
@@ -98,7 +100,6 @@ import org.apache.hadoop.mapred.OutputFo
import org.apache.hadoop.mapred.RecordWriter;
import org.apache.hadoop.mapred.Reporter;
import org.apache.hadoop.util.Progressable;
-import org.apache.hadoop.util.StringUtils;
import org.junit.Before;
import org.junit.Rule;
import org.junit.Test;
@@ -394,22 +395,9 @@ public class TestInputOutputFormat {
OrcInputFormat.getInputPaths(conf));
}
- static class TestContext extends OrcInputFormat.Context {
- List<Runnable> queue = new ArrayList<Runnable>();
-
- TestContext(Configuration conf) {
- super(conf);
- }
-
- @Override
- public void schedule(Runnable runnable) {
- queue.add(runnable);
- }
- }
-
@Test
public void testFileGenerator() throws Exception {
- TestContext context = new TestContext(conf);
+ OrcInputFormat.Context context = new OrcInputFormat.Context(conf);
MockFileSystem fs = new MockFileSystem(conf,
new MockFile("mock:/a/b/part-00", 1000, new byte[0]),
new MockFile("mock:/a/b/part-01", 1000, new byte[0]),
@@ -419,21 +407,22 @@ public class TestInputOutputFormat {
OrcInputFormat.FileGenerator gen =
new OrcInputFormat.FileGenerator(context, fs,
new MockPath(fs, "mock:/a/b"), false);
- gen.run();
- if (context.getErrors().size() > 0) {
- for(Throwable th: context.getErrors()) {
- System.out.println(StringUtils.stringifyException(th));
- }
- throw new IOException("Errors during file generation");
- }
- assertEquals(-1, context.getSchedulers());
- assertEquals(3, context.queue.size());
- assertEquals(new Path("mock:/a/b/part-00"),
- ((OrcInputFormat.SplitGenerator) context.queue.get(0)).getPath());
- assertEquals(new Path("mock:/a/b/part-01"),
- ((OrcInputFormat.SplitGenerator) context.queue.get(1)).getPath());
- assertEquals(new Path("mock:/a/b/part-04"),
- ((OrcInputFormat.SplitGenerator) context.queue.get(2)).getPath());
+ SplitStrategy splitStrategy = gen.call();
+ assertEquals(true, splitStrategy instanceof OrcInputFormat.BISplitStrategy);
+
+ conf.set("mapreduce.input.fileinputformat.split.maxsize", "500");
+ context = new OrcInputFormat.Context(conf);
+ fs = new MockFileSystem(conf,
+ new MockFile("mock:/a/b/part-00", 1000, new byte[1000]),
+ new MockFile("mock:/a/b/part-01", 1000, new byte[1000]),
+ new MockFile("mock:/a/b/_part-02", 1000, new byte[1000]),
+ new MockFile("mock:/a/b/.part-03", 1000, new byte[1000]),
+ new MockFile("mock:/a/b/part-04", 1000, new byte[1000]));
+ gen = new OrcInputFormat.FileGenerator(context, fs,
+ new MockPath(fs, "mock:/a/b"), false);
+ splitStrategy = gen.call();
+ assertEquals(true, splitStrategy instanceof OrcInputFormat.ETLSplitStrategy);
+
}
public static class MockBlock {
@@ -848,11 +837,10 @@ public class TestInputOutputFormat {
new MockBlock("host5-1", "host5-2", "host5-3")));
OrcInputFormat.Context context = new OrcInputFormat.Context(conf);
OrcInputFormat.SplitGenerator splitter =
- new OrcInputFormat.SplitGenerator(context, fs,
- fs.getFileStatus(new Path("/a/file")), null, true,
- new ArrayList<Long>(), true);
- splitter.createSplit(0, 200, null);
- OrcSplit result = context.getResult(-1);
+ new OrcInputFormat.SplitGenerator(new OrcInputFormat.SplitInfo(context, fs,
+ AcidUtils.createOriginalObj(null, fs.getFileStatus(new Path("/a/file"))), null, true,
+ new ArrayList<Long>(), true, null, null));
+ OrcSplit result = splitter.createSplit(0, 200, null);
assertEquals(0, result.getStart());
assertEquals(200, result.getLength());
assertEquals("mock:/a/file", result.getPath().toString());
@@ -861,15 +849,13 @@ public class TestInputOutputFormat {
assertEquals("host1-1", locs[0]);
assertEquals("host1-2", locs[1]);
assertEquals("host1-3", locs[2]);
- splitter.createSplit(500, 600, null);
- result = context.getResult(-1);
+ result = splitter.createSplit(500, 600, null);
locs = result.getLocations();
assertEquals(3, locs.length);
assertEquals("host2-1", locs[0]);
assertEquals("host0", locs[1]);
assertEquals("host2-3", locs[2]);
- splitter.createSplit(0, 2500, null);
- result = context.getResult(-1);
+ result = splitter.createSplit(0, 2500, null);
locs = result.getLocations();
assertEquals(1, locs.length);
assertEquals("host0", locs[0]);
@@ -892,48 +878,36 @@ public class TestInputOutputFormat {
conf.setInt(OrcInputFormat.MIN_SPLIT_SIZE, 200);
OrcInputFormat.Context context = new OrcInputFormat.Context(conf);
OrcInputFormat.SplitGenerator splitter =
- new OrcInputFormat.SplitGenerator(context, fs,
- fs.getFileStatus(new Path("/a/file")), null, true,
- new ArrayList<Long>(), true);
- splitter.run();
- if (context.getErrors().size() > 0) {
- for(Throwable th: context.getErrors()) {
- System.out.println(StringUtils.stringifyException(th));
- }
- throw new IOException("Errors during splitting");
- }
- OrcSplit result = context.getResult(0);
+ new OrcInputFormat.SplitGenerator(new OrcInputFormat.SplitInfo(context, fs,
+ AcidUtils.createOriginalObj(null, fs.getFileStatus(new Path("/a/file"))), null, true,
+ new ArrayList<Long>(), true, null, null));
+ List<OrcSplit> results = splitter.call();
+ OrcSplit result = results.get(0);
assertEquals(3, result.getStart());
assertEquals(497, result.getLength());
- result = context.getResult(1);
+ result = results.get(1);
assertEquals(500, result.getStart());
assertEquals(600, result.getLength());
- result = context.getResult(2);
+ result = results.get(2);
assertEquals(1100, result.getStart());
assertEquals(400, result.getLength());
- result = context.getResult(3);
+ result = results.get(3);
assertEquals(1500, result.getStart());
assertEquals(300, result.getLength());
- result = context.getResult(4);
+ result = results.get(4);
assertEquals(1800, result.getStart());
assertEquals(200, result.getLength());
// test min = 0, max = 0 generates each stripe
conf.setInt(OrcInputFormat.MIN_SPLIT_SIZE, 0);
conf.setInt(OrcInputFormat.MAX_SPLIT_SIZE, 0);
context = new OrcInputFormat.Context(conf);
- splitter = new OrcInputFormat.SplitGenerator(context, fs,
- fs.getFileStatus(new Path("/a/file")), null, true, new ArrayList<Long>(),
- true);
- splitter.run();
- if (context.getErrors().size() > 0) {
- for(Throwable th: context.getErrors()) {
- System.out.println(StringUtils.stringifyException(th));
- }
- throw new IOException("Errors during splitting");
- }
+ splitter = new OrcInputFormat.SplitGenerator(new OrcInputFormat.SplitInfo(context, fs,
+ AcidUtils.createOriginalObj(null, fs.getFileStatus(new Path("/a/file"))), null, true,
+ new ArrayList<Long>(), true, null, null));
+ results = splitter.call();
for(int i=0; i < stripeSizes.length; ++i) {
assertEquals("checking stripe " + i + " size",
- stripeSizes[i], context.getResult(i).getLength());
+ stripeSizes[i], results.get(i).getLength());
}
}
Modified: hive/branches/llap/ql/src/test/queries/clientpositive/groupby3_map.q
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/test/queries/clientpositive/groupby3_map.q?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/test/queries/clientpositive/groupby3_map.q (original)
+++ hive/branches/llap/ql/src/test/queries/clientpositive/groupby3_map.q Tue Apr 7 23:01:18 2015
@@ -29,6 +29,16 @@ INSERT OVERWRITE TABLE dest1 SELECT
variance(substr(src.value,5)),
var_samp(substr(src.value,5));
-SELECT dest1.* FROM dest1;
+SELECT
+c1,
+c2,
+round(c3, 11) c3,
+c4,
+c5,
+round(c6, 11) c6,
+round(c7, 11) c7,
+round(c8, 5) c8,
+round(c9, 9) c9
+FROM dest1;
Modified: hive/branches/llap/ql/src/test/results/clientpositive/groupby3_map.q.out
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/test/results/clientpositive/groupby3_map.q.out?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/test/results/clientpositive/groupby3_map.q.out (original)
+++ hive/branches/llap/ql/src/test/results/clientpositive/groupby3_map.q.out Tue Apr 7 23:01:18 2015
@@ -128,12 +128,32 @@ POSTHOOK: Lineage: dest1.c6 EXPRESSION [
POSTHOOK: Lineage: dest1.c7 EXPRESSION [(src)src.FieldSchema(name:value, type:string, comment:default), ]
POSTHOOK: Lineage: dest1.c8 EXPRESSION [(src)src.FieldSchema(name:value, type:string, comment:default), ]
POSTHOOK: Lineage: dest1.c9 EXPRESSION [(src)src.FieldSchema(name:value, type:string, comment:default), ]
-PREHOOK: query: SELECT dest1.* FROM dest1
+PREHOOK: query: SELECT
+c1,
+c2,
+round(c3, 11) c3,
+c4,
+c5,
+round(c6, 11) c6,
+round(c7, 11) c7,
+round(c8, 5) c8,
+round(c9, 9) c9
+FROM dest1
PREHOOK: type: QUERY
PREHOOK: Input: default@dest1
#### A masked pattern was here ####
-POSTHOOK: query: SELECT dest1.* FROM dest1
+POSTHOOK: query: SELECT
+c1,
+c2,
+round(c3, 11) c3,
+c4,
+c5,
+round(c6, 11) c6,
+round(c7, 11) c7,
+round(c8, 5) c8,
+round(c9, 9) c9
+FROM dest1
POSTHOOK: type: QUERY
POSTHOOK: Input: default@dest1
#### A masked pattern was here ####
-130091.0 260.182 256.10355987055016 98.0 0.0 142.9268095075238 143.06995106518906 20428.072876 20469.01089779559
+130091.0 260.182 256.10355987055 98.0 0.0 142.92680950752 143.06995106519 20428.07288 20469.010897796
Modified: hive/branches/llap/ql/src/test/results/clientpositive/spark/groupby3_map.q.out
URL: http://svn.apache.org/viewvc/hive/branches/llap/ql/src/test/results/clientpositive/spark/groupby3_map.q.out?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/ql/src/test/results/clientpositive/spark/groupby3_map.q.out (original)
+++ hive/branches/llap/ql/src/test/results/clientpositive/spark/groupby3_map.q.out Tue Apr 7 23:01:18 2015
@@ -134,12 +134,32 @@ POSTHOOK: Lineage: dest1.c6 EXPRESSION [
POSTHOOK: Lineage: dest1.c7 EXPRESSION [(src)src.FieldSchema(name:value, type:string, comment:default), ]
POSTHOOK: Lineage: dest1.c8 EXPRESSION [(src)src.FieldSchema(name:value, type:string, comment:default), ]
POSTHOOK: Lineage: dest1.c9 EXPRESSION [(src)src.FieldSchema(name:value, type:string, comment:default), ]
-PREHOOK: query: SELECT dest1.* FROM dest1
+PREHOOK: query: SELECT
+c1,
+c2,
+round(c3, 11) c3,
+c4,
+c5,
+round(c6, 11) c6,
+round(c7, 11) c7,
+round(c8, 5) c8,
+round(c9, 9) c9
+FROM dest1
PREHOOK: type: QUERY
PREHOOK: Input: default@dest1
#### A masked pattern was here ####
-POSTHOOK: query: SELECT dest1.* FROM dest1
+POSTHOOK: query: SELECT
+c1,
+c2,
+round(c3, 11) c3,
+c4,
+c5,
+round(c6, 11) c6,
+round(c7, 11) c7,
+round(c8, 5) c8,
+round(c9, 9) c9
+FROM dest1
POSTHOOK: type: QUERY
POSTHOOK: Input: default@dest1
#### A masked pattern was here ####
-130091.0 260.182 256.10355987055016 98.0 0.0 142.9268095075238 143.06995106518906 20428.072876 20469.01089779559
+130091.0 260.182 256.10355987055 98.0 0.0 142.92680950752 143.06995106519 20428.07288 20469.010897796
Modified: hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/ColumnProjectionUtils.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/ColumnProjectionUtils.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/ColumnProjectionUtils.java (original)
+++ hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/ColumnProjectionUtils.java Tue Apr 7 23:01:18 2015
@@ -152,9 +152,11 @@ public final class ColumnProjectionUtils
}
public static String[] getReadColumnNames(Configuration conf) {
- String names = conf.get(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR);
- if (names == null) return null;
- return names.split(",");
+ String colNames = conf.get(READ_COLUMN_NAMES_CONF_STR, READ_COLUMN_IDS_CONF_STR_DEFAULT);
+ if (colNames != null && !colNames.isEmpty()) {
+ return colNames.split(",");
+ }
+ return new String[] {};
}
private static void setReadColumnIDConf(Configuration conf, String id) {
Modified: hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/avro/AvroDeserializer.java
URL: http://svn.apache.org/viewvc/hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/avro/AvroDeserializer.java?rev=1671970&r1=1671969&r2=1671970&view=diff
==============================================================================
--- hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/avro/AvroDeserializer.java (original)
+++ hive/branches/llap/serde/src/java/org/apache/hadoop/hive/serde2/avro/AvroDeserializer.java Tue Apr 7 23:01:18 2015
@@ -189,7 +189,7 @@ class AvroDeserializer {
String columnName = columnNames.get(i);
Object datum = record.get(columnName);
Schema datumSchema = record.getSchema().getField(columnName).schema();
- Schema.Field field = fileSchema.getField(columnName);
+ Schema.Field field = AvroSerdeUtils.isNullableType(fileSchema)?AvroSerdeUtils.getOtherTypeFromNullableType(fileSchema).getField(columnName):fileSchema.getField(columnName);
objectRow.add(worker(datum, field == null ? null : field.schema(), datumSchema, columnType));
}