diff --git a/presto-hive/src/main/java/io/prestosql/plugin/hive/parquet/ParquetPageSourceFactory.java b/presto-hive/src/main/java/io/prestosql/plugin/hive/parquet/ParquetPageSourceFactory.java index 3e1466a4e..ef01fa05a 100644 --- a/presto-hive/src/main/java/io/prestosql/plugin/hive/parquet/ParquetPageSourceFactory.java +++ b/presto-hive/src/main/java/io/prestosql/plugin/hive/parquet/ParquetPageSourceFactory.java @@ -213,6 +213,7 @@ public class ParquetPageSourceFactory messageColumnIO, blocks.build(), dataSource, + timeZone, systemMemoryContext, maxReadBlockSize); diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileFormats.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileFormats.java index a099705a5..4cca5c90e 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileFormats.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/AbstractTestHiveFileFormats.java @@ -96,7 +96,6 @@ import static io.prestosql.plugin.hive.HiveTestUtils.SESSION; import static io.prestosql.plugin.hive.HiveTestUtils.TYPE_MANAGER; import static io.prestosql.plugin.hive.HiveTestUtils.isDistinctFrom; import static io.prestosql.plugin.hive.HiveTestUtils.mapType; -import static io.prestosql.plugin.hive.HiveType.HIVE_TIMESTAMP; import static io.prestosql.plugin.hive.HiveUtil.isStructuralType; import static io.prestosql.plugin.hive.util.SerDeUtils.serializeObject; import static io.prestosql.spi.type.BigintType.BIGINT; diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveFileFormats.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveFileFormats.java index 82491f583..2476601ac 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveFileFormats.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/TestHiveFileFormats.java @@ -447,8 +447,10 @@ public class TestHiveFileFormats // TODO: empty arrays or maps with null keys don't seem to work // Parquet does not support DATE return TEST_COLUMNS.stream() - .filter(column -> !ImmutableSet.of("t_null_array_int", "t_array_empty", "t_map_null_key", "t_map_null_key_complex_value", "t_map_null_key_complex_key_value") - .contains(column.getName())) + .filter(TestHiveFileFormats::withoutTimestamps) + .filter(TestHiveFileFormats::withoutNullMapKeyTests) + .filter(column -> !column.getName().equals("t_null_array_int")) + .filter(column -> !column.getName().equals("t_array_empty")) .filter(column -> column.isPartitionKey() || ( !hasType(column.getObjectInspector(), PrimitiveCategory.DATE)) && !hasType(column.getObjectInspector(), PrimitiveCategory.SHORT) && diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/benchmark/FileFormat.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/benchmark/FileFormat.java index baaa5760b..ca59c97b4 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/benchmark/FileFormat.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/benchmark/FileFormat.java @@ -166,7 +166,7 @@ public enum FileFormat @Override public ConnectorPageSource createFileFormatReader(ConnectorSession session, HdfsEnvironment hdfsEnvironment, File targetFile, List columnNames, List columnTypes) { - HivePageSourceFactory pageSourceFactory = new ParquetPageSourceFactory(TYPE_MANAGER, hdfsEnvironment, new FileFormatDataSourceStats(), new HiveConfig()); + HivePageSourceFactory pageSourceFactory = new ParquetPageSourceFactory(TYPE_MANAGER, hdfsEnvironment, new FileFormatDataSourceStats(), new HiveConfig().setParquetTimeZone("UTC")); return createPageSource(pageSourceFactory, session, targetFile, columnNames, columnTypes, HiveStorageFormat.PARQUET); } @@ -246,7 +246,7 @@ public enum FileFormat @Override public ConnectorPageSource createFileFormatReader(ConnectorSession session, HdfsEnvironment hdfsEnvironment, File targetFile, List columnNames, List columnTypes) { - HivePageSourceFactory pageSourceFactory = new ParquetPageSourceFactory(TYPE_MANAGER, hdfsEnvironment, new FileFormatDataSourceStats(), new HiveConfig()); + HivePageSourceFactory pageSourceFactory = new ParquetPageSourceFactory(TYPE_MANAGER, hdfsEnvironment, new FileFormatDataSourceStats(), new HiveConfig().setParquetTimeZone("UTC")); return createPageSource(pageSourceFactory, session, targetFile, columnNames, columnTypes, HiveStorageFormat.PARQUET); } diff --git a/presto-hive/src/test/java/io/prestosql/plugin/hive/parquet/AbstractTestParquetReader.java b/presto-hive/src/test/java/io/prestosql/plugin/hive/parquet/AbstractTestParquetReader.java index bb4fed2d0..61cb16b51 100644 --- a/presto-hive/src/test/java/io/prestosql/plugin/hive/parquet/AbstractTestParquetReader.java +++ b/presto-hive/src/test/java/io/prestosql/plugin/hive/parquet/AbstractTestParquetReader.java @@ -41,7 +41,6 @@ import org.testng.annotations.Test; import java.math.BigDecimal; import java.math.BigInteger; -import java.time.LocalDate; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -1772,7 +1771,7 @@ public abstract class AbstractTestParquetReader if (input == null) { return null; } - Timestamp timestamp = new Timestamp(); + long seconds = (input / 1000); int nanos = ((input % 1000) * 1_000_000); @@ -1787,9 +1786,7 @@ public abstract class AbstractTestParquetReader nanos -= 1_000_000_000; seconds += 1; } - timestamp.setTimeInMillis(seconds * 1000); - timestamp.setNanos(nanos); - return timestamp; + return Timestamp.ofEpochSecond(seconds, nanos); } private static SqlTimestamp intToSqlTimestamp(Integer input) @@ -1797,7 +1794,7 @@ public abstract class AbstractTestParquetReader if (input == null) { return null; } - return sqlTimestampOf(input); + return sqlTimestampOf((long) input); } private static Date intToDate(Integer input) @@ -1805,7 +1802,7 @@ public abstract class AbstractTestParquetReader if (input == null) { return null; } - return Date.valueOf(LocalDate.ofEpochDay(input).toString()); + return Date.ofEpochDay(input); } private static SqlDate intToSqlDate(Integer input) diff --git a/presto-parquet/pom.xml b/presto-parquet/pom.xml index badc73f7b..993ddc3a3 100644 --- a/presto-parquet/pom.xml +++ b/presto-parquet/pom.xml @@ -52,6 +52,11 @@ fastutil + + joda-time + joda-time + + org.xerial.snappy snappy-java diff --git a/presto-parquet/src/main/java/io/prestosql/parquet/reader/ParquetReader.java b/presto-parquet/src/main/java/io/prestosql/parquet/reader/ParquetReader.java index 4d10a6a7b..08a7bb5dd 100644 --- a/presto-parquet/src/main/java/io/prestosql/parquet/reader/ParquetReader.java +++ b/presto-parquet/src/main/java/io/prestosql/parquet/reader/ParquetReader.java @@ -39,6 +39,7 @@ import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; import org.apache.parquet.hadoop.metadata.ColumnPath; import org.apache.parquet.io.MessageColumnIO; import org.apache.parquet.io.PrimitiveColumnIO; +import org.joda.time.DateTimeZone; import java.io.Closeable; import java.io.IOException; @@ -67,6 +68,7 @@ public class ParquetReader private final List blocks; private final List columns; private final ParquetDataSource dataSource; + private final DateTimeZone timeZone; private final AggregatedMemoryContext systemMemoryContext; private int currentBlock; @@ -87,11 +89,13 @@ public class ParquetReader public ParquetReader(MessageColumnIO messageColumnIO, List blocks, ParquetDataSource dataSource, + DateTimeZone timeZone, AggregatedMemoryContext systemMemoryContext, DataSize maxReadBlockSize) { this.blocks = blocks; this.dataSource = requireNonNull(dataSource, "dataSource is null"); + this.timeZone = requireNonNull(timeZone, "timeZone is null"); this.systemMemoryContext = requireNonNull(systemMemoryContext, "systemMemoryContext is null"); this.currentRowGroupMemoryContext = systemMemoryContext.newAggregatedMemoryContext(); this.maxReadBlockBytes = requireNonNull(maxReadBlockSize, "maxReadBlockSize is null").toBytes(); @@ -257,7 +261,7 @@ public class ParquetReader { for (PrimitiveColumnIO columnIO : columns) { RichColumnDescriptor column = new RichColumnDescriptor(columnIO.getColumnDescriptor(), columnIO.getType().asPrimitiveType()); - columnReaders[columnIO.getId()] = PrimitiveColumnReader.createReader(column); + columnReaders[columnIO.getId()] = PrimitiveColumnReader.createReader(column, timeZone); } } diff --git a/presto-parquet/src/main/java/io/prestosql/parquet/reader/PrimitiveColumnReader.java b/presto-parquet/src/main/java/io/prestosql/parquet/reader/PrimitiveColumnReader.java index ba9d5686d..03888091f 100644 --- a/presto-parquet/src/main/java/io/prestosql/parquet/reader/PrimitiveColumnReader.java +++ b/presto-parquet/src/main/java/io/prestosql/parquet/reader/PrimitiveColumnReader.java @@ -35,6 +35,7 @@ import org.apache.parquet.column.ColumnDescriptor; import org.apache.parquet.column.values.ValuesReader; import org.apache.parquet.column.values.rle.RunLengthBitPackingHybridDecoder; import org.apache.parquet.io.ParquetDecodingException; +import org.joda.time.DateTimeZone; import java.io.ByteArrayInputStream; import java.io.IOException; @@ -80,7 +81,7 @@ public abstract class PrimitiveColumnReader return ParquetTypeUtils.isValueNull(columnDescriptor.isRequired(), definitionLevel, columnDescriptor.getMaxDefinitionLevel()); } - public static PrimitiveColumnReader createReader(RichColumnDescriptor descriptor) + public static PrimitiveColumnReader createReader(RichColumnDescriptor descriptor, DateTimeZone timeZone) { switch (descriptor.getType()) { case BOOLEAN: @@ -90,7 +91,7 @@ public abstract class PrimitiveColumnReader case INT64: return createDecimalColumnReader(descriptor).orElse(new LongColumnReader(descriptor)); case INT96: - return new TimestampColumnReader(descriptor); + return new TimestampColumnReader(descriptor, timeZone); case FLOAT: return new FloatColumnReader(descriptor); case DOUBLE: diff --git a/presto-parquet/src/main/java/io/prestosql/parquet/reader/TimestampColumnReader.java b/presto-parquet/src/main/java/io/prestosql/parquet/reader/TimestampColumnReader.java index 4149f849d..2cd93477f 100644 --- a/presto-parquet/src/main/java/io/prestosql/parquet/reader/TimestampColumnReader.java +++ b/presto-parquet/src/main/java/io/prestosql/parquet/reader/TimestampColumnReader.java @@ -15,25 +15,38 @@ package io.prestosql.parquet.reader; import io.prestosql.parquet.RichColumnDescriptor; import io.prestosql.spi.block.BlockBuilder; +import io.prestosql.spi.type.TimestampWithTimeZoneType; import io.prestosql.spi.type.Type; -import org.apache.parquet.io.api.Binary; +import org.joda.time.DateTimeZone; import static io.prestosql.parquet.ParquetTimestampUtils.getTimestampMillis; +import static io.prestosql.spi.type.DateTimeEncoding.packDateTimeWithZone; +import static io.prestosql.spi.type.TimeZoneKey.UTC_KEY; +import static java.util.Objects.requireNonNull; public class TimestampColumnReader extends PrimitiveColumnReader { - public TimestampColumnReader(RichColumnDescriptor descriptor) + private final DateTimeZone timeZone; + + public TimestampColumnReader(RichColumnDescriptor descriptor, DateTimeZone timeZone) { super(descriptor); + this.timeZone = requireNonNull(timeZone, "timeZone is null"); } @Override protected void readValue(BlockBuilder blockBuilder, Type type) { if (definitionLevel == columnDescriptor.getMaxDefinitionLevel()) { - Binary binary = valuesReader.readBytes(); - type.writeLong(blockBuilder, getTimestampMillis(binary)); + long utcMillis = getTimestampMillis(valuesReader.readBytes()); + if (type instanceof TimestampWithTimeZoneType) { + type.writeLong(blockBuilder, packDateTimeWithZone(utcMillis, UTC_KEY)); + } + else { + utcMillis = timeZone.convertUTCToLocal(utcMillis); + type.writeLong(blockBuilder, utcMillis); + } } else if (isValueNull()) { blockBuilder.appendNull();