You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by gu...@apache.org on 2017/12/09 02:53:20 UTC
spark git commit: [SPARK-12297][SQL] Adjust timezone for int96 data
from impala
Repository: spark
Updated Branches:
refs/heads/master e4639fa68 -> acf7ef315
[SPARK-12297][SQL] Adjust timezone for int96 data from impala
## What changes were proposed in this pull request?
Int96 data written by impala vs data written by hive & spark is stored slightly differently -- they use a different offset for the timezone. This adds an option "spark.sql.parquet.int96TimestampConversion" (false by default) to adjust timestamps if and only if the writer is impala (or more precisely, if the parquet file's "createdBy" metadata does not start with "parquet-mr"). This matches the existing behavior in hive from HIVE-9482.
## How was this patch tested?
Unit test added, existing tests run via jenkins.
Author: Imran Rashid <ir...@cloudera.com>
Author: Henry Robinson <he...@apache.org>
Closes #19769 from squito/SPARK-12297_skip_conversion.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/acf7ef31
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/acf7ef31
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/acf7ef31
Branch: refs/heads/master
Commit: acf7ef3154e094875fa89f30a78ab111b267db91
Parents: e4639fa
Author: Imran Rashid <ir...@cloudera.com>
Authored: Sat Dec 9 11:53:15 2017 +0900
Committer: hyukjinkwon <gu...@gmail.com>
Committed: Sat Dec 9 11:53:15 2017 +0900
----------------------------------------------------------------------
.../spark/sql/catalyst/util/DateTimeUtils.scala | 1 +
.../org/apache/spark/sql/internal/SQLConf.scala | 9 ++
.../parquet/VectorizedColumnReader.java | 59 +++++++---
.../parquet/VectorizedParquetRecordReader.java | 18 ++-
.../datasources/parquet/ParquetFileFormat.scala | 30 ++++-
.../parquet/ParquetReadSupport.scala | 15 ++-
.../parquet/ParquetRecordMaterializer.scala | 7 +-
.../parquet/ParquetRowConverter.scala | 11 +-
.../resources/test-data/impala_timestamp.parq | Bin 0 -> 374 bytes
.../parquet/ParquetInteroperabilitySuite.scala | 114 +++++++++++++++++++
10 files changed, 237 insertions(+), 27 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
----------------------------------------------------------------------
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
index 746c3e8..b1ed256 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/util/DateTimeUtils.scala
@@ -61,6 +61,7 @@ object DateTimeUtils {
final val YearZero = -17999
final val toYearZero = to2001 + 7304850
final val TimeZoneGMT = TimeZone.getTimeZone("GMT")
+ final val TimeZoneUTC = TimeZone.getTimeZone("UTC")
final val MonthOf31Days = Set(1, 3, 5, 7, 8, 10, 12)
val TIMEZONE_OPTION = "timeZone"
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
----------------------------------------------------------------------
diff --git a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
index ce9cc56..1121444 100644
--- a/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
+++ b/sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala
@@ -291,6 +291,13 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val PARQUET_INT96_TIMESTAMP_CONVERSION = buildConf("spark.sql.parquet.int96TimestampConversion")
+ .doc("This controls whether timestamp adjustments should be applied to INT96 data when " +
+ "converting to timestamps, for data written by Impala. This is necessary because Impala " +
+ "stores INT96 data with a different timezone offset than Hive & Spark.")
+ .booleanConf
+ .createWithDefault(false)
+
object ParquetOutputTimestampType extends Enumeration {
val INT96, TIMESTAMP_MICROS, TIMESTAMP_MILLIS = Value
}
@@ -1206,6 +1213,8 @@ class SQLConf extends Serializable with Logging {
def isParquetINT96AsTimestamp: Boolean = getConf(PARQUET_INT96_AS_TIMESTAMP)
+ def isParquetINT96TimestampConversion: Boolean = getConf(PARQUET_INT96_TIMESTAMP_CONVERSION)
+
def isParquetINT64AsTimestampMillis: Boolean = getConf(PARQUET_INT64_AS_TIMESTAMP_MILLIS)
def parquetOutputTimestampType: ParquetOutputTimestampType.Value = {
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
index 71ca8b1..b764696 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedColumnReader.java
@@ -18,6 +18,7 @@
package org.apache.spark.sql.execution.datasources.parquet;
import java.io.IOException;
+import java.util.TimeZone;
import org.apache.parquet.bytes.BytesUtils;
import org.apache.parquet.column.ColumnDescriptor;
@@ -93,13 +94,18 @@ public class VectorizedColumnReader {
private final PageReader pageReader;
private final ColumnDescriptor descriptor;
private final OriginalType originalType;
+ // The timezone conversion to apply to int96 timestamps. Null if no conversion.
+ private final TimeZone convertTz;
+ private final static TimeZone UTC = DateTimeUtils.TimeZoneUTC();
public VectorizedColumnReader(
ColumnDescriptor descriptor,
OriginalType originalType,
- PageReader pageReader) throws IOException {
+ PageReader pageReader,
+ TimeZone convertTz) throws IOException {
this.descriptor = descriptor;
this.pageReader = pageReader;
+ this.convertTz = convertTz;
this.originalType = originalType;
this.maxDefLevel = descriptor.getMaxDefinitionLevel();
@@ -222,6 +228,10 @@ public class VectorizedColumnReader {
}
}
+ private boolean shouldConvertTimestamps() {
+ return convertTz != null && !convertTz.equals(UTC);
+ }
+
/**
* Reads `num` values into column, decoding the values from `dictionaryIds` and `dictionary`.
*/
@@ -294,11 +304,21 @@ public class VectorizedColumnReader {
break;
case INT96:
if (column.dataType() == DataTypes.TimestampType) {
- for (int i = rowId; i < rowId + num; ++i) {
- // TODO: Convert dictionary of Binaries to dictionary of Longs
- if (!column.isNullAt(i)) {
- Binary v = dictionary.decodeToBinary(dictionaryIds.getDictId(i));
- column.putLong(i, ParquetRowConverter.binaryToSQLTimestamp(v));
+ if (!shouldConvertTimestamps()) {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ Binary v = dictionary.decodeToBinary(dictionaryIds.getDictId(i));
+ column.putLong(i, ParquetRowConverter.binaryToSQLTimestamp(v));
+ }
+ }
+ } else {
+ for (int i = rowId; i < rowId + num; ++i) {
+ if (!column.isNullAt(i)) {
+ Binary v = dictionary.decodeToBinary(dictionaryIds.getDictId(i));
+ long rawTime = ParquetRowConverter.binaryToSQLTimestamp(v);
+ long adjTime = DateTimeUtils.convertTz(rawTime, convertTz, UTC);
+ column.putLong(i, adjTime);
+ }
}
}
} else {
@@ -428,13 +448,26 @@ public class VectorizedColumnReader {
if (column.dataType() == DataTypes.StringType || column.dataType() == DataTypes.BinaryType) {
defColumn.readBinarys(num, column, rowId, maxDefLevel, data);
} else if (column.dataType() == DataTypes.TimestampType) {
- for (int i = 0; i < num; i++) {
- if (defColumn.readInteger() == maxDefLevel) {
- column.putLong(rowId + i,
- // Read 12 bytes for INT96
- ParquetRowConverter.binaryToSQLTimestamp(data.readBinary(12)));
- } else {
- column.putNull(rowId + i);
+ if (!shouldConvertTimestamps()) {
+ for (int i = 0; i < num; i++) {
+ if (defColumn.readInteger() == maxDefLevel) {
+ // Read 12 bytes for INT96
+ long rawTime = ParquetRowConverter.binaryToSQLTimestamp(data.readBinary(12));
+ column.putLong(rowId + i, rawTime);
+ } else {
+ column.putNull(rowId + i);
+ }
+ }
+ } else {
+ for (int i = 0; i < num; i++) {
+ if (defColumn.readInteger() == maxDefLevel) {
+ // Read 12 bytes for INT96
+ long rawTime = ParquetRowConverter.binaryToSQLTimestamp(data.readBinary(12));
+ long adjTime = DateTimeUtils.convertTz(rawTime, convertTz, UTC);
+ column.putLong(rowId + i, adjTime);
+ } else {
+ column.putNull(rowId + i);
+ }
}
}
} else {
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java
----------------------------------------------------------------------
diff --git a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java
index 669d71e..14f2a58 100644
--- a/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java
+++ b/sql/core/src/main/java/org/apache/spark/sql/execution/datasources/parquet/VectorizedParquetRecordReader.java
@@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.datasources.parquet;
import java.io.IOException;
import java.util.Arrays;
import java.util.List;
+import java.util.TimeZone;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
@@ -78,6 +79,12 @@ public class VectorizedParquetRecordReader extends SpecificParquetRecordReaderBa
private boolean[] missingColumns;
/**
+ * The timezone that timestamp INT96 values should be converted to. Null if no conversion. Here to workaround
+ * incompatibilities between different engines when writing timestamp values.
+ */
+ private TimeZone convertTz = null;
+
+ /**
* columnBatch object that is used for batch decoding. This is created on first use and triggers
* batched decoding. It is not valid to interleave calls to the batched interface with the row
* by row RecordReader APIs.
@@ -105,10 +112,15 @@ public class VectorizedParquetRecordReader extends SpecificParquetRecordReaderBa
*/
private final MemoryMode MEMORY_MODE;
- public VectorizedParquetRecordReader(boolean useOffHeap) {
+ public VectorizedParquetRecordReader(TimeZone convertTz, boolean useOffHeap) {
+ this.convertTz = convertTz;
MEMORY_MODE = useOffHeap ? MemoryMode.OFF_HEAP : MemoryMode.ON_HEAP;
}
+ public VectorizedParquetRecordReader(boolean useOffHeap) {
+ this(null, useOffHeap);
+ }
+
/**
* Implementation of RecordReader API.
*/
@@ -291,8 +303,8 @@ public class VectorizedParquetRecordReader extends SpecificParquetRecordReaderBa
columnReaders = new VectorizedColumnReader[columns.size()];
for (int i = 0; i < columns.size(); ++i) {
if (missingColumns[i]) continue;
- columnReaders[i] = new VectorizedColumnReader(
- columns.get(i), types.get(i).getOriginalType(), pages.getPageReader(columns.get(i)));
+ columnReaders[i] = new VectorizedColumnReader(columns.get(i), types.get(i).getOriginalType(),
+ pages.getPageReader(columns.get(i)), convertTz);
}
totalCountLoadedSoFar += pages.getRowCount();
}
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
index 2b10649..45bedf7 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetFileFormat.scala
@@ -45,6 +45,7 @@ import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.expressions._
import org.apache.spark.sql.catalyst.expressions.codegen.GenerateUnsafeProjection
import org.apache.spark.sql.catalyst.parser.LegacyTypeStringParser
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
import org.apache.spark.sql.execution.datasources._
import org.apache.spark.sql.execution.vectorized.{OffHeapColumnVector, OnHeapColumnVector}
import org.apache.spark.sql.internal.SQLConf
@@ -307,6 +308,9 @@ class ParquetFileFormat
hadoopConf.set(
ParquetWriteSupport.SPARK_ROW_SCHEMA,
requiredSchema.json)
+ hadoopConf.set(
+ SQLConf.SESSION_LOCAL_TIMEZONE.key,
+ sparkSession.sessionState.conf.sessionLocalTimeZone)
ParquetWriteSupport.setSchema(requiredSchema, hadoopConf)
@@ -345,6 +349,8 @@ class ParquetFileFormat
resultSchema.forall(_.dataType.isInstanceOf[AtomicType])
val enableRecordFilter: Boolean =
sparkSession.sessionState.conf.parquetRecordFilterEnabled
+ val timestampConversion: Boolean =
+ sparkSession.sessionState.conf.isParquetINT96TimestampConversion
// Whole stage codegen (PhysicalRDD) is able to deal with batches directly
val returningBatch = supportBatch(sparkSession, resultSchema)
@@ -363,6 +369,22 @@ class ParquetFileFormat
fileSplit.getLocations,
null)
+ val sharedConf = broadcastedHadoopConf.value.value
+ // PARQUET_INT96_TIMESTAMP_CONVERSION says to apply timezone conversions to int96 timestamps'
+ // *only* if the file was created by something other than "parquet-mr", so check the actual
+ // writer here for this file. We have to do this per-file, as each file in the table may
+ // have different writers.
+ def isCreatedByParquetMr(): Boolean = {
+ val footer = ParquetFileReader.readFooter(sharedConf, fileSplit.getPath, SKIP_ROW_GROUPS)
+ footer.getFileMetaData().getCreatedBy().startsWith("parquet-mr")
+ }
+ val convertTz =
+ if (timestampConversion && !isCreatedByParquetMr()) {
+ Some(DateTimeUtils.getTimeZone(sharedConf.get(SQLConf.SESSION_LOCAL_TIMEZONE.key)))
+ } else {
+ None
+ }
+
val attemptId = new TaskAttemptID(new TaskID(new JobID(), TaskType.MAP, 0), 0)
val hadoopAttemptContext =
new TaskAttemptContextImpl(broadcastedHadoopConf.value.value, attemptId)
@@ -374,8 +396,8 @@ class ParquetFileFormat
}
val taskContext = Option(TaskContext.get())
val parquetReader = if (enableVectorizedReader) {
- val vectorizedReader =
- new VectorizedParquetRecordReader(enableOffHeapColumnVector && taskContext.isDefined)
+ val vectorizedReader = new VectorizedParquetRecordReader(
+ convertTz.orNull, enableOffHeapColumnVector && taskContext.isDefined)
vectorizedReader.initialize(split, hadoopAttemptContext)
logDebug(s"Appending $partitionSchema ${file.partitionValues}")
vectorizedReader.initBatch(partitionSchema, file.partitionValues)
@@ -388,9 +410,9 @@ class ParquetFileFormat
// ParquetRecordReader returns UnsafeRow
val reader = if (pushed.isDefined && enableRecordFilter) {
val parquetFilter = FilterCompat.get(pushed.get, null)
- new ParquetRecordReader[UnsafeRow](new ParquetReadSupport, parquetFilter)
+ new ParquetRecordReader[UnsafeRow](new ParquetReadSupport(convertTz), parquetFilter)
} else {
- new ParquetRecordReader[UnsafeRow](new ParquetReadSupport)
+ new ParquetRecordReader[UnsafeRow](new ParquetReadSupport(convertTz))
}
reader.initialize(split, hadoopAttemptContext)
reader
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala
index 2854cb1..40ce5d5 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetReadSupport.scala
@@ -17,7 +17,7 @@
package org.apache.spark.sql.execution.datasources.parquet
-import java.util.{Map => JMap}
+import java.util.{Map => JMap, TimeZone}
import scala.collection.JavaConverters._
@@ -48,9 +48,17 @@ import org.apache.spark.sql.types._
* Due to this reason, we no longer rely on [[ReadContext]] to pass requested schema from [[init()]]
* to [[prepareForRead()]], but use a private `var` for simplicity.
*/
-private[parquet] class ParquetReadSupport extends ReadSupport[UnsafeRow] with Logging {
+private[parquet] class ParquetReadSupport(val convertTz: Option[TimeZone])
+ extends ReadSupport[UnsafeRow] with Logging {
private var catalystRequestedSchema: StructType = _
+ def this() {
+ // We need a zero-arg constructor for SpecificParquetRecordReaderBase. But that is only
+ // used in the vectorized reader, where we get the convertTz value directly, and the value here
+ // is ignored.
+ this(None)
+ }
+
/**
* Called on executor side before [[prepareForRead()]] and instantiating actual Parquet record
* readers. Responsible for figuring out Parquet requested schema used for column pruning.
@@ -95,7 +103,8 @@ private[parquet] class ParquetReadSupport extends ReadSupport[UnsafeRow] with Lo
new ParquetRecordMaterializer(
parquetRequestedSchema,
ParquetReadSupport.expandUDT(catalystRequestedSchema),
- new ParquetToSparkSchemaConverter(conf))
+ new ParquetToSparkSchemaConverter(conf),
+ convertTz)
}
}
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRecordMaterializer.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRecordMaterializer.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRecordMaterializer.scala
index 793755e..b2459dd 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRecordMaterializer.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRecordMaterializer.scala
@@ -17,6 +17,8 @@
package org.apache.spark.sql.execution.datasources.parquet
+import java.util.TimeZone
+
import org.apache.parquet.io.api.{GroupConverter, RecordMaterializer}
import org.apache.parquet.schema.MessageType
@@ -33,11 +35,12 @@ import org.apache.spark.sql.types.StructType
private[parquet] class ParquetRecordMaterializer(
parquetSchema: MessageType,
catalystSchema: StructType,
- schemaConverter: ParquetToSparkSchemaConverter)
+ schemaConverter: ParquetToSparkSchemaConverter,
+ convertTz: Option[TimeZone])
extends RecordMaterializer[UnsafeRow] {
private val rootConverter =
- new ParquetRowConverter(schemaConverter, parquetSchema, catalystSchema, NoopUpdater)
+ new ParquetRowConverter(schemaConverter, parquetSchema, catalystSchema, convertTz, NoopUpdater)
override def getCurrentRecord: UnsafeRow = rootConverter.currentRecord
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
index 10f6c3b..1199725 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetRowConverter.scala
@@ -19,6 +19,7 @@ package org.apache.spark.sql.execution.datasources.parquet
import java.math.{BigDecimal, BigInteger}
import java.nio.ByteOrder
+import java.util.TimeZone
import scala.collection.JavaConverters._
import scala.collection.mutable.ArrayBuffer
@@ -117,12 +118,14 @@ private[parquet] class ParquetPrimitiveConverter(val updater: ParentContainerUpd
* @param parquetType Parquet schema of Parquet records
* @param catalystType Spark SQL schema that corresponds to the Parquet record type. User-defined
* types should have been expanded.
+ * @param convertTz the optional time zone to convert to for int96 data
* @param updater An updater which propagates converted field values to the parent container
*/
private[parquet] class ParquetRowConverter(
schemaConverter: ParquetToSparkSchemaConverter,
parquetType: GroupType,
catalystType: StructType,
+ convertTz: Option[TimeZone],
updater: ParentContainerUpdater)
extends ParquetGroupConverter(updater) with Logging {
@@ -151,6 +154,8 @@ private[parquet] class ParquetRowConverter(
|${catalystType.prettyJson}
""".stripMargin)
+ private val UTC = DateTimeUtils.TimeZoneUTC
+
/**
* Updater used together with field converters within a [[ParquetRowConverter]]. It propagates
* converted filed values to the `ordinal`-th cell in `currentRow`.
@@ -279,7 +284,9 @@ private[parquet] class ParquetRowConverter(
val buf = value.toByteBuffer.order(ByteOrder.LITTLE_ENDIAN)
val timeOfDayNanos = buf.getLong
val julianDay = buf.getInt
- updater.setLong(DateTimeUtils.fromJulianDay(julianDay, timeOfDayNanos))
+ val rawTime = DateTimeUtils.fromJulianDay(julianDay, timeOfDayNanos)
+ val adjTime = convertTz.map(DateTimeUtils.convertTz(rawTime, _, UTC)).getOrElse(rawTime)
+ updater.setLong(adjTime)
}
}
@@ -309,7 +316,7 @@ private[parquet] class ParquetRowConverter(
case t: StructType =>
new ParquetRowConverter(
- schemaConverter, parquetType.asGroupType(), t, new ParentContainerUpdater {
+ schemaConverter, parquetType.asGroupType(), t, convertTz, new ParentContainerUpdater {
override def set(value: Any): Unit = updater.set(value.asInstanceOf[InternalRow].copy())
})
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/core/src/test/resources/test-data/impala_timestamp.parq
----------------------------------------------------------------------
diff --git a/sql/core/src/test/resources/test-data/impala_timestamp.parq b/sql/core/src/test/resources/test-data/impala_timestamp.parq
new file mode 100644
index 0000000..21e5318
Binary files /dev/null and b/sql/core/src/test/resources/test-data/impala_timestamp.parq differ
http://git-wip-us.apache.org/repos/asf/spark/blob/acf7ef31/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetInteroperabilitySuite.scala
----------------------------------------------------------------------
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetInteroperabilitySuite.scala b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetInteroperabilitySuite.scala
index 9dc5629..e3edafa 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetInteroperabilitySuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetInteroperabilitySuite.scala
@@ -19,7 +19,15 @@ package org.apache.spark.sql.execution.datasources.parquet
import java.io.File
+import org.apache.commons.io.FileUtils
+import org.apache.hadoop.fs.{FileSystem, Path, PathFilter}
+import org.apache.parquet.format.converter.ParquetMetadataConverter.NO_FILTER
+import org.apache.parquet.hadoop.ParquetFileReader
+import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
+
import org.apache.spark.sql.Row
+import org.apache.spark.sql.catalyst.util.DateTimeUtils
+import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.test.SharedSQLContext
class ParquetInteroperabilitySuite extends ParquetCompatibilityTest with SharedSQLContext {
@@ -87,4 +95,110 @@ class ParquetInteroperabilitySuite extends ParquetCompatibilityTest with SharedS
Row(Seq(2, 3))))
}
}
+
+ test("parquet timestamp conversion") {
+ // Make a table with one parquet file written by impala, and one parquet file written by spark.
+ // We should only adjust the timestamps in the impala file, and only if the conf is set
+ val impalaFile = "test-data/impala_timestamp.parq"
+
+ // here are the timestamps in the impala file, as they were saved by impala
+ val impalaFileData =
+ Seq(
+ "2001-01-01 01:01:01",
+ "2002-02-02 02:02:02",
+ "2003-03-03 03:03:03"
+ ).map(java.sql.Timestamp.valueOf)
+ val impalaPath = Thread.currentThread().getContextClassLoader.getResource(impalaFile)
+ .toURI.getPath
+ withTempPath { tableDir =>
+ val ts = Seq(
+ "2004-04-04 04:04:04",
+ "2005-05-05 05:05:05",
+ "2006-06-06 06:06:06"
+ ).map { s => java.sql.Timestamp.valueOf(s) }
+ import testImplicits._
+ // match the column names of the file from impala
+ val df = spark.createDataset(ts).toDF().repartition(1).withColumnRenamed("value", "ts")
+ df.write.parquet(tableDir.getAbsolutePath)
+ FileUtils.copyFile(new File(impalaPath), new File(tableDir, "part-00001.parq"))
+
+ Seq(false, true).foreach { int96TimestampConversion =>
+ Seq(false, true).foreach { vectorized =>
+ withSQLConf(
+ (SQLConf.PARQUET_OUTPUT_TIMESTAMP_TYPE.key,
+ SQLConf.ParquetOutputTimestampType.INT96.toString),
+ (SQLConf.PARQUET_INT96_TIMESTAMP_CONVERSION.key, int96TimestampConversion.toString()),
+ (SQLConf.PARQUET_VECTORIZED_READER_ENABLED.key, vectorized.toString())
+ ) {
+ val readBack = spark.read.parquet(tableDir.getAbsolutePath).collect()
+ assert(readBack.size === 6)
+ // if we apply the conversion, we'll get the "right" values, as saved by impala in the
+ // original file. Otherwise, they're off by the local timezone offset, set to
+ // America/Los_Angeles in tests
+ val impalaExpectations = if (int96TimestampConversion) {
+ impalaFileData
+ } else {
+ impalaFileData.map { ts =>
+ DateTimeUtils.toJavaTimestamp(DateTimeUtils.convertTz(
+ DateTimeUtils.fromJavaTimestamp(ts),
+ DateTimeUtils.TimeZoneUTC,
+ DateTimeUtils.getTimeZone(conf.sessionLocalTimeZone)))
+ }
+ }
+ val fullExpectations = (ts ++ impalaExpectations).map(_.toString).sorted.toArray
+ val actual = readBack.map(_.getTimestamp(0).toString).sorted
+ withClue(
+ s"int96TimestampConversion = $int96TimestampConversion; vectorized = $vectorized") {
+ assert(fullExpectations === actual)
+
+ // Now test that the behavior is still correct even with a filter which could get
+ // pushed down into parquet. We don't need extra handling for pushed down
+ // predicates because (a) in ParquetFilters, we ignore TimestampType and (b) parquet
+ // does not read statistics from int96 fields, as they are unsigned. See
+ // scalastyle:off line.size.limit
+ // https://github.com/apache/parquet-mr/blob/2fd62ee4d524c270764e9b91dca72e5cf1a005b7/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java#L419
+ // https://github.com/apache/parquet-mr/blob/2fd62ee4d524c270764e9b91dca72e5cf1a005b7/parquet-hadoop/src/main/java/org/apache/parquet/format/converter/ParquetMetadataConverter.java#L348
+ // scalastyle:on line.size.limit
+ //
+ // Just to be defensive in case anything ever changes in parquet, this test checks
+ // the assumption on column stats, and also the end-to-end behavior.
+
+ val hadoopConf = sparkContext.hadoopConfiguration
+ val fs = FileSystem.get(hadoopConf)
+ val parts = fs.listStatus(new Path(tableDir.getAbsolutePath), new PathFilter {
+ override def accept(path: Path): Boolean = !path.getName.startsWith("_")
+ })
+ // grab the meta data from the parquet file. The next section of asserts just make
+ // sure the test is configured correctly.
+ assert(parts.size == 2)
+ parts.foreach { part =>
+ val oneFooter =
+ ParquetFileReader.readFooter(hadoopConf, part.getPath, NO_FILTER)
+ assert(oneFooter.getFileMetaData.getSchema.getColumns.size === 1)
+ assert(oneFooter.getFileMetaData.getSchema.getColumns.get(0).getType() ===
+ PrimitiveTypeName.INT96)
+ val oneBlockMeta = oneFooter.getBlocks().get(0)
+ val oneBlockColumnMeta = oneBlockMeta.getColumns().get(0)
+ val columnStats = oneBlockColumnMeta.getStatistics
+ // This is the important assert. Column stats are written, but they are ignored
+ // when the data is read back as mentioned above, b/c int96 is unsigned. This
+ // assert makes sure this holds even if we change parquet versions (if eg. there
+ // were ever statistics even on unsigned columns).
+ assert(columnStats.isEmpty)
+ }
+
+ // These queries should return the entire dataset with the conversion applied,
+ // but if the predicates were applied to the raw values in parquet, they would
+ // incorrectly filter data out.
+ val query = spark.read.parquet(tableDir.getAbsolutePath)
+ .where("ts > '2001-01-01 01:00:00'")
+ val countWithFilter = query.count()
+ val exp = if (int96TimestampConversion) 6 else 5
+ assert(countWithFilter === exp, query)
+ }
+ }
+ }
+ }
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org