You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by me...@apache.org on 2014/10/21 23:27:39 UTC
git commit: DRILL-1534: Implement Mappify UDF to transform map into a
repeated map with key, value pairs
Repository: incubator-drill
Updated Branches:
refs/heads/master 8dedd7182 -> fd7aee133
DRILL-1534: Implement Mappify UDF to transform map into a repeated map with key, value pairs
Project: http://git-wip-us.apache.org/repos/asf/incubator-drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-drill/commit/fd7aee13
Tree: http://git-wip-us.apache.org/repos/asf/incubator-drill/tree/fd7aee13
Diff: http://git-wip-us.apache.org/repos/asf/incubator-drill/diff/fd7aee13
Branch: refs/heads/master
Commit: fd7aee133a5c55a61c1b2b31a4e27e074c669278
Parents: 8dedd71
Author: Mehant Baid <me...@gmail.com>
Authored: Sat Oct 11 22:35:22 2014 -0700
Committer: Mehant Baid <me...@gmail.com>
Committed: Tue Oct 21 14:27:18 2014 -0700
----------------------------------------------------------------------
.../apache/drill/exec/expr/fn/impl/Mappify.java | 67 ++++++
.../drill/exec/expr/fn/impl/MappifyUtility.java | 87 +++++++
.../drill/exec/vector/complex/MapUtility.java | 228 +++++++++++++++++++
.../complex/writer/TestComplexTypeReader.java | 5 +
.../src/test/resources/jsoninput/input3.json | 42 ++++
5 files changed, 429 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/fd7aee13/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Mappify.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Mappify.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Mappify.java
new file mode 100644
index 0000000..7f340e1
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/Mappify.java
@@ -0,0 +1,67 @@
+/**
+ * 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.expr.fn.impl;
+
+import io.netty.buffer.DrillBuf;
+import org.apache.drill.exec.expr.DrillSimpleFunc;
+import org.apache.drill.exec.expr.annotations.FunctionTemplate;
+import org.apache.drill.exec.expr.annotations.Output;
+import org.apache.drill.exec.expr.annotations.Param;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.vector.complex.reader.FieldReader;
+import org.apache.drill.exec.vector.complex.writer.BaseWriter.ComplexWriter;
+
+import javax.inject.Inject;
+
+public class Mappify {
+
+ /*
+ * The following function can be invoked when we want to convert a map to a repeated map where every
+ * with two fields in each entry:
+ * key: the name of the field in the original map and
+ * value: value of the field
+ *
+ * For eg, consider the following json file:
+ *
+ * {"foo": {"obj":1, "bar":10}}
+ * {"foo": {"obj":2, "bar":20}}
+ *
+ * Invoking mappify(foo) would result in the following output
+ *
+ * [{"key":"obj", "value":1}, {"key":"bar", "value":10}]
+ * [{"key":"obj", "value":2}, {"key":"bar", "value":20}]
+ *
+ * Currently this function only allows
+ * simple maps as input
+ * scalar value fields
+ * value fields need to be of the same data type
+ */
+ @FunctionTemplate(names = {"mappify", "kvgen"}, scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.NULL_IF_NULL, isRandom = true)
+ public static class ConvertMapToKeyValuePairs implements DrillSimpleFunc {
+ @Param FieldReader reader;
+ @Inject DrillBuf buffer;
+ @Output ComplexWriter writer;
+
+ public void setup(RecordBatch incoming) {
+ }
+
+ public void eval() {
+ org.apache.drill.exec.expr.fn.impl.MappifyUtility.mappify(reader, writer, buffer);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/fd7aee13/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/MappifyUtility.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/MappifyUtility.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/MappifyUtility.java
new file mode 100644
index 0000000..92f62ca
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/MappifyUtility.java
@@ -0,0 +1,87 @@
+/**
+ * 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.expr.fn.impl;
+
+import com.google.common.base.Charsets;
+
+import org.apache.drill.common.exceptions.DrillRuntimeException;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.exec.expr.holders.VarCharHolder;
+import org.apache.drill.exec.vector.complex.MapUtility;
+import org.apache.drill.exec.vector.complex.impl.SingleMapReaderImpl;
+import org.apache.drill.exec.vector.complex.reader.FieldReader;
+import org.apache.drill.exec.vector.complex.writer.BaseWriter;
+
+import io.netty.buffer.DrillBuf;
+
+import java.util.Iterator;
+
+public class MappifyUtility {
+
+ // Default names used in the map.
+ public static final String fieldKey = "key";
+ public static final String fieldValue = "value";
+
+ public static void mappify(FieldReader reader, BaseWriter.ComplexWriter writer, DrillBuf buffer) {
+ // Currently we expect single map as input
+ if (!(reader instanceof SingleMapReaderImpl)) {
+ throw new DrillRuntimeException("Mappify function only supports Simple maps as input");
+ }
+ BaseWriter.ListWriter listWriter = writer.rootAsList();
+ listWriter.start();
+ BaseWriter.MapWriter mapWriter = listWriter.map();
+
+ // Iterate over the fields in the map
+ Iterator<String> fieldIterator = reader.iterator();
+ while (fieldIterator.hasNext()) {
+ String str = fieldIterator.next();
+ FieldReader fieldReader = reader.reader(str);
+
+ // Check if the value field is not repeated
+ if (fieldReader.getType().getMode() == TypeProtos.DataMode.REPEATED) {
+ throw new DrillRuntimeException("Mappify function does not support repeated type values");
+ }
+
+ // writing a new field, start a new map
+ mapWriter.start();
+
+ // write "key":"columnname" into the map
+ VarCharHolder vh = new VarCharHolder();
+ byte[] b = str.getBytes(Charsets.UTF_8);
+ buffer.reallocIfNeeded(b.length);
+ buffer.setBytes(0, b);
+ vh.start = 0;
+ vh.end = b.length;
+ vh.buffer = buffer;
+ mapWriter.varChar(fieldKey).write(vh);
+
+ // Skip the value field if its null
+ if (fieldReader.isSet() == false) {
+ mapWriter.end();
+ continue;
+ }
+
+ // Write the value to the map
+ MapUtility.writeToMapFromReader(fieldReader, mapWriter, buffer);
+
+ mapWriter.end();
+ }
+ listWriter.end();
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/fd7aee13/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/MapUtility.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/MapUtility.java b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/MapUtility.java
new file mode 100644
index 0000000..6c1907a
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/vector/complex/MapUtility.java
@@ -0,0 +1,228 @@
+/**
+ * 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.vector.complex;
+
+import com.google.common.base.Charsets;
+import io.netty.buffer.DrillBuf;
+import org.apache.drill.common.exceptions.DrillRuntimeException;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.common.types.TypeProtos.MajorType;
+import org.apache.drill.exec.expr.fn.impl.MappifyUtility;
+import org.apache.drill.exec.expr.holders.BigIntHolder;
+import org.apache.drill.exec.expr.holders.BitHolder;
+import org.apache.drill.exec.expr.holders.DateHolder;
+import org.apache.drill.exec.expr.holders.Decimal18Holder;
+import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
+import org.apache.drill.exec.expr.holders.Decimal9Holder;
+import org.apache.drill.exec.expr.holders.Float4Holder;
+import org.apache.drill.exec.expr.holders.Float8Holder;
+import org.apache.drill.exec.expr.holders.IntHolder;
+import org.apache.drill.exec.expr.holders.IntervalDayHolder;
+import org.apache.drill.exec.expr.holders.IntervalHolder;
+import org.apache.drill.exec.expr.holders.IntervalYearHolder;
+import org.apache.drill.exec.expr.holders.SmallIntHolder;
+import org.apache.drill.exec.expr.holders.TimeHolder;
+import org.apache.drill.exec.expr.holders.TimeStampHolder;
+import org.apache.drill.exec.expr.holders.TinyIntHolder;
+import org.apache.drill.exec.expr.holders.UInt1Holder;
+import org.apache.drill.exec.expr.holders.UInt2Holder;
+import org.apache.drill.exec.expr.holders.UInt4Holder;
+import org.apache.drill.exec.expr.holders.UInt8Holder;
+import org.apache.drill.exec.expr.holders.VarBinaryHolder;
+import org.apache.drill.exec.expr.holders.VarCharHolder;
+import org.apache.drill.exec.util.DecimalUtility;
+import org.apache.drill.exec.vector.complex.reader.FieldReader;
+import org.apache.drill.exec.vector.complex.writer.BaseWriter;
+import org.joda.time.Period;
+
+public class MapUtility {
+ /*
+ * Function to read a value from the field reader, detect the type, construct the appropriate value holder
+ * and use the value holder to write to the Map.
+ */
+ public static void writeToMapFromReader(FieldReader fieldReader, BaseWriter.MapWriter mapWriter, DrillBuf buffer) {
+
+ MajorType valueMajorType = fieldReader.getType();
+ MinorType valueMinorType = valueMajorType.getMinorType();
+
+ switch (valueMinorType) {
+ case TINYINT:
+ TinyIntHolder tinyIntHolder = new TinyIntHolder();
+ tinyIntHolder.value = fieldReader.readByte();
+ mapWriter.tinyInt(MappifyUtility.fieldValue).write(tinyIntHolder);
+ break;
+ case SMALLINT:
+ SmallIntHolder smallIntHolder = new SmallIntHolder();
+ smallIntHolder.value = fieldReader.readShort();
+ mapWriter.smallInt(MappifyUtility.fieldValue).write(smallIntHolder);
+ break;
+ case BIGINT:
+ BigIntHolder bh = new BigIntHolder();
+ bh.value = fieldReader.readLong();
+ mapWriter.bigInt(MappifyUtility.fieldValue).write(bh);
+ break;
+ case INT:
+ IntHolder ih = new IntHolder();
+ ih.value = fieldReader.readInteger();
+ mapWriter.integer(MappifyUtility.fieldValue).write(ih);
+ break;
+ case UINT1:
+ UInt1Holder uInt1Holder = new UInt1Holder();
+ uInt1Holder.value = fieldReader.readByte();
+ mapWriter.uInt1(MappifyUtility.fieldValue).write(uInt1Holder);
+ break;
+ case UINT2:
+ UInt2Holder uInt2Holder = new UInt2Holder();
+ uInt2Holder.value = fieldReader.readCharacter();
+ mapWriter.uInt2(MappifyUtility.fieldValue).write(uInt2Holder);
+ break;
+ case UINT4:
+ UInt4Holder uInt4Holder = new UInt4Holder();
+ uInt4Holder.value = fieldReader.readInteger();
+ mapWriter.uInt4(MappifyUtility.fieldValue).write(uInt4Holder);
+ break;
+ case UINT8:
+ UInt8Holder uInt8Holder = new UInt8Holder();
+ uInt8Holder.value = fieldReader.readInteger();
+ mapWriter.uInt8(MappifyUtility.fieldValue).write(uInt8Holder);
+ break;
+ case DECIMAL9:
+ Decimal9Holder decimalHolder = new Decimal9Holder();
+ decimalHolder.value = fieldReader.readBigDecimal().intValue();
+ decimalHolder.scale = valueMajorType.getScale();
+ decimalHolder.precision = valueMajorType.getPrecision();
+ mapWriter.decimal9(MappifyUtility.fieldValue).write(decimalHolder);
+ break;
+ case DECIMAL18:
+ Decimal18Holder decimal18Holder = new Decimal18Holder();
+ decimal18Holder.value = fieldReader.readBigDecimal().longValue();
+ decimal18Holder.scale = valueMajorType.getScale();
+ decimal18Holder.precision = valueMajorType.getPrecision();
+ mapWriter.decimal18(MappifyUtility.fieldValue).write(decimal18Holder);
+ break;
+ case DECIMAL28SPARSE:
+ Decimal28SparseHolder decimal28Holder = new Decimal28SparseHolder();
+
+ // Ensure that the buffer used to store decimal is of sufficient length
+ buffer.reallocIfNeeded(decimal28Holder.WIDTH);
+ decimal28Holder.scale = valueMajorType.getScale();
+ decimal28Holder.precision = valueMajorType.getPrecision();
+ decimal28Holder.buffer = buffer;
+ decimal28Holder.start = 0;
+ DecimalUtility.getSparseFromBigDecimal(fieldReader.readBigDecimal(), buffer, 0, decimal28Holder.scale,
+ decimal28Holder.precision, decimal28Holder.nDecimalDigits);
+ mapWriter.decimal28Sparse(MappifyUtility.fieldValue).write(decimal28Holder);
+ break;
+ case DECIMAL38SPARSE:
+ Decimal38SparseHolder decimal38Holder = new Decimal38SparseHolder();
+
+ // Ensure that the buffer used to store decimal is of sufficient length
+ buffer.reallocIfNeeded(decimal38Holder.WIDTH);
+ decimal38Holder.scale = valueMajorType.getScale();
+ decimal38Holder.precision = valueMajorType.getPrecision();
+ decimal38Holder.buffer = buffer;
+ decimal38Holder.start = 0;
+ DecimalUtility.getSparseFromBigDecimal(fieldReader.readBigDecimal(), buffer, 0, decimal38Holder.scale,
+ decimal38Holder.precision, decimal38Holder.nDecimalDigits);
+
+ mapWriter.decimal38Sparse(MappifyUtility.fieldValue).write(decimal38Holder);
+ break;
+ case DATE:
+ DateHolder dateHolder = new DateHolder();
+ dateHolder.value = fieldReader.readLong();
+ mapWriter.date(MappifyUtility.fieldValue).write(dateHolder);
+ break;
+ case TIME:
+ TimeHolder timeHolder = new TimeHolder();
+ timeHolder.value = fieldReader.readInteger();
+ mapWriter.time(MappifyUtility.fieldValue).write(timeHolder);
+ break;
+ case TIMESTAMP:
+ TimeStampHolder timeStampHolder = new TimeStampHolder();
+ timeStampHolder.value = fieldReader.readLong();
+ mapWriter.timeStamp(MappifyUtility.fieldValue).write(timeStampHolder);
+ break;
+ case INTERVAL:
+ IntervalHolder intervalHolder = new IntervalHolder();
+ Period period = fieldReader.readPeriod();
+ intervalHolder.months = (period.getYears() * org.apache.drill.exec.expr.fn.impl.DateUtility.yearsToMonths) + period.getMonths();
+ intervalHolder.days = period.getDays();
+ intervalHolder.milliseconds = (period.getHours() * org.apache.drill.exec.expr.fn.impl.DateUtility.hoursToMillis) +
+ (period.getMinutes() * org.apache.drill.exec.expr.fn.impl.DateUtility.minutesToMillis) +
+ (period.getSeconds() * org.apache.drill.exec.expr.fn.impl.DateUtility.secondsToMillis) +
+ (period.getMillis());
+ mapWriter.interval(MappifyUtility.fieldValue).write(intervalHolder);
+ break;
+ case INTERVALDAY:
+ IntervalDayHolder intervalDayHolder = new IntervalDayHolder();
+ Period periodDay = fieldReader.readPeriod();
+ intervalDayHolder.days = periodDay.getDays();
+ intervalDayHolder.milliseconds = (periodDay.getHours() * org.apache.drill.exec.expr.fn.impl.DateUtility.hoursToMillis) +
+ (periodDay.getMinutes() * org.apache.drill.exec.expr.fn.impl.DateUtility.minutesToMillis) +
+ (periodDay.getSeconds() * org.apache.drill.exec.expr.fn.impl.DateUtility.secondsToMillis) +
+ (periodDay.getMillis());
+ mapWriter.intervalDay(MappifyUtility.fieldValue).write(intervalDayHolder);
+ break;
+ case INTERVALYEAR:
+ IntervalYearHolder intervalYearHolder = new IntervalYearHolder();
+ Period periodYear = fieldReader.readPeriod();
+ intervalYearHolder.value = (periodYear.getYears() * org.apache.drill.exec.expr.fn.impl.DateUtility.yearsToMonths) + periodYear.getMonths();
+ mapWriter.intervalYear(MappifyUtility.fieldValue).write(intervalYearHolder);
+ break;
+ case FLOAT4:
+ Float4Holder float4Holder = new Float4Holder();
+ float4Holder.value = fieldReader.readFloat();
+ mapWriter.float4(MappifyUtility.fieldValue).write(float4Holder);
+ break;
+ case FLOAT8:
+ Float8Holder float8Holder = new Float8Holder();
+ float8Holder.value = fieldReader.readDouble();
+ mapWriter.float8(MappifyUtility.fieldValue).write(float8Holder);
+ break;
+ case BIT:
+ BitHolder bitHolder = new BitHolder();
+ bitHolder.value = (fieldReader.readBoolean() == true) ? 1 : 0;
+ mapWriter.bit(MappifyUtility.fieldValue).write(bitHolder);
+ break;
+ case VARCHAR:
+ VarCharHolder vh1 = new VarCharHolder();
+ byte[] b = fieldReader.readText().toString().getBytes(Charsets.UTF_8);
+ buffer.reallocIfNeeded(b.length);
+ buffer.setBytes(0, b);
+ vh1.start = 0;
+ vh1.end = b.length;
+ vh1.buffer = buffer;
+ mapWriter.varChar(MappifyUtility.fieldValue).write(vh1);
+ break;
+ case VARBINARY:
+ VarBinaryHolder varBinaryHolder = new VarBinaryHolder();
+ byte[] b1 = fieldReader.readByteArray();
+ buffer.reallocIfNeeded(b1.length);
+ buffer.setBytes(0, b1);
+ varBinaryHolder.start = 0;
+ varBinaryHolder.end = b1.length;
+ varBinaryHolder.buffer = buffer;
+ mapWriter.varBinary(MappifyUtility.fieldValue).write(varBinaryHolder);
+ break;
+ default:
+ throw new DrillRuntimeException(String.format("Mappify does not support input of type: %s", valueMinorType));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/fd7aee13/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestComplexTypeReader.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestComplexTypeReader.java b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestComplexTypeReader.java
index 2e62897..df92dde 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestComplexTypeReader.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/vector/complex/writer/TestComplexTypeReader.java
@@ -183,4 +183,9 @@ public class TestComplexTypeReader extends BaseTestQuery{
test("select l, l from cp.`jsoninput/input2.json`;");
}
+ @Test
+ public void testKeyValueGen() throws Exception {
+ test("select kvgen(x) from cp.`jsoninput/input2.json`");
+ test("select kvgen(bigintegercol), kvgen(float8col) from cp.`jsoninput/input3.json`");
+ }
}
http://git-wip-us.apache.org/repos/asf/incubator-drill/blob/fd7aee13/exec/java-exec/src/test/resources/jsoninput/input3.json
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/resources/jsoninput/input3.json b/exec/java-exec/src/test/resources/jsoninput/input3.json
new file mode 100644
index 0000000..8eed227
--- /dev/null
+++ b/exec/java-exec/src/test/resources/jsoninput/input3.json
@@ -0,0 +1,42 @@
+{
+ "bigintegercol" : {
+ "int_1" : 1,
+ "int_2": 2,
+ "int_3": 3
+ },
+ "float8col" : {
+ "f8_1" : 1.1,
+ "f8_2": 2.2
+ }
+}
+{
+ "bigintegercol" : {
+ "int_1" : 1,
+ "int_2": 2
+ },
+ "float8col" : {
+ "f8_1" : 1.1,
+ "f8_2": 2.2,
+ "f8_3": 3.3
+ }
+}
+{
+ "bigintegercol" : {
+ "int_1" : 1,
+ "int_3": 3
+ },
+ "float8col" : {
+ "f8_1" : 1.1,
+ "f8_3": 6.6
+ }
+}
+{
+ "bigintegercol" : {
+ "int_2": 2,
+ "int_3": 3
+ },
+ "float8col" : {
+ "f8_1" : 1.1,
+ "f8_2": 2.2
+ }
+}