You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ja...@apache.org on 2015/11/05 06:56:40 UTC
[2/9] drill git commit: DRILL-951: Add support for csv header row
parsing
DRILL-951: Add support for csv header row parsing
This closes #232
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/745bcd1f
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/745bcd1f
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/745bcd1f
Branch: refs/heads/master
Commit: 745bcd1f378397d921860eeba382530483644cd3
Parents: 9edce15
Author: Abhi P <ab...@dremio.com>
Authored: Mon Nov 2 19:03:04 2015 -0800
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Wed Nov 4 20:55:53 2015 -0800
----------------------------------------------------------------------
.../exec/store/easy/text/TextFormatPlugin.java | 7 +-
.../compliant/CompliantTextRecordReader.java | 132 +++++++++++-
.../easy/text/compliant/FieldVarCharOutput.java | 212 +++++++++++++++++++
.../text/compliant/RepeatedVarCharOutput.java | 25 +++
.../text/compliant/TextParsingSettings.java | 6 +-
.../resources/bootstrap-storage-plugins.json | 14 +-
.../drill/exec/store/text/TestCsvHeader.java | 129 +++++++++++
.../resources/bootstrap-storage-plugins.json | 6 +
.../test/resources/store/text/data/cars.csvh | 6 +
pom.xml | 1 +
10 files changed, 524 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/TextFormatPlugin.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/TextFormatPlugin.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/TextFormatPlugin.java
index d40e1fa..f7551ae 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/TextFormatPlugin.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/TextFormatPlugin.java
@@ -140,7 +140,7 @@ public class TextFormatPlugin extends EasyFormatPlugin<TextFormatPlugin.TextForm
public char escape = '"';
public char comment = '#';
public boolean skipFirstLine = false;
-
+ public boolean extractHeader = false;
public List<String> getExtensions() {
return extensions;
@@ -167,6 +167,11 @@ public class TextFormatPlugin extends EasyFormatPlugin<TextFormatPlugin.TextForm
}
@JsonIgnore
+ public boolean isHeaderExtractionEnabled() {
+ return extractHeader;
+ }
+
+ @JsonIgnore
public String getFieldDelimiterAsString(){
return new String(new char[]{fieldDelimiter});
}
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/CompliantTextRecordReader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/CompliantTextRecordReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/CompliantTextRecordReader.java
index 465a971..f0363da 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/CompliantTextRecordReader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/CompliantTextRecordReader.java
@@ -17,12 +17,14 @@
*/
package org.apache.drill.exec.store.easy.text.compliant;
+import com.google.common.collect.Maps;
import com.univocity.parsers.common.TextParsingException;
import io.netty.buffer.DrillBuf;
import java.io.IOException;
import java.io.InputStream;
import java.util.List;
+import java.util.Map;
import javax.annotation.Nullable;
@@ -34,12 +36,16 @@ import org.apache.drill.exec.exception.SchemaChangeException;
import org.apache.drill.exec.ops.FragmentContext;
import org.apache.drill.exec.ops.OperatorContext;
import org.apache.drill.exec.physical.impl.OutputMutator;
+import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.store.AbstractRecordReader;
import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.util.CallBack;
+import org.apache.drill.exec.vector.ValueVector;
import org.apache.hadoop.mapred.FileSplit;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
+import org.apache.drill.exec.expr.TypeHelper;
// New text reader, complies with the RFC 4180 standard for text/csv files
public class CompliantTextRecordReader extends AbstractRecordReader {
@@ -60,6 +66,8 @@ public class CompliantTextRecordReader extends AbstractRecordReader {
// working buffer to handle whitespaces
private DrillBuf whitespaceBuffer;
private DrillFileSystem dfs;
+ // operator context for OutputMutator
+ private OperatorContext oContext;
public CompliantTextRecordReader(FileSplit split, DrillFileSystem dfs, FragmentContext context, TextParsingSettings settings, List<SchemaPath> columns) {
this.split = split;
@@ -93,24 +101,34 @@ public class CompliantTextRecordReader extends AbstractRecordReader {
@Override
public void setup(OperatorContext context, OutputMutator outputMutator) throws ExecutionSetupException {
-
+ oContext = context;
readBuffer = context.getManagedBuffer(READ_BUFFER);
whitespaceBuffer = context.getManagedBuffer(WHITE_SPACE_BUFFER);
+ // setup Output, Input, and Reader
try {
- InputStream stream = dfs.openPossiblyCompressedStream(split.getPath());
- TextInput input = new TextInput(settings, stream, readBuffer, split.getStart(), split.getStart() + split.getLength());
-
TextOutput output = null;
- if(settings.isUseRepeatedVarChar()){
+ TextInput input = null;
+ InputStream stream = null;
+
+ // setup Output using OutputMutator
+ if (settings.isHeaderExtractionEnabled()){
+ //extract header and use that to setup a set of VarCharVectors
+ String [] fieldNames = extractHeader();
+ output = new FieldVarCharOutput(outputMutator, fieldNames, getColumns(), isStarQuery());
+ } else {
+ //simply use RepeatedVarCharVector
output = new RepeatedVarCharOutput(outputMutator, getColumns(), isStarQuery());
- }else{
- //TODO: Add field output.
- throw new UnsupportedOperationException();
}
- this.reader = new TextReader(settings, input, output, whitespaceBuffer);
+ // setup Input using InputStream
+ stream = dfs.openPossiblyCompressedStream(split.getPath());
+ input = new TextInput(settings, stream, readBuffer, split.getStart(), split.getStart() + split.getLength());
+
+ // setup Reader using Input and Output
+ reader = new TextReader(settings, input, output, whitespaceBuffer);
reader.start();
+
} catch (SchemaChangeException | IOException e) {
throw new ExecutionSetupException(String.format("Failure while setting up text reader for file %s", split.getPath()), e);
} catch (IllegalArgumentException e) {
@@ -118,6 +136,44 @@ public class CompliantTextRecordReader extends AbstractRecordReader {
}
}
+ /**
+ * This method is responsible to implement logic for extracting header from text file
+ * Currently it is assumed to be first line if headerExtractionEnabled is set to true
+ * TODO: enhance to support more common header patterns
+ * @return field name strings
+ */
+ private String [] extractHeader() throws SchemaChangeException, IOException, ExecutionSetupException{
+ assert (settings.isHeaderExtractionEnabled());
+ assert (oContext != null);
+
+ // setup Output using OutputMutator
+ // we should use a separate output mutator to avoid reshaping query output with header data
+ HeaderOutputMutator hOutputMutator = new HeaderOutputMutator();
+ TextOutput hOutput = new RepeatedVarCharOutput(hOutputMutator, getColumns(), true);
+ this.allocate(hOutputMutator.fieldVectorMap);
+
+ // setup Input using InputStream
+ // we should read file header irrespective of split given given to this reader
+ InputStream hStream = dfs.openPossiblyCompressedStream(split.getPath());
+ TextInput hInput = new TextInput(settings, hStream, oContext.getManagedBuffer(READ_BUFFER), 0, split.getLength());
+
+ // setup Reader using Input and Output
+ this.reader = new TextReader(settings, hInput, hOutput, oContext.getManagedBuffer(WHITE_SPACE_BUFFER));
+ reader.start();
+
+ // extract first row only
+ reader.parseNext();
+
+ // grab the field names from output
+ String [] fieldNames = ((RepeatedVarCharOutput)hOutput).getTextOutput();
+
+ // cleanup and set to skip the first line next time we read input
+ reader.close();
+ hOutputMutator.close();
+ settings.setSkipFirstLine(true);
+
+ return fieldNames;
+ }
/**
* Generates the next record batch
@@ -158,4 +214,62 @@ public class CompliantTextRecordReader extends AbstractRecordReader {
logger.warn("Exception while closing stream.", e);
}
}
+
+ /**
+ * TextRecordReader during its first phase read to extract header should pass its own
+ * OutputMutator to avoid reshaping query output.
+ * This class provides OutputMutator for header extraction.
+ */
+ private class HeaderOutputMutator implements OutputMutator {
+ private final Map<MaterializedField.Key, ValueVector> fieldVectorMap = Maps.newHashMap();
+
+ @Override
+ public <T extends ValueVector> T addField(MaterializedField field, Class<T> clazz) throws SchemaChangeException {
+ ValueVector v = fieldVectorMap.get(field.key());
+ if (v == null || v.getClass() != clazz) {
+ // Field does not exist add it to the map
+ v = TypeHelper.getNewVector(field, oContext.getAllocator());
+ if (!clazz.isAssignableFrom(v.getClass())) {
+ throw new SchemaChangeException(String.format(
+ "Class %s was provided, expected %s.", clazz.getSimpleName(), v.getClass().getSimpleName()));
+ }
+ fieldVectorMap.put(field.key(), v);
+ }
+ return clazz.cast(v);
+ }
+
+ @Override
+ public void allocate(int recordCount) {
+ //do nothing for now
+ }
+
+ @Override
+ public boolean isNewSchema() {
+ return false;
+ }
+
+ @Override
+ public DrillBuf getManagedBuffer() {
+ return null;
+ }
+
+ @Override
+ public CallBack getCallBack() {
+ return null;
+ }
+
+ /**
+ * Since this OutputMutator is passed by TextRecordReader to get the header out
+ * the mutator might not get cleaned up elsewhere. TextRecordReader will call
+ * this method to clear any allocations
+ */
+ public void close() {
+ for (final ValueVector v : fieldVectorMap.values()) {
+ v.clear();
+ }
+ fieldVectorMap.clear();
+ }
+
+ }
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/FieldVarCharOutput.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/FieldVarCharOutput.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/FieldVarCharOutput.java
new file mode 100644
index 0000000..f86d40f
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/FieldVarCharOutput.java
@@ -0,0 +1,212 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.easy.text.compliant;
+
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.common.types.Types;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.physical.impl.OutputMutator;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.vector.VarCharVector;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.Arrays;
+import java.util.List;
+
+/**
+ * Class is responsible for generating record batches for text file inputs. We generate
+ * a record batch with a set of varchar vectors. A varchar vector contains all the field
+ * values for a given column. Each record is a single value within each vector of the set.
+ */
+class FieldVarCharOutput extends TextOutput {
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(FieldVarCharOutput.class);
+ static final String COL_NAME = "columns";
+
+ // array of output vector
+ private final VarCharVector [] vectors;
+ // boolean array indicating which fields are selected (if star query entire array is set to true)
+ private final boolean[] selectedFields;
+ // current vector to which field will be added
+ private VarCharVector currentVector;
+ // track which field is getting appended
+ private int currentFieldIndex = -1;
+ // track chars within field
+ private int currentDataPointer = 0;
+ // track if field is still getting appended
+ private boolean fieldOpen = true;
+ // holds chars for a field
+ private byte[] fieldBytes;
+
+ private boolean collect = true;
+ private boolean rowHasData= false;
+ private static final int MAX_FIELD_LENGTH = 1024 * 64;
+ private int recordCount = 0;
+ private int batchIndex = 0;
+ private int maxField = 0;
+
+ /**
+ * We initialize and add the varchar vector for each incoming field in this
+ * constructor.
+ * @param outputMutator Used to create/modify schema
+ * @param fieldNames Incoming field names
+ * @param columns List of columns selected in the query
+ * @param isStarQuery boolean to indicate if all fields are selected or not
+ * @throws SchemaChangeException
+ */
+ public FieldVarCharOutput(OutputMutator outputMutator, String [] fieldNames, Collection<SchemaPath> columns, boolean isStarQuery) throws SchemaChangeException {
+
+ int totalFields = fieldNames.length;
+ this.selectedFields = new boolean[totalFields];
+ this.vectors = new VarCharVector[totalFields];
+
+ if (isStarQuery) {
+ maxField = totalFields - 1;
+ Arrays.fill(selectedFields, true);
+ } else {
+ List<Integer> columnIds = new ArrayList<Integer>();
+ String pathStr;
+ int index;
+
+ for (SchemaPath path : columns) {
+ pathStr = path.getRootSegment().getPath();
+ if (pathStr.equals(COL_NAME) && path.getRootSegment().getChild() != null) {
+ //TODO: support both field names and columns index in predicate pushdown
+ index = path.getRootSegment().getChild().getArraySegment().getIndex();
+ } else {
+ index = Arrays.asList(fieldNames).indexOf(pathStr);
+ }
+ assert index >= 0 && index < totalFields : "Invalid column index encountered";
+ columnIds.add(index);
+ }
+ Collections.sort(columnIds);
+
+ for(Integer i : columnIds) {
+ selectedFields[i] = true;
+ maxField = i;
+ }
+ }
+
+ for (int i = 0; i <= maxField; i++) {
+ if (selectedFields[i]) {
+ MaterializedField field = MaterializedField.create(fieldNames[i], Types.required(TypeProtos.MinorType.VARCHAR));
+ this.vectors[i] = outputMutator.addField(field, VarCharVector.class);
+ }
+ }
+
+ this.fieldBytes = new byte[MAX_FIELD_LENGTH];
+
+ }
+
+ /**
+ * Start a new record batch. Resets all pointers
+ */
+ @Override
+ public void startBatch() {
+ this.recordCount = 0;
+ this.batchIndex = 0;
+ this.currentFieldIndex= -1;
+ this.collect = true;
+ this.fieldOpen = false;
+ }
+
+ @Override
+ public void startField(int index) {
+ currentFieldIndex = index;
+ currentDataPointer = 0;
+ fieldOpen = true;
+ collect = selectedFields[index];
+ currentVector = vectors[index];
+ }
+
+ @Override
+ public void append(byte data) {
+ if (!collect) {
+ return;
+ }
+
+ if (currentDataPointer >= MAX_FIELD_LENGTH -1) {
+ throw UserException
+ .unsupportedError()
+ .message("Trying to write something big in a column")
+ .addContext("columnIndex", currentFieldIndex)
+ .addContext("Limit", MAX_FIELD_LENGTH)
+ .build(logger);
+ }
+
+ fieldBytes[currentDataPointer++] = data;
+ }
+
+ @Override
+ public boolean endField() {
+ fieldOpen = false;
+
+ if(collect) {
+ assert currentVector != null;
+ currentVector.getMutator().setSafe(recordCount, fieldBytes, 0, currentDataPointer);
+ }
+
+ if (currentDataPointer > 0) {
+ this.rowHasData = true;
+ }
+
+ return currentFieldIndex < maxField;
+ }
+
+ @Override
+ public boolean endEmptyField() {
+ return endField();
+ }
+
+ @Override
+ public void finishRecord() {
+ if(fieldOpen){
+ endField();
+ }
+
+ recordCount++;
+ }
+
+ // Sets the record count in this batch within the value vector
+ @Override
+ public void finishBatch() {
+ batchIndex++;
+
+ for (int i = 0; i <= maxField; i++) {
+ if (this.vectors[i] != null) {
+ this.vectors[i].getMutator().setValueCount(batchIndex);
+ }
+ }
+
+ }
+
+ @Override
+ public long getRecordCount() {
+ return recordCount;
+ }
+
+ @Override
+ public boolean rowHasData() {
+ return this.rowHasData;
+ }
+
+ }
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/RepeatedVarCharOutput.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/RepeatedVarCharOutput.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/RepeatedVarCharOutput.java
index 91b1a7f..ca02e97 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/RepeatedVarCharOutput.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/RepeatedVarCharOutput.java
@@ -26,6 +26,7 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.expression.FieldReference;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.types.TypeProtos;
@@ -316,6 +317,30 @@ class RepeatedVarCharOutput extends TextOutput {
}
+ /**
+ * This method is a helper method added for DRILL-951
+ * TextRecordReader to call this method to get field names out
+ * @return array of field data strings
+ */
+ public String [] getTextOutput () throws ExecutionSetupException {
+ if (recordCount == 0 || fieldIndex == -1) {
+ return null;
+ }
+
+ if (this.recordStart != characterData) {
+ throw new ExecutionSetupException("record text was requested before finishing record");
+ }
+
+ //Currently only first line header is supported. Return only first record.
+ int retSize = fieldIndex+1;
+ String [] out = new String [retSize];
+
+ RepeatedVarCharVector.Accessor a = this.vector.getAccessor();
+ for (int i=0; i<retSize; i++){
+ out[i] = a.getSingleObject(0,i).toString();
+ }
+ return out;
+ }
// Sets the record count in this batch within the value vector
@Override
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/TextParsingSettings.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/TextParsingSettings.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/TextParsingSettings.java
index e411461..a366c90 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/TextParsingSettings.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/easy/text/compliant/TextParsingSettings.java
@@ -43,7 +43,6 @@ public class TextParsingSettings {
private String lineSeparatorString = "\n";
private boolean skipFirstLine = false;
- // these options are not yet supported
private boolean headerExtractionEnabled = false;
private boolean useRepeatedVarChar = true;
private int numberOfRecordsToRead = -1;
@@ -57,6 +56,11 @@ public class TextParsingSettings {
this.delimiter = bSafe(config.getFieldDelimiter(), "fieldDelimiter");
this.comment = bSafe(config.getComment(), "comment");
this.skipFirstLine = config.isSkipFirstLine();
+ this.headerExtractionEnabled = config.isHeaderExtractionEnabled();
+ if (this.headerExtractionEnabled) {
+ // In case of header TextRecordReader will use set of VarChar vectors vs RepeatedVarChar
+ this.useRepeatedVarChar = false;
+ }
}
public byte getComment(){
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/exec/java-exec/src/main/resources/bootstrap-storage-plugins.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/resources/bootstrap-storage-plugins.json b/exec/java-exec/src/main/resources/bootstrap-storage-plugins.json
index 8d405d0..f5d3880 100644
--- a/exec/java-exec/src/main/resources/bootstrap-storage-plugins.json
+++ b/exec/java-exec/src/main/resources/bootstrap-storage-plugins.json
@@ -42,6 +42,12 @@
"sequencefile": {
type : "sequencefile",
extensions: [ "seq" ]
+ },
+ "csvh" : {
+ type: "text",
+ extensions: [ "csvh" ],
+ delimiter: ",",
+ extractHeader: true
}
}
},
@@ -70,9 +76,11 @@
"avro" : {
type: "avro"
},
- "sequencefile": {
- type : "sequencefile",
- extensions: [ "seq" ]
+ "csvh" : {
+ type: "text",
+ extensions: [ "csvh" ],
+ delimiter: ",",
+ extractHeader: true
}
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/exec/java-exec/src/test/java/org/apache/drill/exec/store/text/TestCsvHeader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/text/TestCsvHeader.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/text/TestCsvHeader.java
new file mode 100644
index 0000000..5dcc9a8
--- /dev/null
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/text/TestCsvHeader.java
@@ -0,0 +1,129 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.text;
+
+import org.apache.drill.BaseTestQuery;
+import org.apache.drill.common.util.FileUtils;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.rpc.user.QueryDataBatch;
+
+import java.util.List;
+
+import org.junit.Before;
+import org.junit.Test;
+
+
+import static org.junit.Assert.assertEquals;
+
+public class TestCsvHeader extends BaseTestQuery{
+
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TestCsvHeader.class);
+ final String OUTPUT_DELIMITER = "|";
+ String root;
+
+ @Before
+ public void initialize() throws Exception {
+ root = FileUtils.getResourceAsFile("/store/text/data/cars.csvh").toURI().toString();
+ test("alter session set `exec.errors.verbose` = true ");
+ }
+
+ @Test //DRILL-951
+ public void testCsvWithHeader() throws Exception {
+ //Pre DRILL-951: Qry: select * from dfs_test.`%s` LIMIT 2
+ String query = String.format("select * from dfs_test.`%s` LIMIT 2", root);
+ List<QueryDataBatch> batches = testSqlWithResults(query);
+
+ String expectedOutput = "Year|Make|Model|Description|Price\n" +
+ "1997|Ford|E350|ac, abs, moon|3000.00\n" +
+ "1999|Chevy|Venture \"Extended Edition\"||4900.00\n";
+
+ validateResults (batches, expectedOutput);
+ }
+
+ @Test //DRILL-951
+ public void testCsvWhereWithHeader() throws Exception {
+ //Pre DRILL-951: Qry: select * from dfs_test.`%s` where columns[1] = 'Chevy'
+ String query = String.format("select * from dfs_test.`%s` where Make = 'Chevy'", root);
+ List<QueryDataBatch> batches = testSqlWithResults(query);
+
+ String expectedOutput = "Year|Make|Model|Description|Price\n" +
+ "1999|Chevy|Venture \"Extended Edition\"||4900.00\n" +
+ "1999|Chevy|Venture \"Extended Edition, Very Large\"||5000.00\n";
+
+ validateResults (batches, expectedOutput);
+ }
+
+ @Test //DRILL-951
+ public void testCsvStarPlusWithHeader() throws Exception {
+ //Pre DRILL-951: Qry: select *,columns[1] from dfs_test.`%s` where columns[1] = 'Chevy'
+ String query = String.format("select *, Make from dfs_test.`%s` where Make = 'Chevy'", root);
+ List<QueryDataBatch> batches = testSqlWithResults(query);
+
+ String expectedOutput = "Year|Make|Model|Description|Price|Make0\n" +
+ "1999|Chevy|Venture \"Extended Edition\"||4900.00|Chevy\n" +
+ "1999|Chevy|Venture \"Extended Edition, Very Large\"||5000.00|Chevy\n";
+
+ validateResults (batches, expectedOutput);
+ }
+
+ @Test //DRILL-951
+ public void testCsvWhereColumnsWithHeader() throws Exception {
+ //Pre DRILL-951: Qry: select columns[1] from dfs_test.`%s` where columns[1] = 'Chevy'
+ String query = String.format("select Make from dfs_test.`%s` where Make = 'Chevy'", root);
+ List<QueryDataBatch> batches = testSqlWithResults(query);
+
+ String expectedOutput = "Make\n" +
+ "Chevy\n" +
+ "Chevy\n";
+
+ validateResults (batches, expectedOutput);
+ }
+
+ @Test //DRILL-951
+ public void testCsvColumnsWithHeader() throws Exception {
+ //Pre DRILL-951: Qry: select columns[0],columns[2],columns[4] from dfs_test.`%s` where columns[1] = 'Chevy'
+ String query = String.format("select `Year`, Model, Price from dfs_test.`%s` where Make = 'Chevy'", root);
+ List<QueryDataBatch> batches = testSqlWithResults(query);
+
+ String expectedOutput = "Year|Model|Price\n" +
+ "1999|Venture \"Extended Edition\"|4900.00\n" +
+ "1999|Venture \"Extended Edition, Very Large\"|5000.00\n";
+
+ validateResults (batches, expectedOutput);
+ }
+
+ @Test //DRILL-951
+ public void testCsvHeaderShortCircuitReads() throws Exception {
+ String query = String.format("select `Year`, Model from dfs_test.`%s` where Make = 'Chevy'", root);
+ List<QueryDataBatch> batches = testSqlWithResults(query);
+
+ String expectedOutput = "Year|Model\n" +
+ "1999|Venture \"Extended Edition\"\n" +
+ "1999|Venture \"Extended Edition, Very Large\"\n";
+
+ validateResults (batches, expectedOutput);
+ }
+
+ private void validateResults (List<QueryDataBatch> batches, String expectedOutput) throws SchemaChangeException {
+ String actualOutput = getResultString(batches, OUTPUT_DELIMITER);
+ //for your and machine's eyes
+ System.out.println(actualOutput);
+ assertEquals(String.format("Result mismatch.\nExpected:\n%s \nReceived:\n%s",
+ expectedOutput, actualOutput), expectedOutput, actualOutput);
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/exec/java-exec/src/test/resources/bootstrap-storage-plugins.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/bootstrap-storage-plugins.json b/exec/java-exec/src/test/resources/bootstrap-storage-plugins.json
index 57b9fbb..76f3116 100644
--- a/exec/java-exec/src/test/resources/bootstrap-storage-plugins.json
+++ b/exec/java-exec/src/test/resources/bootstrap-storage-plugins.json
@@ -51,6 +51,12 @@
"sequencefile": {
type: "sequencefile",
extensions: [ "seq" ]
+ },
+ "csvh" : {
+ type: "text",
+ extensions: [ "csvh" ],
+ delimiter: ",",
+ extractHeader: true
}
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/exec/java-exec/src/test/resources/store/text/data/cars.csvh
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/store/text/data/cars.csvh b/exec/java-exec/src/test/resources/store/text/data/cars.csvh
new file mode 100644
index 0000000..37b82a0
--- /dev/null
+++ b/exec/java-exec/src/test/resources/store/text/data/cars.csvh
@@ -0,0 +1,6 @@
+Year,Make,Model,Description,Price
+1997,Ford,E350,"ac, abs, moon",3000.00
+1999,Chevy,"Venture ""Extended Edition""","",4900.00
+1999,Chevy,"Venture ""Extended Edition, Very Large""",,5000.00
+1996,Jeep,Grand Cherokee,"MUST SELL!
+air, moon roof, loaded",4799.00
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/drill/blob/745bcd1f/pom.xml
----------------------------------------------------------------------
diff --git a/pom.xml b/pom.xml
index e799783..93b710c 100644
--- a/pom.xml
+++ b/pom.xml
@@ -181,6 +181,7 @@
<exclude>**/*.sql</exclude>
<exclude>**/git.properties</exclude>
<exclude>**/*.csv</exclude>
+ <exclude>**/*.csvh</exclude>
<exclude>**/*.tsv</exclude>
<exclude>**/*.txt</exclude>
<exclude>**/*.ssv</exclude>