Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Support for variable precision timestamps in Hive connector (read path) #4953

Closed
wants to merge 5 commits into from
Closed
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -16,11 +16,15 @@
import io.airlift.slice.Slice;
import io.airlift.slice.Slices;
import io.prestosql.hadoop.TextLineLengthLimitExceededException;
import io.prestosql.plugin.base.type.DecodedTimestamp;
import io.prestosql.plugin.base.type.PrestoTimestampEncoder;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.block.Block;
import io.prestosql.spi.connector.RecordCursor;
import io.prestosql.spi.type.DecimalType;
import io.prestosql.spi.type.Decimals;
import io.prestosql.spi.type.LongTimestamp;
import io.prestosql.spi.type.TimestampType;
import io.prestosql.spi.type.Type;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
Expand All @@ -47,11 +51,14 @@
import java.io.UncheckedIOException;
import java.math.BigInteger;
import java.util.Arrays;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static io.prestosql.plugin.base.type.PrestoTimestampEncoderFactory.createTimestampEncoder;
import static io.prestosql.plugin.hive.HiveColumnHandle.ColumnType.REGULAR;
import static io.prestosql.plugin.hive.HiveErrorCode.HIVE_BAD_DATA;
import static io.prestosql.plugin.hive.HiveErrorCode.HIVE_CURSOR_ERROR;
Expand All @@ -70,8 +77,6 @@
import static io.prestosql.spi.type.IntegerType.INTEGER;
import static io.prestosql.spi.type.RealType.REAL;
import static io.prestosql.spi.type.SmallintType.SMALLINT;
import static io.prestosql.spi.type.TimestampType.TIMESTAMP_MILLIS;
import static io.prestosql.spi.type.Timestamps.MICROSECONDS_PER_MILLISECOND;
import static io.prestosql.spi.type.TinyintType.TINYINT;
import static io.prestosql.spi.type.VarbinaryType.VARBINARY;
import static io.prestosql.spi.type.Varchars.isVarcharType;
Expand All @@ -81,6 +86,7 @@
import static java.lang.Math.min;
import static java.lang.String.format;
import static java.util.Objects.requireNonNull;
import static org.joda.time.DateTimeZone.UTC;

public class GenericHiveRecordCursor<K, V extends Writable>
implements RecordCursor
Expand All @@ -106,6 +112,7 @@ public class GenericHiveRecordCursor<K, V extends Writable>
private final Slice[] slices;
private final Object[] objects;
private final boolean[] nulls;
private final PrestoTimestampEncoder<?>[] timestampEncoders;

private final long totalBytes;

Expand Down Expand Up @@ -151,13 +158,19 @@ public GenericHiveRecordCursor(
this.slices = new Slice[size];
this.objects = new Object[size];
this.nulls = new boolean[size];
this.timestampEncoders = new PrestoTimestampEncoder[size];

Map<Type, PrestoTimestampEncoder<?>> timestampEncodersBuilder = new HashMap<>();
// initialize data columns
for (int i = 0; i < columns.size(); i++) {
HiveColumnHandle column = columns.get(i);
checkState(column.getColumnType() == REGULAR, "column type must be regular");

types[i] = column.getType();
Type columnType = column.getType();
types[i] = columnType;
if (columnType instanceof TimestampType) {
timestampEncoders[i] = createTimestampEncoder((TimestampType) columnType, UTC);
}
hiveTypes[i] = column.getHiveType();

StructField field = rowInspector.getStructFieldRef(column.getName());
Expand Down Expand Up @@ -277,18 +290,18 @@ private void parseLongColumn(int column)
else {
Object fieldValue = ((PrimitiveObjectInspector) fieldInspectors[column]).getPrimitiveJavaObject(fieldData);
checkState(fieldValue != null, "fieldValue should not be null");
longs[column] = getLongExpressedValue(fieldValue);
longs[column] = getLongExpressedValue(fieldValue, column);
nulls[column] = false;
}
}

private long getLongExpressedValue(Object value)
private long getLongExpressedValue(Object value, int column)
{
if (value instanceof Date) {
return ((Date) value).toEpochDay();
}
if (value instanceof Timestamp) {
return ((Timestamp) value).toEpochMilli() * MICROSECONDS_PER_MILLISECOND;
return shortTimestamp((Timestamp) value, column);
}
if (value instanceof Float) {
return floatToRawIntBits(((Float) value));
Expand Down Expand Up @@ -455,7 +468,6 @@ public Object getObject(int fieldId)
{
checkState(!closed, "Cursor is closed");

validateType(fieldId, Block.class);
if (!loaded[fieldId]) {
parseObjectColumn(fieldId);
}
Expand All @@ -472,7 +484,17 @@ private void parseObjectColumn(int column)
nulls[column] = true;
}
else {
objects[column] = getBlockObject(types[column], fieldData, fieldInspectors[column]);
Type type = types[column];
if (type.getJavaType() == Block.class) {
objects[column] = getBlockObject(type, fieldData, fieldInspectors[column]);
}
else if (type instanceof TimestampType) {
Timestamp timestamp = (Timestamp) ((PrimitiveObjectInspector) fieldInspectors[column]).getPrimitiveJavaObject(fieldData);
objects[column] = longTimestamp(timestamp, column);
}
else {
throw new IllegalStateException("Unsupported type: " + type);
}
nulls[column] = false;
}
}
Expand Down Expand Up @@ -524,8 +546,13 @@ else if (isStructuralType(hiveTypes[column])) {
else if (DATE.equals(type)) {
parseLongColumn(column);
}
else if (TIMESTAMP_MILLIS.equals(type)) {
parseLongColumn(column);
else if (type instanceof TimestampType) {
if (((TimestampType) type).isShort()) {
parseLongColumn(column);
}
else {
parseObjectColumn(column);
}
}
else if (type instanceof DecimalType) {
parseDecimalColumn(column);
Expand Down Expand Up @@ -561,4 +588,18 @@ public void close()
throw new UncheckedIOException(e);
}
}

private long shortTimestamp(Timestamp value, int column)
{
@SuppressWarnings("unchecked")
PrestoTimestampEncoder<Long> encoder = (PrestoTimestampEncoder<Long>) timestampEncoders[column];
return encoder.getTimestamp(new DecodedTimestamp(value.toEpochSecond(), value.getNanos()));
}

private LongTimestamp longTimestamp(Timestamp value, int column)
{
@SuppressWarnings("unchecked")
PrestoTimestampEncoder<LongTimestamp> encoder = (PrestoTimestampEncoder<LongTimestamp>) timestampEncoders[column];
return encoder.getTimestamp(new DecodedTimestamp(value.toEpochSecond(), value.getNanos()));
}
}
15 changes: 15 additions & 0 deletions presto-hive/src/main/java/io/prestosql/plugin/hive/HiveConfig.java
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,8 @@ public class HiveConfig

private Duration dynamicFilteringProbeBlockingTimeout = new Duration(0, MINUTES);

private HiveTimestampPrecision timestampPrecision = HiveTimestampPrecision.MILLIS;

public int getMaxInitialSplits()
{
return maxInitialSplits;
Expand Down Expand Up @@ -980,4 +982,17 @@ public HiveConfig setDynamicFilteringProbeBlockingTimeout(Duration dynamicFilter
this.dynamicFilteringProbeBlockingTimeout = dynamicFilteringProbeBlockingTimeout;
return this;
}

public HiveTimestampPrecision getTimestampPrecision()
{
return timestampPrecision;
}

@Config("hive.timestamp-precision")
@ConfigDescription("Precision used to represent timestamps")
public HiveConfig setTimestampPrecision(HiveTimestampPrecision timestampPrecision)
{
this.timestampPrecision = timestampPrecision;
return this;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@
import io.prestosql.spi.statistics.TableStatisticType;
import io.prestosql.spi.statistics.TableStatistics;
import io.prestosql.spi.statistics.TableStatisticsMetadata;
import io.prestosql.spi.type.TimestampType;
import io.prestosql.spi.type.Type;
import io.prestosql.spi.type.TypeManager;
import io.prestosql.spi.type.VarcharType;
Expand Down Expand Up @@ -159,6 +160,7 @@
import static io.prestosql.plugin.hive.HivePartitionManager.extractPartitionValues;
import static io.prestosql.plugin.hive.HiveSessionProperties.getCompressionCodec;
import static io.prestosql.plugin.hive.HiveSessionProperties.getHiveStorageFormat;
import static io.prestosql.plugin.hive.HiveSessionProperties.getTimestampPrecision;
import static io.prestosql.plugin.hive.HiveSessionProperties.isBucketExecutionEnabled;
import static io.prestosql.plugin.hive.HiveSessionProperties.isCollectColumnStatisticsOnWrite;
import static io.prestosql.plugin.hive.HiveSessionProperties.isCreateEmptyBucketFiles;
Expand Down Expand Up @@ -545,7 +547,7 @@ private ConnectorTableMetadata doGetTableMetadata(ConnectorSession session, Sche

Function<HiveColumnHandle, ColumnMetadata> metadataGetter = columnMetadataGetter(table);
ImmutableList.Builder<ColumnMetadata> columns = ImmutableList.builder();
for (HiveColumnHandle columnHandle : hiveColumnHandles(table, typeManager)) {
for (HiveColumnHandle columnHandle : hiveColumnHandles(table, typeManager, getTimestampPrecision(session).getPrecision())) {
columns.add(metadataGetter.apply(columnHandle));
}

Expand Down Expand Up @@ -689,7 +691,7 @@ public Map<String, ColumnHandle> getColumnHandles(ConnectorSession session, Conn
SchemaTableName tableName = ((HiveTableHandle) tableHandle).getSchemaTableName();
Table table = metastore.getTable(new HiveIdentity(session), tableName.getSchemaName(), tableName.getTableName())
.orElseThrow(() -> new TableNotFoundException(tableName));
return hiveColumnHandles(table, typeManager).stream()
return hiveColumnHandles(table, typeManager, getTimestampPrecision(session).getPrecision()).stream()
.collect(toImmutableMap(HiveColumnHandle::getName, identity()));
}

Expand Down Expand Up @@ -1210,7 +1212,8 @@ public void finishStatisticsCollection(ConnectorSession session, ConnectorTableH
List<String> partitionColumnNames = partitionColumns.stream()
.map(Column::getName)
.collect(toImmutableList());
List<HiveColumnHandle> hiveColumnHandles = hiveColumnHandles(table, typeManager);
// TODO: revisit when handling write path
List<HiveColumnHandle> hiveColumnHandles = hiveColumnHandles(table, typeManager, TimestampType.DEFAULT_PRECISION);
aalbu marked this conversation as resolved.
Show resolved Hide resolved
Map<String, Type> columnTypes = hiveColumnHandles.stream()
.filter(columnHandle -> !columnHandle.isHidden())
.collect(toImmutableMap(HiveColumnHandle::getName, column -> column.getHiveType().getType(typeManager)));
Expand Down Expand Up @@ -1527,7 +1530,7 @@ public HiveInsertTableHandle beginInsert(ConnectorSession session, ConnectorTabl
}
}

List<HiveColumnHandle> handles = hiveColumnHandles(table, typeManager).stream()
List<HiveColumnHandle> handles = hiveColumnHandles(table, typeManager, getTimestampPrecision(session).getPrecision()).stream()
.filter(columnHandle -> !columnHandle.isHidden())
.collect(toList());

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -92,6 +92,7 @@ public final class HiveSessionProperties
private static final String IGNORE_ABSENT_PARTITIONS = "ignore_absent_partitions";
private static final String QUERY_PARTITION_FILTER_REQUIRED = "query_partition_filter_required";
private static final String PROJECTION_PUSHDOWN_ENABLED = "projection_pushdown_enabled";
private static final String TIMESTAMP_PRECISION = "timestamp_precision";
private static final String PARQUET_OPTIMIZED_WRITER_ENABLED = "parquet_optimized_writer_enabled";
private static final String DYNAMIC_FILTERING_PROBE_BLOCKING_TIMEOUT = "dynamic_filtering_probe_blocking_timeout";

Expand Down Expand Up @@ -368,6 +369,12 @@ public HiveSessionProperties(
"Projection push down enabled for hive",
hiveConfig.isProjectionPushdownEnabled(),
false),
enumProperty(
TIMESTAMP_PRECISION,
"Precision for timestamp columns in Hive tables",
HiveTimestampPrecision.class,
hiveConfig.getTimestampPrecision(),
false),
aalbu marked this conversation as resolved.
Show resolved Hide resolved
booleanProperty(
PARQUET_OPTIMIZED_WRITER_ENABLED,
"Experimental: Enable optimized writer",
Expand Down Expand Up @@ -639,6 +646,11 @@ public static boolean isProjectionPushdownEnabled(ConnectorSession session)
return session.getProperty(PROJECTION_PUSHDOWN_ENABLED, Boolean.class);
}

public static HiveTimestampPrecision getTimestampPrecision(ConnectorSession session)
{
return session.getProperty(TIMESTAMP_PRECISION, HiveTimestampPrecision.class);
}

public static boolean isParquetOptimizedWriterEnabled(ConnectorSession session)
{
return session.getProperty(PARQUET_OPTIMIZED_WRITER_ENABLED, Boolean.class);
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,31 @@
/*
* 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.plugin.hive;

public enum HiveTimestampPrecision
{
MILLIS(3), MICROS(6), NANOS(9);
aalbu marked this conversation as resolved.
Show resolved Hide resolved

private final int precision;

HiveTimestampPrecision(int precision)
{
this.precision = precision;
}

public int getPrecision()
{
return precision;
}
}
14 changes: 14 additions & 0 deletions presto-hive/src/main/java/io/prestosql/plugin/hive/HiveType.java
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
import io.prestosql.spi.type.NamedTypeSignature;
import io.prestosql.spi.type.RowFieldName;
import io.prestosql.spi.type.StandardTypes;
import io.prestosql.spi.type.TimestampType;
import io.prestosql.spi.type.Type;
import io.prestosql.spi.type.TypeManager;
import io.prestosql.spi.type.TypeSignature;
Expand Down Expand Up @@ -120,11 +121,24 @@ public TypeSignature getTypeSignature()
return getTypeSignature(typeInfo);
}

@Deprecated
public Type getType(TypeManager typeManager)
{
return typeManager.getType(getTypeSignature());
}

public Type getType(TypeManager typeManager, int timestampPrecision)
{
Type tentativeType = typeManager.getType(getTypeSignature());
// TODO: handle timestamps in structural types (https://github.com/prestosql/presto/issues/5195)
if (tentativeType instanceof TimestampType) {
if (((TimestampType) tentativeType).getPrecision() != timestampPrecision) {
aalbu marked this conversation as resolved.
Show resolved Hide resolved
return TimestampType.createTimestampType(timestampPrecision);
}
}
return tentativeType;
}

@Override
public boolean equals(Object o)
{
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -68,6 +68,7 @@
import static io.prestosql.plugin.hive.HiveErrorCode.HIVE_CANNOT_OPEN_SPLIT;
import static io.prestosql.plugin.hive.HiveErrorCode.HIVE_MISSING_DATA;
import static io.prestosql.plugin.hive.HivePageSourceFactory.ReaderPageSourceWithProjections.noProjectionAdaptation;
import static io.prestosql.plugin.hive.HiveSessionProperties.getTimestampPrecision;
import static io.prestosql.plugin.hive.ReaderProjections.projectBaseColumns;
import static io.prestosql.plugin.hive.util.HiveUtil.getDeserializerClassName;
import static io.prestosql.rcfile.text.TextRcFileEncoding.DEFAULT_NULL_SEQUENCE;
Expand Down Expand Up @@ -179,8 +180,9 @@ else if (deserializerClassName.equals(ColumnarSerDe.class.getName())) {

try {
ImmutableMap.Builder<Integer, Type> readColumns = ImmutableMap.builder();
int timestampPrecision = getTimestampPrecision(session).getPrecision();
for (HiveColumnHandle column : projectedReaderColumns) {
readColumns.put(column.getBaseHiveColumnIndex(), column.getHiveType().getType(typeManager));
readColumns.put(column.getBaseHiveColumnIndex(), column.getHiveType().getType(typeManager, timestampPrecision));
}

RcFileReader rcFileReader = new RcFileReader(
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -181,7 +181,9 @@ public static Optional<HiveBucketHandle> getHiveBucketHandle(Table table, TypeMa
return Optional.empty();
}

Map<String, HiveColumnHandle> map = getRegularColumnHandles(table, typeManager).stream()
// Bucketing on timestamp is not allowed, so we do not have to know session's selected timestamp precision
int dummyTimestampPrecision = -42;
Map<String, HiveColumnHandle> map = getRegularColumnHandles(table, typeManager, dummyTimestampPrecision).stream()
.collect(Collectors.toMap(HiveColumnHandle::getName, identity()));

ImmutableList.Builder<HiveColumnHandle> bucketColumns = ImmutableList.builder();
Expand Down
Loading