You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@drill.apache.org by ja...@apache.org on 2013/06/17 18:43:37 UTC
[1/3] DRILL-75 and DRILL-76
Updated Branches:
refs/heads/execwork 719c8beb7 -> 01278ae56
refs/heads/master 3a11d2917 -> 11a1fceec
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/JSONRecordReader.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/JSONRecordReader.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/JSONRecordReader.java
new file mode 100644
index 0000000..42d1be5
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/JSONRecordReader.java
@@ -0,0 +1,423 @@
+package org.apache.drill.exec.store;
+
+import com.carrotsearch.hppc.IntObjectOpenHashMap;
+import com.carrotsearch.hppc.cursors.IntObjectCursor;
+import com.carrotsearch.hppc.cursors.ObjectCursor;
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.core.JsonToken;
+import com.google.common.base.Charsets;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.io.Files;
+import com.google.common.io.InputSupplier;
+import com.google.common.io.Resources;
+import org.apache.drill.common.exceptions.DrillRuntimeException;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.impl.OutputMutator;
+import org.apache.drill.exec.proto.SchemaDefProtos;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.record.vector.*;
+import org.apache.drill.exec.schema.*;
+import org.apache.drill.exec.schema.json.jackson.JacksonHelper;
+
+import java.io.File;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.nio.charset.Charset;
+import java.util.List;
+
+import static com.fasterxml.jackson.core.JsonToken.*;
+
+public class JSONRecordReader implements RecordReader {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(JSONRecordReader.class);
+ private static final int DEFAULT_LENGTH = 256 * 1024; // 256kb
+ public static final Charset UTF_8 = Charset.forName("UTF-8");
+
+ private final String inputPath;
+
+ private final IntObjectOpenHashMap<VectorHolder> valueVectorMap;
+
+ private JsonParser parser;
+ private SchemaIdGenerator generator;
+ private DiffSchema diffSchema;
+ private RecordSchema currentSchema;
+ private List<Field> removedFields;
+ private OutputMutator outputMutator;
+ private BufferAllocator allocator;
+ private int batchSize;
+
+ public JSONRecordReader(FragmentContext fragmentContext, String inputPath, int batchSize) {
+ this.inputPath = inputPath;
+ this.allocator = fragmentContext.getAllocator();
+ this.batchSize = batchSize;
+ valueVectorMap = new IntObjectOpenHashMap<>();
+ }
+
+ public JSONRecordReader(FragmentContext fragmentContext, String inputPath) {
+ this(fragmentContext, inputPath, DEFAULT_LENGTH);
+ }
+
+ private JsonParser getParser() {
+ return parser;
+ }
+
+ @Override
+ public void setup(OutputMutator output) throws ExecutionSetupException {
+ outputMutator = output;
+ currentSchema = new ObjectSchema();
+ diffSchema = new DiffSchema();
+ removedFields = Lists.newArrayList();
+
+ try {
+ InputSupplier<InputStreamReader> input;
+ if (inputPath.startsWith("resource:")) {
+ input = Resources.newReaderSupplier(Resources.getResource(inputPath.substring(9)), Charsets.UTF_8);
+ } else {
+ input = Files.newReaderSupplier(new File(inputPath), Charsets.UTF_8);
+ }
+
+ JsonFactory factory = new JsonFactory();
+ parser = factory.createJsonParser(input.getInput());
+ parser.nextToken(); // Read to the first START_OBJECT token
+ generator = new SchemaIdGenerator();
+ } catch (IOException e) {
+ throw new ExecutionSetupException(e);
+ }
+ }
+
+ @Override
+ public int next() {
+ if (parser.isClosed() || !parser.hasCurrentToken()) {
+ return 0;
+ }
+
+ resetBatch();
+
+ int nextRowIndex = 0;
+
+ try {
+ while (ReadType.OBJECT.readRecord(null, this, null, nextRowIndex++)) {
+ parser.nextToken(); // Read to START_OBJECT token
+
+ if (!parser.hasCurrentToken()) {
+ parser.close();
+ break;
+ }
+ }
+
+ parser.nextToken();
+
+ if (!parser.hasCurrentToken()) {
+ parser.close();
+ }
+
+ // Garbage collect fields never referenced in this batch
+ for (Field field : Iterables.concat(currentSchema.removeUnreadFields(), removedFields)) {
+ diffSchema.addRemovedField(field);
+ outputMutator.removeField(field.getFieldId());
+ }
+
+ } catch (IOException | SchemaChangeException e) {
+ logger.error("Error reading next in Json reader", e);
+ }
+ return nextRowIndex;
+ }
+
+ private void resetBatch() {
+ for (ObjectCursor<VectorHolder> holder : valueVectorMap.values()) {
+ holder.value.reset();
+ }
+
+ currentSchema.resetMarkedFields();
+ diffSchema.reset();
+ removedFields.clear();
+ }
+
+ @Override
+ public void cleanup() {
+ try {
+ parser.close();
+ } catch (IOException e) {
+ logger.warn("Error closing Json parser", e);
+ }
+ }
+
+ private SchemaIdGenerator getGenerator() {
+ return generator;
+ }
+
+ private RecordSchema getCurrentSchema() {
+ return currentSchema;
+ }
+
+ private void setCurrentSchema(RecordSchema schema) {
+ currentSchema = schema;
+ }
+
+ private List<Field> getRemovedFields() {
+ return removedFields;
+ }
+
+ private DiffSchema getDiffSchema() {
+ return diffSchema;
+ }
+
+ public BufferAllocator getAllocator() {
+ return allocator;
+ }
+
+ public OutputMutator getOutputMutator() {
+ return outputMutator;
+ }
+
+ public static enum ReadType {
+ ARRAY(END_ARRAY) {
+ @Override
+ public Field createField(RecordSchema parentSchema, int parentFieldId, IdGenerator<Integer> generator, String prefixFieldName, String fieldName, SchemaDefProtos.MajorType fieldType, int index) {
+ return new OrderedField(parentSchema, parentFieldId, generator, fieldType, prefixFieldName, index);
+ }
+
+ @Override
+ public RecordSchema createSchema() throws IOException {
+ return new ListSchema();
+ }
+ },
+ OBJECT(END_OBJECT) {
+ @Override
+ public Field createField(RecordSchema parentSchema,
+ int parentFieldId,
+ IdGenerator<Integer> generator,
+ String prefixFieldName,
+ String fieldName,
+ SchemaDefProtos.MajorType fieldType,
+ int index) {
+ return new NamedField(parentSchema, parentFieldId, generator, prefixFieldName, fieldName, fieldType);
+ }
+
+ @Override
+ public RecordSchema createSchema() throws IOException {
+ return new ObjectSchema();
+ }
+ };
+
+ private final JsonToken endObject;
+
+ ReadType(JsonToken endObject) {
+ this.endObject = endObject;
+ }
+
+ public JsonToken getEndObject() {
+ return endObject;
+ }
+
+ public boolean readRecord(Field parentField,
+ JSONRecordReader reader,
+ String prefixFieldName,
+ int rowIndex) throws IOException, SchemaChangeException {
+ JsonParser parser = reader.getParser();
+ JsonToken token = parser.nextToken();
+ JsonToken endObject = getEndObject();
+ int colIndex = 0;
+ boolean isFull = false;
+ while (token != endObject) {
+ if (token == FIELD_NAME) {
+ token = parser.nextToken();
+ continue;
+ }
+
+ String fieldName = parser.getCurrentName();
+ SchemaDefProtos.MajorType fieldType = JacksonHelper.getFieldType(token);
+ ReadType readType = null;
+ switch (token) {
+ case START_ARRAY:
+ readType = ReadType.ARRAY;
+ break;
+ case START_OBJECT:
+ readType = ReadType.OBJECT;
+ break;
+ }
+ if (fieldType != null) { // Including nulls
+ isFull = isFull ||
+ !recordData(
+ parentField,
+ readType,
+ reader,
+ fieldType,
+ prefixFieldName,
+ fieldName,
+ rowIndex, colIndex);
+ }
+ token = parser.nextToken();
+ colIndex += 1;
+ }
+ return !isFull;
+ }
+
+ private void removeChildFields(List<Field> removedFields, Field field) {
+ RecordSchema schema = field.getAssignedSchema();
+ if (schema == null) {
+ return;
+ }
+ for (Field childField : schema.getFields()) {
+ removedFields.add(childField);
+ if (childField.hasSchema()) {
+ removeChildFields(removedFields, childField);
+ }
+ }
+ }
+
+ private boolean recordData(Field parentField,
+ JSONRecordReader.ReadType readType,
+ JSONRecordReader reader,
+ SchemaDefProtos.MajorType fieldType,
+ String prefixFieldName,
+ String fieldName,
+ int rowIndex,
+ int colIndex) throws IOException, SchemaChangeException {
+ RecordSchema currentSchema = reader.getCurrentSchema();
+ Field field = currentSchema.getField(fieldName, colIndex);
+ boolean isFieldFound = field != null;
+ List<Field> removedFields = reader.getRemovedFields();
+ int parentFieldId = parentField == null ? 0 : parentField.getFieldId();
+ if (!isFieldFound || !field.getFieldType().equals(fieldType)) {
+ if (isFieldFound) {
+ if (field.hasSchema()) {
+ removeChildFields(removedFields, field);
+ }
+ removedFields.add(field);
+ currentSchema.removeField(field, colIndex);
+ }
+
+ field = createField(
+ currentSchema,
+ parentFieldId,
+ reader.getGenerator(),
+ prefixFieldName,
+ fieldName,
+ fieldType,
+ colIndex
+ );
+
+ reader.recordNewField(field);
+ currentSchema.addField(field);
+ }
+
+ field.setRead(true);
+
+ VectorHolder holder = getOrCreateVectorHolder(reader, field, parentFieldId);
+ if (readType != null) {
+ RecordSchema fieldSchema = field.getAssignedSchema();
+ reader.setCurrentSchema(fieldSchema);
+
+ RecordSchema newSchema = readType.createSchema();
+ field.assignSchemaIfNull(newSchema);
+
+ if (fieldSchema == null) reader.setCurrentSchema(newSchema);
+ readType.readRecord(field, reader, field.getFullFieldName(), rowIndex);
+
+ reader.setCurrentSchema(currentSchema);
+ } else {
+ return addValueToVector(
+ rowIndex,
+ holder,
+ reader.getAllocator(),
+ JacksonHelper.getValueFromFieldType(
+ reader.getParser(),
+ fieldType.getMinorType()
+ ),
+ fieldType.getMinorType()
+ );
+ }
+
+ return true;
+ }
+
+ private static <T> boolean addValueToVector(int index, VectorHolder holder, BufferAllocator allocator, T val, SchemaDefProtos.MinorType minorType) {
+ switch (minorType) {
+ case INT: {
+ holder.incAndCheckLength(32);
+ NullableFixed4 fixed4 = (NullableFixed4) holder.getValueVector();
+ if (val == null) {
+ fixed4.setNull(index);
+ } else {
+ fixed4.setInt(index, (Integer) val);
+ }
+ return holder.hasEnoughSpace(32);
+ }
+ case FLOAT4: {
+ holder.incAndCheckLength(32);
+ NullableFixed4 fixed4 = (NullableFixed4) holder.getValueVector();
+ if (val == null) {
+ fixed4.setNull(index);
+ } else {
+ fixed4.setFloat4(index, (Float) val);
+ }
+ return holder.hasEnoughSpace(32);
+ }
+ case VARCHAR4: {
+ if (val == null) {
+ ((NullableVarLen4) holder.getValueVector()).setNull(index);
+ return (index + 1) * 4 <= holder.getLength();
+ } else {
+ byte[] bytes = ((String) val).getBytes(UTF_8);
+ int length = bytes.length * 8;
+ holder.incAndCheckLength(length);
+ NullableVarLen4 varLen4 = (NullableVarLen4) holder.getValueVector();
+ varLen4.setBytes(index, bytes);
+ return holder.hasEnoughSpace(length);
+ }
+ }
+ case BOOLEAN: {
+ holder.incAndCheckLength(1);
+ NullableBit bit = (NullableBit) holder.getValueVector();
+ if (val == null) {
+ bit.setNull(index);
+ } else if ((Boolean) val) {
+ bit.set(index);
+ }
+ return holder.hasEnoughSpace(1);
+ }
+ default:
+ throw new DrillRuntimeException("Type not supported to add value. Type: " + minorType);
+ }
+ }
+
+ private VectorHolder getOrCreateVectorHolder(JSONRecordReader reader, Field field, int parentFieldId) throws SchemaChangeException {
+ return reader.getOrCreateVectorHolder(field, parentFieldId);
+ }
+
+ public abstract RecordSchema createSchema() throws IOException;
+
+ public abstract Field createField(RecordSchema parentSchema,
+ int parentFieldId,
+ IdGenerator<Integer> generator,
+ String prefixFieldName,
+ String fieldName,
+ SchemaDefProtos.MajorType fieldType,
+ int index);
+ }
+
+ private void recordNewField(Field field) {
+ diffSchema.recordNewField(field);
+ }
+
+ private VectorHolder getOrCreateVectorHolder(Field field, int parentFieldId) throws SchemaChangeException {
+ if (!valueVectorMap.containsKey(field.getFieldId())) {
+ SchemaDefProtos.MajorType type = field.getFieldType();
+ int fieldId = field.getFieldId();
+ MaterializedField f = MaterializedField.create(new SchemaPath(field.getFieldName()), fieldId, parentFieldId, type);
+ ValueVector<?> v = TypeHelper.getNewVector(f, allocator);
+ v.allocateNew(batchSize);
+ VectorHolder holder = new VectorHolder(batchSize, v);
+ valueVectorMap.put(fieldId, holder);
+ outputMutator.addField(fieldId, v);
+ return holder;
+ }
+ return valueVectorMap.lget();
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/RecordReader.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/RecordReader.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/RecordReader.java
index 67c84ed..f0d2daf 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/RecordReader.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/RecordReader.java
@@ -25,10 +25,6 @@ public interface RecordReader {
/**
* Configure the RecordReader with the provided schema and the record batch that should be written to.
*
- * @param knownSchema
- * The set of fields that should be written to as well as the expected types for those fields. In the case
- * that RecordReader has a known schema and the expectedSchema does not match the actual schema, a
- * ExceptionSetupException will be thrown.
* @param output
* The place where output for a particular scan should be written. The record reader is responsible for
* mutating the set of schema values for that particular record.
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/VectorHolder.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/VectorHolder.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/VectorHolder.java
new file mode 100644
index 0000000..59590d2
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/VectorHolder.java
@@ -0,0 +1,59 @@
+/*******************************************************************************
+ * 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;
+
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.record.vector.TypeHelper;
+import org.apache.drill.exec.record.vector.ValueVector;
+
+public class VectorHolder {
+ private int length;
+ private ValueVector vector;
+ private int currentLength;
+
+ VectorHolder(int length, ValueVector<?> vector) {
+ this.length = length;
+ this.vector = vector;
+ }
+
+ public ValueVector getValueVector() {
+ return vector;
+ }
+
+ public void incAndCheckLength(int newLength) {
+ if (!hasEnoughSpace(newLength)) {
+ throw new BatchExceededException(length, currentLength + newLength);
+ }
+ currentLength += newLength;
+ }
+
+ public boolean hasEnoughSpace(int newLength) {
+ return length >= currentLength + newLength;
+ }
+
+ public int getLength() {
+ return length;
+ }
+
+ public void reset() {
+ currentLength = 0;
+ vector.allocateNew(length);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/test/java/org/apache/drill/exec/store/JSONRecordReaderTest.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/test/java/org/apache/drill/exec/store/JSONRecordReaderTest.java b/sandbox/prototype/exec/java-exec/src/test/java/org/apache/drill/exec/store/JSONRecordReaderTest.java
new file mode 100644
index 0000000..a29e9f0
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/test/java/org/apache/drill/exec/store/JSONRecordReaderTest.java
@@ -0,0 +1,290 @@
+package org.apache.drill.exec.store;
+
+import com.beust.jcommander.internal.Lists;
+import mockit.Expectations;
+import mockit.Injectable;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.exception.SchemaChangeException;
+import org.apache.drill.exec.memory.DirectBufferAllocator;
+import org.apache.drill.exec.ops.FragmentContext;
+import org.apache.drill.exec.physical.impl.OutputMutator;
+import org.apache.drill.exec.proto.SchemaDefProtos;
+import org.apache.drill.exec.proto.UserBitShared;
+import org.apache.drill.exec.record.vector.ValueVector;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.nio.charset.Charset;
+import java.util.Arrays;
+import java.util.List;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class JSONRecordReaderTest {
+ private static final Charset UTF_8 = Charset.forName("UTF-8");
+
+ private String getResource(String resourceName) {
+ return "resource:" + resourceName;
+ }
+
+ class MockOutputMutator implements OutputMutator {
+ List<Integer> removedFields = Lists.newArrayList();
+ List<ValueVector> addFields = Lists.newArrayList();
+
+ @Override
+ public void removeField(int fieldId) throws SchemaChangeException {
+ removedFields.add(fieldId);
+ }
+
+ @Override
+ public void addField(int fieldId, ValueVector<?> vector) throws SchemaChangeException {
+ addFields.add(vector);
+ }
+
+ @Override
+ public void setNewSchema() throws SchemaChangeException {
+ }
+
+ List<Integer> getRemovedFields() {
+ return removedFields;
+ }
+
+ List<ValueVector> getAddFields() {
+ return addFields;
+ }
+ }
+
+ private <T> void assertField(ValueVector valueVector, int index, SchemaDefProtos.MinorType expectedMinorType, T value, String name) {
+ assertField(valueVector, index, expectedMinorType, value, name, 0);
+ }
+
+ private <T> void assertField(ValueVector valueVector, int index, SchemaDefProtos.MinorType expectedMinorType, T value, String name, int parentFieldId) {
+ UserBitShared.FieldMetadata metadata = valueVector.getMetadata();
+ SchemaDefProtos.FieldDef def = metadata.getDef();
+ assertEquals(expectedMinorType, def.getMajorType().getMinorType());
+ assertEquals(name, def.getNameList().get(0).getName());
+ assertEquals(parentFieldId, def.getParentId());
+
+ if(expectedMinorType == SchemaDefProtos.MinorType.MAP) {
+ return;
+ }
+
+ T val = (T) valueVector.getObject(index);
+ if (val instanceof byte[]) {
+ assertTrue(Arrays.equals((byte[]) value, (byte[]) val));
+ } else {
+ assertEquals(value, val);
+ }
+ }
+
+ @Test
+ public void testSameSchemaInSameBatch(@Injectable final FragmentContext context) throws IOException, ExecutionSetupException {
+ new Expectations() {
+ {
+ context.getAllocator();
+ returns(new DirectBufferAllocator());
+ }
+ };
+ JSONRecordReader jr = new JSONRecordReader(context, getResource("scan_json_test_1.json"));
+
+ MockOutputMutator mutator = new MockOutputMutator();
+ List<ValueVector> addFields = mutator.getAddFields();
+ jr.setup(mutator);
+ assertEquals(2, jr.next());
+ assertEquals(3, addFields.size());
+ assertField(addFields.get(0), 0, SchemaDefProtos.MinorType.INT, 123, "test");
+ assertField(addFields.get(1), 0, SchemaDefProtos.MinorType.BOOLEAN, 1, "b");
+ assertField(addFields.get(2), 0, SchemaDefProtos.MinorType.VARCHAR4, "hi!".getBytes(UTF_8), "c");
+ assertField(addFields.get(0), 1, SchemaDefProtos.MinorType.INT, 1234, "test");
+ assertField(addFields.get(1), 1, SchemaDefProtos.MinorType.BOOLEAN, 0, "b");
+ assertField(addFields.get(2), 1, SchemaDefProtos.MinorType.VARCHAR4, "drill!".getBytes(UTF_8), "c");
+
+ assertEquals(0, jr.next());
+ assertTrue(mutator.getRemovedFields().isEmpty());
+ }
+
+ @Test
+ public void testChangedSchemaInSameBatch(@Injectable final FragmentContext context) throws IOException, ExecutionSetupException {
+ new Expectations() {
+ {
+ context.getAllocator();
+ returns(new DirectBufferAllocator());
+ }
+ };
+
+ JSONRecordReader jr = new JSONRecordReader(context, getResource("scan_json_test_2.json"));
+ MockOutputMutator mutator = new MockOutputMutator();
+ List<ValueVector> addFields = mutator.getAddFields();
+
+ jr.setup(mutator);
+ assertEquals(3, jr.next());
+ assertEquals(7, addFields.size());
+ assertField(addFields.get(0), 0, SchemaDefProtos.MinorType.INT, 123, "test");
+ assertField(addFields.get(1), 0, SchemaDefProtos.MinorType.INT, 1, "b");
+ assertField(addFields.get(2), 0, SchemaDefProtos.MinorType.FLOAT4, (float) 2.15, "c");
+ assertField(addFields.get(3), 0, SchemaDefProtos.MinorType.BOOLEAN, 1, "bool");
+ assertField(addFields.get(4), 0, SchemaDefProtos.MinorType.VARCHAR4, "test1".getBytes(UTF_8), "str1");
+
+ assertField(addFields.get(0), 1, SchemaDefProtos.MinorType.INT, 1234, "test");
+ assertField(addFields.get(1), 1, SchemaDefProtos.MinorType.INT, 3, "b");
+ assertField(addFields.get(3), 1, SchemaDefProtos.MinorType.BOOLEAN, 0, "bool");
+ assertField(addFields.get(4), 1, SchemaDefProtos.MinorType.VARCHAR4, "test2".getBytes(UTF_8), "str1");
+ assertField(addFields.get(5), 1, SchemaDefProtos.MinorType.INT, 4, "d");
+
+ assertField(addFields.get(0), 2, SchemaDefProtos.MinorType.INT, 12345, "test");
+ assertField(addFields.get(2), 2, SchemaDefProtos.MinorType.FLOAT4, (float) 5.16, "c");
+ assertField(addFields.get(3), 2, SchemaDefProtos.MinorType.BOOLEAN, 1, "bool");
+ assertField(addFields.get(5), 2, SchemaDefProtos.MinorType.INT, 6, "d");
+ assertField(addFields.get(6), 2, SchemaDefProtos.MinorType.VARCHAR4, "test3".getBytes(UTF_8), "str2");
+ assertTrue(mutator.getRemovedFields().isEmpty());
+ assertEquals(0, jr.next());
+ }
+
+ @Test
+ public void testChangedSchemaInTwoBatches(@Injectable final FragmentContext context) throws IOException, ExecutionSetupException {
+ new Expectations() {
+ {
+ context.getAllocator();
+ returns(new DirectBufferAllocator());
+ }
+ };
+
+ JSONRecordReader jr = new JSONRecordReader(context, getResource("scan_json_test_2.json"), 64); // batch only fits 1 int
+ MockOutputMutator mutator = new MockOutputMutator();
+ List<ValueVector> addFields = mutator.getAddFields();
+ List<Integer> removedFields = mutator.getRemovedFields();
+
+ jr.setup(mutator);
+ assertEquals(1, jr.next());
+ assertEquals(5, addFields.size());
+ assertField(addFields.get(0), 0, SchemaDefProtos.MinorType.INT, 123, "test");
+ assertField(addFields.get(1), 0, SchemaDefProtos.MinorType.INT, 1, "b");
+ assertField(addFields.get(2), 0, SchemaDefProtos.MinorType.FLOAT4, (float) 2.15, "c");
+ assertField(addFields.get(3), 0, SchemaDefProtos.MinorType.BOOLEAN, 1, "bool");
+ assertField(addFields.get(4), 0, SchemaDefProtos.MinorType.VARCHAR4, "test1".getBytes(UTF_8), "str1");
+ assertTrue(removedFields.isEmpty());
+ assertEquals(1, jr.next());
+ assertEquals(6, addFields.size());
+ assertField(addFields.get(0), 0, SchemaDefProtos.MinorType.INT, 1234, "test");
+ assertField(addFields.get(1), 0, SchemaDefProtos.MinorType.INT, 3, "b");
+ assertField(addFields.get(3), 0, SchemaDefProtos.MinorType.BOOLEAN, 0, "bool");
+ assertField(addFields.get(4), 0, SchemaDefProtos.MinorType.VARCHAR4, "test2".getBytes(UTF_8), "str1");
+ assertField(addFields.get(5), 0, SchemaDefProtos.MinorType.INT, 4, "d");
+ assertEquals(1, removedFields.size());
+ assertEquals(3, (int) removedFields.get(0));
+ removedFields.clear();
+ assertEquals(1, jr.next());
+ assertEquals(8, addFields.size()); // The reappearing of field 'c' is also included
+ assertField(addFields.get(0), 0, SchemaDefProtos.MinorType.INT, 12345, "test");
+ assertField(addFields.get(3), 0, SchemaDefProtos.MinorType.BOOLEAN, 1, "bool");
+ assertField(addFields.get(5), 0, SchemaDefProtos.MinorType.INT, 6, "d");
+ assertField(addFields.get(6), 0, SchemaDefProtos.MinorType.FLOAT4, (float) 5.16, "c");
+ assertField(addFields.get(7), 0, SchemaDefProtos.MinorType.VARCHAR4, "test3".getBytes(UTF_8), "str2");
+ assertEquals(2, removedFields.size());
+ assertTrue(removedFields.contains(5));
+ assertTrue(removedFields.contains(2));
+ assertEquals(0, jr.next());
+ }
+
+ @Test
+ public void testNestedFieldInSameBatch(@Injectable final FragmentContext context) throws ExecutionSetupException {
+ new Expectations() {
+ {
+ context.getAllocator();
+ returns(new DirectBufferAllocator());
+ }
+ };
+
+ JSONRecordReader jr = new JSONRecordReader(context, getResource("scan_json_test_3.json"));
+
+ MockOutputMutator mutator = new MockOutputMutator();
+ List<ValueVector> addFields = mutator.getAddFields();
+ jr.setup(mutator);
+ assertEquals(2, jr.next());
+ assertEquals(5, addFields.size());
+ assertField(addFields.get(0), 0, SchemaDefProtos.MinorType.INT, 123, "test");
+ assertField(addFields.get(1), 0, SchemaDefProtos.MinorType.MAP, null, "a");
+ assertField(addFields.get(2), 0, SchemaDefProtos.MinorType.VARCHAR4, "test".getBytes(UTF_8), "b", 2);
+ assertField(addFields.get(3), 0, SchemaDefProtos.MinorType.MAP, null, "a", 2);
+ assertField(addFields.get(4), 0, SchemaDefProtos.MinorType.BOOLEAN, 1, "d", 4);
+ assertField(addFields.get(0), 1, SchemaDefProtos.MinorType.INT, 1234, "test");
+ assertField(addFields.get(1), 1, SchemaDefProtos.MinorType.MAP, null, "a");
+ assertField(addFields.get(2), 1, SchemaDefProtos.MinorType.VARCHAR4, "test2".getBytes(UTF_8), "b", 2);
+ assertField(addFields.get(3), 1, SchemaDefProtos.MinorType.MAP, null, "a", 2);
+ assertField(addFields.get(4), 1, SchemaDefProtos.MinorType.BOOLEAN, 0, "d", 4);
+
+ assertEquals(0, jr.next());
+ assertTrue(mutator.getRemovedFields().isEmpty());
+ }
+
+ /*
+
+ @Test
+ public void testScanJsonRemovedOneField() throws IOException {
+ ScanJson sj = new ScanJson(getResource("scan_json_test_3.json"));
+ PhysicalOperatorIterator iterator = sj.getIterator();
+ expectSchemaChanged(iterator);
+ DiffSchema diffSchema = expectSchemaChanged(iterator).getSchemaChanges();
+ assertEquals(0, diffSchema.getAddedFields().size());
+ assertEquals(1, diffSchema.getRemovedFields().size());
+ assertEquals(PhysicalOperatorIterator.NextOutcome.NONE_LEFT, iterator.next());
+ }
+
+ @Test
+ public void testScanJsonAddOneRemoveOne() throws IOException {
+ ScanJson sj = new ScanJson(getResource("scan_json_test_4.json"));
+ PhysicalOperatorIterator iterator = sj.getIterator();
+ expectSchemaChanged(iterator);
+ DiffSchema diffSchema = expectSchemaChanged(iterator).getSchemaChanges();
+ assertEquals(1, diffSchema.getAddedFields().size());
+ assertEquals(1, diffSchema.getRemovedFields().size());
+ assertEquals(PhysicalOperatorIterator.NextOutcome.NONE_LEFT, iterator.next());
+ }
+
+ @Test
+ public void testScanJsonCycleAdditions() throws IOException {
+ ScanJson sj = new ScanJson(getResource("scan_json_test_5.json"));
+ PhysicalOperatorIterator iterator = sj.getIterator();
+ expectSchemaChanged(iterator);
+ DiffSchema diffSchema = expectSchemaChanged(iterator).getSchemaChanges();
+ assertEquals(1, diffSchema.getAddedFields().size());
+ assertEquals(1, diffSchema.getRemovedFields().size());
+ diffSchema = expectSchemaChanged(iterator).getSchemaChanges();
+ assertEquals(1, diffSchema.getAddedFields().size());
+ assertEquals(Field.FieldType.FLOAT, diffSchema.getAddedFields().get(0).getFieldType());
+ assertEquals("test2", diffSchema.getAddedFields().get(0).getFieldName());
+ assertEquals(1, diffSchema.getRemovedFields().size());
+ assertEquals(Field.FieldType.BOOLEAN, diffSchema.getRemovedFields().get(0).getFieldType());
+ assertEquals("test3", diffSchema.getRemovedFields().get(0).getFieldName());
+ assertEquals(PhysicalOperatorIterator.NextOutcome.NONE_LEFT, iterator.next());
+ }
+
+ @Test
+ public void testScanJsonModifiedOneFieldType() throws IOException {
+ ScanJson sj = new ScanJson(getResource("scan_json_test_6.json"));
+ PhysicalOperatorIterator iterator = sj.getIterator();
+ expectSchemaChanged(iterator);
+ DiffSchema diffSchema = expectSchemaChanged(iterator).getSchemaChanges();
+ List<Field> addedFields = diffSchema.getAddedFields();
+ assertEquals(4, addedFields.size());
+ List<Field> removedFields = diffSchema.getRemovedFields();
+ assertEquals(4, removedFields.size());
+ assertFieldExists("test", Field.FieldType.STRING, addedFields);
+ assertFieldExists("test2", Field.FieldType.BOOLEAN, addedFields);
+ assertFieldExists("b", Field.FieldType.ARRAY, addedFields);
+ assertFieldExists("[0]", Field.FieldType.INTEGER, addedFields);
+ assertFieldExists("test", Field.FieldType.INTEGER, removedFields);
+ assertFieldExists("test2", Field.FieldType.ARRAY, removedFields);
+ assertFieldExists("b", Field.FieldType.INTEGER, removedFields);
+ assertFieldExists("[0]", Field.FieldType.INTEGER, removedFields);
+ assertEquals(PhysicalOperatorIterator.NextOutcome.NONE_LEFT, iterator.next());
+ }
+
+ private void expectSchemaChanged(PhysicalOperatorIterator iterator) throws IOException {
+ }
+
+ private void expectDataRecord(PhysicalOperatorIterator iterator) throws IOException {
+ }
+*/
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_1.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_1.json b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_1.json
new file mode 100644
index 0000000..2ab53dd
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_1.json
@@ -0,0 +1,10 @@
+{
+ "test": 123,
+ "b": true,
+ "c": "hi!"
+}
+{
+ "test": 1234,
+ "b": false,
+ "c": "drill!"
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_2.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_2.json b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_2.json
new file mode 100644
index 0000000..ce46bbf
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_2.json
@@ -0,0 +1,21 @@
+{
+ "test": 123,
+ "b": 1,
+ "c": 2.15,
+ "bool": true,
+ "str1": "test1"
+}
+{
+ "test": 1234,
+ "b": 3,
+ "d": 4,
+ "bool": false,
+ "str1": "test2"
+}
+{
+ "test": 12345,
+ "c": 5.16,
+ "d": 6,
+ "bool": true,
+ "str2": "test3"
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_3.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_3.json b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_3.json
new file mode 100644
index 0000000..cb7897e
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_3.json
@@ -0,0 +1,18 @@
+{
+ "test": 123,
+ "a": {
+ "b": "test",
+ "a": {
+ "d": true
+ }
+ }
+}
+{
+ "test": 1234,
+ "a": {
+ "b": "test2",
+ "a": {
+ "d": false
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_4.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_4.json b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_4.json
new file mode 100644
index 0000000..0fb3202
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_4.json
@@ -0,0 +1,14 @@
+{
+ "test": 123,
+ "test2": [1,2,3],
+ "a": {
+ "b": 1
+ }
+}
+{
+ "test": 1234,
+ "test3": false,
+ "a": {
+ "b": 2
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_5.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_5.json b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_5.json
new file mode 100644
index 0000000..ae1aaf2
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_5.json
@@ -0,0 +1,21 @@
+{
+ "test": 123,
+ "test2": [1,2,3],
+ "a": {
+ "b": 1
+ }
+}
+{
+ "test": 1234,
+ "test3": false,
+ "a": {
+ "b": 2
+ }
+}
+{
+ "test": 1234,
+ "test2": 1.5,
+ "a": {
+ "b": 2
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_6.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_6.json b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_6.json
new file mode 100644
index 0000000..68b53d4
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/test/resources/scan_json_test_6.json
@@ -0,0 +1,14 @@
+{
+ "test": 123,
+ "test2": [1,2,3],
+ "a": {
+ "b": 1
+ }
+}
+{
+ "test": "abc",
+ "test2": false,
+ "a": {
+ "b": [1,2,3,4]
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/ref/src/test/resources/donuts.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/ref/src/test/resources/donuts.json b/sandbox/prototype/exec/ref/src/test/resources/donuts.json
index 2d98b59..d16ef46 100644
--- a/sandbox/prototype/exec/ref/src/test/resources/donuts.json
+++ b/sandbox/prototype/exec/ref/src/test/resources/donuts.json
@@ -1,11 +1,11 @@
-
- {
+{
"id": "0001",
"type": "donut",
"name": "Cake",
"ppu": 0.55,
"sales": 35,
+
"batters":
{
"batter":
@@ -56,6 +56,7 @@
"ppu": 0.55,
"sales": 300,
+
"batters":
{
"batter":
@@ -79,6 +80,7 @@
"ppu": 0.69,
"sales": 14,
+
"batters":
{
"batter":
@@ -115,6 +117,7 @@
"ppu": 1.00,
"sales": 700,
+
"batters":
{
"batter":
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/ref/src/test/resources/simple_plan.json
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/ref/src/test/resources/simple_plan.json b/sandbox/prototype/exec/ref/src/test/resources/simple_plan.json
index 0f38476..c0837e6 100644
--- a/sandbox/prototype/exec/ref/src/test/resources/simple_plan.json
+++ b/sandbox/prototype/exec/ref/src/test/resources/simple_plan.json
@@ -78,3 +78,4 @@
}
]
}
+
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/pom.xml
----------------------------------------------------------------------
diff --git a/sandbox/prototype/pom.xml b/sandbox/prototype/pom.xml
index 25f156d..105016e 100644
--- a/sandbox/prototype/pom.xml
+++ b/sandbox/prototype/pom.xml
@@ -28,7 +28,17 @@
<url>http://conjars.org/repo</url>
<layout>default</layout>
</repository>
- </repositories>
+ <repository>
+ <id>sonatype-nexus-snapshots</id>
+ <url>https://oss.sonatype.org/content/repositories/snapshots</url>
+ <releases>
+ <enabled>false</enabled>
+ </releases>
+ <snapshots>
+ <enabled>true</enabled>
+ </snapshots>
+ </repository>
+ </repositories>
<properties>
<dep.junit.version>4.11</dep.junit.version>
@@ -123,6 +133,12 @@
<!-- Test Dependencies -->
<dependency>
+ <groupId>com.googlecode.jmockit</groupId>
+ <artifactId>jmockit</artifactId>
+ <version>1.2</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
<groupId>junit</groupId>
<artifactId>junit</artifactId>
<version>${dep.junit.version}</version>
[2/3] git commit: DRILL-75 and DRILL-76
Posted by ja...@apache.org.
DRILL-75 and DRILL-76
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/01278ae5
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/01278ae5
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/01278ae5
Branch: refs/heads/execwork
Commit: 01278ae56f0ac3d16ff5e05722055fee1332cd9a
Parents: 719c8be
Author: Jacques Nadeau <ja...@apache.org>
Authored: Mon Jun 17 09:10:15 2013 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Mon Jun 17 09:10:15 2013 -0700
----------------------------------------------------------------------
.../apache/drill/common/logical/data/Scan.java | 4 +-
.../org/apache/drill/exec/record/DeadBuf.java | 2 +-
.../record/vector/AbstractFixedValueVector.java | 8 +-
.../exec/record/vector/BaseValueVector.java | 5 +-
.../apache/drill/exec/record/vector/Fixed4.java | 12 +-
.../drill/exec/record/vector/NullableBit.java | 20 +
.../exec/record/vector/NullableFixed4.java | 17 +-
.../exec/record/vector/NullableValueVector.java | 12 +-
.../exec/record/vector/NullableVarLen4.java | 21 +
.../drill/exec/record/vector/TypeHelper.java | 33 +-
.../drill/exec/record/vector/VarLen1.java | 6 +-
.../drill/exec/record/vector/VarLen2.java | 6 +-
.../drill/exec/record/vector/VarLen4.java | 18 +-
.../exec/record/vector/VariableVector.java | 33 +-
.../drill/exec/rpc/InboundRpcMessage.java | 2 +-
.../apache/drill/exec/schema/DiffSchema.java | 1 +
.../org/apache/drill/exec/schema/Field.java | 74 +---
.../apache/drill/exec/schema/IdGenerator.java | 25 +-
.../apache/drill/exec/schema/ListSchema.java | 5 +-
.../apache/drill/exec/schema/NamedField.java | 21 +-
.../apache/drill/exec/schema/OrderedField.java | 19 +-
.../drill/exec/schema/SchemaRecorder.java | 122 ------
.../exec/schema/json/jackson/JacksonHelper.java | 62 ++-
.../schema/json/jackson/PhysicalOperator.java | 36 --
.../json/jackson/PhysicalOperatorIterator.java | 31 --
.../exec/schema/json/jackson/ScanJson.java | 203 ---------
.../transform/ProtobufSchemaTransformer.java | 109 -----
.../schema/transform/SchemaTransformer.java | 30 --
.../drill/exec/server/DrillbitContext.java | 4 +-
.../exec/store/BatchExceededException.java | 9 +
.../drill/exec/store/JSONRecordReader.java | 423 +++++++++++++++++++
.../apache/drill/exec/store/RecordReader.java | 4 -
.../apache/drill/exec/store/VectorHolder.java | 59 +++
.../drill/exec/store/JSONRecordReaderTest.java | 290 +++++++++++++
.../src/test/resources/scan_json_test_1.json | 10 +
.../src/test/resources/scan_json_test_2.json | 21 +
.../src/test/resources/scan_json_test_3.json | 18 +
.../src/test/resources/scan_json_test_4.json | 14 +
.../src/test/resources/scan_json_test_5.json | 21 +
.../src/test/resources/scan_json_test_6.json | 14 +
.../exec/ref/src/test/resources/donuts.json | 7 +-
.../ref/src/test/resources/simple_plan.json | 1 +
sandbox/prototype/pom.xml | 18 +-
43 files changed, 1162 insertions(+), 688 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java
index c8d396b..7933e40 100644
--- a/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java
+++ b/sandbox/prototype/common/src/main/java/org/apache/drill/common/logical/data/Scan.java
@@ -6,9 +6,9 @@
* 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.
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/DeadBuf.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/DeadBuf.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/DeadBuf.java
index dafb68c..0c2adcf 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/DeadBuf.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/DeadBuf.java
@@ -34,7 +34,7 @@ import io.netty.buffer.ByteBufIndexFinder;
public class DeadBuf implements ByteBuf {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DeadBuf.class);
- private static final String ERROR_MESSAGE = "Attemped to access a DeadBuf. This would happen if you attempted to interact with a buffer that has been moved or not yet initialized.";
+ private static final String ERROR_MESSAGE = "Attemped to access a DeadBuf. This would happen if you attempted to interact with a buffer that has been moved or not yet initialized.";
public static final DeadBuf DEAD_BUFFER = new DeadBuf();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/AbstractFixedValueVector.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/AbstractFixedValueVector.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/AbstractFixedValueVector.java
index b32f067..8748285 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/AbstractFixedValueVector.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/AbstractFixedValueVector.java
@@ -37,8 +37,12 @@ abstract class AbstractFixedValueVector<T extends AbstractFixedValueVector<T>> e
super(field, allocator);
this.widthInBits = widthInBits;
}
-
- @Override
+
+ public int getWidthInBits() {
+ return widthInBits;
+ }
+
+ @Override
protected int getAllocationSize(int valueCount) {
return (int) Math.ceil(valueCount*widthInBits*1.0/8);
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/BaseValueVector.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/BaseValueVector.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/BaseValueVector.java
index b001add..69cd628 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/BaseValueVector.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/BaseValueVector.java
@@ -42,8 +42,7 @@ public abstract class BaseValueVector<T extends BaseValueVector<T>> implements V
public final void allocateNew(int valueCount){
int allocationSize = getAllocationSize(valueCount);
- ByteBuf newBuf = allocator.buffer(allocationSize);
- newBuf.retain();
+ ByteBuf newBuf = allocator.buffer(allocationSize);
resetAllocation(valueCount, newBuf);
}
@@ -54,7 +53,7 @@ public abstract class BaseValueVector<T extends BaseValueVector<T>> implements V
/**
* Update the current buffer allocation utilize the provided allocation.
- * @param valueCount
+ * @param maxValueCount
* @param buf
*/
protected final void resetAllocation(int maxValueCount, ByteBuf buf){
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/Fixed4.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/Fixed4.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/Fixed4.java
index 650029d..ff54784 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/Fixed4.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/Fixed4.java
@@ -18,6 +18,7 @@
package org.apache.drill.exec.record.vector;
import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.proto.SchemaDefProtos;
import org.apache.drill.exec.record.MaterializedField;
public class Fixed4 extends AbstractFixedValueVector<Fixed4>{
@@ -38,18 +39,21 @@ public class Fixed4 extends AbstractFixedValueVector<Fixed4>{
}
public final void setFloat4(int index, float value){
- index*=8;
+ index*=4;
data.setFloat(index, value);
}
public final float getFloat4(int index){
- index*=8;
+ index*=4;
return data.getFloat(index);
}
@Override
public Object getObject(int index) {
- return getInt(index);
+ if (field != null && field.getType().getMinorType() == SchemaDefProtos.MinorType.FLOAT4) {
+ return getFloat4(index);
+ } else {
+ return getInt(index);
+ }
}
-
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableBit.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableBit.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableBit.java
new file mode 100644
index 0000000..ac8c6b6
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableBit.java
@@ -0,0 +1,20 @@
+package org.apache.drill.exec.record.vector;
+
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.record.MaterializedField;
+
+public class NullableBit extends NullableValueVector<NullableBit, Bit>{
+ public NullableBit(MaterializedField field, BufferAllocator allocator) {
+ super(field, allocator);
+ }
+
+ @Override
+ protected Bit getNewValueVector(BufferAllocator allocator) {
+ return new Bit(this.field, allocator);
+ }
+
+ public void set(int index) {
+ setNotNull(index);
+ value.set(index);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableFixed4.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableFixed4.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableFixed4.java
index cc18538..d09b04a 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableFixed4.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableFixed4.java
@@ -6,9 +6,9 @@
* 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.
@@ -24,14 +24,21 @@ public final class NullableFixed4 extends NullableValueVector<NullableFixed4, Fi
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(NullableFixed4.class);
public NullableFixed4(MaterializedField field, BufferAllocator allocator) {
- super(field, allocator, NullableFixed4.class);
+ super(field, allocator);
}
@Override
protected Fixed4 getNewValueVector(BufferAllocator allocator) {
- return new Fixed4(null, allocator);
+ return new Fixed4(this.field, allocator);
}
+ public void setInt(int index, int newVal) {
+ setNotNull(index);
+ value.setInt(index, newVal);
+ }
-
+ public void setFloat4(int index, float val) {
+ setNotNull(index);
+ value.setFloat4(index, val);
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableValueVector.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableValueVector.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableValueVector.java
index 692ab87..8c3d7f6 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableValueVector.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableValueVector.java
@@ -32,7 +32,7 @@ abstract class NullableValueVector<T extends NullableValueVector<T, E>, E extend
protected Bit bits;
protected E value;
- public NullableValueVector(MaterializedField field, BufferAllocator allocator, Class<T> valueClass) {
+ public NullableValueVector(MaterializedField field, BufferAllocator allocator) {
super(field, allocator);
bits = new Bit(null, allocator);
value = getNewValueVector(allocator);
@@ -40,6 +40,14 @@ abstract class NullableValueVector<T extends NullableValueVector<T, E>, E extend
protected abstract E getNewValueVector(BufferAllocator allocator);
+ public void setNull(int index) {
+ bits.set(index);
+ }
+
+ public void setNotNull(int index) {
+ bits.clear(index);
+ }
+
public int isNull(int index){
return bits.getBit(index);
}
@@ -89,7 +97,7 @@ abstract class NullableValueVector<T extends NullableValueVector<T, E>, E extend
@Override
public Object getObject(int index) {
- if(isNull(index) == 0){
+ if(isNull(index) == 1){
return null;
}else{
return value.getObject(index);
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableVarLen4.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableVarLen4.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableVarLen4.java
new file mode 100644
index 0000000..f5a2770
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/NullableVarLen4.java
@@ -0,0 +1,21 @@
+package org.apache.drill.exec.record.vector;
+
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.record.MaterializedField;
+
+public class NullableVarLen4 extends NullableValueVector<NullableVarLen4, VarLen4> {
+
+ public NullableVarLen4(MaterializedField field, BufferAllocator allocator) {
+ super(field, allocator);
+ }
+
+ @Override
+ protected VarLen4 getNewValueVector(BufferAllocator allocator) {
+ return new VarLen4(field, allocator);
+ }
+
+ public void setBytes(int index, byte[] bytes) {
+ setNotNull(index);
+ value.setBytes(index, bytes);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/TypeHelper.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/TypeHelper.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/TypeHelper.java
index 8e89c41..543a222 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/TypeHelper.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/TypeHelper.java
@@ -6,9 +6,9 @@
* 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.
@@ -25,11 +25,11 @@ import org.apache.drill.exec.record.MaterializedField;
public class TypeHelper {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(TypeHelper.class);
-
+
private static final int WIDTH_ESTIMATE_1 = 10;
private static final int WIDTH_ESTIMATE_2 = 50000;
private static final int WIDTH_ESTIMATE_4 = 1024*1024;
-
+
public static int getSize(MajorType major){
switch(major.getMinorType()){
case TINYINT: return 1;
@@ -65,11 +65,11 @@ public class TypeHelper {
case PROTO2: return 2 + WIDTH_ESTIMATE_2;
case PROTO4: return 4 + WIDTH_ESTIMATE_4;
case MSGPACK2: return 2 + WIDTH_ESTIMATE_2;
- case MSGPACK4: return 4 + WIDTH_ESTIMATE_4;
+ case MSGPACK4: return 4 + WIDTH_ESTIMATE_4;
}
return 4;
}
-
+
public static Class<?> getValueVectorClass(MinorType type, DataMode mode){
switch(mode){
case OPTIONAL:
@@ -189,12 +189,12 @@ public class TypeHelper {
break;
default:
break;
-
+
}
throw new UnsupportedOperationException();
}
-
-
+
+
public static ValueVector<?> getNewVector(MaterializedField field, BufferAllocator allocator){
MajorType type = field.getType();
switch(type.getMode()){
@@ -233,18 +233,27 @@ public class TypeHelper {
case PROTO2: return new VarLen2(field, allocator);
case PROTO4: return new VarLen4(field, allocator);
case MSGPACK2: return new VarLen2(field, allocator);
- case MSGPACK4: return new VarLen4(field, allocator);
+ case MSGPACK4: return new VarLen4(field, allocator);
}
break;
case REPEATED:
+ switch(type.getMinorType()) {
+ case MAP: return new RepeatMap(field, allocator);
+ }
break;
case OPTIONAL:
+ switch(type.getMinorType()) {
+ case BOOLEAN: return new NullableBit(field, allocator);
+ case INT: return new NullableFixed4(field, allocator);
+ case FLOAT4: return new NullableFixed4(field, allocator);
+ case VARCHAR4: return new NullableVarLen4(field, allocator);
+ }
break;
default:
break;
-
+
}
- throw new UnsupportedOperationException();
+ throw new UnsupportedOperationException(type.getMinorType() + " type is not supported. Mode: " + type.getMode());
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen1.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen1.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen1.java
index d87029d..6c2b39e 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen1.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen1.java
@@ -24,13 +24,13 @@ public class VarLen1 extends VariableVector<VarLen1, Fixed1>{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(VarLen1.class);
public VarLen1(MaterializedField field, BufferAllocator allocator) {
- super(field, allocator);
+ super(field, allocator, 1);
}
@Override
protected Fixed1 getNewLengthVector(BufferAllocator allocator) {
return new Fixed1(null, allocator);
}
-
-
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen2.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen2.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen2.java
index ebd440a..3bcdd94 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen2.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen2.java
@@ -24,13 +24,13 @@ public class VarLen2 extends VariableVector<VarLen2, Fixed2>{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(VarLen2.class);
public VarLen2(MaterializedField field, BufferAllocator allocator) {
- super(field, allocator);
+ super(field, allocator, 2);
}
@Override
protected Fixed2 getNewLengthVector(BufferAllocator allocator) {
return new Fixed2(null, allocator);
}
-
-
+
+
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen4.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen4.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen4.java
index b3cd712..71ed7a1 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen4.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VarLen4.java
@@ -20,17 +20,29 @@ package org.apache.drill.exec.record.vector;
import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.record.MaterializedField;
+import static com.google.common.base.Preconditions.checkArgument;
+
public class VarLen4 extends VariableVector<VarLen4, Fixed4>{
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(VarLen4.class);
public VarLen4(MaterializedField field, BufferAllocator allocator) {
- super(field, allocator);
+ super(field, allocator, 4);
}
@Override
protected Fixed4 getNewLengthVector(BufferAllocator allocator) {
return new Fixed4(null, allocator);
}
-
-
+
+ public void setBytes(int index, byte[] bytes) {
+ checkArgument(index >= 0);
+ if(index == 0) {
+ lengthVector.setInt(0, bytes.length);
+ data.setBytes(0, bytes);
+ } else {
+ int previousOffset = lengthVector.getInt(index - 1);
+ lengthVector.setInt(index, previousOffset + bytes.length);
+ data.setBytes(previousOffset, bytes);
+ }
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VariableVector.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VariableVector.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VariableVector.java
index 4247f14..dc8fec4 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VariableVector.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/record/vector/VariableVector.java
@@ -23,6 +23,9 @@ import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.record.DeadBuf;
import org.apache.drill.exec.record.MaterializedField;
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
/**
* A vector of variable length bytes. Constructed as a vector of lengths or positions and a vector of values. Random access is only possible if the variable vector stores positions as opposed to lengths.
*/
@@ -31,12 +34,12 @@ public abstract class VariableVector<T extends VariableVector<T, E>, E extends B
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(VariableVector.class);
protected final E lengthVector;
- private ByteBuf values = DeadBuf.DEAD_BUFFER;
protected int expectedValueLength;
- public VariableVector(MaterializedField field, BufferAllocator allocator) {
+ public VariableVector(MaterializedField field, BufferAllocator allocator, int expectedValueLength) {
super(field, allocator);
this.lengthVector = getNewLengthVector(allocator);
+ this.expectedValueLength = expectedValueLength;
}
protected abstract E getNewLengthVector(BufferAllocator allocator);
@@ -50,7 +53,7 @@ public abstract class VariableVector<T extends VariableVector<T, E>, E extends B
protected void childResetAllocation(int valueCount, ByteBuf buf) {
int firstSize = lengthVector.getAllocationSize(valueCount);
lengthVector.resetAllocation(valueCount, buf.slice(0, firstSize));
- values = buf.slice(firstSize, expectedValueLength * valueCount);
+ data = buf.slice(firstSize, expectedValueLength * valueCount);
}
@Override
@@ -62,16 +65,15 @@ public abstract class VariableVector<T extends VariableVector<T, E>, E extends B
@Override
protected void childClear() {
lengthVector.clear();
- if(values != DeadBuf.DEAD_BUFFER){
- values.release();
- values = DeadBuf.DEAD_BUFFER;
+ if(data != DeadBuf.DEAD_BUFFER){
+ data.release();
+ data = DeadBuf.DEAD_BUFFER;
}
}
-
@Override
public ByteBuf[] getBuffers() {
- return new ByteBuf[]{lengthVector.data, values};
+ return new ByteBuf[]{lengthVector.data, data};
}
@Override
@@ -81,13 +83,20 @@ public abstract class VariableVector<T extends VariableVector<T, E>, E extends B
}
public void setTotalBytes(int totalBytes){
- values.writerIndex(totalBytes);
+ data.writerIndex(totalBytes);
}
@Override
public Object getObject(int index) {
- return null;
+ checkArgument(index >= 0);
+ int startIdx = 0;
+ if(index > 0) {
+ startIdx = (int) lengthVector.getObject(index - 1);
+ }
+ int size = (int) lengthVector.getObject(index) - startIdx;
+ checkState(size >= 0);
+ byte[] dst = new byte[size];
+ data.getBytes(startIdx, dst, 0, size);
+ return dst;
}
-
-
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java
index be1ff6b..650a8ea 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/rpc/InboundRpcMessage.java
@@ -6,7 +6,7 @@
* 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
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/DiffSchema.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/DiffSchema.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/DiffSchema.java
index 016e097..b654a92 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/DiffSchema.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/DiffSchema.java
@@ -19,6 +19,7 @@
package org.apache.drill.exec.schema;
import com.google.common.collect.Lists;
+import org.apache.drill.exec.physical.impl.OutputMutator;
import java.util.List;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/Field.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/Field.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/Field.java
index e19c099..eef0634 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/Field.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/Field.java
@@ -20,46 +20,47 @@ package org.apache.drill.exec.schema;
import com.google.common.base.Objects;
import com.google.common.base.Strings;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.memory.BufferAllocator;
+import org.apache.drill.exec.proto.SchemaDefProtos;
+import org.apache.drill.exec.record.MaterializedField;
+import org.apache.drill.exec.record.vector.*;
+import org.apache.drill.exec.store.BatchExceededException;
+import org.apache.drill.exec.store.VectorHolder;
+
+import java.nio.charset.Charset;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
+import static org.apache.drill.exec.proto.SchemaDefProtos.*;
public abstract class Field {
- final FieldType fieldType;
- int fieldId;
- String prefixFieldName;
- String fieldName;
+ final MajorType fieldType;
+ final int parentFieldId;
+ final int fieldId;
+ final String prefixFieldName;
RecordSchema schema;
RecordSchema parentSchema;
boolean read;
-
- public Field(RecordSchema parentSchema, IdGenerator<Integer> generator, FieldType fieldType, String prefixFieldName) {
+ public Field(RecordSchema parentSchema, int parentFieldId, IdGenerator<Integer> generator, MajorType type, String prefixFieldName) {
this.fieldId = generator.getNextId();
- this.fieldType = fieldType;
+ fieldType = type;
this.prefixFieldName = prefixFieldName;
this.parentSchema = parentSchema;
+ this.parentFieldId = parentFieldId;
}
- public Field assignSchema(RecordSchema newSchema) {
- checkState(schema == null, "Schema already assigned to field: %s", fieldName);
- checkState(fieldType.isEmbedSchema(), "Schema cannot be assigned to non-embedded types: %s", fieldType);
- schema = newSchema;
- return this;
- }
+ public abstract String getFieldName();
public String getFullFieldName() {
- return Strings.isNullOrEmpty(prefixFieldName) ? fieldName : prefixFieldName + "." + fieldName;
+ return Strings.isNullOrEmpty(prefixFieldName) ? getFieldName() : prefixFieldName + "." + getFieldName();
}
public int getFieldId() {
return fieldId;
}
- public String getFieldName() {
- return fieldName;
- }
-
public void setRead(boolean read) {
this.read = read;
}
@@ -78,18 +79,10 @@ public abstract class Field {
return addAttributesToHelper(getAttributesStringHelper()).toString();
}
- public RecordSchema getParentSchema() {
- return parentSchema;
- }
-
public RecordSchema getAssignedSchema() {
return schema;
}
- public FieldType getFieldType() {
- return fieldType;
- }
-
public void assignSchemaIfNull(RecordSchema newSchema) {
if (!hasSchema()) {
schema = newSchema;
@@ -104,32 +97,7 @@ public abstract class Field {
return schema != null;
}
- public static enum FieldType {
- INTEGER(1),
- FLOAT(2),
- BOOLEAN(3),
- STRING(4),
- ARRAY(5, true),
- MAP(6, true);
-
- byte value;
- boolean embedSchema;
-
- FieldType(int value, boolean embedSchema) {
- this.value = (byte) value;
- this.embedSchema = embedSchema;
- }
-
- FieldType(int value) {
- this(value, false);
- }
-
- public byte value() {
- return value;
- }
-
- public boolean isEmbedSchema() {
- return embedSchema;
- }
+ public MajorType getFieldType() {
+ return fieldType;
}
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/IdGenerator.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/IdGenerator.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/IdGenerator.java
index 728e8e1..d5de4ab 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/IdGenerator.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/IdGenerator.java
@@ -1,12 +1,23 @@
+/*******************************************************************************
+ * 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.schema;
-/**
- * Created with IntelliJ IDEA.
- * User: tnachen
- * Date: 1/2/13
- * Time: 10:50 PM
- * To change this template use File | Settings | File Templates.
- */
public interface IdGenerator<T> {
public T getNextId();
public void reset();
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/ListSchema.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/ListSchema.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/ListSchema.java
index efdc8fd..e5bd1a4 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/ListSchema.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/ListSchema.java
@@ -21,6 +21,7 @@ package org.apache.drill.exec.schema;
import com.google.common.base.Predicate;
import com.google.common.collect.Iterables;
import com.google.common.collect.Lists;
+import org.apache.drill.exec.proto.SchemaDefProtos;
import java.util.List;
@@ -36,7 +37,7 @@ public class ListSchema implements RecordSchema {
@Override
public void addField(Field field) {
- if (field.fieldType.isEmbedSchema() || fields.isEmpty() || !isSingleTyped() ||
+ if (field.getFieldType().getMode() == SchemaDefProtos.DataMode.REPEATED || fields.isEmpty() || !isSingleTyped() ||
!Iterables.getOnlyElement(fields).equals(field.getFieldType())) {
fields.add(field);
}
@@ -96,7 +97,7 @@ public class ListSchema implements RecordSchema {
if (!field.isRead()) {
removedFields.add(field);
return true;
- } else if(field.hasSchema()) {
+ } else if (field.hasSchema()) {
Iterables.addAll(removedFields, field.getAssignedSchema().removeUnreadFields());
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/NamedField.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/NamedField.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/NamedField.java
index aa0d6aa..b975ad7 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/NamedField.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/NamedField.java
@@ -19,18 +19,27 @@
package org.apache.drill.exec.schema;
import com.google.common.base.Objects;
+import org.apache.drill.exec.proto.SchemaDefProtos;
+import org.apache.drill.exec.schema.json.jackson.JacksonHelper;
public class NamedField extends Field {
- final FieldType keyType;
+ final SchemaDefProtos.MajorType keyType;
+ String fieldName;
- public NamedField(RecordSchema parentSchema, IdGenerator<Integer> generator, String prefixFieldName, String fieldName, Field.FieldType fieldType) {
- this(parentSchema, generator, prefixFieldName, fieldName, fieldType, FieldType.STRING);
+ public NamedField(RecordSchema parentSchema, int parentFieldId, IdGenerator<Integer> generator, String prefixFieldName, String fieldName, SchemaDefProtos.MajorType fieldType) {
+ this(parentSchema, parentFieldId, generator, prefixFieldName, fieldName, fieldType, JacksonHelper.STRING_TYPE);
}
- public NamedField(RecordSchema parentSchema, IdGenerator<Integer> generator, String prefixFieldName, String fieldName, Field.FieldType fieldType, FieldType keyType) {
- super(parentSchema, generator, fieldType, prefixFieldName);
+ public NamedField(RecordSchema parentSchema,
+ int parentFieldId,
+ IdGenerator<Integer> generator,
+ String prefixFieldName,
+ String fieldName,
+ SchemaDefProtos.MajorType fieldType,
+ SchemaDefProtos.MajorType keyType) {
+ super(parentSchema, parentFieldId, generator, fieldType, prefixFieldName);
this.fieldName = fieldName;
- this.keyType = FieldType.STRING;
+ this.keyType = keyType;
}
public String getFieldName() {
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/OrderedField.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/OrderedField.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/OrderedField.java
index 67fd2fa..5f514af 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/OrderedField.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/OrderedField.java
@@ -19,11 +19,24 @@
package org.apache.drill.exec.schema;
import com.google.common.base.Objects;
+import org.apache.drill.exec.proto.SchemaDefProtos;
public class OrderedField extends Field {
- public OrderedField(RecordSchema parentSchema, IdGenerator<Integer> generator, FieldType fieldType, String prefixFieldName, int index) {
- super(parentSchema, generator, fieldType, prefixFieldName);
- this.fieldName = "[" + index + "]";
+ private final int index;
+
+ public OrderedField(RecordSchema parentSchema,
+ int parentFieldId,
+ IdGenerator<Integer> generator,
+ SchemaDefProtos.MajorType type,
+ String prefixFieldName,
+ int index) {
+ super(parentSchema, parentFieldId, generator, type, prefixFieldName);
+ this.index = index;
+ }
+
+ @Override
+ public String getFieldName() {
+ return "[" + index + "]";
}
@Override
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/SchemaRecorder.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/SchemaRecorder.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/SchemaRecorder.java
deleted file mode 100644
index 54a4e0e..0000000
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/SchemaRecorder.java
+++ /dev/null
@@ -1,122 +0,0 @@
-/*******************************************************************************
- * 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.schema;
-
-import com.fasterxml.jackson.core.JsonParser;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import org.apache.drill.exec.schema.json.jackson.JacksonHelper;
-import org.apache.drill.exec.schema.json.jackson.ScanJson;
-
-import java.io.IOException;
-import java.util.List;
-
-public class SchemaRecorder {
- DiffSchema diffSchema;
- RecordSchema currentSchema;
- List<Field> removedFields;
-
- public SchemaRecorder() {
- currentSchema = new ObjectSchema();
- diffSchema = new DiffSchema();
- removedFields = Lists.newArrayList();
- }
-
- public RecordSchema getCurrentSchema() {
- return currentSchema;
- }
-
- public void recordData(ScanJson.ReadType currentReadType, ScanJson.ReadType readType, JsonParser parser, IdGenerator generator, DataRecord record, Field.FieldType fieldType, String prefixFieldName, String fieldName, int index) throws IOException {
- Field field = currentSchema.getField(fieldName, index);
-
- if (field == null || field.getFieldType() != fieldType) {
- if (field != null) {
- removeStaleField(index, field);
- }
- field = currentReadType.createField(currentSchema, generator, prefixFieldName, fieldName, fieldType, index);
- field.setRead(true);
- diffSchema.recordNewField(field);
- currentSchema.addField(field);
- } else {
- field.setRead(true);
- }
-
- if (readType != null) {
- RecordSchema origSchema = currentSchema;
- if (field != null) {
- currentSchema = field.getAssignedSchema();
- }
-
- RecordSchema newSchema = readType.createSchema();
- field.assignSchemaIfNull(newSchema);
- setCurrentSchemaIfNull(newSchema);
- readType.readRecord(parser, generator, this, record, field.getFullFieldName());
-
- currentSchema = origSchema;
- } else {
- RecordSchema schema = field.getParentSchema();
- record.addData(field.getFieldId(), JacksonHelper.getValueFromFieldType(parser, fieldType), schema != null && schema instanceof ListSchema);
- }
- }
-
- private void removeStaleField(int index, Field field) {
- if (field.hasSchema()) {
- removeChildFields(field);
- }
- removedFields.add(field);
- currentSchema.removeField(field, index);
- }
-
- private void removeChildFields(Field field) {
- RecordSchema schema = field.getAssignedSchema();
- if(schema == null) { return; }
- for (Field childField : schema.getFields()) {
- removedFields.add(childField);
- if (childField.hasSchema()) {
- removeChildFields(childField);
- }
- }
- }
-
- public boolean hasDiffs() {
- return diffSchema.hasDiffFields();
- }
-
- public DiffSchema getDiffSchema() {
- return hasDiffs() ? diffSchema : null;
- }
-
- public void setCurrentSchemaIfNull(RecordSchema newSchema) {
- if (currentSchema == null) {
- currentSchema = newSchema;
- }
- }
-
- public void reset() {
- currentSchema.resetMarkedFields();
- diffSchema.reset();
- removedFields.clear();
- }
-
- public void addMissingFields() {
- for (Field field : Iterables.concat(currentSchema.removeUnreadFields(), removedFields)) {
- diffSchema.addRemovedField(field);
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/JacksonHelper.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/JacksonHelper.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/JacksonHelper.java
index 0643710..64a9d58 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/JacksonHelper.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/JacksonHelper.java
@@ -20,42 +20,58 @@ package org.apache.drill.exec.schema.json.jackson;
import com.fasterxml.jackson.core.JsonParser;
import com.fasterxml.jackson.core.JsonToken;
-import com.google.common.collect.Maps;
-
-import org.apache.drill.exec.schema.Field;
+import org.apache.drill.exec.proto.SchemaDefProtos;
import java.io.IOException;
-import java.util.EnumMap;
-import static com.google.common.base.Preconditions.checkNotNull;
+import static org.apache.drill.exec.proto.SchemaDefProtos.MajorType;
+import static org.apache.drill.exec.proto.SchemaDefProtos.MinorType;
public class JacksonHelper {
- private static final EnumMap<JsonToken, Field.FieldType> TYPE_LOOKUP = Maps.newEnumMap(JsonToken.class);
-
- static {
- TYPE_LOOKUP.put(JsonToken.VALUE_STRING, Field.FieldType.STRING);
- TYPE_LOOKUP.put(JsonToken.VALUE_FALSE, Field.FieldType.BOOLEAN);
- TYPE_LOOKUP.put(JsonToken.VALUE_TRUE, Field.FieldType.BOOLEAN);
- TYPE_LOOKUP.put(JsonToken.START_ARRAY, Field.FieldType.ARRAY);
- TYPE_LOOKUP.put(JsonToken.START_OBJECT, Field.FieldType.MAP);
- TYPE_LOOKUP.put(JsonToken.VALUE_NUMBER_INT, Field.FieldType.INTEGER);
- TYPE_LOOKUP.put(JsonToken.VALUE_NUMBER_FLOAT, Field.FieldType.FLOAT);
- }
- public static Field.FieldType getFieldType(JsonToken token) {
- return TYPE_LOOKUP.get(token);
+ public static final MajorType STRING_TYPE = MajorType.newBuilder().setMinorType(MinorType.VARCHAR4).setMode(SchemaDefProtos.DataMode.OPTIONAL).build();
+ public static final MajorType BOOLEAN_TYPE = MajorType.newBuilder().setMinorType(MinorType.BOOLEAN).setMode(SchemaDefProtos.DataMode.OPTIONAL).build();
+ public static final MajorType ARRAY_TYPE = MajorType.newBuilder().setMinorType(MinorType.LATE).setMode(SchemaDefProtos.DataMode.REPEATED).build();
+ public static final MajorType MAP_TYPE = MajorType.newBuilder().setMinorType(MinorType.MAP).setMode(SchemaDefProtos.DataMode.REPEATED).build();
+ public static final MajorType INT_TYPE = MajorType.newBuilder().setMinorType(MinorType.INT).setMode(SchemaDefProtos.DataMode.OPTIONAL).build();
+ public static final MajorType FLOAT_TYPE = MajorType.newBuilder().setMinorType(MinorType.FLOAT4).setMode(SchemaDefProtos.DataMode.OPTIONAL).build();
+ public static final MajorType NULL_TYPE = MajorType.newBuilder().setMinorType(MinorType.LATE).setMode(SchemaDefProtos.DataMode.OPTIONAL).build();
+
+ public static MajorType getFieldType(JsonToken token) {
+ switch(token) {
+ case VALUE_STRING:
+ return STRING_TYPE;
+ case VALUE_FALSE:
+ return BOOLEAN_TYPE;
+ case VALUE_TRUE:
+ return BOOLEAN_TYPE;
+ case START_ARRAY:
+ return ARRAY_TYPE;
+ case START_OBJECT:
+ return MAP_TYPE;
+ case VALUE_NUMBER_INT:
+ return INT_TYPE;
+ case VALUE_NUMBER_FLOAT:
+ return FLOAT_TYPE;
+ case VALUE_NULL:
+ return NULL_TYPE;
+ }
+
+ throw new UnsupportedOperationException("Unsupported Jackson type: " + token);
}
- public static Object getValueFromFieldType(JsonParser parser, Field.FieldType fieldType) throws IOException {
- switch(fieldType) {
- case INTEGER:
+ public static Object getValueFromFieldType(JsonParser parser, MinorType fieldType) throws IOException {
+ switch (fieldType) {
+ case INT:
return parser.getIntValue();
- case STRING:
+ case VARCHAR4:
return parser.getValueAsString();
- case FLOAT:
+ case FLOAT4:
return parser.getFloatValue();
case BOOLEAN:
return parser.getBooleanValue();
+ case LATE:
+ return null;
default:
throw new RuntimeException("Unexpected Field type to return value: " + fieldType.toString());
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/PhysicalOperator.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/PhysicalOperator.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/PhysicalOperator.java
deleted file mode 100644
index e450ee9..0000000
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/PhysicalOperator.java
+++ /dev/null
@@ -1,36 +0,0 @@
-/*******************************************************************************
- * 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.schema.json.jackson;
-
-import com.google.common.collect.Lists;
-
-import org.apache.drill.exec.schema.Record;
-
-import java.io.IOException;
-import java.util.List;
-
-public abstract class PhysicalOperator {
- List<PhysicalOperatorIterator> parents;
-
- public PhysicalOperator(PhysicalOperatorIterator... parents) {
- this.parents = Lists.newArrayList(parents);
- }
-
- public abstract PhysicalOperatorIterator getIterator();
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/PhysicalOperatorIterator.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/PhysicalOperatorIterator.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/PhysicalOperatorIterator.java
deleted file mode 100644
index bf4053e..0000000
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/PhysicalOperatorIterator.java
+++ /dev/null
@@ -1,31 +0,0 @@
-/*******************************************************************************
- * 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.schema.json.jackson;
-
-import org.apache.drill.exec.schema.Record;
-
-import java.io.IOException;
-
-public interface PhysicalOperatorIterator{
- public enum NextOutcome {NONE_LEFT, INCREMENTED_SCHEMA_UNCHANGED, INCREMENTED_SCHEMA_CHANGED}
- public Record getRecord();
- public NextOutcome next() throws IOException;
-}
-
-
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/ScanJson.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/ScanJson.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/ScanJson.java
deleted file mode 100644
index a1c30e9..0000000
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/json/jackson/ScanJson.java
+++ /dev/null
@@ -1,203 +0,0 @@
-/*******************************************************************************
- * 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.schema.json.jackson;
-
-import com.fasterxml.jackson.core.JsonFactory;
-import com.fasterxml.jackson.core.JsonParser;
-import com.fasterxml.jackson.core.JsonToken;
-import com.google.common.base.Charsets;
-import com.google.common.collect.Maps;
-import com.google.common.io.Files;
-import com.google.common.io.InputSupplier;
-import com.google.common.io.Resources;
-
-import org.apache.drill.exec.schema.*;
-
-import java.io.*;
-import java.util.Map;
-
-public class ScanJson extends PhysicalOperator {
- private ScanJsonIterator iterator;
-
- private static final Map<JsonToken, ReadType> READ_TYPES = Maps.newHashMap();
-
- static {
- READ_TYPES.put(JsonToken.START_ARRAY, ReadType.ARRAY);
- READ_TYPES.put(JsonToken.START_OBJECT, ReadType.OBJECT);
- }
-
- public ScanJson(String inputName) throws IOException {
- super();
- this.iterator = new ScanJsonIterator(inputName);
- }
-
- @Override
- public PhysicalOperatorIterator getIterator() {
- return iterator;
- }
-
- class ScanJsonIterator implements PhysicalOperatorIterator {
- private JsonParser parser;
- private SchemaRecorder recorder;
- private BackedRecord record;
- private IdGenerator generator;
-
- private ScanJsonIterator(String inputName) throws IOException {
- InputSupplier<InputStreamReader> input;
- if (inputName.startsWith("resource:")) {
- input = Resources.newReaderSupplier(Resources.getResource(inputName.substring(inputName.indexOf(':') + 1)), Charsets.UTF_8);
- } else {
- input = Files.newReaderSupplier(new File(inputName), Charsets.UTF_8);
- }
-
- JsonFactory factory = new JsonFactory();
- parser = factory.createJsonParser(input.getInput());
- parser.nextToken(); // Read to the first START_OBJECT token
- recorder = new SchemaRecorder();
- generator = new SchemaIdGenerator();
- }
-
- @Override
- public Record getRecord() {
- return record;
- }
-
- @Override
- public NextOutcome next() throws IOException {
- if (parser.isClosed() || !parser.hasCurrentToken()) {
- return NextOutcome.NONE_LEFT;
- }
-
- recorder.reset();
-
- DataRecord dataRecord = new DataRecord();
- ReadType.OBJECT.readRecord(parser, generator, recorder, dataRecord, null);
-
- parser.nextToken(); // Read to START_OBJECT token
-
- if (!parser.hasCurrentToken()) {
- parser.close();
- }
-
- recorder.addMissingFields();
- if (record == null) {
- record = new BackedRecord(recorder.getDiffSchema(), dataRecord);
- } else {
- record.setBackend(recorder.getDiffSchema(), dataRecord);
- }
- return recorder.hasDiffs() ? NextOutcome.INCREMENTED_SCHEMA_CHANGED : NextOutcome.INCREMENTED_SCHEMA_UNCHANGED;
- }
-
- public RecordSchema getCurrentSchema() {
- return recorder.getCurrentSchema();
- }
- }
-
- public static enum ReadType {
- ARRAY(JsonToken.END_ARRAY) {
- @Override
- public Field createField(RecordSchema parentSchema, IdGenerator<Integer> generator, String prefixFieldName, String fieldName, Field.FieldType fieldType, int index) {
- return new OrderedField(parentSchema, generator, fieldType, prefixFieldName, index);
- }
-
- @Override
- public RecordSchema createSchema() throws IOException {
- return new ListSchema();
- }
- },
- OBJECT(JsonToken.END_OBJECT) {
- @Override
- public Field createField(RecordSchema parentSchema, IdGenerator<Integer> generator, String prefixFieldName, String fieldName, Field.FieldType fieldType, int index) {
- return new NamedField(parentSchema, generator, prefixFieldName, fieldName, fieldType);
- }
-
- @Override
- public RecordSchema createSchema() throws IOException {
- return new ObjectSchema();
- }
- };
-
- private final JsonToken endObject;
-
- ReadType(JsonToken endObject) {
- this.endObject = endObject;
- }
-
- public JsonToken getEndObject() {
- return endObject;
- }
-
- public void readRecord(JsonParser parser, IdGenerator generator, SchemaRecorder recorder, DataRecord record, String prefixFieldName) throws IOException {
- JsonToken token = parser.nextToken();
- JsonToken endObject = getEndObject();
- int index = 0;
- while (token != endObject) {
- if (token == JsonToken.FIELD_NAME) {
- token = parser.nextToken();
- continue;
- }
-
- String fieldName = parser.getCurrentName();
- Field.FieldType fieldType = JacksonHelper.getFieldType(token);
- ReadType readType = READ_TYPES.get(token);
- if (fieldType != null) { // Including nulls
- recorder.recordData(this, readType, parser, generator, record, fieldType, prefixFieldName, fieldName, index);
- }
- token = parser.nextToken();
- ++index;
- }
- }
-
- public abstract RecordSchema createSchema() throws IOException;
-
- public abstract Field createField(RecordSchema parentSchema, IdGenerator<Integer> generator, String prefixFieldName, String fieldName, Field.FieldType fieldType, int index);
- }
-
- public static void main(String[] args) throws IOException {
- if (args.length != 1) {
- System.err.println("Requires json path: ScanJson <json_path>");
- return;
- }
-
- String jsonPath = args[0];
-
- System.out.println("Reading json input...");
- ScanJson sj = new ScanJson(jsonPath);
- ScanJsonIterator iterator = (ScanJsonIterator) sj.getIterator();
- long count = 0;
-
- while (iterator.next() != PhysicalOperatorIterator.NextOutcome.NONE_LEFT) {
- Record record = iterator.getRecord();
- System.out.println("Record " + ++count);
- System.out.println("Schema: ");
- System.out.println(iterator.getCurrentSchema().toSchemaString());
- System.out.println();
- System.out.println("Changes since last record: ");
- System.out.println();
- System.out.println(record.getSchemaChanges());
- System.out.println();
- }
- }
-}
-
-
-
-
-
-
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/transform/ProtobufSchemaTransformer.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/transform/ProtobufSchemaTransformer.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/transform/ProtobufSchemaTransformer.java
deleted file mode 100644
index a81a9d9..0000000
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/transform/ProtobufSchemaTransformer.java
+++ /dev/null
@@ -1,109 +0,0 @@
-/*******************************************************************************
- * 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.schema.transform;
-
-import com.google.common.base.Function;
-import com.google.common.collect.Iterables;
-import com.google.common.collect.Lists;
-import com.google.common.collect.Maps;
-import com.google.common.io.ByteArrayDataOutput;
-import com.google.common.io.ByteStreams;
-import com.google.protobuf.DescriptorProtos;
-
-import org.apache.drill.exec.schema.Field;
-import org.apache.drill.exec.schema.ListSchema;
-import org.apache.drill.exec.schema.ObjectSchema;
-import org.apache.drill.exec.schema.RecordSchema;
-
-import java.util.EnumMap;
-import java.util.List;
-import java.util.Map;
-
-import static com.google.common.base.Preconditions.checkArgument;
-import static com.google.common.base.Preconditions.checkNotNull;
-
-public class ProtobufSchemaTransformer implements SchemaTransformer<DescriptorProtos.DescriptorProto> {
- private static final Map<Field.FieldType, Function<Field, Object>> FIELD_MAP = Maps.newEnumMap(Field.FieldType.class);
- private static final Map<Field.FieldType, DescriptorProtos.FieldDescriptorProto.Type> TYPE_MAP = Maps.newEnumMap(Field.FieldType.class);
- private int fieldIndex = 0;
- public static final String LIST_SCHEMA_NAME = "_EmbeddedList"; //Read from config?
-
- static {
- TYPE_MAP.put(Field.FieldType.BOOLEAN, DescriptorProtos.FieldDescriptorProto.Type.TYPE_BOOL);
- TYPE_MAP.put(Field.FieldType.STRING, DescriptorProtos.FieldDescriptorProto.Type.TYPE_STRING);
- TYPE_MAP.put(Field.FieldType.FLOAT, DescriptorProtos.FieldDescriptorProto.Type.TYPE_FLOAT);
- TYPE_MAP.put(Field.FieldType.INTEGER, DescriptorProtos.FieldDescriptorProto.Type.TYPE_INT32);
- }
-
- private DescriptorProtos.DescriptorProto.Builder transformSchema(String name, DescriptorProtos.DescriptorProto.Builder parentBuilder, RecordSchema schema) {
- if (schema instanceof ObjectSchema) {
- return addObjectSchema(name, parentBuilder, ObjectSchema.class.cast(schema));
- } else if (schema instanceof ListSchema) {
- return addListSchema(name, ListSchema.class.cast(schema));
- } else {
- throw new RuntimeException("Unknown schema passed to transformer: " + schema);
- }
- }
-
- public DescriptorProtos.DescriptorProto transformSchema(String name, RecordSchema schema) {
- return transformSchema(name, null, schema).build();
- }
-
- private DescriptorProtos.DescriptorProto.Builder addListSchema(String name, ListSchema schema) {
- DescriptorProtos.DescriptorProto.Builder builder = DescriptorProtos.DescriptorProto.newBuilder().setName(name);
- DescriptorProtos.FieldDescriptorProto.Builder builderForValue = DescriptorProtos.FieldDescriptorProto.newBuilder();
- builderForValue.setTypeName(LIST_SCHEMA_NAME).setType(DescriptorProtos.FieldDescriptorProto.Type.TYPE_MESSAGE);
- builder.addField(builderForValue);
- return builder;
- }
-
- private DescriptorProtos.DescriptorProto.Builder addObjectSchema(String name,
- DescriptorProtos.DescriptorProto.Builder parentBuilder,
- ObjectSchema schema) {
- DescriptorProtos.DescriptorProto.Builder builder = DescriptorProtos.DescriptorProto.newBuilder().setName(name);
- for (Field field : schema.getFields()) {
- DescriptorProtos.FieldDescriptorProto.Builder builderForValue = DescriptorProtos.FieldDescriptorProto.newBuilder();
- String fieldName = field.getFieldName();
- builderForValue.setName(fieldName).setLabel(DescriptorProtos.FieldDescriptorProto.Label.LABEL_OPTIONAL).setNumber(++fieldIndex);
- if (field.hasSchema()) {
- RecordSchema innerSchema = field.getAssignedSchema();
- if (innerSchema instanceof ObjectSchema) {
- addObjectSchema(fieldName, builder, (ObjectSchema) innerSchema);
- DescriptorProtos.DescriptorProto innerProto = Iterables.getLast(builder.getNestedTypeList());
- builderForValue.setTypeName(innerProto.getName()).setType(DescriptorProtos.FieldDescriptorProto.Type.TYPE_MESSAGE);
- } else if (innerSchema instanceof ListSchema) {
- builderForValue.setTypeName(LIST_SCHEMA_NAME).setType(DescriptorProtos.FieldDescriptorProto.Type.TYPE_MESSAGE);
- }
- } else {
- builderForValue.setType(getProtoType(field.getFieldType()));
- }
- builder.addField(builderForValue);
- }
-
- if (parentBuilder != null) {
- parentBuilder.addNestedType(builder);
- }
-
- return builder;
- }
-
- private DescriptorProtos.FieldDescriptorProto.Type getProtoType(Field.FieldType fieldType) {
- return checkNotNull(TYPE_MAP.get(fieldType));
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/transform/SchemaTransformer.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/transform/SchemaTransformer.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/transform/SchemaTransformer.java
deleted file mode 100644
index 54d851f..0000000
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/schema/transform/SchemaTransformer.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/*******************************************************************************
- * 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.schema.transform;
-
-import org.apache.drill.exec.schema.Field;
-import org.apache.drill.exec.schema.ListSchema;
-import org.apache.drill.exec.schema.ObjectSchema;
-import org.apache.drill.exec.schema.RecordSchema;
-
-import java.util.List;
-
-public interface SchemaTransformer<T> {
- public T transformSchema(String name, RecordSchema schema);
-}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
index d5aaab2..c933594 100644
--- a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java
@@ -51,10 +51,8 @@ public class DrillbitContext {
super();
Preconditions.checkNotNull(endpoint);
Preconditions.checkNotNull(context);
- Preconditions.checkNotNull(context);
- Preconditions.checkNotNull(com);
Preconditions.checkNotNull(com);
-
+
this.context = context;
this.coord = coord;
this.com = com;
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/01278ae5/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/BatchExceededException.java
----------------------------------------------------------------------
diff --git a/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/BatchExceededException.java b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/BatchExceededException.java
new file mode 100644
index 0000000..d47caf7
--- /dev/null
+++ b/sandbox/prototype/exec/java-exec/src/main/java/org/apache/drill/exec/store/BatchExceededException.java
@@ -0,0 +1,9 @@
+package org.apache.drill.exec.store;
+
+import org.apache.drill.common.exceptions.DrillRuntimeException;
+
+public class BatchExceededException extends DrillRuntimeException {
+ public BatchExceededException(int capacity, int attempted) {
+ super("Batch exceeded in size. Capacity: " + capacity + ", Attempted: " + attempted);
+ }
+}
[3/3] git commit: Adding parquet repository to project root POM.xml
file.
Posted by ja...@apache.org.
Adding parquet repository to project root POM.xml file.
Signed-off-by: Jacques Nadeau <ja...@apache.org>
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/11a1fcee
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/11a1fcee
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/11a1fcee
Branch: refs/heads/master
Commit: 11a1fceeca7c3c81c42a2bf876cd8468e239e6bd
Parents: 3a11d29
Author: Jason Altekruse <al...@gmial.com>
Authored: Mon Jun 17 09:37:17 2013 -0700
Committer: Jacques Nadeau <ja...@apache.org>
Committed: Mon Jun 17 09:43:12 2013 -0700
----------------------------------------------------------------------
sandbox/prototype/pom.xml | 26 ++++++++++++++++++++++++--
1 file changed, 24 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/11a1fcee/sandbox/prototype/pom.xml
----------------------------------------------------------------------
diff --git a/sandbox/prototype/pom.xml b/sandbox/prototype/pom.xml
index a4abf8b..efa1346 100644
--- a/sandbox/prototype/pom.xml
+++ b/sandbox/prototype/pom.xml
@@ -15,7 +15,29 @@
<packaging>pom</packaging>
<name>prototype-parent</name>
-
+ <repositories>
+ <repository>
+ <releases>
+ <enabled>true</enabled>
+ <updatePolicy>always</updatePolicy>
+ <checksumPolicy>warn</checksumPolicy>
+ </releases>
+ <id>conjars</id>
+ <name>Conjars</name>
+ <url>http://conjars.org/repo</url>
+ <layout>default</layout>
+ </repository>
+ <repository>
+ <id>sonatype-nexus-snapshots</id>
+ <url>https://oss.sonatype.org/content/repositories/snapshots</url>
+ <releases>
+ <enabled>false</enabled>
+ </releases>
+ <snapshots>
+ <enabled>true</enabled>
+ </snapshots>
+ </repository>
+ </repositories>
<properties>
<dep.junit.version>4.11</dep.junit.version>
<dep.slf4j.version>1.7.2</dep.slf4j.version>
@@ -144,4 +166,4 @@
<module>planner</module>
<module>sqlparser</module>
</modules>
-</project>
\ No newline at end of file
+</project>