You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@hive.apache.org by gu...@apache.org on 2013/08/26 23:42:21 UTC
svn commit: r1517707 [6/17] - in /hive/branches/tez: ./
beeline/src/java/org/apache/hive/beeline/ bin/ bin/ext/
common/src/java/org/apache/hadoop/hive/common/
common/src/java/org/apache/hadoop/hive/conf/ conf/
contrib/src/java/org/apache/hadoop/hive/co...
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/exec/persistence/RowContainer.java Mon Aug 26 21:42:12 2013
@@ -30,8 +30,8 @@ import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalFileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
-import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.exec.FileSinkOperator.RecordWriter;
+import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.io.HiveFileFormatUtils;
import org.apache.hadoop.hive.ql.io.HiveOutputFormat;
import org.apache.hadoop.hive.ql.metadata.HiveException;
@@ -71,18 +71,18 @@ import org.apache.hadoop.util.Reflection
* reading.
*
*/
-public class RowContainer<Row extends List<Object>> extends AbstractRowContainer<Row> {
+public class RowContainer<ROW extends List<Object>> extends AbstractRowContainer<ROW> {
protected static Log LOG = LogFactory.getLog(RowContainer.class);
// max # of rows can be put into one block
private static final int BLOCKSIZE = 25000;
- private Row[] currentWriteBlock; // the last block that add() should append to
- private Row[] currentReadBlock; // the current block where the cursor is in
+ private ROW[] currentWriteBlock; // the last block that add() should append to
+ private ROW[] currentReadBlock; // the current block where the cursor is in
// since currentReadBlock may assigned to currentWriteBlock, we need to store
// original read block
- private Row[] firstReadBlockPointer;
+ private ROW[] firstReadBlockPointer;
private int blockSize; // number of objects in the block before it is spilled
// to disk
private int numFlushedBlocks; // total # of blocks
@@ -108,7 +108,7 @@ public class RowContainer<Row extends Li
RecordWriter rw = null;
InputFormat<WritableComparable, Writable> inputFormat = null;
InputSplit[] inputSplits = null;
- private Row dummyRow = null;
+ private ROW dummyRow = null;
private final Reporter reporter;
Writable val = null; // cached to use serialize data
@@ -130,7 +130,7 @@ public class RowContainer<Row extends Li
this.addCursor = 0;
this.numFlushedBlocks = 0;
this.tmpFile = null;
- this.currentWriteBlock = (Row[]) new ArrayList[blockSize];
+ this.currentWriteBlock = (ROW[]) new ArrayList[blockSize];
this.currentReadBlock = this.currentWriteBlock;
this.firstReadBlockPointer = currentReadBlock;
this.serde = null;
@@ -142,7 +142,7 @@ public class RowContainer<Row extends Li
this.reporter = reporter;
}
}
-
+
private JobConf getLocalFSJobConfClone(Configuration jc) {
if (this.jobCloneUsingLocalFs == null) {
this.jobCloneUsingLocalFs = new JobConf(jc);
@@ -158,13 +158,13 @@ public class RowContainer<Row extends Li
}
@Override
- public void add(Row t) throws HiveException {
+ public void add(ROW t) throws HiveException {
if (this.tblDesc != null) {
- if (addCursor >= blockSize) { // spill the current block to tmp file
+ if (willSpill()) { // spill the current block to tmp file
spillBlock(currentWriteBlock, addCursor);
addCursor = 0;
if (numFlushedBlocks == 1) {
- currentWriteBlock = (Row[]) new ArrayList[blockSize];
+ currentWriteBlock = (ROW[]) new ArrayList[blockSize];
}
}
currentWriteBlock[addCursor++] = t;
@@ -178,7 +178,7 @@ public class RowContainer<Row extends Li
}
@Override
- public Row first() throws HiveException {
+ public ROW first() throws HiveException {
if (size == 0) {
return null;
}
@@ -221,10 +221,10 @@ public class RowContainer<Row extends Li
localJc, reporter);
currentSplitPointer++;
- nextBlock();
+ nextBlock(0);
}
// we are guaranteed that we can get data here (since 'size' is not zero)
- Row ret = currentReadBlock[itrCursor++];
+ ROW ret = currentReadBlock[itrCursor++];
removeKeys(ret);
return ret;
} catch (Exception e) {
@@ -234,7 +234,7 @@ public class RowContainer<Row extends Li
}
@Override
- public Row next() throws HiveException {
+ public ROW next() throws HiveException {
if (!firstCalled) {
throw new RuntimeException("Call first() then call next().");
@@ -252,19 +252,16 @@ public class RowContainer<Row extends Li
return null;
}
- Row ret;
+ ROW ret;
if (itrCursor < this.readBlockSize) {
ret = this.currentReadBlock[itrCursor++];
removeKeys(ret);
return ret;
} else {
- nextBlock();
+ nextBlock(0);
if (this.readBlockSize == 0) {
if (currentWriteBlock != null && currentReadBlock != currentWriteBlock) {
- this.itrCursor = 0;
- this.readBlockSize = this.addCursor;
- this.firstReadBlockPointer = this.currentReadBlock;
- currentReadBlock = currentWriteBlock;
+ setWriteBlockAsReadBlock();
} else {
return null;
}
@@ -273,7 +270,7 @@ public class RowContainer<Row extends Li
}
}
- private void removeKeys(Row ret) {
+ private void removeKeys(ROW ret) {
if (this.keyObject != null && this.currentReadBlock != this.currentWriteBlock) {
int len = this.keyObject.size();
int rowSize = ((ArrayList) ret).size();
@@ -285,39 +282,10 @@ public class RowContainer<Row extends Li
private final ArrayList<Object> row = new ArrayList<Object>(2);
- private void spillBlock(Row[] block, int length) throws HiveException {
+ private void spillBlock(ROW[] block, int length) throws HiveException {
try {
if (tmpFile == null) {
-
- String suffix = ".tmp";
- if (this.keyObject != null) {
- suffix = "." + this.keyObject.toString() + suffix;
- }
-
- while (true) {
- parentFile = File.createTempFile("hive-rowcontainer", "");
- boolean success = parentFile.delete() && parentFile.mkdir();
- if (success) {
- break;
- }
- LOG.debug("retry creating tmp row-container directory...");
- }
-
- tmpFile = File.createTempFile("RowContainer", suffix, parentFile);
- LOG.info("RowContainer created temp file " + tmpFile.getAbsolutePath());
- // Delete the temp file if the JVM terminate normally through Hadoop job
- // kill command.
- // Caveat: it won't be deleted if JVM is killed by 'kill -9'.
- parentFile.deleteOnExit();
- tmpFile.deleteOnExit();
-
- // rFile = new RandomAccessFile(tmpFile, "rw");
- HiveOutputFormat<?, ?> hiveOutputFormat = tblDesc.getOutputFileFormatClass().newInstance();
- tempOutPath = new Path(tmpFile.toString());
- JobConf localJc = getLocalFSJobConfClone(jc);
- rw = HiveFileFormatUtils.getRecordWriter(this.jobCloneUsingLocalFs,
- hiveOutputFormat, serde.getSerializedClass(), false,
- tblDesc.getProperties(), tempOutPath, reporter);
+ setupWriter();
} else if (rw == null) {
throw new HiveException("RowContainer has already been closed for writing.");
}
@@ -329,14 +297,14 @@ public class RowContainer<Row extends Li
if (this.keyObject != null) {
row.set(1, this.keyObject);
for (int i = 0; i < length; ++i) {
- Row currentValRow = block[i];
+ ROW currentValRow = block[i];
row.set(0, currentValRow);
Writable outVal = serde.serialize(row, standardOI);
rw.write(outVal);
}
} else {
for (int i = 0; i < length; ++i) {
- Row currentValRow = block[i];
+ ROW currentValRow = block[i];
Writable outVal = serde.serialize(currentValRow, standardOI);
rw.write(outVal);
}
@@ -350,6 +318,9 @@ public class RowContainer<Row extends Li
} catch (Exception e) {
clear();
LOG.error(e.toString(), e);
+ if ( e instanceof HiveException ) {
+ throw (HiveException) e;
+ }
throw new HiveException(e);
}
}
@@ -364,7 +335,7 @@ public class RowContainer<Row extends Li
return size;
}
- private boolean nextBlock() throws HiveException {
+ protected boolean nextBlock(int readIntoOffset) throws HiveException {
itrCursor = 0;
this.readBlockSize = 0;
if (this.numFlushedBlocks == 0) {
@@ -376,13 +347,13 @@ public class RowContainer<Row extends Li
val = serde.getSerializedClass().newInstance();
}
boolean nextSplit = true;
- int i = 0;
+ int i = readIntoOffset;
if (rr != null) {
Object key = rr.createKey();
while (i < this.currentReadBlock.length && rr.next(key, val)) {
nextSplit = false;
- this.currentReadBlock[i++] = (Row) ObjectInspectorUtils.copyToStandardObject(serde
+ this.currentReadBlock[i++] = (ROW) ObjectInspectorUtils.copyToStandardObject(serde
.deserialize(val), serde.getObjectInspector(), ObjectInspectorCopyOption.WRITABLE);
}
}
@@ -393,7 +364,7 @@ public class RowContainer<Row extends Li
rr = inputFormat.getRecordReader(inputSplits[currentSplitPointer], jobCloneUsingLocalFs,
reporter);
currentSplitPointer++;
- return nextBlock();
+ return nextBlock(0);
}
this.readBlockSize = i;
@@ -504,4 +475,119 @@ public class RowContainer<Row extends Li
this.tblDesc = tblDesc;
}
+ protected int getAddCursor() {
+ return addCursor;
+ }
+
+ protected final boolean willSpill() {
+ return addCursor >= blockSize;
+ }
+
+ protected int getBlockSize() {
+ return blockSize;
+ }
+
+ protected void setupWriter() throws HiveException {
+ try {
+
+ if ( tmpFile != null ) {
+ return;
+ }
+
+ String suffix = ".tmp";
+ if (this.keyObject != null) {
+ suffix = "." + this.keyObject.toString() + suffix;
+ }
+
+ while (true) {
+ parentFile = File.createTempFile("hive-rowcontainer", "");
+ boolean success = parentFile.delete() && parentFile.mkdir();
+ if (success) {
+ break;
+ }
+ LOG.debug("retry creating tmp row-container directory...");
+ }
+
+ tmpFile = File.createTempFile("RowContainer", suffix, parentFile);
+ LOG.info("RowContainer created temp file " + tmpFile.getAbsolutePath());
+ // Delete the temp file if the JVM terminate normally through Hadoop job
+ // kill command.
+ // Caveat: it won't be deleted if JVM is killed by 'kill -9'.
+ parentFile.deleteOnExit();
+ tmpFile.deleteOnExit();
+
+ // rFile = new RandomAccessFile(tmpFile, "rw");
+ HiveOutputFormat<?, ?> hiveOutputFormat = tblDesc.getOutputFileFormatClass().newInstance();
+ tempOutPath = new Path(tmpFile.toString());
+ JobConf localJc = getLocalFSJobConfClone(jc);
+ rw = HiveFileFormatUtils.getRecordWriter(this.jobCloneUsingLocalFs,
+ hiveOutputFormat, serde.getSerializedClass(), false,
+ tblDesc.getProperties(), tempOutPath, reporter);
+ } catch (Exception e) {
+ clear();
+ LOG.error(e.toString(), e);
+ throw new HiveException(e);
+ }
+
+ }
+
+ protected RecordWriter getRecordWriter() {
+ return rw;
+ }
+
+ protected InputSplit[] getInputSplits() {
+ return inputSplits;
+ }
+
+ protected boolean endOfCurrentReadBlock() {
+ if (tblDesc == null) {
+ return false;
+ }
+ return itrCursor >= this.readBlockSize;
+ }
+
+ protected int getCurrentReadBlockSize() {
+ return readBlockSize;
+ }
+
+ protected void setWriteBlockAsReadBlock() {
+ this.itrCursor = 0;
+ this.readBlockSize = this.addCursor;
+ this.firstReadBlockPointer = this.currentReadBlock;
+ currentReadBlock = currentWriteBlock;
+ }
+
+ protected org.apache.hadoop.mapred.RecordReader setReaderAtSplit(int splitNum)
+ throws IOException {
+ JobConf localJc = getLocalFSJobConfClone(jc);
+ currentSplitPointer = splitNum;
+ if ( rr != null ) {
+ rr.close();
+ }
+ // open record reader to read next split
+ rr = inputFormat.getRecordReader(inputSplits[currentSplitPointer], jobCloneUsingLocalFs,
+ reporter);
+ currentSplitPointer++;
+ return rr;
+ }
+
+ protected ROW getReadBlockRow(int rowOffset) {
+ itrCursor = rowOffset + 1;
+ return currentReadBlock[rowOffset];
+ }
+
+ protected void resetCurrentReadBlockToFirstReadBlock() {
+ currentReadBlock = firstReadBlockPointer;
+ }
+
+ protected void resetReadBlocks() {
+ this.currentReadBlock = this.currentWriteBlock;
+ this.firstReadBlockPointer = currentReadBlock;
+ }
+
+ protected void close() throws HiveException {
+ clear();
+ currentReadBlock = firstReadBlockPointer = currentWriteBlock = null;
+ }
+
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/history/HiveHistory.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/history/HiveHistory.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/history/HiveHistory.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/history/HiveHistory.java Mon Aug 26 21:42:12 2013
@@ -18,55 +18,21 @@
package org.apache.hadoop.hive.ql.history;
-import java.io.BufferedReader;
-import java.io.File;
-import java.io.FileInputStream;
-import java.io.FileNotFoundException;
import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.PrintWriter;
import java.io.Serializable;
import java.util.HashMap;
import java.util.Map;
-import java.util.Random;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.hive.conf.HiveConf;
+
import org.apache.hadoop.hive.ql.QueryPlan;
import org.apache.hadoop.hive.ql.exec.Task;
-import org.apache.hadoop.hive.ql.session.SessionState;
-import org.apache.hadoop.hive.ql.session.SessionState.LogHelper;
-import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.mapred.Counters;
-import org.apache.hadoop.mapred.Counters.Counter;
-import org.apache.hadoop.mapred.Counters.Group;
/**
- * HiveHistory.
- *
+ * HiveHistory. Logs information such as query, query plan, runtime statistics
+ * into a file.
+ * Each session uses a new object, which creates a new file.
*/
-public class HiveHistory {
-
- PrintWriter histStream; // History File stream
-
- String histFileName; // History file name
-
- private static final Log LOG = LogFactory.getLog("hive.ql.exec.HiveHistory");
-
- private LogHelper console;
-
- private Map<String, String> idToTableMap = null;
-
- // Job Hash Map
- private final HashMap<String, QueryInfo> queryInfoMap = new HashMap<String, QueryInfo>();
-
- // Task Hash Map
- private final HashMap<String, TaskInfo> taskInfoMap = new HashMap<String, TaskInfo>();
-
- private static final String DELIMITER = " ";
+public interface HiveHistory {
/**
* RecordTypes.
@@ -105,20 +71,11 @@ public class HiveHistory {
ROWS_INSERTED
};
- private static final String KEY = "(\\w+)";
- private static final String VALUE = "[[^\"]?]+"; // anything but a " in ""
- private static final String ROW_COUNT_PATTERN = "TABLE_ID_(\\d+)_ROWCOUNT";
-
- private static final Pattern pattern = Pattern.compile(KEY + "=" + "\""
- + VALUE + "\"");
-
- private static final Pattern rowCountPattern = Pattern.compile(ROW_COUNT_PATTERN);
-
- // temp buffer for parsed dataa
- private static Map<String, String> parseBuffer = new HashMap<String, String>();
-
/**
- * Listner interface Parser will call handle function for each record type.
+ * Listener interface.
+ * Parser will call handle function for each history record row, specifying
+ * the record type and its values
+ *
*/
public static interface Listener {
@@ -126,63 +83,6 @@ public class HiveHistory {
}
/**
- * Parses history file and calls call back functions.
- *
- * @param path
- * @param l
- * @throws IOException
- */
- public static void parseHiveHistory(String path, Listener l) throws IOException {
- FileInputStream fi = new FileInputStream(path);
- BufferedReader reader = new BufferedReader(new InputStreamReader(fi));
- try {
- String line = null;
- StringBuilder buf = new StringBuilder();
- while ((line = reader.readLine()) != null) {
- buf.append(line);
- // if it does not end with " then it is line continuation
- if (!line.trim().endsWith("\"")) {
- continue;
- }
- parseLine(buf.toString(), l);
- buf = new StringBuilder();
- }
- } finally {
- try {
- reader.close();
- } catch (IOException ex) {
- }
- }
- }
-
- /**
- * Parse a single line of history.
- *
- * @param line
- * @param l
- * @throws IOException
- */
- private static void parseLine(String line, Listener l) throws IOException {
- // extract the record type
- int idx = line.indexOf(' ');
- String recType = line.substring(0, idx);
- String data = line.substring(idx + 1, line.length());
-
- Matcher matcher = pattern.matcher(data);
-
- while (matcher.find()) {
- String tuple = matcher.group(0);
- String[] parts = tuple.split("=");
-
- parseBuffer.put(parts[0], parts[1].substring(1, parts[1].length() - 1));
- }
-
- l.handle(RecordTypes.valueOf(recType), parseBuffer);
-
- parseBuffer.clear();
- }
-
- /**
* Info.
*
*/
@@ -216,122 +116,25 @@ public class HiveHistory {
};
- /**
- * Construct HiveHistory object an open history log file.
- *
- * @param ss
- */
- public HiveHistory(SessionState ss) {
-
- try {
- console = new LogHelper(LOG);
- String conf_file_loc = ss.getConf().getVar(
- HiveConf.ConfVars.HIVEHISTORYFILELOC);
- if ((conf_file_loc == null) || conf_file_loc.length() == 0) {
- console.printError("No history file location given");
- return;
- }
-
- // Create directory
- File f = new File(conf_file_loc);
- if (!f.exists()) {
- if (!f.mkdirs()) {
- console.printError("Unable to create log directory " + conf_file_loc);
- return;
- }
- }
- Random randGen = new Random();
- do {
- histFileName = conf_file_loc + "/hive_job_log_" + ss.getSessionId() + "_"
- + Math.abs(randGen.nextInt()) + ".txt";
- } while (new File(histFileName).exists());
- console.printInfo("Hive history file=" + histFileName);
- histStream = new PrintWriter(histFileName);
-
- HashMap<String, String> hm = new HashMap<String, String>();
- hm.put(Keys.SESSION_ID.name(), ss.getSessionId());
- log(RecordTypes.SessionStart, hm);
- } catch (FileNotFoundException e) {
- console.printError("FAILED: Failed to open Query Log : " + histFileName
- + " " + e.getMessage(), "\n"
- + org.apache.hadoop.util.StringUtils.stringifyException(e));
- }
-
- }
/**
* @return historyFileName
*/
- public String getHistFileName() {
- return histFileName;
- }
-
- /**
- * Write the a history record to history file.
- *
- * @param rt
- * @param keyValMap
- */
- void log(RecordTypes rt, Map<String, String> keyValMap) {
-
- if (histStream == null) {
- return;
- }
-
- StringBuilder sb = new StringBuilder();
- sb.append(rt.name());
-
- for (Map.Entry<String, String> ent : keyValMap.entrySet()) {
-
- sb.append(DELIMITER);
- String key = ent.getKey();
- String val = ent.getValue();
- if(val != null) {
- val = val.replace(System.getProperty("line.separator"), " ");
- }
- sb.append(key + "=\"" + val + "\"");
-
- }
- sb.append(DELIMITER);
- sb.append(Keys.TIME.name() + "=\"" + System.currentTimeMillis() + "\"");
- histStream.println(sb);
- histStream.flush();
-
- }
+ public String getHistFileName();
/**
- * Called at the start of job Driver.execute().
+ * Called at the start of query execution in Driver.execute().
*/
- public void startQuery(String cmd, String id) {
- SessionState ss = SessionState.get();
- if (ss == null) {
- return;
- }
- QueryInfo ji = new QueryInfo();
-
- ji.hm.put(Keys.QUERY_ID.name(), id);
- ji.hm.put(Keys.QUERY_STRING.name(), cmd);
-
- queryInfoMap.put(id, ji);
-
- log(RecordTypes.QueryStart, ji.hm);
-
- }
+ public void startQuery(String cmd, String id);
/**
- * Used to set job status and other attributes of a job.
+ * Used to set query status and other attributes of a query
*
* @param queryId
* @param propName
* @param propValue
*/
- public void setQueryProperty(String queryId, Keys propName, String propValue) {
- QueryInfo ji = queryInfoMap.get(queryId);
- if (ji == null) {
- return;
- }
- ji.hm.put(propName.name(), propValue);
- }
+ public void setQueryProperty(String queryId, Keys propName, String propValue);
/**
* Used to set task properties.
@@ -341,14 +144,7 @@ public class HiveHistory {
* @param propValue
*/
public void setTaskProperty(String queryId, String taskId, Keys propName,
- String propValue) {
- String id = queryId + ":" + taskId;
- TaskInfo ti = taskInfoMap.get(id);
- if (ti == null) {
- return;
- }
- ti.hm.put(propName.name(), propValue);
- }
+ String propValue);
/**
* Serialize the task counters and set as a task property.
@@ -357,190 +153,62 @@ public class HiveHistory {
* @param taskId
* @param ctrs
*/
- public void setTaskCounters(String queryId, String taskId, Counters ctrs) {
- String id = queryId + ":" + taskId;
- QueryInfo ji = queryInfoMap.get(queryId);
- StringBuilder sb1 = new StringBuilder("");
- TaskInfo ti = taskInfoMap.get(id);
- if ((ti == null) || (ctrs == null)) {
- return;
- }
- StringBuilder sb = new StringBuilder("");
- try {
-
- boolean first = true;
- for (Group group : ctrs) {
- for (Counter counter : group) {
- if (first) {
- first = false;
- } else {
- sb.append(',');
- }
- sb.append(group.getDisplayName());
- sb.append('.');
- sb.append(counter.getDisplayName());
- sb.append(':');
- sb.append(counter.getCounter());
- String tab = getRowCountTableName(counter.getDisplayName());
- if (tab != null) {
- if (sb1.length() > 0) {
- sb1.append(",");
- }
- sb1.append(tab);
- sb1.append('~');
- sb1.append(counter.getCounter());
- ji.rowCountMap.put(tab, counter.getCounter());
-
- }
- }
- }
-
- } catch (Exception e) {
- LOG.warn(org.apache.hadoop.util.StringUtils.stringifyException(e));
- }
- if (sb1.length() > 0) {
- taskInfoMap.get(id).hm.put(Keys.ROWS_INSERTED.name(), sb1.toString());
- queryInfoMap.get(queryId).hm.put(Keys.ROWS_INSERTED.name(), sb1
- .toString());
- }
- if (sb.length() > 0) {
- taskInfoMap.get(id).hm.put(Keys.TASK_COUNTERS.name(), sb.toString());
- }
- }
+ public void setTaskCounters(String queryId, String taskId, Counters ctrs);
- public void printRowCount(String queryId) {
- QueryInfo ji = queryInfoMap.get(queryId);
- if (ji == null) {
- return;
- }
- for (String tab : ji.rowCountMap.keySet()) {
- console.printInfo(ji.rowCountMap.get(tab) + " Rows loaded to " + tab);
- }
- }
+ public void printRowCount(String queryId);
/**
- * Called at the end of Job. A Job is sql query.
+ * Called at the end of a query
*
* @param queryId
*/
- public void endQuery(String queryId) {
- QueryInfo ji = queryInfoMap.get(queryId);
- if (ji == null) {
- return;
- }
- log(RecordTypes.QueryEnd, ji.hm);
- queryInfoMap.remove(queryId);
- }
+ public void endQuery(String queryId);
/**
- * Called at the start of a task. Called by Driver.run() A Job can have
+ * Called at the start of a task. Called by Driver.run() A query can have
* multiple tasks. Tasks will have multiple operator.
*
* @param task
*/
public void startTask(String queryId, Task<? extends Serializable> task,
- String taskName) {
- SessionState ss = SessionState.get();
- if (ss == null) {
- return;
- }
- TaskInfo ti = new TaskInfo();
-
- ti.hm.put(Keys.QUERY_ID.name(), ss.getQueryId());
- ti.hm.put(Keys.TASK_ID.name(), task.getId());
- ti.hm.put(Keys.TASK_NAME.name(), taskName);
-
- String id = queryId + ":" + task.getId();
- taskInfoMap.put(id, ti);
-
- log(RecordTypes.TaskStart, ti.hm);
-
- }
+ String taskName);
/**
* Called at the end of a task.
*
* @param task
*/
- public void endTask(String queryId, Task<? extends Serializable> task) {
- String id = queryId + ":" + task.getId();
- TaskInfo ti = taskInfoMap.get(id);
-
- if (ti == null) {
- return;
- }
- log(RecordTypes.TaskEnd, ti.hm);
- taskInfoMap.remove(id);
- }
+ public void endTask(String queryId, Task<? extends Serializable> task);
/**
- * Called at the end of a task.
+ * Logs progress of a task if ConfVars.HIVE_LOG_INCREMENTAL_PLAN_PROGRESS is
+ * set to true
*
* @param task
*/
- public void progressTask(String queryId, Task<? extends Serializable> task) {
- String id = queryId + ":" + task.getId();
- TaskInfo ti = taskInfoMap.get(id);
- if (ti == null) {
- return;
- }
- log(RecordTypes.TaskProgress, ti.hm);
+ public void progressTask(String queryId, Task<? extends Serializable> task);
- }
/**
- * write out counters.
+ * Logs the current plan state
+ * @param plan
+ * @throws IOException
*/
- static ThreadLocal<Map<String,String>> ctrMapFactory =
- new ThreadLocal<Map<String, String>>() {
- @Override
- protected Map<String,String> initialValue() {
- return new HashMap<String,String>();
- }
- };
-
- public void logPlanProgress(QueryPlan plan) throws IOException {
- Map<String,String> ctrmap = ctrMapFactory.get();
- ctrmap.put("plan", plan.toString());
- log(RecordTypes.Counters, ctrmap);
- }
+ public void logPlanProgress(QueryPlan plan) throws IOException;
+
/**
- * Set the table to id map.
+ * Set the id to table name map
*
* @param map
*/
- public void setIdToTableMap(Map<String, String> map) {
- idToTableMap = map;
- }
+ public void setIdToTableMap(Map<String, String> map);
/**
- * Returns table name for the counter name.
- *
- * @param name
- * @return tableName
+ * Close the log file stream
*/
- String getRowCountTableName(String name) {
- if (idToTableMap == null) {
- return null;
- }
- Matcher m = rowCountPattern.matcher(name);
-
- if (m.find()) {
- String tuple = m.group(1);
- return idToTableMap.get(tuple);
- }
- return null;
+ public void closeStream();
- }
- public void closeStream() {
- IOUtils.cleanup(LOG, histStream);
- }
- @Override
- public void finalize() throws Throwable {
- closeStream();
- super.finalize();
- }
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/history/HiveHistoryViewer.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/history/HiveHistoryViewer.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/history/HiveHistoryViewer.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/history/HiveHistoryViewer.java Mon Aug 26 21:42:12 2013
@@ -22,6 +22,8 @@ import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.hive.ql.history.HiveHistory.Keys;
import org.apache.hadoop.hive.ql.history.HiveHistory.Listener;
import org.apache.hadoop.hive.ql.history.HiveHistory.QueryInfo;
@@ -35,8 +37,8 @@ import org.apache.hadoop.hive.ql.history
public class HiveHistoryViewer implements Listener {
String historyFile;
-
String sessionId;
+ private static final Log LOG = LogFactory.getLog(HiveHistoryViewer.class);
// Job Hash Map
private final HashMap<String, QueryInfo> jobInfoMap = new HashMap<String, QueryInfo>();
@@ -65,19 +67,18 @@ public class HiveHistoryViewer implement
* Parse history files.
*/
void init() {
-
try {
- HiveHistory.parseHiveHistory(historyFile, this);
+ HiveHistoryUtil.parseHiveHistory(historyFile, this);
} catch (IOException e) {
- // TODO Auto-generated catch block
+ // TODO pass on this exception
e.printStackTrace();
+ LOG.error("Error parsing hive history log file", e);
}
-
}
/**
- * Implementation Listner interface function.
- *
+ * Implementation Listener interface function.
+ *
* @see org.apache.hadoop.hive.ql.history.HiveHistory.Listener#handle(org.apache.hadoop.hive.ql.history.HiveHistory.RecordTypes,
* java.util.Map)
*/
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/HiveInputFormat.java Mon Aug 26 21:42:12 2013
@@ -428,6 +428,8 @@ public class HiveInputFormat<K extends W
} else {
ColumnProjectionUtils.setFullyReadColumns(jobConf);
}
+ ColumnProjectionUtils.appendReadColumnNames(jobConf,
+ tableScan.getNeededColumns());
pushFilters(jobConf, tableScan);
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/RCFile.java Mon Aug 26 21:42:12 2013
@@ -1601,7 +1601,7 @@ public class RCFile {
}
}
/* move the last 16 bytes to the prefix area */
- System.arraycopy(buffer, buffer.length - prefix - 1, buffer, 0, prefix);
+ System.arraycopy(buffer, buffer.length - prefix, buffer, 0, prefix);
n = (int)Math.min(n, end - in.getPos());
}
} catch (ChecksumException e) { // checksum failure
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/BitFieldReader.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/BitFieldReader.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/BitFieldReader.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/BitFieldReader.java Mon Aug 26 21:42:12 2013
@@ -39,7 +39,7 @@ class BitFieldReader {
current = 0xff & input.next();
bitsLeft = 8;
} else {
- throw new EOFException("Read past end of bit field from " + input);
+ throw new EOFException("Read past end of bit field from " + this);
}
}
@@ -85,4 +85,10 @@ class BitFieldReader {
bitsLeft = (int) (8 - (totalBits % 8));
}
}
+
+ @Override
+ public String toString() {
+ return "bit reader current: " + current + " bits left: " + bitsLeft +
+ " bit size: " + bitSize + " from " + input;
+ }
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/InStream.java Mon Aug 26 21:42:12 2013
@@ -25,97 +25,149 @@ abstract class InStream extends InputStr
private static class UncompressedStream extends InStream {
private final String name;
- private byte[] array;
- private int offset;
- private final int base;
- private final int limit;
+ private final ByteBuffer[] bytes;
+ private final long[] offsets;
+ private final long length;
+ private long currentOffset;
+ private byte[] range;
+ private int currentRange;
+ private int offsetInRange;
+ private int limitInRange;
- public UncompressedStream(String name, ByteBuffer input) {
+ public UncompressedStream(String name, ByteBuffer[] input, long[] offsets,
+ long length) {
this.name = name;
- this.array = input.array();
- base = input.arrayOffset() + input.position();
- offset = base;
- limit = input.arrayOffset() + input.limit();
+ this.bytes = input;
+ this.offsets = offsets;
+ this.length = length;
+ currentRange = 0;
+ offsetInRange = 0;
+ limitInRange = 0;
+ currentOffset = 0;
}
@Override
public int read() {
- if (offset == limit) {
- return -1;
+ if (offsetInRange >= limitInRange) {
+ if (currentOffset == length) {
+ return -1;
+ }
+ seek(currentOffset);
}
- return 0xff & array[offset++];
+ currentOffset += 1;
+ return 0xff & range[offsetInRange++];
}
@Override
public int read(byte[] data, int offset, int length) {
- if (this.offset == limit) {
- return -1;
+ if (offsetInRange >= limitInRange) {
+ if (currentOffset == this.length) {
+ return -1;
+ }
+ seek(currentOffset);
}
- int actualLength = Math.min(length, limit - this.offset);
- System.arraycopy(array, this.offset, data, offset, actualLength);
- this.offset += actualLength;
+ int actualLength = Math.min(length, limitInRange - offsetInRange);
+ System.arraycopy(range, offsetInRange, data, offset, actualLength);
+ offsetInRange += actualLength;
+ currentOffset += actualLength;
return actualLength;
}
@Override
public int available() {
- return limit - offset;
+ if (offsetInRange < limitInRange) {
+ return limitInRange - offsetInRange;
+ }
+ return (int) (length - currentOffset);
}
@Override
public void close() {
- array = null;
- offset = 0;
+ currentRange = bytes.length;
+ currentOffset = length;
}
@Override
public void seek(PositionProvider index) throws IOException {
- offset = base + (int) index.getNext();
+ seek(index.getNext());
+ }
+
+ public void seek(long desired) {
+ for(int i = 0; i < bytes.length; ++i) {
+ if (offsets[i] <= desired &&
+ desired - offsets[i] < bytes[i].remaining()) {
+ currentOffset = desired;
+ currentRange = i;
+ this.range = bytes[i].array();
+ offsetInRange = bytes[i].arrayOffset() + bytes[i].position();
+ limitInRange = bytes[i].arrayOffset() + bytes[i].limit();
+ offsetInRange += desired - offsets[i];
+ return;
+ }
+ }
+ throw new IllegalArgumentException("Seek in " + name + " to " +
+ desired + " is outside of the data");
}
@Override
public String toString() {
- return "uncompressed stream " + name + " base: " + base +
- " offset: " + offset + " limit: " + limit;
+ return "uncompressed stream " + name + " position: " + currentOffset +
+ " length: " + length + " range: " + currentRange +
+ " offset: " + offsetInRange + " limit: " + limitInRange;
}
}
private static class CompressedStream extends InStream {
private final String name;
- private byte[] array;
+ private final ByteBuffer[] bytes;
+ private final long[] offsets;
private final int bufferSize;
+ private final long length;
private ByteBuffer uncompressed = null;
private final CompressionCodec codec;
- private int offset;
- private final int base;
- private final int limit;
+ private byte[] compressed = null;
+ private long currentOffset;
+ private int currentRange;
+ private int offsetInCompressed;
+ private int limitInCompressed;
private boolean isUncompressedOriginal;
- public CompressedStream(String name, ByteBuffer input,
+ public CompressedStream(String name, ByteBuffer[] input,
+ long[] offsets, long length,
CompressionCodec codec, int bufferSize
) {
- this.array = input.array();
+ this.bytes = input;
this.name = name;
this.codec = codec;
+ this.length = length;
+ this.offsets = offsets;
this.bufferSize = bufferSize;
- base = input.arrayOffset() + input.position();
- offset = base;
- limit = input.arrayOffset() + input.limit();
+ currentOffset = 0;
+ currentRange = 0;
+ offsetInCompressed = 0;
+ limitInCompressed = 0;
}
private void readHeader() throws IOException {
- if (limit - offset > OutStream.HEADER_SIZE) {
- int chunkLength = ((0xff & array[offset + 2]) << 15) |
- ((0xff & array[offset + 1]) << 7) | ((0xff & array[offset]) >> 1);
+ if (compressed == null || offsetInCompressed >= limitInCompressed) {
+ seek(currentOffset);
+ }
+ if (limitInCompressed - offsetInCompressed > OutStream.HEADER_SIZE) {
+ int chunkLength = ((0xff & compressed[offsetInCompressed + 2]) << 15) |
+ ((0xff & compressed[offsetInCompressed + 1]) << 7) |
+ ((0xff & compressed[offsetInCompressed]) >> 1);
if (chunkLength > bufferSize) {
throw new IllegalArgumentException("Buffer size too small. size = " +
bufferSize + " needed = " + chunkLength);
}
- boolean isOriginal = (array[offset] & 0x01) == 1;
- offset += OutStream.HEADER_SIZE;
+ boolean isOriginal = (compressed[offsetInCompressed] & 0x01) == 1;
+ offsetInCompressed += OutStream.HEADER_SIZE;
if (isOriginal) {
isUncompressedOriginal = true;
- uncompressed = ByteBuffer.wrap(array, offset, chunkLength);
+ uncompressed = bytes[currentRange].duplicate();
+ uncompressed.position(offsetInCompressed -
+ bytes[currentRange].arrayOffset());
+ uncompressed.limit(offsetInCompressed + chunkLength);
} else {
if (isUncompressedOriginal) {
uncompressed = ByteBuffer.allocate(bufferSize);
@@ -125,19 +177,21 @@ abstract class InStream extends InputStr
} else {
uncompressed.clear();
}
- codec.decompress(ByteBuffer.wrap(array, offset, chunkLength),
+ codec.decompress(ByteBuffer.wrap(compressed, offsetInCompressed,
+ chunkLength),
uncompressed);
}
- offset += chunkLength;
+ offsetInCompressed += chunkLength;
+ currentOffset += chunkLength + OutStream.HEADER_SIZE;
} else {
- throw new IllegalStateException("Can't read header");
+ throw new IllegalStateException("Can't read header at " + this);
}
}
@Override
public int read() throws IOException {
if (uncompressed == null || uncompressed.remaining() == 0) {
- if (offset == limit) {
+ if (currentOffset == length) {
return -1;
}
readHeader();
@@ -148,7 +202,7 @@ abstract class InStream extends InputStr
@Override
public int read(byte[] data, int offset, int length) throws IOException {
if (uncompressed == null || uncompressed.remaining() == 0) {
- if (this.offset == this.limit) {
+ if (currentOffset == this.length) {
return -1;
}
readHeader();
@@ -164,7 +218,7 @@ abstract class InStream extends InputStr
@Override
public int available() throws IOException {
if (uncompressed == null || uncompressed.remaining() == 0) {
- if (offset == limit) {
+ if (currentOffset == length) {
return 0;
}
readHeader();
@@ -174,27 +228,74 @@ abstract class InStream extends InputStr
@Override
public void close() {
- array = null;
uncompressed = null;
- offset = 0;
+ currentRange = bytes.length;
+ offsetInCompressed = 0;
+ limitInCompressed = 0;
+ currentOffset = length;
}
@Override
public void seek(PositionProvider index) throws IOException {
- offset = base + (int) index.getNext();
- int uncompBytes = (int) index.getNext();
- if (uncompBytes != 0) {
+ seek(index.getNext());
+ long uncompressedBytes = index.getNext();
+ if (uncompressedBytes != 0) {
readHeader();
- uncompressed.position(uncompressed.position() + uncompBytes);
+ uncompressed.position(uncompressed.position() +
+ (int) uncompressedBytes);
} else if (uncompressed != null) {
+ // mark the uncompressed buffer as done
uncompressed.position(uncompressed.limit());
}
}
+ private void seek(long desired) throws IOException {
+ for(int i = 0; i < bytes.length; ++i) {
+ if (offsets[i] <= desired &&
+ desired - offsets[i] < bytes[i].remaining()) {
+ currentRange = i;
+ compressed = bytes[i].array();
+ offsetInCompressed = (int) (bytes[i].arrayOffset() +
+ bytes[i].position() + (desired - offsets[i]));
+ currentOffset = desired;
+ limitInCompressed = bytes[i].arrayOffset() + bytes[i].limit();
+ return;
+ }
+ }
+ // if they are seeking to the precise end, go ahead and let them go there
+ int segments = bytes.length;
+ if (segments != 0 &&
+ desired == offsets[segments - 1] + bytes[segments - 1].remaining()) {
+ currentRange = segments - 1;
+ compressed = bytes[currentRange].array();
+ offsetInCompressed = bytes[currentRange].arrayOffset() +
+ bytes[currentRange].limit();
+ currentOffset = desired;
+ limitInCompressed = offsetInCompressed;
+ return;
+ }
+ throw new IOException("Seek outside of data in " + this + " to " +
+ desired);
+ }
+
+ private String rangeString() {
+ StringBuilder builder = new StringBuilder();
+ for(int i=0; i < offsets.length; ++i) {
+ if (i != 0) {
+ builder.append("; ");
+ }
+ builder.append(" range " + i + " = " + offsets[i] + " to " +
+ bytes[i].remaining());
+ }
+ return builder.toString();
+ }
+
@Override
public String toString() {
- return "compressed stream " + name + " base: " + base +
- " offset: " + offset + " limit: " + limit +
+ return "compressed stream " + name + " position: " + currentOffset +
+ " length: " + length + " range: " + currentRange +
+ " offset: " + offsetInCompressed + " limit: " + limitInCompressed +
+ rangeString() +
(uncompressed == null ? "" :
" uncompressed: " + uncompressed.position() + " to " +
uncompressed.limit());
@@ -203,14 +304,29 @@ abstract class InStream extends InputStr
public abstract void seek(PositionProvider index) throws IOException;
+ /**
+ * Create an input stream from a list of buffers.
+ * @param name the name of the stream
+ * @param input the list of ranges of bytes for the stream
+ * @param offsets a list of offsets (the same length as input) that must
+ * contain the first offset of the each set of bytes in input
+ * @param length the length in bytes of the stream
+ * @param codec the compression codec
+ * @param bufferSize the compression buffer size
+ * @return an input stream
+ * @throws IOException
+ */
public static InStream create(String name,
- ByteBuffer input,
+ ByteBuffer[] input,
+ long[] offsets,
+ long length,
CompressionCodec codec,
int bufferSize) throws IOException {
if (codec == null) {
- return new UncompressedStream(name, input);
+ return new UncompressedStream(name, input, offsets, length);
} else {
- return new CompressedStream(name, input, codec, bufferSize);
+ return new CompressedStream(name, input, offsets, length, codec,
+ bufferSize);
}
}
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcFile.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcFile.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcFile.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcFile.java Mon Aug 26 21:42:12 2013
@@ -97,7 +97,7 @@ public final class OrcFile {
CompressionKind compress,
int bufferSize,
int rowIndexStride) throws IOException {
- return new WriterImpl(fs, path, inspector, stripeSize, compress,
+ return new WriterImpl(fs, path, conf, inspector, stripeSize, compress,
bufferSize, rowIndexStride, getMemoryManager(conf));
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcInputFormat.java Mon Aug 26 21:42:12 2013
@@ -18,13 +18,22 @@
package org.apache.hadoop.hive.ql.io.orc;
+import com.google.common.collect.Lists;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hive.conf.HiveConf;
+import org.apache.hadoop.hive.ql.exec.Utilities;
import org.apache.hadoop.hive.ql.io.InputFormatChecker;
+import org.apache.hadoop.hive.ql.io.sarg.PredicateLeaf;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
+import org.apache.hadoop.hive.ql.plan.TableScanDesc;
import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapred.FileInputFormat;
import org.apache.hadoop.mapred.FileSplit;
@@ -43,6 +52,8 @@ import java.util.List;
public class OrcInputFormat extends FileInputFormat<NullWritable, OrcStruct>
implements InputFormatChecker {
+ private static final Log LOG = LogFactory.getLog(OrcInputFormat.class);
+
private static class OrcRecordReader
implements RecordReader<NullWritable, OrcStruct> {
private final org.apache.hadoop.hive.ql.io.orc.RecordReader reader;
@@ -53,14 +64,38 @@ public class OrcInputFormat extends Fil
OrcRecordReader(Reader file, Configuration conf,
long offset, long length) throws IOException {
- this.reader = file.rows(offset, length,
- findIncludedColumns(file.getTypes(), conf));
+ String serializedPushdown = conf.get(TableScanDesc.FILTER_EXPR_CONF_STR);
+ String columnNamesString =
+ conf.get(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR);
+ String[] columnNames = null;
+ SearchArgument sarg = null;
List<OrcProto.Type> types = file.getTypes();
if (types.size() == 0) {
numColumns = 0;
} else {
numColumns = types.get(0).getSubtypesCount();
}
+ columnNames = new String[types.size()];
+ LOG.info("included column ids = " +
+ conf.get(ColumnProjectionUtils.READ_COLUMN_IDS_CONF_STR, "null"));
+ LOG.info("included columns names = " +
+ conf.get(ColumnProjectionUtils.READ_COLUMN_NAMES_CONF_STR, "null"));
+ boolean[] includeColumn = findIncludedColumns(types, conf);
+ if (serializedPushdown != null && columnNamesString != null) {
+ sarg = SearchArgument.FACTORY.create
+ (Utilities.deserializeExpression(serializedPushdown, conf));
+ LOG.info("ORC pushdown predicate: " + sarg);
+ String[] neededColumnNames = columnNamesString.split(",");
+ int i = 0;
+ for(int columnId: types.get(0).getSubtypesList()) {
+ if (includeColumn[columnId]) {
+ columnNames[columnId] = neededColumnNames[i++];
+ }
+ }
+ } else {
+ LOG.info("No ORC pushdown predicate");
+ }
+ this.reader = file.rows(offset, length,includeColumn, sarg, columnNames);
this.offset = offset;
this.length = length;
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSerde.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSerde.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSerde.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OrcSerde.java Mon Aug 26 21:42:12 2013
@@ -17,6 +17,8 @@
*/
package org.apache.hadoop.hive.ql.io.orc;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hive.serde2.SerDe;
import org.apache.hadoop.hive.serde2.SerDeException;
@@ -38,6 +40,9 @@ import java.util.Properties;
* It transparently passes the object to/from the ORC file reader/writer.
*/
public class OrcSerde implements SerDe {
+
+ private static final Log LOG = LogFactory.getLog(OrcSerde.class);
+
private final OrcSerdeRow row = new OrcSerdeRow();
private ObjectInspector inspector = null;
@@ -129,4 +134,5 @@ public class OrcSerde implements SerDe {
public SerDeStats getSerDeStats() {
return null;
}
+
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OutStream.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OutStream.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OutStream.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/OutStream.java Mon Aug 26 21:42:12 2013
@@ -76,11 +76,7 @@ class OutStream extends PositionedOutput
}
public void clear() throws IOException {
- uncompressedBytes = 0;
- compressedBytes = 0;
- compressed = null;
- overflow = null;
- current = null;
+ flush();
suppress = false;
}
@@ -246,7 +242,10 @@ class OutStream extends PositionedOutput
receiver.output(compressed);
compressed = null;
}
- clear();
+ uncompressedBytes = 0;
+ compressedBytes = 0;
+ overflow = null;
+ current = null;
}
@Override
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/Reader.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/Reader.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/Reader.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/Reader.java Mon Aug 26 21:42:12 2013
@@ -18,6 +18,7 @@
package org.apache.hadoop.hive.ql.io.orc;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import java.io.IOException;
@@ -118,8 +119,26 @@ public interface Reader {
* @param include true for each column that should be included
* @return a new RecordReader that will read the specified rows.
* @throws IOException
+ * @deprecated
*/
RecordReader rows(long offset, long length,
boolean[] include) throws IOException;
+ /**
+ * Create a RecordReader that will read a section of a file. It starts reading
+ * at the first stripe after the offset and continues to the stripe that
+ * starts at offset + length. It also accepts a list of columns to read and a
+ * search argument.
+ * @param offset the minimum offset of the first stripe to read
+ * @param length the distance from offset of the first address to stop reading
+ * at
+ * @param include true for each column that should be included
+ * @param sarg a search argument that limits the rows that should be read.
+ * @param neededColumns the names of the included columns
+ * @return the record reader for the rows
+ */
+ RecordReader rows(long offset, long length,
+ boolean[] include, SearchArgument sarg,
+ String[] neededColumns) throws IOException;
+
}
Modified: hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ReaderImpl.java
URL: http://svn.apache.org/viewvc/hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ReaderImpl.java?rev=1517707&r1=1517706&r2=1517707&view=diff
==============================================================================
--- hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ReaderImpl.java (original)
+++ hive/branches/tez/ql/src/java/org/apache/hadoop/hive/ql/io/orc/ReaderImpl.java Mon Aug 26 21:42:12 2013
@@ -24,6 +24,7 @@ import org.apache.commons.logging.LogFac
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.hive.ql.io.sarg.SearchArgument;
import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
import org.apache.hadoop.io.Text;
@@ -307,7 +308,8 @@ final class ReaderImpl implements Reader
buffer.position(psOffset - footerSize);
buffer.limit(psOffset);
}
- InputStream instream = InStream.create("footer", buffer, codec, bufferSize);
+ InputStream instream = InStream.create("footer", new ByteBuffer[]{buffer},
+ new long[]{0L}, footerSize, codec, bufferSize);
footer = OrcProto.Footer.parseFrom(instream);
inspector = OrcStruct.createObjectInspector(0, footer.getTypesList());
file.close();
@@ -315,15 +317,22 @@ final class ReaderImpl implements Reader
@Override
public RecordReader rows(boolean[] include) throws IOException {
- return rows(0, Long.MAX_VALUE, include);
+ return rows(0, Long.MAX_VALUE, include, null, null);
}
@Override
public RecordReader rows(long offset, long length, boolean[] include
) throws IOException {
+ return rows(offset, length, include, null, null);
+ }
+
+ @Override
+ public RecordReader rows(long offset, long length, boolean[] include,
+ SearchArgument sarg, String[] columnNames
+ ) throws IOException {
return new RecordReaderImpl(this.getStripes(), fileSystem, path, offset,
- length, footer.getTypesList(), codec, bufferSize,
- include, footer.getRowIndexStride());
+ length, footer.getTypesList(), codec, bufferSize,
+ include, footer.getRowIndexStride(), sarg, columnNames);
}
}