Add Hive schemas into InputFormat#getSplits
@@ -85,6 +85,7 @@ import static com.facebook.presto.spi.StandardErrorCode.NOT_SUPPORTED;
import static com.google.common.base.Preconditions.checkArgument;
import static com.google.common.base.Preconditions.checkState;
import static com.google.common.collect.ImmutableList.toImmutableList;
+import static com.google.common.collect.Maps.fromProperties;
import static com.google.common.collect.Streams.stream;
import static com.google.common.util.concurrent.Futures.immediateFuture;
import static java.lang.Math.max;
@@ -145,6 +146,7 @@ public class StoragePartitionLoader
{
String partitionName = partition.getHivePartition().getPartitionId();
Storage storage = partition.getPartition().map(Partition::getStorage).orElse(table.getStorage());
+ Properties schema = getPartitionSchema(table, partition.getPartition());
String inputFormatName = storage.getStorageFormat().getInputFormat();
int partitionDataColumnCount = partition.getPartition()
.map(p -> p.getColumns().size())
@@ -248,6 +250,8 @@ public class StoragePartitionLoader
}
JobConf jobConf = toJobConf(configuration);
FileInputFormat.setInputPaths(jobConf, path);
+ // SerDes parameters and Table parameters passing into input format
+ fromProperties(schema).forEach(jobConf::set);
InputSplit[] splits = inputFormat.getSplits(jobConf, 0);
return addSplitsToSource(splits, splitFactory, hiveSplitSource, stopped);
@@ -256,7 +260,6 @@ public class StoragePartitionLoader
// S3 Select pushdown works at the granularity of individual S3 objects,
// Partial aggregation pushdown works at the granularity of individual files
// therefore we must not split files when either is enabled.
- Properties schema = getHiveSchema(storage.getSerdeParameters(), table.getParameters());
// Skip header / footer lines are not splittable except for a special case when skip.header.line.count=1
boolean splittable = !s3SelectPushdownEnabled && !partialAggregationsPushedDown && getFooterCount(schema) == 0 && getHeaderCount(schema) <= 1;
@@ -472,6 +475,11 @@ public class StoragePartitionLoader
}
}
+ private static Properties getPartitionSchema(Table table, Optional<Partition> partition)
+ {
+ return partition.map(value -> getHiveSchema(value, table)).orElseGet(() -> getHiveSchema(table));
+ }
+
public static class BucketSplitInfo
{
private final List<HiveColumnHandle> bucketColumns;
