You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by jn...@apache.org on 2017/03/02 20:59:52 UTC
[25/27] drill git commit: DRILL-5034: Select timestamp from hive
generated parquet always return in UTC
DRILL-5034: Select timestamp from hive generated parquet always return in UTC
- TIMESTAMP_IMPALA function is reverted to retaine local timezone
- TIMESTAMP_IMPALA_LOCALTIMEZONE is deleted
- Retain local timezone for the INT96 timestamp values in the parquet files while
PARQUET_READER_INT96_AS_TIMESTAMP option is on
Minor changes according to the review
Fix for the test, which relies on particular timezone
close #656
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/4a0fd56c
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/4a0fd56c
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/4a0fd56c
Branch: refs/heads/master
Commit: 4a0fd56c106550eee26ca68eaed6108f0dbad798
Parents: 8cded5a
Author: Vitalii Diravka <vi...@gmail.com>
Authored: Mon Nov 14 21:13:28 2016 +0000
Committer: Jinfeng Ni <jn...@apache.org>
Committed: Thu Mar 2 10:37:23 2017 -0800
----------------------------------------------------------------------
...onvertHiveParquetScanToDrillParquetScan.java | 2 +-
.../impl/conv/ConvertFromImpalaTimestamp.java | 28 +++----------------
.../store/parquet/ParquetReaderUtility.java | 25 +++++++++++++----
.../NullableFixedByteAlignedReaders.java | 19 +++++--------
.../parquet2/DrillParquetGroupConverter.java | 2 +-
.../test/java/org/apache/drill/TestBuilder.java | 15 ++++++++--
.../physical/impl/writer/TestParquetWriter.java | 29 +++++++++++---------
7 files changed, 61 insertions(+), 59 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/4a0fd56c/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java
index 46c0ff0..bb59600 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/ConvertHiveParquetScanToDrillParquetScan.java
@@ -68,7 +68,7 @@ public class ConvertHiveParquetScanToDrillParquetScan extends StoragePluginOptim
public static final ConvertHiveParquetScanToDrillParquetScan INSTANCE = new ConvertHiveParquetScanToDrillParquetScan();
private static final DrillSqlOperator INT96_TO_TIMESTAMP =
- new DrillSqlOperator("convert_fromTIMESTAMP_IMPALA_LOCALTIMEZONE", 1, true, false);
+ new DrillSqlOperator("convert_fromTIMESTAMP_IMPALA", 1, true, false);
private static final DrillSqlOperator RTRIM = new DrillSqlOperator("RTRIM", 1, true, false);
http://git-wip-us.apache.org/repos/asf/drill/blob/4a0fd56c/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/ConvertFromImpalaTimestamp.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/ConvertFromImpalaTimestamp.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/ConvertFromImpalaTimestamp.java
index 38e0514..4d3d46b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/ConvertFromImpalaTimestamp.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/fn/impl/conv/ConvertFromImpalaTimestamp.java
@@ -28,29 +28,6 @@ import org.apache.drill.exec.expr.holders.VarBinaryHolder;
public class ConvertFromImpalaTimestamp {
- @FunctionTemplate(name = "convert_fromTIMESTAMP_IMPALA_LOCALTIMEZONE", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.NULL_IF_NULL)
- public static class ImpalaTimestampConvertFromWithLocalTimezone implements DrillSimpleFunc {
-
- @Param VarBinaryHolder in;
- @Output TimeStampHolder out;
-
-
- @Override
- public void setup() { }
-
- @Override
- public void eval() {
- org.apache.drill.exec.util.ByteBufUtil.checkBufferLength(in.buffer, in.start, in.end, 12);
-
- in.buffer.readerIndex(in.start);
- long nanosOfDay = in.buffer.readLong();
- int julianDay = in.buffer.readInt();
- long dateTime = (julianDay - org.apache.drill.exec.store.parquet.ParquetReaderUtility.JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH) *
- org.joda.time.DateTimeConstants.MILLIS_PER_DAY + (nanosOfDay / org.apache.drill.exec.store.parquet.ParquetReaderUtility.NanoTimeUtils.NANOS_PER_MILLISECOND);
- out.value = new org.joda.time.DateTime(dateTime, org.joda.time.chrono.JulianChronology.getInstance()).withZoneRetainFields(org.joda.time.DateTimeZone.UTC).getMillis();
- }
- }
-
@FunctionTemplate(name = "convert_fromTIMESTAMP_IMPALA", scope = FunctionTemplate.FunctionScope.SIMPLE, nulls = FunctionTemplate.NullHandling.NULL_IF_NULL)
public static class ImpalaTimestampConvertFrom implements DrillSimpleFunc {
@@ -68,8 +45,11 @@ public class ConvertFromImpalaTimestamp {
in.buffer.readerIndex(in.start);
long nanosOfDay = in.buffer.readLong();
int julianDay = in.buffer.readInt();
- out.value = (julianDay - org.apache.drill.exec.store.parquet.ParquetReaderUtility.JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH) *
+ long dateTime = (julianDay - org.apache.drill.exec.store.parquet.ParquetReaderUtility.JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH) *
org.joda.time.DateTimeConstants.MILLIS_PER_DAY + (nanosOfDay / org.apache.drill.exec.store.parquet.ParquetReaderUtility.NanoTimeUtils.NANOS_PER_MILLISECOND);
+ /* Note: This function uses local timezone for drill backward compatibility
+ and to avoid issues while reading hive parquet files */
+ out.value = org.joda.time.DateTimeZone.getDefault().convertUTCToLocal(dateTime);
}
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/4a0fd56c/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java
index a94e220..4247d41 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java
@@ -20,6 +20,7 @@ package org.apache.drill.exec.store.parquet;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.common.expression.PathSegment;
import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.expr.holders.NullableTimeStampHolder;
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.store.AbstractRecordReader;
@@ -41,6 +42,7 @@ import org.joda.time.Chronology;
import org.joda.time.DateTimeConstants;
import org.apache.parquet.example.data.simple.NanoTime;
import org.apache.parquet.io.api.Binary;
+import org.joda.time.DateTimeZone;
import java.util.Arrays;
import java.util.HashMap;
@@ -323,18 +325,29 @@ public class ParquetReaderUtility {
* @param binaryTimeStampValue
* hive, impala timestamp values with nanoseconds precision
* are stored in parquet Binary as INT96 (12 constant bytes)
- *
- * @return Unix Timestamp - the number of milliseconds since January 1, 1970, 00:00:00 GMT
- * represented by @param binaryTimeStampValue .
+ * @param retainLocalTimezone
+ * parquet files don't keep local timeZone according to the
+ * <a href="https://github.com/Parquet/parquet-format/blob/master/LogicalTypes.md#timestamp">Parquet spec</a>,
+ * but some tools (hive, for example) retain local timezone for parquet files by default
+ * Note: Impala doesn't retain local timezone by default
+ * @return Timestamp in milliseconds - the number of milliseconds since January 1, 1970, 00:00:00 GMT
+ * represented by @param binaryTimeStampValue.
+ * The nanos precision is cut to millis. Therefore the length of single timestamp value is
+ * {@value NullableTimeStampHolder#WIDTH} bytes instead of 12 bytes.
*/
- public static long getDateTimeValueFromBinary(Binary binaryTimeStampValue) {
+ public static long getDateTimeValueFromBinary(Binary binaryTimeStampValue, boolean retainLocalTimezone) {
// This method represents binaryTimeStampValue as ByteBuffer, where timestamp is stored as sum of
- // julian day number (32-bit) and nanos of day (64-bit)
+ // julian day number (4 bytes) and nanos of day (8 bytes)
NanoTime nt = NanoTime.fromBinary(binaryTimeStampValue);
int julianDay = nt.getJulianDay();
long nanosOfDay = nt.getTimeOfDayNanos();
- return (julianDay - JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH) * DateTimeConstants.MILLIS_PER_DAY
+ long dateTime = (julianDay - JULIAN_DAY_NUMBER_FOR_UNIX_EPOCH) * DateTimeConstants.MILLIS_PER_DAY
+ nanosOfDay / NANOS_PER_MILLISECOND;
+ if (retainLocalTimezone) {
+ return DateTimeZone.getDefault().convertUTCToLocal(dateTime);
+ } else {
+ return dateTime;
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/4a0fd56c/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/NullableFixedByteAlignedReaders.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/NullableFixedByteAlignedReaders.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/NullableFixedByteAlignedReaders.java
index b233a65..759b0f2 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/NullableFixedByteAlignedReaders.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/columnreaders/NullableFixedByteAlignedReaders.java
@@ -23,6 +23,7 @@ import java.nio.ByteBuffer;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.exec.expr.holders.NullableDecimal28SparseHolder;
import org.apache.drill.exec.expr.holders.NullableDecimal38SparseHolder;
+import org.apache.drill.exec.expr.holders.NullableTimeStampHolder;
import org.apache.drill.exec.store.parquet.ParquetReaderUtility;
import org.apache.drill.exec.util.DecimalUtility;
import org.apache.drill.exec.vector.NullableBigIntVector;
@@ -110,14 +111,9 @@ public class NullableFixedByteAlignedReaders {
/**
* Class for reading parquet fixed binary type INT96, which is used for storing hive,
- * impala timestamp values with nanoseconds precision (12 bytes). So it reads such values as a drill timestamp (8 bytes).
+ * impala timestamp values with nanoseconds precision (12 bytes). It reads such values as a drill timestamp (8 bytes).
*/
static class NullableFixedBinaryAsTimeStampReader extends NullableFixedByteAlignedReader<NullableTimeStampVector> {
- /**
- * The width of each element of the TimeStampVector is 8 byte(s).
- */
- private static final int TIMESTAMP_LENGTH_IN_BITS = 64;
-
NullableFixedBinaryAsTimeStampReader(ParquetRecordReader parentReader, int allocateSize, ColumnDescriptor descriptor,
ColumnChunkMetaData columnChunkMetaData, boolean fixedLength, NullableTimeStampVector v, SchemaElement schemaElement) throws ExecutionSetupException {
super(parentReader, allocateSize, descriptor, columnChunkMetaData, fixedLength, v, schemaElement);
@@ -127,19 +123,18 @@ public class NullableFixedByteAlignedReaders {
protected void readField(long recordsToReadInThisPass) {
this.bytebuf = pageReader.pageData;
if (usingDictionary) {
- for (int i = 0; i < recordsToReadInThisPass; i++){
+ for (int i = 0; i < recordsToReadInThisPass; i++) {
Binary binaryTimeStampValue = pageReader.dictionaryValueReader.readBytes();
- valueVec.getMutator().setSafe(valuesReadInCurrentPass + i, getDateTimeValueFromBinary(binaryTimeStampValue));
+ valueVec.getMutator().setSafe(valuesReadInCurrentPass + i, getDateTimeValueFromBinary(binaryTimeStampValue, true));
}
} else {
for (int i = 0; i < recordsToReadInThisPass; i++) {
Binary binaryTimeStampValue = pageReader.valueReader.readBytes();
- valueVec.getMutator().setSafe(valuesReadInCurrentPass + i, getDateTimeValueFromBinary(binaryTimeStampValue));
+ valueVec.getMutator().setSafe(valuesReadInCurrentPass + i, getDateTimeValueFromBinary(binaryTimeStampValue, true));
}
}
- // The nanos precision is cut to millis. Therefore the length of single timestamp value is 8 bytes(s)
- // instead of 12 byte(s).
- dataTypeLengthInBits = TIMESTAMP_LENGTH_IN_BITS;
+ // The width of each element of the TimeStampVector is 8 bytes (64 bits) instead of 12 bytes.
+ dataTypeLengthInBits = NullableTimeStampHolder.WIDTH * 8;
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/4a0fd56c/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetGroupConverter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetGroupConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetGroupConverter.java
index 2f2db05..79dc740 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetGroupConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet2/DrillParquetGroupConverter.java
@@ -646,7 +646,7 @@ public class DrillParquetGroupConverter extends GroupConverter {
@Override
public void addBinary(Binary value) {
- holder.value = getDateTimeValueFromBinary(value);
+ holder.value = getDateTimeValueFromBinary(value, true);
writer.write(holder);
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/4a0fd56c/exec/java-exec/src/test/java/org/apache/drill/TestBuilder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/TestBuilder.java b/exec/java-exec/src/test/java/org/apache/drill/TestBuilder.java
index 619959b..36a713f 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/TestBuilder.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/TestBuilder.java
@@ -14,12 +14,13 @@
* 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;
import static org.junit.Assert.assertEquals;
import java.io.IOException;
+import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
@@ -35,7 +36,6 @@ import org.apache.drill.common.expression.parser.ExprLexer;
import org.apache.drill.common.expression.parser.ExprParser;
import org.apache.drill.common.types.TypeProtos;
import org.apache.drill.common.types.Types;
-import org.apache.drill.exec.memory.BufferAllocator;
import org.apache.drill.exec.proto.UserBitShared;
import org.apache.drill.exec.proto.UserBitShared.QueryType;
import org.apache.drill.exec.proto.UserProtos.PreparedStatementHandle;
@@ -45,6 +45,7 @@ import org.apache.drill.exec.util.Text;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
+import org.joda.time.DateTimeZone;
public class TestBuilder {
@@ -665,4 +666,14 @@ public class TestBuilder {
}
return map;
}
+
+ /**
+ * Helper method for the timestamp values that depend on the local timezone
+ * @param value expected timestamp value in UTC
+ * @return timestamp value for the local timezone
+ */
+ public static Timestamp convertToLocalTimestamp(String value) {
+ long UTCTimestamp = Timestamp.valueOf(value).getTime();
+ return new Timestamp(DateTimeZone.getDefault().convertUTCToLocal(UTCTimestamp));
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/4a0fd56c/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java
index 362c943..65e9c38 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/writer/TestParquetWriter.java
@@ -1,4 +1,4 @@
-/**
+/*
* 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
@@ -18,6 +18,7 @@
package org.apache.drill.exec.physical.impl.writer;
import static org.apache.drill.exec.store.parquet.ParquetRecordWriter.DRILL_VERSION_PROPERTY;
+import static org.apache.drill.TestBuilder.convertToLocalTimestamp;
import static org.apache.parquet.format.converter.ParquetMetadataConverter.SKIP_ROW_GROUPS;
import static org.junit.Assert.assertEquals;
@@ -25,7 +26,6 @@ import java.io.File;
import java.io.FileWriter;
import java.math.BigDecimal;
import java.sql.Date;
-import java.sql.Timestamp;
import java.util.Arrays;
import java.util.ArrayList;
import java.util.Collection;
@@ -45,8 +45,6 @@ import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
-import org.apache.log4j.Level;
-import org.apache.parquet.Log;
import org.apache.parquet.hadoop.ParquetFileReader;
import org.apache.parquet.hadoop.metadata.ParquetMetadata;
import org.joda.time.DateTime;
@@ -57,7 +55,6 @@ import org.junit.Ignore;
import org.junit.Rule;
import org.junit.Test;
import org.junit.rules.TemporaryFolder;
-import org.junit.rules.Timeout;
import org.junit.runner.RunWith;
import org.junit.runners.Parameterized;
@@ -759,7 +756,7 @@ public class TestParquetWriter extends BaseTestQuery {
compareParquetReadersColumnar("field_impala_ts", "cp.`parquet/int96_impala_1.parquet`");
} finally {
test("alter session reset %s", ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP);
- }
+ }
}
/*
@@ -774,6 +771,7 @@ public class TestParquetWriter extends BaseTestQuery {
Test the reading of a binary field as drill timestamp where data is in dicationary _and_ non-dictionary encoded pages
*/
@Test
+ @Ignore("relies on particular time zone, works for UTC")
public void testImpalaParquetBinaryAsTimeStamp_DictChange() throws Exception {
final String WORKING_PATH = TestTools.getWorkingPath();
final String TEST_RES_PATH = WORKING_PATH + "/src/test/resources";
@@ -837,6 +835,7 @@ public class TestParquetWriter extends BaseTestQuery {
Test the conversion from int96 to impala timestamp with hive data including nulls. Validate against expected values
*/
@Test
+ @Ignore("relies on particular time zone")
public void testHiveParquetTimestampAsInt96_basic() throws Exception {
final String q = "SELECT cast(convert_from(timestamp_field, 'TIMESTAMP_IMPALA') as varchar(19)) as timestamp_field "
+ "from cp.`parquet/part1/hive_all_types.parquet` ";
@@ -845,7 +844,7 @@ public class TestParquetWriter extends BaseTestQuery {
.unOrdered()
.sqlQuery(q)
.baselineColumns("timestamp_field")
- .baselineValues("2013-07-06 00:01:00")
+ .baselineValues("2013-07-05 17:01:00")
.baselineValues((Object)null)
.go();
}
@@ -917,10 +916,11 @@ public class TestParquetWriter extends BaseTestQuery {
try {
testBuilder()
.ordered()
- .sqlQuery("select `%s` from %s", selection, table)
+ .sqlQuery("select `%1$s` from %2$s order by `%1$s`", selection, table)
.optionSettingQueriesForTestQuery(
"alter session set `%s` = true", ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP)
- .sqlBaselineQuery("select convert_from(`%1$s`, 'TIMESTAMP_IMPALA') as `%1$s` from %2$s", selection, table)
+ .sqlBaselineQuery("select convert_from(`%1$s`, 'TIMESTAMP_IMPALA') as `%1$s` from %2$s order by `%1$s`",
+ selection, table)
.optionSettingQueriesForBaseline(
"alter session set `%s` = false", ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP)
.build()
@@ -968,13 +968,16 @@ public class TestParquetWriter extends BaseTestQuery {
public void testInt96TimeStampValueWidth() throws Exception {
try {
testBuilder()
- .ordered()
- .sqlQuery("select c, d from cp.`parquet/data.snappy.parquet` where d = '2015-07-18 13:52:51'")
+ .unOrdered()
+ .sqlQuery("select c, d from cp.`parquet/data.snappy.parquet` " +
+ "where `a` is not null and `c` is not null and `d` is not null")
.optionSettingQueriesForTestQuery(
"alter session set `%s` = true", ExecConstants.PARQUET_READER_INT96_AS_TIMESTAMP)
.baselineColumns("c", "d")
- .baselineValues(new DateTime(Date.valueOf("2011-04-11").getTime()),
- new DateTime(Timestamp.valueOf("2015-07-18 13:52:51").getTime()))
+ .baselineValues(new DateTime(Date.valueOf("2012-12-15").getTime()),
+ new DateTime(convertToLocalTimestamp("2016-04-24 20:06:28")))
+ .baselineValues(new DateTime(Date.valueOf("2011-07-09").getTime()),
+ new DateTime(convertToLocalTimestamp("2015-04-15 22:35:49")))
.build()
.run();
} finally {