Skip to content

Commit

Permalink
Pushdown nested fields (#3)
Browse files Browse the repository at this point in the history 
* Pushdown dereference expression to paruqet reader (qqibrow)

* Fix TestMergeNestedColumns
  • Loading branch information
James Taylor authored and Puneet Jaiswal committed Jul 31, 2019
1 parent f1f77c6 commit e0a3478
Show file tree
Hide file tree
Showing 45 changed files with 1,736 additions and 89 deletions.
23 changes: 17 additions & 6 deletions presto-hive/src/main/java/io/prestosql/plugin/hive/HiveColumnHandle.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonProperty;
import io.prestosql.spi.NestedColumn;
import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.connector.ColumnMetadata;
import io.prestosql.spi.type.TypeManager;
Expand Down Expand Up @@ -60,6 +61,7 @@ public enum ColumnType
private final int hiveColumnIndex;
private final ColumnType columnType;
private final Optional<String> comment;
private final Optional<NestedColumn> nestedColumn;

@JsonCreator
public HiveColumnHandle(
Expand All @@ -68,7 +70,8 @@ public HiveColumnHandle(
@JsonProperty("typeSignature") TypeSignature typeSignature,
@JsonProperty("hiveColumnIndex") int hiveColumnIndex,
@JsonProperty("columnType") ColumnType columnType,
@JsonProperty("comment") Optional<String> comment)
@JsonProperty("comment") Optional<String> comment,
@JsonProperty("nestedColumn") Optional<NestedColumn> nestedColumn)
{
this.name = requireNonNull(name, "name is null");
checkArgument(hiveColumnIndex >= 0 || columnType == PARTITION_KEY || columnType == SYNTHESIZED, "hiveColumnIndex is negative");
Expand All @@ -77,6 +80,7 @@ public HiveColumnHandle(
this.typeName = requireNonNull(typeSignature, "type is null");
this.columnType = requireNonNull(columnType, "columnType is null");
this.comment = requireNonNull(comment, "comment is null");
this.nestedColumn = requireNonNull(nestedColumn, "nestedColumn is null");
}

@JsonProperty
Expand All @@ -97,6 +101,12 @@ public int getHiveColumnIndex()
return hiveColumnIndex;
}

@JsonProperty
public Optional<NestedColumn> getNestedColumn()
{
return nestedColumn;
}

public boolean isPartitionKey()
{
return columnType == PARTITION_KEY;
Expand Down Expand Up @@ -133,7 +143,7 @@ public ColumnType getColumnType()
@Override
public int hashCode()
{
return Objects.hash(name, hiveColumnIndex, hiveType, columnType, comment);
return Objects.hash(name, hiveColumnIndex, hiveType, columnType, comment, nestedColumn);
}

@Override
Expand All @@ -150,7 +160,8 @@ public boolean equals(Object obj)
Objects.equals(this.hiveColumnIndex, other.hiveColumnIndex) &&
Objects.equals(this.hiveType, other.hiveType) &&
Objects.equals(this.columnType, other.columnType) &&
Objects.equals(this.comment, other.comment);
Objects.equals(this.comment, other.comment) &&
Objects.equals(this.nestedColumn, other.nestedColumn);
}

@Override
Expand All @@ -167,12 +178,12 @@ public static HiveColumnHandle updateRowIdHandle()
// plan-time support for row-by-row delete so that planning doesn't fail. This is why we need
// rowid handle. Note that in Hive connector, rowid handle is not implemented beyond plan-time.

return new HiveColumnHandle(UPDATE_ROW_ID_COLUMN_NAME, HIVE_LONG, BIGINT.getTypeSignature(), -1, SYNTHESIZED, Optional.empty());
return new HiveColumnHandle(UPDATE_ROW_ID_COLUMN_NAME, HIVE_LONG, BIGINT.getTypeSignature(), -1, SYNTHESIZED, Optional.empty(), Optional.empty());
}

public static HiveColumnHandle pathColumnHandle()
{
return new HiveColumnHandle(PATH_COLUMN_NAME, PATH_HIVE_TYPE, PATH_TYPE_SIGNATURE, PATH_COLUMN_INDEX, SYNTHESIZED, Optional.empty());
return new HiveColumnHandle(PATH_COLUMN_NAME, PATH_HIVE_TYPE, PATH_TYPE_SIGNATURE, PATH_COLUMN_INDEX, SYNTHESIZED, Optional.empty(), Optional.empty());
}

/**
Expand All @@ -182,7 +193,7 @@ public static HiveColumnHandle pathColumnHandle()
*/
public static HiveColumnHandle bucketColumnHandle()
{
return new HiveColumnHandle(BUCKET_COLUMN_NAME, BUCKET_HIVE_TYPE, BUCKET_TYPE_SIGNATURE, BUCKET_COLUMN_INDEX, SYNTHESIZED, Optional.empty());
return new HiveColumnHandle(BUCKET_COLUMN_NAME, BUCKET_HIVE_TYPE, BUCKET_TYPE_SIGNATURE, BUCKET_COLUMN_INDEX, SYNTHESIZED, Optional.empty(), Optional.empty());
}

public static boolean isPathColumnHandle(HiveColumnHandle column)
Expand Down
39 changes: 38 additions & 1 deletion presto-hive/src/main/java/io/prestosql/plugin/hive/HiveMetadata.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@

import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Joiner;
import com.google.common.base.Preconditions;
import com.google.common.base.Splitter;
import com.google.common.base.Suppliers;
import com.google.common.base.Verify;
Expand Down Expand Up @@ -45,6 +46,7 @@
import io.prestosql.plugin.hive.metastore.Table;
import io.prestosql.plugin.hive.metastore.thrift.ThriftMetastoreUtil;
import io.prestosql.plugin.hive.statistics.HiveStatisticsProvider;
import io.prestosql.spi.NestedColumn;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.StandardErrorCode;
import io.prestosql.spi.block.Block;
Expand Down Expand Up @@ -176,6 +178,7 @@
import static io.prestosql.plugin.hive.HiveUtil.decodeViewData;
import static io.prestosql.plugin.hive.HiveUtil.encodeViewData;
import static io.prestosql.plugin.hive.HiveUtil.getPartitionKeyColumnHandles;
import static io.prestosql.plugin.hive.HiveUtil.getRegularColumnHandles;
import static io.prestosql.plugin.hive.HiveUtil.hiveColumnHandles;
import static io.prestosql.plugin.hive.HiveUtil.schemaTableName;
import static io.prestosql.plugin.hive.HiveUtil.toPartitionValues;
Expand Down Expand Up @@ -633,6 +636,39 @@ public ColumnMetadata getColumnMetadata(ConnectorSession session, ConnectorTable
return ((HiveColumnHandle) columnHandle).getColumnMetadata(typeManager);
}

@Override
public Map<NestedColumn, ColumnHandle> getNestedColumnHandles(ConnectorSession session, ConnectorTableHandle tableHandle, Collection<NestedColumn> nestedColumns)
{
if (!HiveSessionProperties.isStatisticsEnabled(session)) {
return ImmutableMap.of();
}

SchemaTableName tableName = schemaTableName(tableHandle);
Optional<Table> table = metastore.getTable(tableName.getSchemaName(), tableName.getTableName());

if (!table.isPresent()) {
throw new TableNotFoundException(tableName);
}

// Only pushdown nested column for parquet table for now
if (!extractHiveStorageFormat(table.get()).equals(HiveStorageFormat.PARQUET)) {
return ImmutableMap.of();
}

List<HiveColumnHandle> regularColumnHandles = getRegularColumnHandles(table.get());
Map<String, HiveColumnHandle> regularHiveColumnHandles = regularColumnHandles.stream().collect(Collectors.toMap(HiveColumnHandle::getName, identity()));
ImmutableMap.Builder<NestedColumn, ColumnHandle> columnHandles = ImmutableMap.builder();
for (NestedColumn nestedColumn : nestedColumns) {
HiveColumnHandle hiveColumnHandle = regularHiveColumnHandles.get(nestedColumn.getBase());
Optional<HiveType> childType = hiveColumnHandle.getHiveType().findChildType(nestedColumn);
Preconditions.checkArgument(childType.isPresent(), "%s doesn't exist in parent type %s", nestedColumn, hiveColumnHandle.getHiveType());
if (hiveColumnHandle != null) {
columnHandles.put(nestedColumn, new HiveColumnHandle(nestedColumn.getName(), childType.get(), childType.get().getTypeSignature(), hiveColumnHandle.getHiveColumnIndex(), hiveColumnHandle.getColumnType(), hiveColumnHandle.getComment(), Optional.of(nestedColumn)));
}
}
return columnHandles.build();
}

@Override
public void createSchema(ConnectorSession session, String schemaName, Map<String, Object> properties)
{
Expand Down Expand Up @@ -2148,7 +2184,8 @@ else if (column.isHidden()) {
column.getType().getTypeSignature(),
ordinal,
columnType,
Optional.ofNullable(column.getComment())));
Optional.ofNullable(column.getComment()),
Optional.empty()));
ordinal++;
}

Expand Down
18 changes: 15 additions & 3 deletions presto-hive/src/main/java/io/prestosql/plugin/hive/HivePageSourceProvider.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
import com.google.common.collect.ImmutableSet;
import io.prestosql.plugin.hive.HdfsEnvironment.HdfsContext;
import io.prestosql.plugin.hive.HiveSplit.BucketConversion;
import io.prestosql.spi.NestedColumn;
import io.prestosql.spi.connector.ColumnHandle;
import io.prestosql.spi.connector.ConnectorPageSource;
import io.prestosql.spi.connector.ConnectorPageSourceProvider;
Expand Down Expand Up @@ -320,8 +321,13 @@ public static List<ColumnMapping> buildColumnMappings(
for (HiveColumnHandle column : columns) {
Optional<HiveType> coercionFrom = Optional.ofNullable(columnCoercions.get(column.getHiveColumnIndex()));
if (column.getColumnType() == REGULAR) {
checkArgument(regularColumnIndices.add(column.getHiveColumnIndex()), "duplicate hiveColumnIndex in columns list");
columnMappings.add(regular(column, regularIndex, coercionFrom));
if (column.getNestedColumn().isPresent()) {
columnMappings.add(regular(column, regularIndex, getHiveType(coercionFrom, column.getNestedColumn().get())));
}
else {
checkArgument(regularColumnIndices.add(column.getHiveColumnIndex()), "duplicate hiveColumnIndex in columns list");
columnMappings.add(regular(column, regularIndex, coercionFrom));
}
regularIndex++;
}
else {
Expand All @@ -345,6 +351,11 @@ public static List<ColumnMapping> buildColumnMappings(
return columnMappings.build();
}

private static Optional<HiveType> getHiveType(Optional<HiveType> baseType, NestedColumn nestedColumn)
{
return baseType.flatMap(type -> type.findChildType(nestedColumn));
}

public static List<ColumnMapping> extractRegularAndInterimColumnMappings(List<ColumnMapping> columnMappings)
{
return columnMappings.stream()
Expand All @@ -366,7 +377,8 @@ public static List<HiveColumnHandle> toColumnHandles(List<ColumnMapping> regular
columnMapping.getCoercionFrom().get().getTypeSignature(),
columnHandle.getHiveColumnIndex(),
columnHandle.getColumnType(),
Optional.empty());
Optional.empty(),
columnHandle.getNestedColumn());
})
.collect(toList());
}
Expand Down
18 changes: 18 additions & 0 deletions presto-hive/src/main/java/io/prestosql/plugin/hive/HiveType.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,9 @@

import com.fasterxml.jackson.annotation.JsonCreator;
import com.fasterxml.jackson.annotation.JsonValue;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import io.prestosql.spi.NestedColumn;
import io.prestosql.spi.PrestoException;
import io.prestosql.spi.type.NamedTypeSignature;
import io.prestosql.spi.type.RowFieldName;
Expand Down Expand Up @@ -175,6 +177,22 @@ public static boolean isSupportedType(TypeInfo typeInfo)
return false;
}

public Optional<HiveType> findChildType(NestedColumn nestedColumn)
{
TypeInfo typeInfo = getTypeInfo();
for (String part : nestedColumn.getRest()) {
Preconditions.checkArgument(typeInfo instanceof StructTypeInfo, "typeinfo is not struct type", typeInfo);
StructTypeInfo structTypeInfo = (StructTypeInfo) typeInfo;
try {
typeInfo = structTypeInfo.getStructFieldTypeInfo(part);
}
catch (RuntimeException e) {
return Optional.empty();
}
}
return Optional.of(toHiveType(typeInfo));
}

@JsonCreator
public static HiveType valueOf(String hiveTypeName)
{
Expand Down
4 changes: 2 additions & 2 deletions presto-hive/src/main/java/io/prestosql/plugin/hive/HiveUtil.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -825,7 +825,7 @@ public static List<HiveColumnHandle> getRegularColumnHandles(Table table)
// ignore unsupported types rather than failing
HiveType hiveType = field.getType();
if (hiveType.isSupportedType()) {
columns.add(new HiveColumnHandle(field.getName(), hiveType, hiveType.getTypeSignature(), hiveColumnIndex, REGULAR, field.getComment()));
columns.add(new HiveColumnHandle(field.getName(), hiveType, hiveType.getTypeSignature(), hiveColumnIndex, REGULAR, field.getComment(), Optional.empty()));
}
hiveColumnIndex++;
}
Expand All @@ -843,7 +843,7 @@ public static List<HiveColumnHandle> getPartitionKeyColumnHandles(Table table)
if (!hiveType.isSupportedType()) {
throw new PrestoException(NOT_SUPPORTED, format("Unsupported Hive type %s found in partition keys of table %s.%s", hiveType, table.getDatabaseName(), table.getTableName()));
}
columns.add(new HiveColumnHandle(field.getName(), hiveType, hiveType.getTypeSignature(), -1, PARTITION_KEY, field.getComment()));
columns.add(new HiveColumnHandle(field.getName(), hiveType, hiveType.getTypeSignature(), -1, PARTITION_KEY, field.getComment(), Optional.empty()));
}

return columns.build();
Expand Down
2 changes: 1 addition & 1 deletion presto-hive/src/main/java/io/prestosql/plugin/hive/orc/OrcPageSourceFactory.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -260,7 +260,7 @@ private static List<HiveColumnHandle> getPhysicalHiveColumnHandles(List<HiveColu
physicalOrdinal = nextMissingColumnIndex;
nextMissingColumnIndex++;
}
physicalColumns.add(new HiveColumnHandle(column.getName(), column.getHiveType(), column.getTypeSignature(), physicalOrdinal, column.getColumnType(), column.getComment()));
physicalColumns.add(new HiveColumnHandle(column.getName(), column.getHiveType(), column.getTypeSignature(), physicalOrdinal, column.getColumnType(), column.getComment(), column.getNestedColumn()));
}
return physicalColumns.build();
}
Expand Down
Loading

0 comments on commit e0a3478

Please sign in to comment.