/*
 * Licensed 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 io.prestosql.parquet.reader;

import io.prestosql.parquet.RichColumnDescriptor;
import io.prestosql.spi.block.BlockBuilder;
import io.prestosql.spi.type.Type;
import org.apache.parquet.io.api.Binary;

import static io.prestosql.parquet.ParquetTimestampUtils.getTimestampMillis;

public class TimestampColumnReader
        extends PrimitiveColumnReader
{
    public TimestampColumnReader(RichColumnDescriptor descriptor)
    {
        super(descriptor);
    }

    @Override
    protected void readValue(BlockBuilder blockBuilder, Type type)
    {
        if (definitionLevel == columnDescriptor.getMaxDefinitionLevel()) {
            Binary binary = valuesReader.readBytes();
            type.writeLong(blockBuilder, getTimestampMillis(binary));
        }
        else if (isValueNull()) {
            blockBuilder.appendNull();
        }
    }

    @Override
    protected void skipValue()
    {
        if (definitionLevel == columnDescriptor.getMaxDefinitionLevel()) {
            valuesReader.readBytes();
        }
    }
}

@@ -721,52 +721,67 @@
 pe.T
-ype;%0Aimport org.apache.parquet.io.api.Binary
+imestampWithTimeZoneType;%0Aimport io.prestosql.spi.type.Type
 ;%0A%0Ai
@@ -855,16 +855,148 @@
 pMillis;
+%0Aimport static io.prestosql.spi.type.DateTimeEncoding.packDateTimeWithZone;%0Aimport static io.prestosql.spi.type.TimeZoneKey.UTC_KEY;
 %0A%0Apublic
@@ -1348,24 +1348,44 @@
     
-Binary binary = 
+long utcMillis = getTimestampMillis(
 valu
@@ -1396,32 +1396,33 @@
 ader.readBytes()
+)
 ;%0A            ty
@@ -1423,65 +1423,248 @@
     
-type.writeLong(blockBuilder, getTimestampMillis(binary));
+if (type instanceof TimestampWithTimeZoneType) %7B%0A                type.writeLong(blockBuilder, packDateTimeWithZone(utcMillis, UTC_KEY));%0A            %7D%0A            else %7B%0A                type.writeLong(blockBuilder, utcMillis);%0A            %7D
 %0A   

Add timestamp with UTC time zone support to Parquet column reader