http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetGroupScan.java ---------------------------------------------------------------------- diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetGroupScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetGroupScan.java index a53587c..564b3bd 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetGroupScan.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetGroupScan.java @@ -20,196 +20,106 @@ package org.apache.drill.exec.store.parquet; import java.io.IOException; import java.util.ArrayList; import java.util.Collection; -import java.util.HashMap; -import java.util.HashSet; +import java.util.LinkedHashMap; import java.util.List; import java.util.Map; -import java.util.Set; +import java.util.function.Function; +import java.util.stream.Collectors; import org.apache.drill.common.exceptions.ExecutionSetupException; import org.apache.drill.common.exceptions.UserException; -import org.apache.drill.common.expression.ErrorCollector; -import org.apache.drill.common.expression.ErrorCollectorImpl; -import org.apache.drill.common.expression.ExpressionStringBuilder; import org.apache.drill.common.expression.LogicalExpression; import org.apache.drill.common.expression.SchemaPath; import org.apache.drill.common.expression.ValueExpressions; import org.apache.drill.common.logical.FormatPluginConfig; import org.apache.drill.common.logical.StoragePluginConfig; -import org.apache.drill.common.types.TypeProtos; -import org.apache.drill.common.types.TypeProtos.MajorType; -import org.apache.drill.common.types.TypeProtos.MinorType; -import org.apache.drill.common.types.Types; -import org.apache.drill.exec.compile.sig.ConstantExpressionIdentifier; -import org.apache.drill.exec.expr.ExpressionTreeMaterializer; -import org.apache.drill.exec.expr.fn.FunctionImplementationRegistry; -import org.apache.drill.exec.expr.stat.ParquetFilterPredicate; -import org.apache.drill.exec.ops.UdfUtilities; -import org.apache.drill.exec.physical.EndpointAffinity; -import org.apache.drill.exec.physical.PhysicalOperatorSetupException; -import org.apache.drill.exec.physical.base.AbstractFileGroupScan; import org.apache.drill.exec.physical.base.GroupScan; import org.apache.drill.exec.physical.base.PhysicalOperator; -import org.apache.drill.exec.physical.base.ScanStats; -import org.apache.drill.exec.physical.base.ScanStats.GroupScanProperty; -import org.apache.drill.exec.planner.physical.PlannerSettings; import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint; -import org.apache.drill.exec.server.options.OptionManager; import org.apache.drill.exec.store.ColumnExplorer; import org.apache.drill.exec.store.StoragePluginRegistry; import org.apache.drill.exec.store.dfs.DrillFileSystem; import org.apache.drill.exec.store.dfs.FileSelection; +import org.apache.drill.exec.store.parquet.metadata.Metadata; import org.apache.drill.exec.util.DrillFileSystemUtil; import org.apache.drill.exec.store.dfs.MetadataContext; import org.apache.drill.exec.store.dfs.MetadataContext.PruneStatus; -import org.apache.drill.exec.store.dfs.ReadEntryFromHDFS; import org.apache.drill.exec.store.dfs.ReadEntryWithPath; -import org.apache.drill.exec.store.dfs.easy.FileWork; -import org.apache.drill.exec.store.parquet.Metadata.ColumnMetadata; -import org.apache.drill.exec.store.parquet.Metadata.ParquetFileMetadata; -import org.apache.drill.exec.store.parquet.Metadata.ParquetTableMetadataBase; -import org.apache.drill.exec.store.parquet.Metadata.RowGroupMetadata; -import org.apache.drill.exec.store.parquet.stat.ColumnStatistics; -import org.apache.drill.exec.store.parquet.stat.ParquetMetaStatCollector; -import org.apache.drill.exec.store.schedule.AffinityCreator; -import org.apache.drill.exec.store.schedule.AssignmentCreator; -import org.apache.drill.exec.store.schedule.CompleteWork; -import org.apache.drill.exec.store.schedule.EndpointByteMap; -import org.apache.drill.exec.store.schedule.EndpointByteMapImpl; -import org.apache.drill.exec.util.DecimalUtility; +import org.apache.drill.exec.store.parquet.metadata.MetadataBase.ParquetFileMetadata; +import org.apache.drill.exec.store.parquet.metadata.MetadataBase.ParquetTableMetadataBase; import org.apache.drill.exec.util.ImpersonationUtil; -import org.apache.drill.exec.vector.NullableBitVector; -import org.apache.drill.exec.vector.NullableBigIntVector; -import org.apache.drill.exec.vector.NullableDateVector; -import org.apache.drill.exec.vector.NullableDecimal18Vector; -import org.apache.drill.exec.vector.NullableFloat4Vector; -import org.apache.drill.exec.vector.NullableFloat8Vector; -import org.apache.drill.exec.vector.NullableIntVector; -import org.apache.drill.exec.vector.NullableIntervalVector; -import org.apache.drill.exec.vector.NullableSmallIntVector; -import org.apache.drill.exec.vector.NullableTimeStampVector; -import org.apache.drill.exec.vector.NullableTimeVector; -import org.apache.drill.exec.vector.NullableTinyIntVector; -import org.apache.drill.exec.vector.NullableUInt1Vector; -import org.apache.drill.exec.vector.NullableUInt2Vector; -import org.apache.drill.exec.vector.NullableUInt4Vector; -import org.apache.drill.exec.vector.NullableVarBinaryVector; -import org.apache.drill.exec.vector.NullableVarCharVector; -import org.apache.drill.exec.vector.ValueVector; import org.apache.hadoop.fs.FileStatus; +import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; -import org.joda.time.DateTimeConstants; -import org.apache.parquet.io.api.Binary; -import org.apache.parquet.schema.OriginalType; -import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName; import com.fasterxml.jackson.annotation.JacksonInject; import com.fasterxml.jackson.annotation.JsonCreator; -import com.fasterxml.jackson.annotation.JsonIgnore; import com.fasterxml.jackson.annotation.JsonProperty; import com.fasterxml.jackson.annotation.JsonTypeName; import com.google.common.base.Preconditions; -import com.google.common.collect.ArrayListMultimap; -import com.google.common.collect.ListMultimap; -import com.google.common.collect.Lists; -import com.google.common.collect.Maps; import com.google.common.collect.Sets; @JsonTypeName("parquet-scan") -public class ParquetGroupScan extends AbstractFileGroupScan { - static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ParquetGroupScan.class); +public class ParquetGroupScan extends AbstractParquetGroupScan { + private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ParquetGroupScan.class); - private final List<ReadEntryWithPath> entries; private final ParquetFormatPlugin formatPlugin; private final ParquetFormatConfig formatConfig; private final DrillFileSystem fs; private final MetadataContext metaContext; + private boolean usedMetadataCache; // false by default + // may change when filter push down / partition pruning is applied private String selectionRoot; - - private boolean usedMetadataCache = false; - private List<EndpointAffinity> endpointAffinities; - private List<SchemaPath> columns; - private ListMultimap<Integer, RowGroupInfo> mappings; - private List<RowGroupInfo> rowGroupInfos; - private LogicalExpression filter; - - /** - * The parquet table metadata may have already been read - * from a metadata cache file earlier; we can re-use during - * the ParquetGroupScan and avoid extra loading time. - */ - private Metadata.ParquetTableMetadataBase parquetTableMetadata = null; - private String cacheFileRoot = null; - - /* - * total number of rows (obtained from parquet footer) - */ - private long rowCount; - - /* - * total number of non-null value for each column in parquet files. - */ - private Map<SchemaPath, Long> columnValueCounts; - - @JsonCreator public ParquetGroupScan( // - @JsonProperty("userName") String userName, - @JsonProperty("entries") List<ReadEntryWithPath> entries,// - @JsonProperty("storage") StoragePluginConfig storageConfig, // - @JsonProperty("format") FormatPluginConfig formatConfig, // - @JacksonInject StoragePluginRegistry engineRegistry, // - @JsonProperty("columns") List<SchemaPath> columns, // - @JsonProperty("selectionRoot") String selectionRoot, // - @JsonProperty("cacheFileRoot") String cacheFileRoot, // - @JsonProperty("filter") LogicalExpression filter - ) throws IOException, ExecutionSetupException { - super(ImpersonationUtil.resolveUserName(userName)); - this.columns = columns; + private String cacheFileRoot; + + @JsonCreator + public ParquetGroupScan(@JacksonInject StoragePluginRegistry engineRegistry, + @JsonProperty("userName") String userName, + @JsonProperty("entries") List<ReadEntryWithPath> entries, + @JsonProperty("storage") StoragePluginConfig storageConfig, + @JsonProperty("format") FormatPluginConfig formatConfig, + @JsonProperty("columns") List<SchemaPath> columns, + @JsonProperty("selectionRoot") String selectionRoot, + @JsonProperty("cacheFileRoot") String cacheFileRoot, + @JsonProperty("filter") LogicalExpression filter) throws IOException, ExecutionSetupException { + super(ImpersonationUtil.resolveUserName(userName), columns, entries, filter); Preconditions.checkNotNull(storageConfig); Preconditions.checkNotNull(formatConfig); this.formatPlugin = (ParquetFormatPlugin) engineRegistry.getFormatPlugin(storageConfig, formatConfig); Preconditions.checkNotNull(formatPlugin); this.fs = ImpersonationUtil.createFileSystem(getUserName(), formatPlugin.getFsConf()); this.formatConfig = formatPlugin.getConfig(); - this.entries = entries; this.selectionRoot = selectionRoot; this.cacheFileRoot = cacheFileRoot; - this.filter = filter; this.metaContext = new MetadataContext(); init(); } - public ParquetGroupScan( // - String userName, - FileSelection selection, // - ParquetFormatPlugin formatPlugin, // - List<SchemaPath> columns) throws IOException { + public ParquetGroupScan(String userName, + FileSelection selection, + ParquetFormatPlugin formatPlugin, + List<SchemaPath> columns) throws IOException { this(userName, selection, formatPlugin, columns, ValueExpressions.BooleanExpression.TRUE); } - public ParquetGroupScan( // - String userName, - FileSelection selection, // - ParquetFormatPlugin formatPlugin, // - List<SchemaPath> columns, - LogicalExpression filter) // - throws IOException { - super(userName); + public ParquetGroupScan(String userName, + FileSelection selection, + ParquetFormatPlugin formatPlugin, + List<SchemaPath> columns, + LogicalExpression filter) throws IOException { + super(userName, columns, new ArrayList<>(), filter); + this.formatPlugin = formatPlugin; - this.columns = columns; this.formatConfig = formatPlugin.getConfig(); this.fs = ImpersonationUtil.createFileSystem(userName, formatPlugin.getFsConf()); - this.selectionRoot = selection.getSelectionRoot(); this.cacheFileRoot = selection.getCacheFileRoot(); MetadataContext metadataContext = selection.getMetaContext(); this.metaContext = metadataContext != null ? metadataContext : new MetadataContext(); - final FileSelection fileSelection = expandIfNecessary(selection); - - this.entries = Lists.newArrayList(); - + FileSelection fileSelection = expandIfNecessary(selection); if (fileSelection != null) { if (checkForInitializingEntriesWithSelectionRoot()) { // The fully expanded list is already stored as part of the fileSet @@ -220,41 +130,20 @@ public class ParquetGroupScan extends AbstractFileGroupScan { } } - this.filter = filter; - init(); } } - /* - * This is used to clone another copy of the group scan. - */ private ParquetGroupScan(ParquetGroupScan that) { this(that, null); } - /* - * This is used to clone another copy of the group scan, but keep the metadata caching info from the file selection - */ private ParquetGroupScan(ParquetGroupScan that, FileSelection selection) { super(that); - this.columns = that.columns == null ? null : Lists.newArrayList(that.columns); - this.endpointAffinities = that.endpointAffinities == null ? null : Lists.newArrayList(that.endpointAffinities); - this.entries = that.entries == null ? null : Lists.newArrayList(that.entries); this.formatConfig = that.formatConfig; this.formatPlugin = that.formatPlugin; this.fs = that.fs; - this.mappings = that.mappings == null ? null : ArrayListMultimap.create(that.mappings); - this.rowCount = that.rowCount; - this.rowGroupInfos = that.rowGroupInfos == null ? null : Lists.newArrayList(that.rowGroupInfos); this.selectionRoot = that.selectionRoot; - this.columnValueCounts = that.columnValueCounts == null ? null : new HashMap<>(that.columnValueCounts); - this.partitionColTypeMap = that.partitionColTypeMap == null ? null : new HashMap<>(that.partitionColTypeMap); - this.partitionValueMap = that.partitionValueMap == null ? null : new HashMap<>(that.partitionValueMap); - this.fileSet = that.fileSet == null ? null : new HashSet<>(that.fileSet); - this.usedMetadataCache = that.usedMetadataCache; - this.parquetTableMetadata = that.parquetTableMetadata; - this.filter = that.filter; if (selection != null) { this.cacheFileRoot = selection.getCacheFileRoot(); MetadataContext metaContext = selection.getMetaContext(); @@ -263,542 +152,206 @@ public class ParquetGroupScan extends AbstractFileGroupScan { this.cacheFileRoot = that.cacheFileRoot; this.metaContext = that.metaContext; } + this.usedMetadataCache = that.usedMetadataCache; } - /** - * expands the selection's folders if metadata cache is found for the selection root.<br> - * If the selection has already been expanded or no metadata cache was found, does nothing - * - * @param selection actual selection before expansion - * @return new selection after expansion, if no expansion was done returns the input selection - * - * @throws IOException - */ - private FileSelection expandIfNecessary(FileSelection selection) throws IOException { - if (selection.isExpandedFully()) { - return selection; - } - - // use the cacheFileRoot if provided (e.g after partition pruning) - Path metaFilePath = new Path(cacheFileRoot != null ? cacheFileRoot : selectionRoot, Metadata.METADATA_FILENAME); - if (!fs.exists(metaFilePath)) { // no metadata cache - if (selection.isExpandedPartial()) { - logger.error("'{}' metadata file does not exist, but metadata directories cache file is present", metaFilePath); - metaContext.setMetadataCacheCorrupted(true); - } - - return selection; - } - - return expandSelectionFromMetadataCache(selection, metaFilePath); - } - - /** - * For two cases the entries should be initialized with just the selection root instead of the fully expanded list: - * <ul> - * <li> When metadata caching is corrupted (to use correct file selection) - * <li> Metadata caching is correct and used, but pruning was not applicable or was attempted and nothing was pruned - * (to reduce overhead in parquet group scan). - * </ul> - * - * @return true if entries should be initialized with selection root, false otherwise - */ - private boolean checkForInitializingEntriesWithSelectionRoot() { - // TODO: at some point we should examine whether the list of entries is absolutely needed. - return metaContext.isMetadataCacheCorrupted() || (parquetTableMetadata != null && - (metaContext.getPruneStatus() == PruneStatus.NOT_STARTED || metaContext.getPruneStatus() == PruneStatus.NOT_PRUNED)); - } - - public List<ReadEntryWithPath> getEntries() { - return entries; - } - + // getters for serialization / deserialization start @JsonProperty("format") public ParquetFormatConfig getFormatConfig() { - return this.formatConfig; + return formatConfig; } @JsonProperty("storage") public StoragePluginConfig getEngineConfig() { - return this.formatPlugin.getStorageConfig(); + return formatPlugin.getStorageConfig(); } + @JsonProperty public String getSelectionRoot() { return selectionRoot; } - public Set<String> getFileSet() { - return fileSet; + @JsonProperty + public String getCacheFileRoot() { + return cacheFileRoot; } + // getters for serialization / deserialization end - public LogicalExpression getFilter() { - return this.filter; - } - - public void setFilter(LogicalExpression filter) { - this.filter = filter; + @Override + public ParquetRowGroupScan getSpecificScan(int minorFragmentId) { + return new ParquetRowGroupScan(getUserName(), formatPlugin, getReadEntries(minorFragmentId), columns, selectionRoot, filter); } @Override - public boolean hasFiles() { - return true; + public PhysicalOperator getNewWithChildren(List<PhysicalOperator> children) { + Preconditions.checkArgument(children.isEmpty()); + return new ParquetGroupScan(this); } - @JsonIgnore @Override - public Collection<String> getFiles() { - return fileSet; + public GroupScan clone(List<SchemaPath> columns) { + ParquetGroupScan newScan = new ParquetGroupScan(this); + newScan.columns = columns; + return newScan; } - private Set<String> fileSet; - - @JsonIgnore - // only for partition columns : value is unique for each partition - private Map<SchemaPath, MajorType> partitionColTypeMap = Maps.newHashMap(); - - /** - * When reading the very first footer, any column is a potential partition column. So for the first footer, we check - * every column to see if it is single valued, and if so, add it to the list of potential partition columns. For the - * remaining footers, we will not find any new partition columns, but we may discover that what was previously a - * potential partition column now no longer qualifies, so it needs to be removed from the list. - * @return whether column is a potential partition column - */ - private boolean checkForPartitionColumn(ColumnMetadata columnMetadata, boolean first, long rowCount) { - SchemaPath schemaPath = SchemaPath.getCompoundPath(columnMetadata.getName()); - final PrimitiveTypeName primitiveType; - final OriginalType originalType; - int precision = 0; - int scale = 0; - if (this.parquetTableMetadata.hasColumnMetadata()) { - // only ColumnTypeMetadata_v3 stores information about scale and precision - if (parquetTableMetadata instanceof Metadata.ParquetTableMetadata_v3) { - Metadata.ColumnTypeMetadata_v3 columnTypeInfo = ((Metadata.ParquetTableMetadata_v3) parquetTableMetadata) - .getColumnTypeInfo(columnMetadata.getName()); - scale = columnTypeInfo.scale; - precision = columnTypeInfo.precision; - } - primitiveType = this.parquetTableMetadata.getPrimitiveType(columnMetadata.getName()); - originalType = this.parquetTableMetadata.getOriginalType(columnMetadata.getName()); - } else { - primitiveType = columnMetadata.getPrimitiveType(); - originalType = columnMetadata.getOriginalType(); - } - if (first) { - if (hasSingleValue(columnMetadata, rowCount)) { - partitionColTypeMap.put(schemaPath, getType(primitiveType, originalType, scale, precision)); - return true; - } else { - return false; - } - } else { - if (!partitionColTypeMap.keySet().contains(schemaPath)) { - return false; - } else { - if (!hasSingleValue(columnMetadata, rowCount)) { - partitionColTypeMap.remove(schemaPath); - return false; - } - if (!getType(primitiveType, originalType, scale, precision).equals(partitionColTypeMap.get(schemaPath))) { - partitionColTypeMap.remove(schemaPath); - return false; - } - } - } - return true; + @Override + public ParquetGroupScan clone(FileSelection selection) throws IOException { + ParquetGroupScan newScan = new ParquetGroupScan(this, selection); + newScan.modifyFileSelection(selection); + newScan.init(); + return newScan; } - /** - * Builds major type using given {@code OriginalType originalType} or {@code PrimitiveTypeName type}. - * For DECIMAL will be returned major type with scale and precision. - * - * @param type parquet primitive type - * @param originalType parquet original type - * @param scale type scale (used for DECIMAL type) - * @param precision type precision (used for DECIMAL type) - * @return major type - */ - public static MajorType getType(PrimitiveTypeName type, OriginalType originalType, int scale, int precision) { - if (originalType != null) { - switch (originalType) { - case DECIMAL: - return Types.withScaleAndPrecision(MinorType.DECIMAL18, TypeProtos.DataMode.OPTIONAL, scale, precision); - case DATE: - return Types.optional(MinorType.DATE); - case TIME_MILLIS: - return Types.optional(MinorType.TIME); - case TIMESTAMP_MILLIS: - return Types.optional(MinorType.TIMESTAMP); - case UTF8: - return Types.optional(MinorType.VARCHAR); - case UINT_8: - return Types.optional(MinorType.UINT1); - case UINT_16: - return Types.optional(MinorType.UINT2); - case UINT_32: - return Types.optional(MinorType.UINT4); - case UINT_64: - return Types.optional(MinorType.UINT8); - case INT_8: - return Types.optional(MinorType.TINYINT); - case INT_16: - return Types.optional(MinorType.SMALLINT); - case INTERVAL: - return Types.optional(MinorType.INTERVAL); - } - } + @Override + public String toString() { + StringBuilder builder = new StringBuilder(); + builder.append("ParquetGroupScan ["); + builder.append("entries=").append(entries); + builder.append(", selectionRoot=").append(selectionRoot); + builder.append(", numFiles=").append(getEntries().size()); + builder.append(", numRowGroups=").append(rowGroupInfos.size()); + builder.append(", usedMetadataFile=").append(usedMetadataCache); - switch (type) { - case BOOLEAN: - return Types.optional(MinorType.BIT); - case INT32: - return Types.optional(MinorType.INT); - case INT64: - return Types.optional(MinorType.BIGINT); - case FLOAT: - return Types.optional(MinorType.FLOAT4); - case DOUBLE: - return Types.optional(MinorType.FLOAT8); - case BINARY: - case FIXED_LEN_BYTE_ARRAY: - case INT96: - return Types.optional(MinorType.VARBINARY); - default: - // Should never hit this - throw new UnsupportedOperationException("Unsupported type:" + type); + String filterString = getFilterString(); + if (!filterString.isEmpty()) { + builder.append(", filter=").append(filterString); } - } - /** - * Checks that the column chunk has a single value. - * - * @param columnChunkMetaData metadata to check - * @param rowCount rows count in column chunk - * @return true if column has single value - */ - private boolean hasSingleValue(ColumnMetadata columnChunkMetaData, long rowCount) { - // ColumnMetadata will have a non-null value iff the minValue and the maxValue for the - // rowgroup are the same - return (columnChunkMetaData != null) && (columnChunkMetaData.hasSingleValue(rowCount)); - } - - @Override public void modifyFileSelection(FileSelection selection) { - entries.clear(); - fileSet = Sets.newHashSet(); - for (String fileName : selection.getFiles()) { - entries.add(new ReadEntryWithPath(fileName)); - fileSet.add(fileName); + if (usedMetadataCache) { + // For EXPLAIN, remove the URI prefix from cacheFileRoot. If cacheFileRoot is null, we + // would have read the cache file from selectionRoot + String cacheFileRootString = (cacheFileRoot == null) ? + Path.getPathWithoutSchemeAndAuthority(new Path(selectionRoot)).toString() : + Path.getPathWithoutSchemeAndAuthority(new Path(cacheFileRoot)).toString(); + builder.append(", cacheFileRoot=").append(cacheFileRootString); } - List<RowGroupInfo> newRowGroupList = Lists.newArrayList(); - for (RowGroupInfo rowGroupInfo : rowGroupInfos) { - if (fileSet.contains(rowGroupInfo.getPath())) { - newRowGroupList.add(rowGroupInfo); - } - } - this.rowGroupInfos = newRowGroupList; - } + builder.append(", columns=").append(columns); + builder.append("]"); - public MajorType getTypeForColumn(SchemaPath schemaPath) { - return partitionColTypeMap.get(schemaPath); + return builder.toString(); } - // Map from file names to maps of column name to partition value mappings - private Map<String, Map<SchemaPath, Object>> partitionValueMap = Maps.newHashMap(); - - public void populatePruningVector(ValueVector v, int index, SchemaPath column, String file) { - String f = Path.getPathWithoutSchemeAndAuthority(new Path(file)).toString(); - MajorType majorType = getTypeForColumn(column); - MinorType type = majorType.getMinorType(); - switch (type) { - case BIT: { - NullableBitVector bitVector = (NullableBitVector) v; - Boolean value = (Boolean) partitionValueMap.get(f).get(column); - if (value == null) { - bitVector.getMutator().setNull(index); - } else { - bitVector.getMutator().setSafe(index, value ? 1 : 0); - } - return; - } - case INT: { - NullableIntVector intVector = (NullableIntVector) v; - Integer value = (Integer) partitionValueMap.get(f).get(column); - if (value == null) { - intVector.getMutator().setNull(index); - } else { - intVector.getMutator().setSafe(index, value); - } - return; - } - case SMALLINT: { - NullableSmallIntVector smallIntVector = (NullableSmallIntVector) v; - Integer value = (Integer) partitionValueMap.get(f).get(column); - if (value == null) { - smallIntVector.getMutator().setNull(index); - } else { - smallIntVector.getMutator().setSafe(index, value.shortValue()); - } - return; - } - case TINYINT: { - NullableTinyIntVector tinyIntVector = (NullableTinyIntVector) v; - Integer value = (Integer) partitionValueMap.get(f).get(column); - if (value == null) { - tinyIntVector.getMutator().setNull(index); - } else { - tinyIntVector.getMutator().setSafe(index, value.byteValue()); - } - return; - } - case UINT1: { - NullableUInt1Vector intVector = (NullableUInt1Vector) v; - Integer value = (Integer) partitionValueMap.get(f).get(column); - if (value == null) { - intVector.getMutator().setNull(index); - } else { - intVector.getMutator().setSafe(index, value.byteValue()); - } - return; - } - case UINT2: { - NullableUInt2Vector intVector = (NullableUInt2Vector) v; - Integer value = (Integer) partitionValueMap.get(f).get(column); - if (value == null) { - intVector.getMutator().setNull(index); - } else { - intVector.getMutator().setSafe(index, (char) value.shortValue()); - } - return; - } - case UINT4: { - NullableUInt4Vector intVector = (NullableUInt4Vector) v; - Integer value = (Integer) partitionValueMap.get(f).get(column); - if (value == null) { - intVector.getMutator().setNull(index); - } else { - intVector.getMutator().setSafe(index, value); - } - return; - } - case BIGINT: { - NullableBigIntVector bigIntVector = (NullableBigIntVector) v; - Long value = (Long) partitionValueMap.get(f).get(column); - if (value == null) { - bigIntVector.getMutator().setNull(index); - } else { - bigIntVector.getMutator().setSafe(index, value); - } - return; - } - case FLOAT4: { - NullableFloat4Vector float4Vector = (NullableFloat4Vector) v; - Float value = (Float) partitionValueMap.get(f).get(column); - if (value == null) { - float4Vector.getMutator().setNull(index); - } else { - float4Vector.getMutator().setSafe(index, value); - } - return; - } - case FLOAT8: { - NullableFloat8Vector float8Vector = (NullableFloat8Vector) v; - Double value = (Double) partitionValueMap.get(f).get(column); - if (value == null) { - float8Vector.getMutator().setNull(index); - } else { - float8Vector.getMutator().setSafe(index, value); - } - return; - } - case VARBINARY: { - NullableVarBinaryVector varBinaryVector = (NullableVarBinaryVector) v; - Object s = partitionValueMap.get(f).get(column); - byte[] bytes; - if (s == null) { - varBinaryVector.getMutator().setNull(index); - return; - } else { - bytes = getBytes(type, s); - } - varBinaryVector.getMutator().setSafe(index, bytes, 0, bytes.length); - return; - } - case DECIMAL18: { - NullableDecimal18Vector decimalVector = (NullableDecimal18Vector) v; - Object s = partitionValueMap.get(f).get(column); - byte[] bytes; - if (s == null) { - decimalVector.getMutator().setNull(index); - return; - } else if (s instanceof Integer) { - long value = DecimalUtility.getBigDecimalFromPrimitiveTypes( - (Integer) s, - majorType.getScale(), - majorType.getPrecision()).longValue(); - decimalVector.getMutator().setSafe(index, value); - return; - } else if (s instanceof Long) { - long value = DecimalUtility.getBigDecimalFromPrimitiveTypes( - (Long) s, - majorType.getScale(), - majorType.getPrecision()).longValue(); - decimalVector.getMutator().setSafe(index, value); - return; - } else { - bytes = getBytes(type, s); - } - long value = DecimalUtility.getBigDecimalFromByteArray(bytes, 0, bytes.length, majorType.getScale()).longValue(); - decimalVector.getMutator().setSafe(index, value); - return; + // overridden protected methods block start + @Override + protected void initInternal() throws IOException { + FileSystem processUserFileSystem = ImpersonationUtil.createFileSystem(ImpersonationUtil.getProcessUserName(), fs.getConf()); + Path metaPath = null; + if (entries.size() == 1 && parquetTableMetadata == null) { + Path p = Path.getPathWithoutSchemeAndAuthority(new Path(entries.get(0).getPath())); + if (fs.isDirectory(p)) { + // Using the metadata file makes sense when querying a directory; otherwise + // if querying a single file we can look up the metadata directly from the file + metaPath = new Path(p, Metadata.METADATA_FILENAME); } - case DATE: { - NullableDateVector dateVector = (NullableDateVector) v; - Integer value = (Integer) partitionValueMap.get(f).get(column); - if (value == null) { - dateVector.getMutator().setNull(index); - } else { - dateVector.getMutator().setSafe(index, value * (long) DateTimeConstants.MILLIS_PER_DAY); + if (!metaContext.isMetadataCacheCorrupted() && metaPath != null && fs.exists(metaPath)) { + parquetTableMetadata = Metadata.readBlockMeta(processUserFileSystem, metaPath, metaContext, formatConfig); + if (parquetTableMetadata != null) { + usedMetadataCache = true; } - return; } - case TIME: { - NullableTimeVector timeVector = (NullableTimeVector) v; - Integer value = (Integer) partitionValueMap.get(f).get(column); - if (value == null) { - timeVector.getMutator().setNull(index); - } else { - timeVector.getMutator().setSafe(index, value); - } - return; + if (!usedMetadataCache) { + parquetTableMetadata = Metadata.getParquetTableMetadata(processUserFileSystem, p.toString(), formatConfig); } - case TIMESTAMP: { - NullableTimeStampVector timeStampVector = (NullableTimeStampVector) v; - Long value = (Long) partitionValueMap.get(f).get(column); - if (value == null) { - timeStampVector.getMutator().setNull(index); - } else { - timeStampVector.getMutator().setSafe(index, value); + } else { + Path p = Path.getPathWithoutSchemeAndAuthority(new Path(selectionRoot)); + metaPath = new Path(p, Metadata.METADATA_FILENAME); + if (!metaContext.isMetadataCacheCorrupted() && fs.isDirectory(new Path(selectionRoot)) + && fs.exists(metaPath)) { + if (parquetTableMetadata == null) { + parquetTableMetadata = Metadata.readBlockMeta(processUserFileSystem, metaPath, metaContext, formatConfig); } - return; - } - case VARCHAR: { - NullableVarCharVector varCharVector = (NullableVarCharVector) v; - Object s = partitionValueMap.get(f).get(column); - byte[] bytes; - if (s == null) { - varCharVector.getMutator().setNull(index); - return; - } else { - bytes = getBytes(type, s); + if (parquetTableMetadata != null) { + usedMetadataCache = true; + if (fileSet != null) { + parquetTableMetadata = removeUnneededRowGroups(parquetTableMetadata); + } } - varCharVector.getMutator().setSafe(index, bytes, 0, bytes.length); - return; } - case INTERVAL: { - NullableIntervalVector intervalVector = (NullableIntervalVector) v; - Object s = partitionValueMap.get(f).get(column); - byte[] bytes; - if (s == null) { - intervalVector.getMutator().setNull(index); - return; - } else { - bytes = getBytes(type, s); + if (!usedMetadataCache) { + final List<FileStatus> fileStatuses = new ArrayList<>(); + for (ReadEntryWithPath entry : entries) { + fileStatuses.addAll( + DrillFileSystemUtil.listFiles(fs, Path.getPathWithoutSchemeAndAuthority(new Path(entry.getPath())), true)); } - intervalVector.getMutator().setSafe(index, 1, - ParquetReaderUtility.getIntFromLEBytes(bytes, 0), - ParquetReaderUtility.getIntFromLEBytes(bytes, 4), - ParquetReaderUtility.getIntFromLEBytes(bytes, 8)); - return; + + Map<FileStatus, FileSystem> statusMap = fileStatuses.stream() + .collect( + Collectors.toMap( + Function.identity(), + s -> processUserFileSystem, + (oldFs, newFs) -> newFs, + LinkedHashMap::new)); + + parquetTableMetadata = Metadata.getParquetTableMetadata(statusMap, formatConfig); } - default: - throw new UnsupportedOperationException("Unsupported type: " + type); } } - /** - * Returns the sequence of bytes received from {@code Object source}. - * - * @param type the column type - * @param source the source of the bytes sequence - * @return bytes sequence obtained from {@code Object source} - */ - private byte[] getBytes(MinorType type, Object source) { - byte[] bytes; - if (source instanceof Binary) { - bytes = ((Binary) source).getBytes(); - } else if (source instanceof byte[]) { - bytes = (byte[]) source; - } else { - throw new UnsupportedOperationException("Unable to create column data for type: " + type); - } - return bytes; + @Override + protected AbstractParquetGroupScan cloneWithFileSelection(Collection<String> filePaths) throws IOException { + FileSelection newSelection = new FileSelection(null, new ArrayList<>(filePaths), getSelectionRoot(), cacheFileRoot, false); + return clone(newSelection); } - public static class RowGroupInfo extends ReadEntryFromHDFS implements CompleteWork, FileWork { - - private EndpointByteMap byteMap; - private int rowGroupIndex; - private List<? extends ColumnMetadata> columns; - private long rowCount; // rowCount = -1 indicates to include all rows. - private long numRecordsToRead; - - @JsonCreator - public RowGroupInfo(@JsonProperty("path") String path, @JsonProperty("start") long start, - @JsonProperty("length") long length, @JsonProperty("rowGroupIndex") int rowGroupIndex, long rowCount) { - super(path, start, length); - this.rowGroupIndex = rowGroupIndex; - this.rowCount = rowCount; - this.numRecordsToRead = rowCount; - } - - public RowGroupReadEntry getRowGroupReadEntry() { - return new RowGroupReadEntry(this.getPath(), this.getStart(), this.getLength(), - this.rowGroupIndex, this.getNumRecordsToRead()); - } - - public int getRowGroupIndex() { - return this.rowGroupIndex; - } - - @Override - public int compareTo(CompleteWork o) { - return Long.compare(getTotalBytes(), o.getTotalBytes()); - } - - @Override - public long getTotalBytes() { - return this.getLength(); - } - - @Override - public EndpointByteMap getByteMap() { - return byteMap; - } + @Override + protected Collection<DrillbitEndpoint> getDrillbits() { + return formatPlugin.getContext().getBits(); + } - public long getNumRecordsToRead() { - return numRecordsToRead; - } + @Override + protected boolean supportsFileImplicitColumns() { + return selectionRoot != null; + } - public void setNumRecordsToRead(long numRecords) { - numRecordsToRead = numRecords; - } + @Override + protected List<String> getPartitionValues(RowGroupInfo rowGroupInfo) { + return ColumnExplorer.listPartitionValues(rowGroupInfo.getPath(), selectionRoot); + } + // overridden protected methods block end - public void setEndpointByteMap(EndpointByteMap byteMap) { - this.byteMap = byteMap; + // private methods block start + /** + * Expands the selection's folders if metadata cache is found for the selection root.<br> + * If the selection has already been expanded or no metadata cache was found, does nothing + * + * @param selection actual selection before expansion + * @return new selection after expansion, if no expansion was done returns the input selection + */ + private FileSelection expandIfNecessary(FileSelection selection) throws IOException { + if (selection.isExpandedFully()) { + return selection; } - public long getRowCount() { - return rowCount; - } + // use the cacheFileRoot if provided (e.g after partition pruning) + Path metaFilePath = new Path(cacheFileRoot != null ? cacheFileRoot : selectionRoot, Metadata.METADATA_FILENAME); + if (!fs.exists(metaFilePath)) { // no metadata cache + if (selection.isExpandedPartial()) { + logger.error("'{}' metadata file does not exist, but metadata directories cache file is present", metaFilePath); + metaContext.setMetadataCacheCorrupted(true); + } - public List<? extends ColumnMetadata> getColumns() { - return columns; + return selection; } - public void setColumns(List<? extends ColumnMetadata> columns) { - this.columns = columns; - } + return expandSelectionFromMetadataCache(selection, metaFilePath); + } + /** + * For two cases the entries should be initialized with just the selection root instead of the fully expanded list: + * <ul> + * <li> When metadata caching is corrupted (to use correct file selection) + * <li> Metadata caching is correct and used, but pruning was not applicable or was attempted and nothing was pruned + * (to reduce overhead in parquet group scan). + * </ul> + * + * @return true if entries should be initialized with selection root, false otherwise + */ + private boolean checkForInitializingEntriesWithSelectionRoot() { + // TODO: at some point we should examine whether the list of entries is absolutely needed. + return metaContext.isMetadataCacheCorrupted() || (parquetTableMetadata != null && + (metaContext.getPruneStatus() == PruneStatus.NOT_STARTED || metaContext.getPruneStatus() == PruneStatus.NOT_PRUNED)); } /** @@ -810,17 +363,16 @@ public class ParquetGroupScan extends AbstractFileGroupScan { * @param metaFilePath metadata cache file path * @return file selection read from cache * - * @throws IOException * @throws UserException when the updated selection is empty, this happens if the user selects an empty folder. */ - private FileSelection - expandSelectionFromMetadataCache(FileSelection selection, Path metaFilePath) throws IOException { + private FileSelection expandSelectionFromMetadataCache(FileSelection selection, Path metaFilePath) throws IOException { // get the metadata for the root directory by reading the metadata file // parquetTableMetadata contains the metadata for all files in the selection root folder, but we need to make sure // we only select the files that are part of selection (by setting fileSet appropriately) // get (and set internal field) the metadata for the directory by reading the metadata file - parquetTableMetadata = Metadata.readBlockMeta(fs, metaFilePath, metaContext, formatConfig); + FileSystem processUserFileSystem = ImpersonationUtil.createFileSystem(ImpersonationUtil.getProcessUserName(), fs.getConf()); + parquetTableMetadata = Metadata.readBlockMeta(processUserFileSystem, metaFilePath, metaContext, formatConfig); if (ignoreExpandingSelection(parquetTableMetadata)) { return selection; } @@ -837,7 +389,7 @@ public class ParquetGroupScan extends AbstractFileGroupScan { final Path first = fileStatuses.get(0).getPath(); if (fileStatuses.size() == 1 && selection.getSelectionRoot().equals(first.toString())) { // we are selecting all files from selection root. Expand the file list from the cache - for (Metadata.ParquetFileMetadata file : parquetTableMetadata.getFiles()) { + for (ParquetFileMetadata file : parquetTableMetadata.getFiles()) { fileSet.add(file.getPath()); } @@ -850,7 +402,7 @@ public class ParquetGroupScan extends AbstractFileGroupScan { // list of files from the metadata cache file that is present in the cacheFileRoot directory and populate // the fileSet. However, this is *not* the final list of files that will be scanned in execution since the // second phase of partition pruning will apply on the files and modify the file selection appropriately. - for (Metadata.ParquetFileMetadata file : this.parquetTableMetadata.getFiles()) { + for (ParquetFileMetadata file : this.parquetTableMetadata.getFiles()) { fileSet.add(file.getPath()); } } @@ -861,11 +413,11 @@ public class ParquetGroupScan extends AbstractFileGroupScan { if (status.isDirectory()) { //TODO [DRILL-4496] read the metadata cache files in parallel final Path metaPath = new Path(cacheFileRoot, Metadata.METADATA_FILENAME); - final Metadata.ParquetTableMetadataBase metadata = Metadata.readBlockMeta(fs, metaPath, metaContext, formatConfig); + final ParquetTableMetadataBase metadata = Metadata.readBlockMeta(processUserFileSystem, metaPath, metaContext, formatConfig); if (ignoreExpandingSelection(metadata)) { return selection; } - for (Metadata.ParquetFileMetadata file : metadata.getFiles()) { + for (ParquetFileMetadata file : metadata.getFiles()) { fileSet.add(file.getPath()); } } else { @@ -881,7 +433,7 @@ public class ParquetGroupScan extends AbstractFileGroupScan { return null; } - List<String> fileNames = Lists.newArrayList(fileSet); + List<String> fileNames = new ArrayList<>(fileSet); // when creating the file selection, set the selection root without the URI prefix // The reason is that the file names above have been created in the form @@ -902,141 +454,8 @@ public class ParquetGroupScan extends AbstractFileGroupScan { return newSelection; } - private void init() throws IOException { - Path metaPath = null; - if (entries.size() == 1 && parquetTableMetadata == null) { - Path p = Path.getPathWithoutSchemeAndAuthority(new Path(entries.get(0).getPath())); - if (fs.isDirectory(p)) { - // Using the metadata file makes sense when querying a directory; otherwise - // if querying a single file we can look up the metadata directly from the file - metaPath = new Path(p, Metadata.METADATA_FILENAME); - } - if (!metaContext.isMetadataCacheCorrupted() && metaPath != null && fs.exists(metaPath)) { - parquetTableMetadata = Metadata.readBlockMeta(fs, metaPath, metaContext, formatConfig); - if (parquetTableMetadata != null) { - usedMetadataCache = true; - } - } - if (!usedMetadataCache) { - parquetTableMetadata = Metadata.getParquetTableMetadata(fs, p.toString(), formatConfig); - } - } else { - Path p = Path.getPathWithoutSchemeAndAuthority(new Path(selectionRoot)); - metaPath = new Path(p, Metadata.METADATA_FILENAME); - if (!metaContext.isMetadataCacheCorrupted() && fs.isDirectory(new Path(selectionRoot)) - && fs.exists(metaPath)) { - if (parquetTableMetadata == null) { - parquetTableMetadata = Metadata.readBlockMeta(fs, metaPath, metaContext, formatConfig); - } - if (parquetTableMetadata != null) { - usedMetadataCache = true; - if (fileSet != null) { - parquetTableMetadata = removeUnneededRowGroups(parquetTableMetadata); - } - } - } - if (!usedMetadataCache) { - final List<FileStatus> fileStatuses = Lists.newArrayList(); - for (ReadEntryWithPath entry : entries) { - fileStatuses.addAll(DrillFileSystemUtil.listFiles(fs, Path.getPathWithoutSchemeAndAuthority(new Path(entry.getPath())), true)); - } - parquetTableMetadata = Metadata.getParquetTableMetadata(fs, fileStatuses, formatConfig); - } - } - - if (fileSet == null) { - fileSet = Sets.newHashSet(); - for (ParquetFileMetadata file : parquetTableMetadata.getFiles()) { - fileSet.add(file.getPath()); - } - } - - Map<String, DrillbitEndpoint> hostEndpointMap = Maps.newHashMap(); - - for (DrillbitEndpoint endpoint : formatPlugin.getContext().getBits()) { - hostEndpointMap.put(endpoint.getAddress(), endpoint); - } - - rowGroupInfos = Lists.newArrayList(); - for (ParquetFileMetadata file : parquetTableMetadata.getFiles()) { - int rgIndex = 0; - for (RowGroupMetadata rg : file.getRowGroups()) { - RowGroupInfo rowGroupInfo = - new RowGroupInfo(file.getPath(), rg.getStart(), rg.getLength(), rgIndex, rg.getRowCount()); - EndpointByteMap endpointByteMap = new EndpointByteMapImpl(); - for (String host : rg.getHostAffinity().keySet()) { - if (hostEndpointMap.containsKey(host)) { - endpointByteMap - .add(hostEndpointMap.get(host), (long) (rg.getHostAffinity().get(host) * rg.getLength())); - } - } - rowGroupInfo.setEndpointByteMap(endpointByteMap); - rowGroupInfo.setColumns(rg.getColumns()); - rgIndex++; - rowGroupInfos.add(rowGroupInfo); - } - } - - this.endpointAffinities = AffinityCreator.getAffinityMap(rowGroupInfos); - updatePartitionColTypeMap(); - } - - private void updatePartitionColTypeMap() { - columnValueCounts = Maps.newHashMap(); - this.rowCount = 0; - boolean first = true; - for (RowGroupInfo rowGroup : this.rowGroupInfos) { - long rowCount = rowGroup.getRowCount(); - for (ColumnMetadata column : rowGroup.getColumns()) { - SchemaPath schemaPath = SchemaPath.getCompoundPath(column.getName()); - Long previousCount = columnValueCounts.get(schemaPath); - if (previousCount != null) { - if (previousCount != GroupScan.NO_COLUMN_STATS) { - if (column.getNulls() != null) { - Long newCount = rowCount - column.getNulls(); - columnValueCounts.put(schemaPath, columnValueCounts.get(schemaPath) + newCount); - } - } - } else { - if (column.getNulls() != null) { - Long newCount = rowCount - column.getNulls(); - columnValueCounts.put(schemaPath, newCount); - } else { - columnValueCounts.put(schemaPath, GroupScan.NO_COLUMN_STATS); - } - } - boolean partitionColumn = checkForPartitionColumn(column, first, rowCount); - if (partitionColumn) { - Map<SchemaPath, Object> map = partitionValueMap.get(rowGroup.getPath()); - if (map == null) { - map = Maps.newHashMap(); - partitionValueMap.put(rowGroup.getPath(), map); - } - Object value = map.get(schemaPath); - Object currentValue = column.getMaxValue(); - if (value != null) { - if (value != currentValue) { - partitionColTypeMap.remove(schemaPath); - } - } else { - // the value of a column with primitive type can not be null, - // so checks that there are really null value and puts it to the map - if (rowCount == column.getNulls()) { - map.put(schemaPath, null); - } else { - map.put(schemaPath, currentValue); - } - } - } else { - partitionColTypeMap.remove(schemaPath); - } - } - this.rowCount += rowGroup.getRowCount(); - first = false; - } - } private ParquetTableMetadataBase removeUnneededRowGroups(ParquetTableMetadataBase parquetTableMetadata) { - List<ParquetFileMetadata> newFileMetadataList = Lists.newArrayList(); + List<ParquetFileMetadata> newFileMetadataList = new ArrayList<>(); for (ParquetFileMetadata file : parquetTableMetadata.getFiles()) { if (fileSet.contains(file.getPath())) { newFileMetadataList.add(file); @@ -1049,284 +468,6 @@ public class ParquetGroupScan extends AbstractFileGroupScan { } /** - * Calculates the affinity each endpoint has for this scan, by adding up the affinity each endpoint has for each - * rowGroup - * - * @return a list of EndpointAffinity objects - */ - @Override - public List<EndpointAffinity> getOperatorAffinity() { - return this.endpointAffinities; - } - - @Override - public void applyAssignments(List<DrillbitEndpoint> incomingEndpoints) throws PhysicalOperatorSetupException { - - this.mappings = AssignmentCreator.getMappings(incomingEndpoints, rowGroupInfos); - } - - @Override public ParquetRowGroupScan getSpecificScan(int minorFragmentId) { - assert minorFragmentId < mappings.size() : String - .format("Mappings length [%d] should be longer than minor fragment id [%d] but it isn't.", - mappings.size(), minorFragmentId); - - List<RowGroupInfo> rowGroupsForMinor = mappings.get(minorFragmentId); - - Preconditions.checkArgument(!rowGroupsForMinor.isEmpty(), - String.format("MinorFragmentId %d has no read entries assigned", minorFragmentId)); - - return new ParquetRowGroupScan( - getUserName(), formatPlugin, convertToReadEntries(rowGroupsForMinor), columns, selectionRoot, filter); - } - - private List<RowGroupReadEntry> convertToReadEntries(List<RowGroupInfo> rowGroups) { - List<RowGroupReadEntry> entries = Lists.newArrayList(); - for (RowGroupInfo rgi : rowGroups) { - RowGroupReadEntry entry = new RowGroupReadEntry(rgi.getPath(), rgi.getStart(), rgi.getLength(), rgi.getRowGroupIndex(), rgi.getNumRecordsToRead()); - entries.add(entry); - } - return entries; - } - - @Override - public int getMaxParallelizationWidth() { - return rowGroupInfos.size(); - } - - public List<SchemaPath> getColumns() { - return columns; - } - - @Override - public ScanStats getScanStats() { - int columnCount = columns == null ? 20 : columns.size(); - return new ScanStats(GroupScanProperty.EXACT_ROW_COUNT, rowCount, 1, rowCount * columnCount); - } - - @Override - @JsonIgnore - public PhysicalOperator getNewWithChildren(List<PhysicalOperator> children) { - Preconditions.checkArgument(children.isEmpty()); - return new ParquetGroupScan(this); - } - - @Override - public String getDigest() { - return toString(); - } - - @Override - public String toString() { - String cacheFileString = ""; - if (usedMetadataCache) { - // For EXPLAIN, remove the URI prefix from cacheFileRoot. If cacheFileRoot is null, we - // would have read the cache file from selectionRoot - String str = (cacheFileRoot == null) ? - Path.getPathWithoutSchemeAndAuthority(new Path(selectionRoot)).toString() : - Path.getPathWithoutSchemeAndAuthority(new Path(cacheFileRoot)).toString(); - cacheFileString = ", cacheFileRoot=" + str; - } - final String filterStr = filter == null || filter.equals(ValueExpressions.BooleanExpression.TRUE) ? "" : ", filter=" + ExpressionStringBuilder.toString(this.filter); - - return "ParquetGroupScan [entries=" + entries - + ", selectionRoot=" + selectionRoot - + ", numFiles=" + getEntries().size() - + ", numRowGroups=" + rowGroupInfos.size() - + ", usedMetadataFile=" + usedMetadataCache - + filterStr - + cacheFileString - + ", columns=" + columns - + "]"; - } - - @Override - public GroupScan clone(List<SchemaPath> columns) { - ParquetGroupScan newScan = new ParquetGroupScan(this); - newScan.columns = columns; - return newScan; - } - - // Based on maxRecords to read for the scan, - // figure out how many rowGroups to read and update number of records to read for each of them. - // Returns total number of rowGroups to read. - private int updateRowGroupInfo(long maxRecords) { - long count = 0; - int index = 0; - for (RowGroupInfo rowGroupInfo : rowGroupInfos) { - long rowCount = rowGroupInfo.getRowCount(); - if (count + rowCount <= maxRecords) { - count += rowCount; - rowGroupInfo.setNumRecordsToRead(rowCount); - index++; - continue; - } else if (count < maxRecords) { - rowGroupInfo.setNumRecordsToRead(maxRecords - count); - index++; - } - break; - } - - return index; - } - - @Override - public ParquetGroupScan clone(FileSelection selection) throws IOException { - ParquetGroupScan newScan = new ParquetGroupScan(this, selection); - newScan.modifyFileSelection(selection); - newScan.init(); - return newScan; - } - - public ParquetGroupScan clone(FileSelection selection, long maxRecords) throws IOException { - ParquetGroupScan newScan = clone(selection); - newScan.updateRowGroupInfo(maxRecords); - return newScan; - } - - @Override - public boolean supportsLimitPushdown() { - return true; - } - - @Override - public GroupScan applyLimit(int maxRecords) { - Preconditions.checkArgument(rowGroupInfos.size() >= 0); - - maxRecords = Math.max(maxRecords, 1); // Make sure it request at least 1 row -> 1 rowGroup. - // further optimization : minimize # of files chosen, or the affinity of files chosen. - - // Calculate number of rowGroups to read based on maxRecords and update - // number of records to read for each of those rowGroups. - int index = updateRowGroupInfo(maxRecords); - - Set<String> fileNames = Sets.newHashSet(); // HashSet keeps a fileName unique. - for (RowGroupInfo rowGroupInfo : rowGroupInfos.subList(0, index)) { - fileNames.add(rowGroupInfo.getPath()); - } - - // If there is no change in fileSet, no need to create new groupScan. - if (fileNames.size() == fileSet.size() ) { - // There is no reduction of rowGroups. Return the original groupScan. - logger.debug("applyLimit() does not apply!"); - return null; - } - - try { - FileSelection newSelection = new FileSelection(null, Lists.newArrayList(fileNames), getSelectionRoot(), cacheFileRoot, false); - logger.debug("applyLimit() reduce parquet file # from {} to {}", fileSet.size(), fileNames.size()); - return this.clone(newSelection, maxRecords); - } catch (IOException e) { - logger.warn("Could not apply rowcount based prune due to Exception : {}", e); - return null; - } - } - - @Override - @JsonIgnore - public boolean canPushdownProjects(List<SchemaPath> columns) { - return true; - } - - /** - * Return column value count for the specified column. If does not contain such column, return 0. - */ - @Override - public long getColumnValueCount(SchemaPath column) { - return columnValueCounts.containsKey(column) ? columnValueCounts.get(column) : 0; - } - - @Override - public List<SchemaPath> getPartitionColumns() { - return new ArrayList<>(partitionColTypeMap.keySet()); - } - - public GroupScan applyFilter(LogicalExpression filterExpr, UdfUtilities udfUtilities, - FunctionImplementationRegistry functionImplementationRegistry, OptionManager optionManager) { - if (rowGroupInfos.size() == 1 || - ! (parquetTableMetadata.isRowGroupPrunable()) || - rowGroupInfos.size() > optionManager.getOption(PlannerSettings.PARQUET_ROWGROUP_FILTER_PUSHDOWN_PLANNING_THRESHOLD) - ) { - // Stop pruning for 3 cases: - // - 1 single parquet file, - // - metadata does not have proper format to support row group level filter pruning, - // - # of row groups is beyond PARQUET_ROWGROUP_FILTER_PUSHDOWN_PLANNING_THRESHOLD. - return null; - } - - final Set<SchemaPath> schemaPathsInExpr = filterExpr.accept(new ParquetRGFilterEvaluator.FieldReferenceFinder(), null); - - final List<RowGroupInfo> qualifiedRGs = new ArrayList<>(rowGroupInfos.size()); - Set<String> qualifiedFileNames = Sets.newHashSet(); // HashSet keeps a fileName unique. - - ParquetFilterPredicate filterPredicate = null; - - for (RowGroupInfo rowGroup : rowGroupInfos) { - final ColumnExplorer columnExplorer = new ColumnExplorer(optionManager, this.columns); - Map<String, String> implicitColValues = columnExplorer.populateImplicitColumns(rowGroup.getPath(), selectionRoot); - - ParquetMetaStatCollector statCollector = new ParquetMetaStatCollector( - parquetTableMetadata, - rowGroup.getColumns(), - implicitColValues); - - Map<SchemaPath, ColumnStatistics> columnStatisticsMap = statCollector.collectColStat(schemaPathsInExpr); - - if (filterPredicate == null) { - ErrorCollector errorCollector = new ErrorCollectorImpl(); - LogicalExpression materializedFilter = ExpressionTreeMaterializer.materializeFilterExpr( - filterExpr, columnStatisticsMap, errorCollector, functionImplementationRegistry); - - if (errorCollector.hasErrors()) { - logger.error("{} error(s) encountered when materialize filter expression : {}", - errorCollector.getErrorCount(), errorCollector.toErrorString()); - return null; - } - // logger.debug("materializedFilter : {}", ExpressionStringBuilder.toString(materializedFilter)); - - Set<LogicalExpression> constantBoundaries = ConstantExpressionIdentifier.getConstantExpressionSet(materializedFilter); - filterPredicate = (ParquetFilterPredicate) ParquetFilterBuilder.buildParquetFilterPredicate( - materializedFilter, constantBoundaries, udfUtilities); - - if (filterPredicate == null) { - return null; - } - } - - if (ParquetRGFilterEvaluator.canDrop(filterPredicate, columnStatisticsMap, rowGroup.getRowCount())) { - continue; - } - - qualifiedRGs.add(rowGroup); - qualifiedFileNames.add(rowGroup.getPath()); // TODO : optimize when 1 file contains m row groups. - } - - - if (qualifiedRGs.size() == rowGroupInfos.size() ) { - // There is no reduction of rowGroups. Return the original groupScan. - logger.debug("applyFilter does not have any pruning!"); - return null; - } else if (qualifiedFileNames.size() == 0) { - logger.warn("All rowgroups have been filtered out. Add back one to get schema from scannner"); - RowGroupInfo rg = rowGroupInfos.iterator().next(); - qualifiedFileNames.add(rg.getPath()); - qualifiedRGs.add(rg); - } - - try { - FileSelection newSelection = new FileSelection(null, Lists.newArrayList(qualifiedFileNames), getSelectionRoot(), cacheFileRoot, false); - logger.info("applyFilter {} reduce parquet rowgroup # from {} to {}", ExpressionStringBuilder.toString(filterExpr), rowGroupInfos.size(), qualifiedRGs.size()); - ParquetGroupScan clonegroupscan = this.clone(newSelection); - clonegroupscan.rowGroupInfos = qualifiedRGs; - clonegroupscan.updatePartitionColTypeMap(); - return clonegroupscan; - - } catch (IOException e) { - logger.warn("Could not apply filter prune due to Exception : {}", e); - return null; - } - } - - /** * If metadata is corrupted, ignore expanding selection and reset parquetTableMetadata and fileSet fields * * @param metadata parquet table metadata @@ -1341,5 +482,6 @@ public class ParquetGroupScan extends AbstractFileGroupScan { } return false; } + // private methods block end }
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetGroupScanStatistics.java ---------------------------------------------------------------------- diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetGroupScanStatistics.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetGroupScanStatistics.java new file mode 100644 index 0000000..5a3e0c2 --- /dev/null +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetGroupScanStatistics.java @@ -0,0 +1,215 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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 org.apache.drill.exec.store.parquet; + +import org.apache.drill.common.expression.SchemaPath; +import org.apache.drill.common.types.TypeProtos; +import org.apache.drill.exec.physical.base.GroupScan; +import org.apache.parquet.schema.OriginalType; +import org.apache.parquet.schema.PrimitiveType; + +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ColumnMetadata; +import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ParquetTableMetadataBase; +import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ColumnTypeMetadata_v3; +import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ParquetTableMetadata_v3; + +/** + * Holds common statistics about data in parquet group scan, + * including information about total row count, columns counts, partition columns. + */ +public class ParquetGroupScanStatistics { + + // map from file names to maps of column name to partition value mappings + private Map<String, Map<SchemaPath, Object>> partitionValueMap; + // only for partition columns : value is unique for each partition + private Map<SchemaPath, TypeProtos.MajorType> partitionColTypeMap; + // total number of non-null value for each column in parquet files + private Map<SchemaPath, Long> columnValueCounts; + // total number of rows (obtained from parquet footer) + private long rowCount; + + + public ParquetGroupScanStatistics(List<RowGroupInfo> rowGroupInfos, ParquetTableMetadataBase parquetTableMetadata) { + collect(rowGroupInfos, parquetTableMetadata); + } + + public ParquetGroupScanStatistics(ParquetGroupScanStatistics that) { + this.partitionValueMap = new HashMap<>(that.partitionValueMap); + this.partitionColTypeMap = new HashMap<>(that.partitionColTypeMap); + this.columnValueCounts = new HashMap<>(that.columnValueCounts); + this.rowCount = that.rowCount; + } + + public long getColumnValueCount(SchemaPath column) { + return columnValueCounts.containsKey(column) ? columnValueCounts.get(column) : 0; + } + + public List<SchemaPath> getPartitionColumns() { + return new ArrayList<>(partitionColTypeMap.keySet()); + } + + public TypeProtos.MajorType getTypeForColumn(SchemaPath schemaPath) { + return partitionColTypeMap.get(schemaPath); + } + + public long getRowCount() { + return rowCount; + } + + public Object getPartitionValue(String path, SchemaPath column) { + return partitionValueMap.get(path).get(column); + } + + public void collect(List<RowGroupInfo> rowGroupInfos, ParquetTableMetadataBase parquetTableMetadata) { + resetHolders(); + boolean first = true; + for (RowGroupInfo rowGroup : rowGroupInfos) { + long rowCount = rowGroup.getRowCount(); + for (ColumnMetadata column : rowGroup.getColumns()) { + SchemaPath schemaPath = SchemaPath.getCompoundPath(column.getName()); + Long previousCount = columnValueCounts.get(schemaPath); + if (previousCount != null) { + if (previousCount != GroupScan.NO_COLUMN_STATS && column.getNulls() != null) { + Long newCount = rowCount - column.getNulls(); + columnValueCounts.put(schemaPath, columnValueCounts.get(schemaPath) + newCount); + } + } else { + if (column.getNulls() != null) { + Long newCount = rowCount - column.getNulls(); + columnValueCounts.put(schemaPath, newCount); + } else { + columnValueCounts.put(schemaPath, GroupScan.NO_COLUMN_STATS); + } + } + boolean partitionColumn = checkForPartitionColumn(column, first, rowCount, parquetTableMetadata); + if (partitionColumn) { + Map<SchemaPath, Object> map = partitionValueMap.computeIfAbsent(rowGroup.getPath(), key -> new HashMap<>()); + Object value = map.get(schemaPath); + Object currentValue = column.getMaxValue(); + if (value != null) { + if (value != currentValue) { + partitionColTypeMap.remove(schemaPath); + } + } else { + // the value of a column with primitive type can not be null, + // so checks that there are really null value and puts it to the map + if (rowCount == column.getNulls()) { + map.put(schemaPath, null); + } else { + map.put(schemaPath, currentValue); + } + } + } else { + partitionColTypeMap.remove(schemaPath); + } + } + this.rowCount += rowGroup.getRowCount(); + first = false; + } + } + + /** + * Re-init holders eigther during first instance creation or statistics update based on updated list of row groups. + */ + private void resetHolders() { + this.partitionValueMap = new HashMap<>(); + this.partitionColTypeMap = new HashMap<>(); + this.columnValueCounts = new HashMap<>(); + this.rowCount = 0; + } + + /** + * When reading the very first footer, any column is a potential partition column. So for the first footer, we check + * every column to see if it is single valued, and if so, add it to the list of potential partition columns. For the + * remaining footers, we will not find any new partition columns, but we may discover that what was previously a + * potential partition column now no longer qualifies, so it needs to be removed from the list. + * + * @param columnMetadata column metadata + * @param first if columns first appears in row group + * @param rowCount row count + * @param parquetTableMetadata parquet table metadata + * + * @return whether column is a potential partition column + */ + private boolean checkForPartitionColumn(ColumnMetadata columnMetadata, + boolean first, + long rowCount, + ParquetTableMetadataBase parquetTableMetadata) { + SchemaPath schemaPath = SchemaPath.getCompoundPath(columnMetadata.getName()); + final PrimitiveType.PrimitiveTypeName primitiveType; + final OriginalType originalType; + int precision = 0; + int scale = 0; + if (parquetTableMetadata.hasColumnMetadata()) { + // only ColumnTypeMetadata_v3 stores information about scale and precision + if (parquetTableMetadata instanceof ParquetTableMetadata_v3) { + ColumnTypeMetadata_v3 columnTypeInfo = ((ParquetTableMetadata_v3) parquetTableMetadata) + .getColumnTypeInfo(columnMetadata.getName()); + scale = columnTypeInfo.scale; + precision = columnTypeInfo.precision; + } + primitiveType = parquetTableMetadata.getPrimitiveType(columnMetadata.getName()); + originalType = parquetTableMetadata.getOriginalType(columnMetadata.getName()); + } else { + primitiveType = columnMetadata.getPrimitiveType(); + originalType = columnMetadata.getOriginalType(); + } + if (first) { + if (hasSingleValue(columnMetadata, rowCount)) { + partitionColTypeMap.put(schemaPath, ParquetReaderUtility.getType(primitiveType, originalType, scale, precision)); + return true; + } else { + return false; + } + } else { + if (!partitionColTypeMap.keySet().contains(schemaPath)) { + return false; + } else { + if (!hasSingleValue(columnMetadata, rowCount)) { + partitionColTypeMap.remove(schemaPath); + return false; + } + if (!ParquetReaderUtility.getType(primitiveType, originalType, scale, precision).equals(partitionColTypeMap.get(schemaPath))) { + partitionColTypeMap.remove(schemaPath); + return false; + } + } + } + return true; + } + + /** + * Checks that the column chunk has a single value. + * ColumnMetadata will have a non-null value iff the minValue and + * the maxValue for the rowgroup are the same. + * + * @param columnChunkMetaData metadata to check + * @param rowCount rows count in column chunk + * + * @return true if column has single value + */ + private boolean hasSingleValue(ColumnMetadata columnChunkMetaData, long rowCount) { + return (columnChunkMetaData != null) && (columnChunkMetaData.hasSingleValue(rowCount)); + } + +} http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java ---------------------------------------------------------------------- diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java index 45fa51d..83ce4d2 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetPushDownFilter.java @@ -65,7 +65,7 @@ public abstract class ParquetPushDownFilter extends StoragePluginOptimizerRule { @Override public boolean matches(RelOptRuleCall call) { final ScanPrel scan = call.rel(2); - if (scan.getGroupScan() instanceof ParquetGroupScan) { + if (scan.getGroupScan() instanceof AbstractParquetGroupScan) { return super.matches(call); } return false; @@ -90,7 +90,7 @@ public abstract class ParquetPushDownFilter extends StoragePluginOptimizerRule { @Override public boolean matches(RelOptRuleCall call) { final ScanPrel scan = call.rel(1); - if (scan.getGroupScan() instanceof ParquetGroupScan) { + if (scan.getGroupScan() instanceof AbstractParquetGroupScan) { return super.matches(call); } return false; @@ -114,7 +114,7 @@ public abstract class ParquetPushDownFilter extends StoragePluginOptimizerRule { } protected void doOnMatch(RelOptRuleCall call, FilterPrel filter, ProjectPrel project, ScanPrel scan) { - ParquetGroupScan groupScan = (ParquetGroupScan) scan.getGroupScan(); + AbstractParquetGroupScan groupScan = (AbstractParquetGroupScan) scan.getGroupScan(); if (groupScan.getFilter() != null && !groupScan.getFilter().equals(ValueExpressions.BooleanExpression.TRUE)) { return; } @@ -152,25 +152,26 @@ public abstract class ParquetPushDownFilter extends StoragePluginOptimizerRule { LogicalExpression conditionExp = DrillOptiq.toDrill( new DrillParseContext(PrelUtil.getPlannerSettings(call.getPlanner())), scan, qualifedPred); - Stopwatch timer = Stopwatch.createStarted(); + + Stopwatch timer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null; final GroupScan newGroupScan = groupScan.applyFilter(conditionExp,optimizerContext, optimizerContext.getFunctionRegistry(), optimizerContext.getPlannerSettings().getOptions()); - logger.info("Took {} ms to apply filter on parquet row groups. ", timer.elapsed(TimeUnit.MILLISECONDS)); + if (timer != null) { + logger.debug("Took {} ms to apply filter on parquet row groups. ", timer.elapsed(TimeUnit.MILLISECONDS)); + timer.stop(); + } if (newGroupScan == null ) { return; } - final ScanPrel newScanRel = ScanPrel.create(scan, scan.getTraitSet(), newGroupScan, scan.getRowType()); - RelNode inputRel = newScanRel; + RelNode newScan = ScanPrel.create(scan, scan.getTraitSet(), newGroupScan, scan.getRowType());; if (project != null) { - inputRel = project.copy(project.getTraitSet(), ImmutableList.of(inputRel)); + newScan = project.copy(project.getTraitSet(), ImmutableList.of(newScan)); } - - final RelNode newFilter = filter.copy(filter.getTraitSet(), ImmutableList.of(inputRel)); - + final RelNode newFilter = filter.copy(filter.getTraitSet(), ImmutableList.<RelNode>of(newScan)); call.transformTo(newFilter); } } http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java ---------------------------------------------------------------------- diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java index c2580cb..9157590 100644 --- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java +++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetReaderUtility.java @@ -22,9 +22,12 @@ import org.apache.commons.codec.binary.Base64; import org.apache.drill.common.exceptions.UserException; import org.apache.drill.common.expression.PathSegment; import org.apache.drill.common.expression.SchemaPath; +import org.apache.drill.common.types.TypeProtos; +import org.apache.drill.common.types.Types; import org.apache.drill.exec.expr.holders.NullableTimeStampHolder; import org.apache.drill.exec.planner.physical.PlannerSettings; import org.apache.drill.exec.server.options.OptionManager; +import org.apache.drill.exec.store.parquet.metadata.MetadataVersion; import org.apache.drill.exec.util.Utilities; import org.apache.drill.exec.work.ExecErrorConstants; import org.apache.parquet.SemanticVersion; @@ -53,6 +56,15 @@ import java.util.List; import java.util.Map; import java.util.Set; +import static org.apache.drill.exec.store.parquet.metadata.Metadata_V2.ColumnTypeMetadata_v2; +import static org.apache.drill.exec.store.parquet.metadata.Metadata_V2.ParquetTableMetadata_v2; +import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ColumnTypeMetadata_v3; +import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ParquetTableMetadata_v3; +import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ColumnMetadata; +import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ParquetTableMetadataBase; +import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ParquetFileMetadata; +import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.RowGroupMetadata; + /** * Utility class where we can capture common logic between the two parquet readers */ @@ -142,7 +154,7 @@ public class ParquetReaderUtility { return (int) (corruptedDate - CORRECT_CORRUPT_DATE_SHIFT); } - public static void correctDatesInMetadataCache(Metadata.ParquetTableMetadataBase parquetTableMetadata) { + public static void correctDatesInMetadataCache(ParquetTableMetadataBase parquetTableMetadata) { DateCorruptionStatus cacheFileCanContainsCorruptDates = new MetadataVersion(parquetTableMetadata.getMetadataVersion()).compareTo(new MetadataVersion(3, 0)) >= 0 ? DateCorruptionStatus.META_SHOWS_NO_CORRUPTION : DateCorruptionStatus.META_UNCLEAR_TEST_VALUES; @@ -150,19 +162,19 @@ public class ParquetReaderUtility { // Looking for the DATE data type of column names in the metadata cache file ("metadata_version" : "v2") String[] names = new String[0]; if (new MetadataVersion(2, 0).equals(new MetadataVersion(parquetTableMetadata.getMetadataVersion()))) { - for (Metadata.ColumnTypeMetadata_v2 columnTypeMetadata : - ((Metadata.ParquetTableMetadata_v2) parquetTableMetadata).columnTypeInfo.values()) { + for (ColumnTypeMetadata_v2 columnTypeMetadata : + ((ParquetTableMetadata_v2) parquetTableMetadata).columnTypeInfo.values()) { if (OriginalType.DATE.equals(columnTypeMetadata.originalType)) { names = columnTypeMetadata.name; } } } - for (Metadata.ParquetFileMetadata file : parquetTableMetadata.getFiles()) { + for (ParquetFileMetadata file : parquetTableMetadata.getFiles()) { // Drill has only ever written a single row group per file, only need to correct the statistics // on the first row group - Metadata.RowGroupMetadata rowGroupMetadata = file.getRowGroups().get(0); + RowGroupMetadata rowGroupMetadata = file.getRowGroups().get(0); Long rowCount = rowGroupMetadata.getRowCount(); - for (Metadata.ColumnMetadata columnMetadata : rowGroupMetadata.getColumns()) { + for (ColumnMetadata columnMetadata : rowGroupMetadata.getColumns()) { // Setting Min/Max values for ParquetTableMetadata_v1 if (new MetadataVersion(1, 0).equals(new MetadataVersion(parquetTableMetadata.getMetadataVersion()))) { OriginalType originalType = columnMetadata.getOriginalType(); @@ -192,15 +204,15 @@ public class ParquetReaderUtility { * * @param parquetTableMetadata table metadata that should be corrected */ - public static void correctBinaryInMetadataCache(Metadata.ParquetTableMetadataBase parquetTableMetadata) { + public static void correctBinaryInMetadataCache(ParquetTableMetadataBase parquetTableMetadata) { // Looking for the names of the columns with BINARY data type // in the metadata cache file for V2 and all v3 versions Set<List<String>> columnsNames = getBinaryColumnsNames(parquetTableMetadata); - for (Metadata.ParquetFileMetadata file : parquetTableMetadata.getFiles()) { - for (Metadata.RowGroupMetadata rowGroupMetadata : file.getRowGroups()) { + for (ParquetFileMetadata file : parquetTableMetadata.getFiles()) { + for (RowGroupMetadata rowGroupMetadata : file.getRowGroups()) { Long rowCount = rowGroupMetadata.getRowCount(); - for (Metadata.ColumnMetadata columnMetadata : rowGroupMetadata.getColumns()) { + for (ColumnMetadata columnMetadata : rowGroupMetadata.getColumns()) { // Setting Min/Max values for ParquetTableMetadata_v1 if (new MetadataVersion(1, 0).equals(new MetadataVersion(parquetTableMetadata.getMetadataVersion()))) { if (columnMetadata.getPrimitiveType() == PrimitiveTypeName.BINARY @@ -231,19 +243,19 @@ public class ParquetReaderUtility { * @param parquetTableMetadata table metadata the source of the columns to check * @return set of the lists with column names */ - private static Set<List<String>> getBinaryColumnsNames(Metadata.ParquetTableMetadataBase parquetTableMetadata) { + private static Set<List<String>> getBinaryColumnsNames(ParquetTableMetadataBase parquetTableMetadata) { Set<List<String>> names = Sets.newHashSet(); - if (parquetTableMetadata instanceof Metadata.ParquetTableMetadata_v2) { - for (Metadata.ColumnTypeMetadata_v2 columnTypeMetadata : - ((Metadata.ParquetTableMetadata_v2) parquetTableMetadata).columnTypeInfo.values()) { + if (parquetTableMetadata instanceof ParquetTableMetadata_v2) { + for (ColumnTypeMetadata_v2 columnTypeMetadata : + ((ParquetTableMetadata_v2) parquetTableMetadata).columnTypeInfo.values()) { if (columnTypeMetadata.primitiveType == PrimitiveTypeName.BINARY || columnTypeMetadata.primitiveType == PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) { names.add(Arrays.asList(columnTypeMetadata.name)); } } - } else if (parquetTableMetadata instanceof Metadata.ParquetTableMetadata_v3) { - for (Metadata.ColumnTypeMetadata_v3 columnTypeMetadata : - ((Metadata.ParquetTableMetadata_v3) parquetTableMetadata).columnTypeInfo.values()) { + } else if (parquetTableMetadata instanceof ParquetTableMetadata_v3) { + for (ColumnTypeMetadata_v3 columnTypeMetadata : + ((ParquetTableMetadata_v3) parquetTableMetadata).columnTypeInfo.values()) { if (columnTypeMetadata.primitiveType == PrimitiveTypeName.BINARY || columnTypeMetadata.primitiveType == PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY) { names.add(Arrays.asList(columnTypeMetadata.name)); @@ -260,7 +272,7 @@ public class ParquetReaderUtility { * @param columnMetadata column metadata that should be changed * @param rowCount rows count in column chunk */ - private static void setMinMaxValues(Metadata.ColumnMetadata columnMetadata, long rowCount) { + private static void setMinMaxValues(ColumnMetadata columnMetadata, long rowCount) { if (columnMetadata.hasSingleValue(rowCount)) { Object minValue = columnMetadata.getMinValue(); if (minValue != null && minValue instanceof String) { @@ -278,7 +290,7 @@ public class ParquetReaderUtility { * @param columnMetadata column metadata that should be changed * @param rowCount rows count in column chunk */ - private static void convertMinMaxValues(Metadata.ColumnMetadata columnMetadata, long rowCount) { + private static void convertMinMaxValues(ColumnMetadata columnMetadata, long rowCount) { if (columnMetadata.hasSingleValue(rowCount)) { Object minValue = columnMetadata.getMinValue(); if (minValue != null && minValue instanceof String) { @@ -362,8 +374,8 @@ public class ParquetReaderUtility { * This method only checks the first Row Group, because Drill has only ever written * a single Row Group per file. * - * @param footer - * @param columns + * @param footer parquet footer + * @param columns list of columns schema path * @param autoCorrectCorruptDates user setting to allow enabling/disabling of auto-correction * of corrupt dates. There are some rare cases (storing dates thousands * of years into the future, with tools other than Drill writing files) @@ -459,4 +471,66 @@ public class ParquetReaderUtility { } } } + + /** + * Builds major type using given {@code OriginalType originalType} or {@code PrimitiveTypeName type}. + * For DECIMAL will be returned major type with scale and precision. + * + * @param type parquet primitive type + * @param originalType parquet original type + * @param scale type scale (used for DECIMAL type) + * @param precision type precision (used for DECIMAL type) + * @return major type + */ + public static TypeProtos.MajorType getType(PrimitiveTypeName type, OriginalType originalType, int scale, int precision) { + if (originalType != null) { + switch (originalType) { + case DECIMAL: + return Types.withScaleAndPrecision(TypeProtos.MinorType.DECIMAL18, TypeProtos.DataMode.OPTIONAL, scale, precision); + case DATE: + return Types.optional(TypeProtos.MinorType.DATE); + case TIME_MILLIS: + return Types.optional(TypeProtos.MinorType.TIME); + case TIMESTAMP_MILLIS: + return Types.optional(TypeProtos.MinorType.TIMESTAMP); + case UTF8: + return Types.optional(TypeProtos.MinorType.VARCHAR); + case UINT_8: + return Types.optional(TypeProtos.MinorType.UINT1); + case UINT_16: + return Types.optional(TypeProtos.MinorType.UINT2); + case UINT_32: + return Types.optional(TypeProtos.MinorType.UINT4); + case UINT_64: + return Types.optional(TypeProtos.MinorType.UINT8); + case INT_8: + return Types.optional(TypeProtos.MinorType.TINYINT); + case INT_16: + return Types.optional(TypeProtos.MinorType.SMALLINT); + case INTERVAL: + return Types.optional(TypeProtos.MinorType.INTERVAL); + } + } + + switch (type) { + case BOOLEAN: + return Types.optional(TypeProtos.MinorType.BIT); + case INT32: + return Types.optional(TypeProtos.MinorType.INT); + case INT64: + return Types.optional(TypeProtos.MinorType.BIGINT); + case FLOAT: + return Types.optional(TypeProtos.MinorType.FLOAT4); + case DOUBLE: + return Types.optional(TypeProtos.MinorType.FLOAT8); + case BINARY: + case FIXED_LEN_BYTE_ARRAY: + case INT96: + return Types.optional(TypeProtos.MinorType.VARBINARY); + default: + // Should never hit this + throw new UnsupportedOperationException("Unsupported type:" + type); + } + } + }
