You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by ar...@apache.org on 2018/04/27 09:41:45 UTC
[3/8] drill git commit: DRILL-6331: Revisit Hive Drill native parquet
implementation to be exposed to Drill optimizations (filter / limit push down,
count to direct scan)
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRowGroupScan.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRowGroupScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRowGroupScan.java
index f1fb1e9..3ccdad0 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRowGroupScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetRowGroupScan.java
@@ -17,7 +17,6 @@
*/
package org.apache.drill.exec.store.parquet;
-import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
@@ -26,12 +25,9 @@ import org.apache.drill.common.expression.LogicalExpression;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.logical.FormatPluginConfig;
import org.apache.drill.common.logical.StoragePluginConfig;
-import org.apache.drill.exec.physical.base.AbstractBase;
-import org.apache.drill.exec.physical.base.GroupScan;
import org.apache.drill.exec.physical.base.PhysicalOperator;
-import org.apache.drill.exec.physical.base.PhysicalVisitor;
-import org.apache.drill.exec.physical.base.SubScan;
import org.apache.drill.exec.proto.UserBitShared.CoreOperatorType;
+import org.apache.drill.exec.store.ColumnExplorer;
import org.apache.drill.exec.store.StoragePluginRegistry;
import com.fasterxml.jackson.annotation.JacksonInject;
@@ -40,31 +36,26 @@ 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.Iterators;
+import org.apache.hadoop.conf.Configuration;
-// Class containing information for reading a single parquet row group form HDFS
+// Class containing information for reading a single parquet row group from HDFS
@JsonTypeName("parquet-row-group-scan")
-public class ParquetRowGroupScan extends AbstractBase implements SubScan {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ParquetRowGroupScan.class);
+public class ParquetRowGroupScan extends AbstractParquetRowGroupScan {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ParquetRowGroupScan.class);
- private final ParquetFormatConfig formatConfig;
private final ParquetFormatPlugin formatPlugin;
- private final List<RowGroupReadEntry> rowGroupReadEntries;
- private final List<SchemaPath> columns;
- private LogicalExpression filter;
- private String selectionRoot;
+ private final ParquetFormatConfig formatConfig;
+ private final String selectionRoot;
@JsonCreator
- public ParquetRowGroupScan( //
- @JacksonInject StoragePluginRegistry registry, //
- @JsonProperty("userName") String userName, //
- @JsonProperty("storage") StoragePluginConfig storageConfig, //
- @JsonProperty("format") FormatPluginConfig formatConfig, //
- @JsonProperty("entries") LinkedList<RowGroupReadEntry> rowGroupReadEntries, //
- @JsonProperty("columns") List<SchemaPath> columns, //
- @JsonProperty("selectionRoot") String selectionRoot, //
- @JsonProperty("filter") LogicalExpression filter
- ) throws ExecutionSetupException {
+ public ParquetRowGroupScan(@JacksonInject StoragePluginRegistry registry,
+ @JsonProperty("userName") String userName,
+ @JsonProperty("storageConfig") StoragePluginConfig storageConfig,
+ @JsonProperty("formatConfig") FormatPluginConfig formatConfig,
+ @JsonProperty("rowGroupReadEntries") LinkedList<RowGroupReadEntry> rowGroupReadEntries,
+ @JsonProperty("columns") List<SchemaPath> columns,
+ @JsonProperty("selectionRoot") String selectionRoot,
+ @JsonProperty("filter") LogicalExpression filter) throws ExecutionSetupException {
this(userName,
(ParquetFormatPlugin) registry.getFormatPlugin(Preconditions.checkNotNull(storageConfig), Preconditions.checkNotNull(formatConfig)),
rowGroupReadEntries,
@@ -73,82 +64,72 @@ public class ParquetRowGroupScan extends AbstractBase implements SubScan {
filter);
}
- public ParquetRowGroupScan( //
- String userName, //
- ParquetFormatPlugin formatPlugin, //
- List<RowGroupReadEntry> rowGroupReadEntries, //
- List<SchemaPath> columns, //
- String selectionRoot, //
- LogicalExpression filter
- ) {
- super(userName);
+ public ParquetRowGroupScan(String userName,
+ ParquetFormatPlugin formatPlugin,
+ List<RowGroupReadEntry> rowGroupReadEntries,
+ List<SchemaPath> columns,
+ String selectionRoot,
+ LogicalExpression filter) {
+ super(userName, rowGroupReadEntries, columns, filter);
this.formatPlugin = Preconditions.checkNotNull(formatPlugin, "Could not find format config for the given configuration");
this.formatConfig = formatPlugin.getConfig();
- this.rowGroupReadEntries = rowGroupReadEntries;
- this.columns = columns == null ? GroupScan.ALL_COLUMNS : columns;
this.selectionRoot = selectionRoot;
- this.filter = filter;
- }
-
- @JsonProperty("entries")
- public List<RowGroupReadEntry> getRowGroupReadEntries() {
- return rowGroupReadEntries;
}
- @JsonProperty("storage")
- public StoragePluginConfig getEngineConfig() {
+ @JsonProperty
+ public StoragePluginConfig getStorageConfig() {
return formatPlugin.getStorageConfig();
}
- /**
- * @return Parquet plugin format config
- */
- @JsonProperty("format")
+ @JsonProperty
public ParquetFormatConfig getFormatConfig() {
return formatConfig;
}
+ @JsonProperty
public String getSelectionRoot() {
return selectionRoot;
}
- @Override
- public boolean isExecutable() {
- return false;
- }
-
@JsonIgnore
public ParquetFormatPlugin getStorageEngine() {
return formatPlugin;
}
@Override
- public <T, X, E extends Throwable> T accept(PhysicalVisitor<T, X, E> physicalVisitor, X value) throws E {
- return physicalVisitor.visitSubScan(this, value);
+ public PhysicalOperator getNewWithChildren(List<PhysicalOperator> children) {
+ Preconditions.checkArgument(children.isEmpty());
+ return new ParquetRowGroupScan(getUserName(), formatPlugin, rowGroupReadEntries, columns, selectionRoot, filter);
}
@Override
- public PhysicalOperator getNewWithChildren(List<PhysicalOperator> children) throws ExecutionSetupException {
- Preconditions.checkArgument(children.isEmpty());
- return new ParquetRowGroupScan(getUserName(), formatPlugin, rowGroupReadEntries, columns, selectionRoot, filter);
+ public int getOperatorType() {
+ return CoreOperatorType.PARQUET_ROW_GROUP_SCAN_VALUE;
}
@Override
- public Iterator<PhysicalOperator> iterator() {
- return Iterators.emptyIterator();
+ public AbstractParquetRowGroupScan copy(List<SchemaPath> columns) {
+ return new ParquetRowGroupScan(getUserName(), formatPlugin, rowGroupReadEntries, columns, selectionRoot, filter);
}
- public List<SchemaPath> getColumns() {
- return columns;
+ @Override
+ public boolean areCorruptDatesAutoCorrected() {
+ return formatConfig.areCorruptDatesAutoCorrected();
}
- public LogicalExpression getFilter() {
- return filter;
+ @Override
+ public Configuration getFsConf(RowGroupReadEntry rowGroupReadEntry) {
+ return formatPlugin.getFsConf();
}
@Override
- public int getOperatorType() {
- return CoreOperatorType.PARQUET_ROW_GROUP_SCAN_VALUE;
+ public boolean supportsFileImplicitColumns() {
+ return selectionRoot != null;
}
+ @Override
+ public List<String> getPartitionValues(RowGroupReadEntry rowGroupReadEntry) {
+ return ColumnExplorer.listPartitionValues(rowGroupReadEntry.getPath(), selectionRoot);
+ }
}
+
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetScanBatchCreator.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetScanBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetScanBatchCreator.java
index e928ebb..c71e3c3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetScanBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/ParquetScanBatchCreator.java
@@ -17,15 +17,7 @@
*/
package org.apache.drill.exec.store.parquet;
-import java.io.IOException;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Map;
-import java.util.concurrent.TimeUnit;
-
-import com.google.common.base.Functions;
-import com.google.common.base.Stopwatch;
-import com.google.common.collect.Maps;
+import com.google.common.base.Preconditions;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.ops.ExecutorFragmentContext;
@@ -33,144 +25,52 @@ import org.apache.drill.exec.ops.OperatorContext;
import org.apache.drill.exec.physical.impl.BatchCreator;
import org.apache.drill.exec.physical.impl.ScanBatch;
import org.apache.drill.exec.record.RecordBatch;
-import org.apache.drill.exec.store.ColumnExplorer;
-import org.apache.drill.exec.store.RecordReader;
+import org.apache.drill.exec.server.options.OptionManager;
import org.apache.drill.exec.store.dfs.DrillFileSystem;
-import org.apache.drill.exec.store.parquet.columnreaders.ParquetRecordReader;
-import org.apache.drill.exec.store.parquet2.DrillParquetReader;
import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.Path;
-import org.apache.parquet.column.ColumnDescriptor;
-import org.apache.parquet.hadoop.CodecFactory;
-import org.apache.parquet.hadoop.ParquetFileReader;
-import org.apache.parquet.hadoop.metadata.ParquetMetadata;
-import org.apache.parquet.schema.MessageType;
-import org.apache.parquet.schema.Type;
-import com.google.common.base.Preconditions;
-import com.google.common.collect.Lists;
-
-public class ParquetScanBatchCreator implements BatchCreator<ParquetRowGroupScan>{
- private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(ParquetScanBatchCreator.class);
+import java.io.IOException;
+import java.util.List;
- private static final String ENABLE_BYTES_READ_COUNTER = "parquet.benchmark.bytes.read";
- private static final String ENABLE_BYTES_TOTAL_COUNTER = "parquet.benchmark.bytes.total";
- private static final String ENABLE_TIME_READ_COUNTER = "parquet.benchmark.time.read";
+public class ParquetScanBatchCreator extends AbstractParquetScanBatchCreator implements BatchCreator<ParquetRowGroupScan> {
- @SuppressWarnings("resource")
@Override
- public ScanBatch getBatch(ExecutorFragmentContext context, ParquetRowGroupScan rowGroupScan, List<RecordBatch> children)
- throws ExecutionSetupException {
+ public ScanBatch getBatch(ExecutorFragmentContext context, ParquetRowGroupScan rowGroupScan, List<RecordBatch> children) throws ExecutionSetupException {
Preconditions.checkArgument(children.isEmpty());
OperatorContext oContext = context.newOperatorContext(rowGroupScan);
+ return getBatch(context, rowGroupScan, oContext);
+ }
- final ColumnExplorer columnExplorer = new ColumnExplorer(context.getOptions(), rowGroupScan.getColumns());
-
- if (!columnExplorer.isStarQuery()) {
- rowGroupScan = new ParquetRowGroupScan(rowGroupScan.getUserName(), rowGroupScan.getStorageEngine(),
- rowGroupScan.getRowGroupReadEntries(), columnExplorer.getTableColumns(), rowGroupScan.getSelectionRoot(),
- rowGroupScan.getFilter());
- rowGroupScan.setOperatorId(rowGroupScan.getOperatorId());
- }
-
- DrillFileSystem fs;
- try {
- boolean useAsyncPageReader =
- context.getOptions().getOption(ExecConstants.PARQUET_PAGEREADER_ASYNC).bool_val;
- if (useAsyncPageReader) {
- fs = oContext.newNonTrackingFileSystem(rowGroupScan.getStorageEngine().getFsConf());
- } else {
- fs = oContext.newFileSystem(rowGroupScan.getStorageEngine().getFsConf());
- }
- } catch (IOException e) {
- throw new ExecutionSetupException(
- String.format("Failed to create DrillFileSystem: %s", e.getMessage()), e);
- }
-
- Configuration conf = new Configuration(fs.getConf());
- conf.setBoolean(ENABLE_BYTES_READ_COUNTER, false);
- conf.setBoolean(ENABLE_BYTES_TOTAL_COUNTER, false);
- conf.setBoolean(ENABLE_TIME_READ_COUNTER, false);
+ @Override
+ protected AbstractDrillFileSystemManager getDrillFileSystemCreator(OperatorContext operatorContext, OptionManager optionManager) {
+ return new ParquetDrillFileSystemManager(operatorContext, optionManager.getOption(ExecConstants.PARQUET_PAGEREADER_ASYNC).bool_val);
+ }
- // keep footers in a map to avoid re-reading them
- Map<String, ParquetMetadata> footers = Maps.newHashMap();
- List<RecordReader> readers = new LinkedList<>();
- List<Map<String, String>> implicitColumns = Lists.newArrayList();
- Map<String, String> mapWithMaxColumns = Maps.newLinkedHashMap();
- for(RowGroupReadEntry e : rowGroupScan.getRowGroupReadEntries()){
- /*
- Here we could store a map from file names to footers, to prevent re-reading the footer for each row group in a file
- TODO - to prevent reading the footer again in the parquet record reader (it is read earlier in the ParquetStorageEngine)
- we should add more information to the RowGroupInfo that will be populated upon the first read to
- provide the reader with all of th file meta-data it needs
- These fields will be added to the constructor below
- */
- try {
- Stopwatch timer = Stopwatch.createUnstarted();
- if (!footers.containsKey(e.getPath())){
- timer.start();
- ParquetMetadata footer = ParquetFileReader.readFooter(conf, new Path(e.getPath()));
- long timeToRead = timer.elapsed(TimeUnit.MICROSECONDS);
- logger.trace("ParquetTrace,Read Footer,{},{},{},{},{},{},{}", "", e.getPath(), "", 0, 0, 0, timeToRead);
- footers.put(e.getPath(), footer );
- }
- boolean autoCorrectCorruptDates = rowGroupScan.getFormatConfig().areCorruptDatesAutoCorrected();
- ParquetReaderUtility.DateCorruptionStatus containsCorruptDates = ParquetReaderUtility.detectCorruptDates(footers.get(e.getPath()), rowGroupScan.getColumns(),
- autoCorrectCorruptDates);
- if (logger.isDebugEnabled()) {
- logger.debug(containsCorruptDates.toString());
- }
- if (!context.getOptions().getBoolean(ExecConstants.PARQUET_NEW_RECORD_READER) && !isComplex(footers.get(e.getPath()))) {
- readers.add(
- new ParquetRecordReader(
- context, e.getPath(), e.getRowGroupIndex(), e.getNumRecordsToRead(), fs,
- CodecFactory.createDirectCodecFactory(
- fs.getConf(),
- new ParquetDirectByteBufferAllocator(oContext.getAllocator()), 0),
- footers.get(e.getPath()),
- rowGroupScan.getColumns(),
- containsCorruptDates
- )
- );
- } else {
- ParquetMetadata footer = footers.get(e.getPath());
- readers.add(new DrillParquetReader(context, footer, e, columnExplorer.getTableColumns(), fs, containsCorruptDates));
- }
- Map<String, String> implicitValues = columnExplorer.populateImplicitColumns(e, rowGroupScan.getSelectionRoot());
- implicitColumns.add(implicitValues);
- if (implicitValues.size() > mapWithMaxColumns.size()) {
- mapWithMaxColumns = implicitValues;
- }
+ /**
+ * Creates file system only if it was not created before, otherwise returns already created instance.
+ */
+ private class ParquetDrillFileSystemManager extends AbstractDrillFileSystemManager {
- } catch (IOException e1) {
- throw new ExecutionSetupException(e1);
- }
- }
+ private final boolean useAsyncPageReader;
+ private DrillFileSystem fs;
- // all readers should have the same number of implicit columns, add missing ones with value null
- Map<String, String> diff = Maps.transformValues(mapWithMaxColumns, Functions.constant((String) null));
- for (Map<String, String> map : implicitColumns) {
- map.putAll(Maps.difference(map, diff).entriesOnlyOnRight());
+ ParquetDrillFileSystemManager(OperatorContext operatorContext, boolean useAsyncPageReader) {
+ super(operatorContext);
+ this.useAsyncPageReader = useAsyncPageReader;
}
- return new ScanBatch(context, oContext, readers, implicitColumns);
- }
-
- private static boolean isComplex(ParquetMetadata footer) {
- MessageType schema = footer.getFileMetaData().getSchema();
-
- for (Type type : schema.getFields()) {
- if (!type.isPrimitive()) {
- return true;
- }
- }
- for (ColumnDescriptor col : schema.getColumns()) {
- if (col.getMaxRepetitionLevel() > 0) {
- return true;
+ @Override
+ protected DrillFileSystem get(Configuration config, String path) throws ExecutionSetupException {
+ if (fs == null) {
+ try {
+ fs = useAsyncPageReader ? operatorContext.newNonTrackingFileSystem(config) : operatorContext.newFileSystem(config);
+ } catch (IOException e) {
+ throw new ExecutionSetupException(String.format("Failed to create DrillFileSystem: %s", e.getMessage()), e);
+ }
}
+ return fs;
}
- return false;
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/RowGroupInfo.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/RowGroupInfo.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/RowGroupInfo.java
new file mode 100644
index 0000000..af436d8
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/RowGroupInfo.java
@@ -0,0 +1,98 @@
+/*
+ * 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 com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.drill.exec.store.dfs.ReadEntryFromHDFS;
+import org.apache.drill.exec.store.dfs.easy.FileWork;
+import org.apache.drill.exec.store.schedule.CompleteWork;
+import org.apache.drill.exec.store.schedule.EndpointByteMap;
+
+import java.util.List;
+
+import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ColumnMetadata;
+
+public 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;
+ }
+
+ public long getNumRecordsToRead() {
+ return numRecordsToRead;
+ }
+
+ public void setNumRecordsToRead(long numRecords) {
+ numRecordsToRead = numRecords;
+ }
+
+ public void setEndpointByteMap(EndpointByteMap byteMap) {
+ this.byteMap = byteMap;
+ }
+
+ public long getRowCount() {
+ return rowCount;
+ }
+
+ public List<? extends ColumnMetadata> getColumns() {
+ return columns;
+ }
+
+ public void setColumns(List<? extends ColumnMetadata> columns) {
+ this.columns = columns;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
new file mode 100644
index 0000000..cdb28c2
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata.java
@@ -0,0 +1,694 @@
+/*
+ * 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.metadata;
+
+import com.fasterxml.jackson.core.JsonFactory;
+import com.fasterxml.jackson.core.JsonGenerator.Feature;
+import com.fasterxml.jackson.core.JsonParser;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.module.SimpleModule;
+import com.fasterxml.jackson.module.afterburner.AfterburnerModule;
+import com.google.common.base.Stopwatch;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Maps;
+import org.apache.commons.lang3.tuple.Pair;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.util.DrillVersionInfo;
+import org.apache.drill.exec.store.TimedRunnable;
+import org.apache.drill.exec.store.dfs.MetadataContext;
+import org.apache.drill.exec.store.parquet.ParquetFormatConfig;
+import org.apache.drill.exec.store.parquet.ParquetReaderUtility;
+import org.apache.drill.exec.util.DrillFileSystemUtil;
+import org.apache.drill.exec.util.ImpersonationUtil;
+import org.apache.hadoop.fs.BlockLocation;
+import org.apache.hadoop.fs.FSDataInputStream;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.parquet.column.statistics.Statistics;
+import org.apache.parquet.format.converter.ParquetMetadataConverter;
+import org.apache.parquet.hadoop.ParquetFileReader;
+import org.apache.parquet.hadoop.metadata.BlockMetaData;
+import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData;
+import org.apache.parquet.hadoop.metadata.ParquetMetadata;
+import org.apache.parquet.schema.GroupType;
+import org.apache.parquet.schema.MessageType;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+import org.apache.parquet.schema.Type;
+
+import javax.annotation.Nullable;
+import java.io.IOException;
+import java.security.PrivilegedExceptionAction;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ParquetFileMetadata;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ParquetTableMetadataBase;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.RowGroupMetadata;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataVersion.Constants.SUPPORTED_VERSIONS;
+import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ColumnMetadata_v3;
+import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ColumnTypeMetadata_v3;
+import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ParquetFileMetadata_v3;
+import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ParquetTableMetadata_v3;
+import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.RowGroupMetadata_v3;
+
+public class Metadata {
+ private static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(Metadata.class);
+
+ public static final String[] OLD_METADATA_FILENAMES = {".drill.parquet_metadata.v2"};
+ public static final String METADATA_FILENAME = ".drill.parquet_metadata";
+ public static final String METADATA_DIRECTORIES_FILENAME = ".drill.parquet_metadata_directories";
+
+ private final ParquetFormatConfig formatConfig;
+
+ private ParquetTableMetadataBase parquetTableMetadata;
+ private ParquetTableMetadataDirs parquetTableMetadataDirs;
+
+
+ private Metadata(ParquetFormatConfig formatConfig) {
+ this.formatConfig = formatConfig;
+ }
+
+ /**
+ * Create the parquet metadata file for the directory at the given path, and for any subdirectories.
+ *
+ * @param fs file system
+ * @param path path
+ */
+ public static void createMeta(FileSystem fs, String path, ParquetFormatConfig formatConfig) throws IOException {
+ Metadata metadata = new Metadata(formatConfig);
+ metadata.createMetaFilesRecursively(path, fs);
+ }
+
+ /**
+ * Get the parquet metadata for the parquet files in the given directory, including those in subdirectories.
+ *
+ * @param fs file system
+ * @param path path
+ * @return parquet table metadata
+ */
+ public static ParquetTableMetadata_v3 getParquetTableMetadata(FileSystem fs, String path, ParquetFormatConfig formatConfig)
+ throws IOException {
+ Metadata metadata = new Metadata(formatConfig);
+ return metadata.getParquetTableMetadata(path, fs);
+ }
+
+ /**
+ * Get the parquet metadata for a list of parquet files.
+ *
+ * @param fileStatusMap file statuses and corresponding file systems
+ * @param formatConfig parquet format config
+ * @return parquet table metadata
+ */
+ public static ParquetTableMetadata_v3 getParquetTableMetadata(Map<FileStatus, FileSystem> fileStatusMap,
+ ParquetFormatConfig formatConfig) throws IOException {
+ Metadata metadata = new Metadata(formatConfig);
+ return metadata.getParquetTableMetadata(fileStatusMap);
+ }
+
+ /**
+ * Get the parquet metadata for the table by reading the metadata file
+ *
+ * @param fs current file system
+ * @param path The path to the metadata file, located in the directory that contains the parquet files
+ * @param metaContext metadata context
+ * @param formatConfig parquet format plugin configs
+ * @return parquet table metadata. Null if metadata cache is missing, unsupported or corrupted
+ */
+ public static @Nullable ParquetTableMetadataBase readBlockMeta(FileSystem fs, Path path, MetadataContext metaContext,
+ ParquetFormatConfig formatConfig) {
+ if (ignoreReadingMetadata(metaContext, path)) {
+ return null;
+ }
+ Metadata metadata = new Metadata(formatConfig);
+ metadata.readBlockMeta(path, false, metaContext, fs);
+ return metadata.parquetTableMetadata;
+ }
+
+ /**
+ * Get the parquet metadata for all subdirectories by reading the metadata file
+ *
+ * @param fs current file system
+ * @param path The path to the metadata file, located in the directory that contains the parquet files
+ * @param metaContext metadata context
+ * @param formatConfig parquet format plugin configs
+ * @return parquet metadata for a directory. Null if metadata cache is missing, unsupported or corrupted
+ */
+ public static @Nullable ParquetTableMetadataDirs readMetadataDirs(FileSystem fs, Path path,
+ MetadataContext metaContext, ParquetFormatConfig formatConfig) {
+ if (ignoreReadingMetadata(metaContext, path)) {
+ return null;
+ }
+ Metadata metadata = new Metadata(formatConfig);
+ metadata.readBlockMeta(path, true, metaContext, fs);
+ return metadata.parquetTableMetadataDirs;
+ }
+
+ /**
+ * Ignore reading metadata files, if metadata is missing, unsupported or corrupted
+ *
+ * @param metaContext Metadata context
+ * @param path The path to the metadata file, located in the directory that contains the parquet files
+ * @return true if parquet metadata is missing or corrupted, false otherwise
+ */
+ private static boolean ignoreReadingMetadata(MetadataContext metaContext, Path path) {
+ if (metaContext.isMetadataCacheCorrupted()) {
+ logger.warn("Ignoring of reading '{}' metadata file. Parquet metadata cache files are unsupported or corrupted. " +
+ "Query performance may be slow. Make sure the cache files are up-to-date by running the 'REFRESH TABLE " +
+ "METADATA' command", path);
+ return true;
+ }
+ return false;
+ }
+
+ /**
+ * Create the parquet metadata files for the directory at the given path and for any subdirectories.
+ * Metadata cache files written to the disk contain relative paths. Returned Pair of metadata contains absolute paths.
+ *
+ * @param path to the directory of the parquet table
+ * @param fs file system
+ * @return Pair of parquet metadata. The left one is a parquet metadata for the table. The right one of the Pair is
+ * a metadata for all subdirectories (if they are present and there are no any parquet files in the
+ * {@code path} directory).
+ * @throws IOException if parquet metadata can't be serialized and written to the json file
+ */
+ private Pair<ParquetTableMetadata_v3, ParquetTableMetadataDirs> createMetaFilesRecursively(final String path, FileSystem fs) throws IOException {
+ Stopwatch timer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
+ List<ParquetFileMetadata_v3> metaDataList = Lists.newArrayList();
+ List<String> directoryList = Lists.newArrayList();
+ ConcurrentHashMap<ColumnTypeMetadata_v3.Key, ColumnTypeMetadata_v3> columnTypeInfoSet =
+ new ConcurrentHashMap<>();
+ Path p = new Path(path);
+ FileStatus fileStatus = fs.getFileStatus(p);
+ assert fileStatus.isDirectory() : "Expected directory";
+
+ final Map<FileStatus, FileSystem> childFiles = new LinkedHashMap<>();
+
+ for (final FileStatus file : DrillFileSystemUtil.listAll(fs, p, false)) {
+ if (file.isDirectory()) {
+ ParquetTableMetadata_v3 subTableMetadata = (createMetaFilesRecursively(file.getPath().toString(), fs)).getLeft();
+ metaDataList.addAll(subTableMetadata.files);
+ directoryList.addAll(subTableMetadata.directories);
+ directoryList.add(file.getPath().toString());
+ // Merge the schema from the child level into the current level
+ //TODO: We need a merge method that merges two columns with the same name but different types
+ columnTypeInfoSet.putAll(subTableMetadata.columnTypeInfo);
+ } else {
+ childFiles.put(file, fs);
+ }
+ }
+ ParquetTableMetadata_v3 parquetTableMetadata = new ParquetTableMetadata_v3(SUPPORTED_VERSIONS.last().toString(),
+ DrillVersionInfo.getVersion());
+ if (childFiles.size() > 0) {
+ List<ParquetFileMetadata_v3 > childFilesMetadata = getParquetFileMetadata_v3(parquetTableMetadata, childFiles);
+ metaDataList.addAll(childFilesMetadata);
+ // Note that we do not need to merge the columnInfo at this point. The columnInfo is already added
+ // to the parquetTableMetadata.
+ }
+
+ parquetTableMetadata.directories = directoryList;
+ parquetTableMetadata.files = metaDataList;
+ // TODO: We need a merge method that merges two columns with the same name but different types
+ if (parquetTableMetadata.columnTypeInfo == null) {
+ parquetTableMetadata.columnTypeInfo = new ConcurrentHashMap<>();
+ }
+ parquetTableMetadata.columnTypeInfo.putAll(columnTypeInfoSet);
+
+ for (String oldName : OLD_METADATA_FILENAMES) {
+ fs.delete(new Path(p, oldName), false);
+ }
+ // relative paths in the metadata are only necessary for meta cache files.
+ ParquetTableMetadata_v3 metadataTableWithRelativePaths =
+ MetadataPathUtils.createMetadataWithRelativePaths(parquetTableMetadata, path);
+ writeFile(metadataTableWithRelativePaths, new Path(p, METADATA_FILENAME), fs);
+
+ if (directoryList.size() > 0 && childFiles.size() == 0) {
+ ParquetTableMetadataDirs parquetTableMetadataDirsRelativePaths =
+ new ParquetTableMetadataDirs(metadataTableWithRelativePaths.directories);
+ writeFile(parquetTableMetadataDirsRelativePaths, new Path(p, METADATA_DIRECTORIES_FILENAME), fs);
+ if (timer != null) {
+ logger.debug("Creating metadata files recursively took {} ms", timer.elapsed(TimeUnit.MILLISECONDS));
+ }
+ ParquetTableMetadataDirs parquetTableMetadataDirs = new ParquetTableMetadataDirs(directoryList);
+ return Pair.of(parquetTableMetadata, parquetTableMetadataDirs);
+ }
+ List<String> emptyDirList = Lists.newArrayList();
+ if (timer != null) {
+ logger.debug("Creating metadata files recursively took {} ms", timer.elapsed(TimeUnit.MILLISECONDS));
+ timer.stop();
+ }
+ return Pair.of(parquetTableMetadata, new ParquetTableMetadataDirs(emptyDirList));
+ }
+
+ /**
+ * Get the parquet metadata for the parquet files in a directory.
+ *
+ * @param path the path of the directory
+ * @return metadata object for an entire parquet directory structure
+ * @throws IOException in case of problems during accessing files
+ */
+ private ParquetTableMetadata_v3 getParquetTableMetadata(String path, FileSystem fs) throws IOException {
+ Path p = new Path(path);
+ FileStatus fileStatus = fs.getFileStatus(p);
+ Stopwatch watch = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
+ List<FileStatus> fileStatuses = new ArrayList<>();
+ if (fileStatus.isFile()) {
+ fileStatuses.add(fileStatus);
+ } else {
+ fileStatuses.addAll(DrillFileSystemUtil.listFiles(fs, p, true));
+ }
+ if (watch != null) {
+ logger.debug("Took {} ms to get file statuses", watch.elapsed(TimeUnit.MILLISECONDS));
+ watch.reset();
+ watch.start();
+ }
+
+ Map<FileStatus, FileSystem> fileStatusMap = fileStatuses.stream()
+ .collect(
+ Collectors.toMap(
+ Function.identity(),
+ s -> fs,
+ (oldFs, newFs) -> newFs,
+ LinkedHashMap::new));
+
+ ParquetTableMetadata_v3 metadata_v3 = getParquetTableMetadata(fileStatusMap);
+ if (watch != null) {
+ logger.debug("Took {} ms to read file metadata", watch.elapsed(TimeUnit.MILLISECONDS));
+ watch.stop();
+ }
+ return metadata_v3;
+ }
+
+ /**
+ * Get the parquet metadata for a list of parquet files
+ *
+ * @param fileStatusMap file statuses and corresponding file systems
+ * @return parquet table metadata object
+ * @throws IOException if parquet file metadata can't be obtained
+ */
+ private ParquetTableMetadata_v3 getParquetTableMetadata(Map<FileStatus, FileSystem> fileStatusMap)
+ throws IOException {
+ ParquetTableMetadata_v3 tableMetadata = new ParquetTableMetadata_v3(SUPPORTED_VERSIONS.last().toString(),
+ DrillVersionInfo.getVersion());
+ tableMetadata.files = getParquetFileMetadata_v3(tableMetadata, fileStatusMap);
+ tableMetadata.directories = new ArrayList<>();
+ return tableMetadata;
+ }
+
+ /**
+ * Get a list of file metadata for a list of parquet files
+ *
+ * @param parquetTableMetadata_v3 can store column schema info from all the files and row groups
+ * @param fileStatusMap parquet files statuses and corresponding file systems
+ *
+ * @return list of the parquet file metadata with absolute paths
+ * @throws IOException is thrown in case of issues while executing the list of runnables
+ */
+ private List<ParquetFileMetadata_v3> getParquetFileMetadata_v3(
+ ParquetTableMetadata_v3 parquetTableMetadata_v3, Map<FileStatus, FileSystem> fileStatusMap) throws IOException {
+
+ List<TimedRunnable<ParquetFileMetadata_v3>> gatherers = fileStatusMap.entrySet().stream()
+ .map(e -> new MetadataGatherer(parquetTableMetadata_v3, e.getKey(), e.getValue()))
+ .collect(Collectors.toList());
+
+ List<ParquetFileMetadata_v3> metaDataList = new ArrayList<>();
+ metaDataList.addAll(TimedRunnable.run("Fetch parquet metadata", logger, gatherers, 16));
+ return metaDataList;
+ }
+
+ /**
+ * TimedRunnable that reads the footer from parquet and collects file metadata
+ */
+ private class MetadataGatherer extends TimedRunnable<ParquetFileMetadata_v3> {
+
+ private final ParquetTableMetadata_v3 parquetTableMetadata;
+ private final FileStatus fileStatus;
+ private final FileSystem fs;
+
+ MetadataGatherer(ParquetTableMetadata_v3 parquetTableMetadata, FileStatus fileStatus, FileSystem fs) {
+ this.parquetTableMetadata = parquetTableMetadata;
+ this.fileStatus = fileStatus;
+ this.fs = fs;
+ }
+
+ @Override
+ protected ParquetFileMetadata_v3 runInner() throws Exception {
+ return getParquetFileMetadata_v3(parquetTableMetadata, fileStatus, fs);
+ }
+
+ @Override
+ protected IOException convertToIOException(Exception e) {
+ if (e instanceof IOException) {
+ return (IOException) e;
+ } else {
+ return new IOException(e);
+ }
+ }
+ }
+
+ private ColTypeInfo getColTypeInfo(MessageType schema, Type type, String[] path, int depth) {
+ if (type.isPrimitive()) {
+ PrimitiveType primitiveType = (PrimitiveType) type;
+ int precision = 0;
+ int scale = 0;
+ if (primitiveType.getDecimalMetadata() != null) {
+ precision = primitiveType.getDecimalMetadata().getPrecision();
+ scale = primitiveType.getDecimalMetadata().getScale();
+ }
+
+ int repetitionLevel = schema.getMaxRepetitionLevel(path);
+ int definitionLevel = schema.getMaxDefinitionLevel(path);
+
+ return new ColTypeInfo(type.getOriginalType(), precision, scale, repetitionLevel, definitionLevel);
+ }
+ Type t = ((GroupType) type).getType(path[depth]);
+ return getColTypeInfo(schema, t, path, depth + 1);
+ }
+
+ private class ColTypeInfo {
+ public OriginalType originalType;
+ public int precision;
+ public int scale;
+ public int repetitionLevel;
+ public int definitionLevel;
+
+ ColTypeInfo(OriginalType originalType, int precision, int scale, int repetitionLevel, int definitionLevel) {
+ this.originalType = originalType;
+ this.precision = precision;
+ this.scale = scale;
+ this.repetitionLevel = repetitionLevel;
+ this.definitionLevel = definitionLevel;
+ }
+ }
+
+ /**
+ * Get the metadata for a single file
+ */
+ private ParquetFileMetadata_v3 getParquetFileMetadata_v3(ParquetTableMetadata_v3 parquetTableMetadata,
+ final FileStatus file, final FileSystem fs) throws IOException, InterruptedException {
+ final ParquetMetadata metadata;
+ final UserGroupInformation processUserUgi = ImpersonationUtil.getProcessUserUGI();
+ try {
+ metadata = processUserUgi.doAs((PrivilegedExceptionAction<ParquetMetadata>)
+ () -> ParquetFileReader.readFooter(fs.getConf(), file, ParquetMetadataConverter.NO_FILTER));
+ } catch(Exception e) {
+ logger.error("Exception while reading footer of parquet file [Details - path: {}, owner: {}] as process user {}",
+ file.getPath(), file.getOwner(), processUserUgi.getShortUserName(), e);
+ throw e;
+ }
+
+ MessageType schema = metadata.getFileMetaData().getSchema();
+
+// Map<SchemaPath, OriginalType> originalTypeMap = Maps.newHashMap();
+ Map<SchemaPath, ColTypeInfo> colTypeInfoMap = Maps.newHashMap();
+ schema.getPaths();
+ for (String[] path : schema.getPaths()) {
+ colTypeInfoMap.put(SchemaPath.getCompoundPath(path), getColTypeInfo(schema, schema, path, 0));
+ }
+
+ List<RowGroupMetadata_v3> rowGroupMetadataList = Lists.newArrayList();
+
+ ArrayList<SchemaPath> ALL_COLS = new ArrayList<>();
+ ALL_COLS.add(SchemaPath.STAR_COLUMN);
+ boolean autoCorrectCorruptDates = formatConfig.areCorruptDatesAutoCorrected();
+ ParquetReaderUtility.DateCorruptionStatus containsCorruptDates = ParquetReaderUtility.detectCorruptDates(metadata, ALL_COLS, autoCorrectCorruptDates);
+ if (logger.isDebugEnabled()) {
+ logger.debug(containsCorruptDates.toString());
+ }
+ for (BlockMetaData rowGroup : metadata.getBlocks()) {
+ List<ColumnMetadata_v3> columnMetadataList = Lists.newArrayList();
+ long length = 0;
+ for (ColumnChunkMetaData col : rowGroup.getColumns()) {
+ ColumnMetadata_v3 columnMetadata;
+
+ boolean statsAvailable = (col.getStatistics() != null && !col.getStatistics().isEmpty());
+
+ Statistics<?> stats = col.getStatistics();
+ String[] columnName = col.getPath().toArray();
+ SchemaPath columnSchemaName = SchemaPath.getCompoundPath(columnName);
+ ColTypeInfo colTypeInfo = colTypeInfoMap.get(columnSchemaName);
+
+ ColumnTypeMetadata_v3 columnTypeMetadata =
+ new ColumnTypeMetadata_v3(columnName, col.getType(), colTypeInfo.originalType,
+ colTypeInfo.precision, colTypeInfo.scale, colTypeInfo.repetitionLevel, colTypeInfo.definitionLevel);
+
+ if (parquetTableMetadata.columnTypeInfo == null) {
+ parquetTableMetadata.columnTypeInfo = new ConcurrentHashMap<>();
+ }
+ // Save the column schema info. We'll merge it into one list
+ parquetTableMetadata.columnTypeInfo
+ .put(new ColumnTypeMetadata_v3.Key(columnTypeMetadata.name), columnTypeMetadata);
+ if (statsAvailable) {
+ // Write stats when they are not null
+ Object minValue = null;
+ Object maxValue = null;
+ if (stats.genericGetMax() != null && stats.genericGetMin() != null ) {
+ minValue = stats.genericGetMin();
+ maxValue = stats.genericGetMax();
+ if (containsCorruptDates == ParquetReaderUtility.DateCorruptionStatus.META_SHOWS_CORRUPTION
+ && columnTypeMetadata.originalType == OriginalType.DATE) {
+ minValue = ParquetReaderUtility.autoCorrectCorruptedDate((Integer) minValue);
+ maxValue = ParquetReaderUtility.autoCorrectCorruptedDate((Integer) maxValue);
+ }
+
+ }
+ columnMetadata =
+ new ColumnMetadata_v3(columnTypeMetadata.name, col.getType(), minValue, maxValue, stats.getNumNulls());
+ } else {
+ columnMetadata = new ColumnMetadata_v3(columnTypeMetadata.name, col.getType(), null, null, null);
+ }
+ columnMetadataList.add(columnMetadata);
+ length += col.getTotalSize();
+ }
+
+ // DRILL-5009: Skip the RowGroup if it is empty
+ // Note we still read the schema even if there are no values in the RowGroup
+ if (rowGroup.getRowCount() == 0) {
+ continue;
+ }
+ RowGroupMetadata_v3 rowGroupMeta =
+ new RowGroupMetadata_v3(rowGroup.getStartingPos(), length, rowGroup.getRowCount(),
+ getHostAffinity(file, fs, rowGroup.getStartingPos(), length), columnMetadataList);
+
+ rowGroupMetadataList.add(rowGroupMeta);
+ }
+ String path = Path.getPathWithoutSchemeAndAuthority(file.getPath()).toString();
+
+ return new ParquetFileMetadata_v3(path, file.getLen(), rowGroupMetadataList);
+ }
+
+ /**
+ * Get the host affinity for a row group.
+ *
+ * @param fileStatus the parquet file
+ * @param start the start of the row group
+ * @param length the length of the row group
+ * @return host affinity for the row group
+ */
+ private Map<String, Float> getHostAffinity(FileStatus fileStatus, FileSystem fs, long start, long length)
+ throws IOException {
+ BlockLocation[] blockLocations = fs.getFileBlockLocations(fileStatus, start, length);
+ Map<String, Float> hostAffinityMap = Maps.newHashMap();
+ for (BlockLocation blockLocation : blockLocations) {
+ for (String host : blockLocation.getHosts()) {
+ Float currentAffinity = hostAffinityMap.get(host);
+ float blockStart = blockLocation.getOffset();
+ float blockEnd = blockStart + blockLocation.getLength();
+ float rowGroupEnd = start + length;
+ Float newAffinity = (blockLocation.getLength() - (blockStart < start ? start - blockStart : 0) -
+ (blockEnd > rowGroupEnd ? blockEnd - rowGroupEnd : 0)) / length;
+ if (currentAffinity != null) {
+ hostAffinityMap.put(host, currentAffinity + newAffinity);
+ } else {
+ hostAffinityMap.put(host, newAffinity);
+ }
+ }
+ }
+ return hostAffinityMap;
+ }
+
+ /**
+ * Serialize parquet metadata to json and write to a file.
+ *
+ * @param parquetTableMetadata parquet table metadata
+ * @param p file path
+ */
+ private void writeFile(ParquetTableMetadata_v3 parquetTableMetadata, Path p, FileSystem fs) throws IOException {
+ JsonFactory jsonFactory = new JsonFactory();
+ jsonFactory.configure(Feature.AUTO_CLOSE_TARGET, false);
+ jsonFactory.configure(JsonParser.Feature.AUTO_CLOSE_SOURCE, false);
+ ObjectMapper mapper = new ObjectMapper(jsonFactory);
+ SimpleModule module = new SimpleModule();
+ module.addSerializer(ColumnMetadata_v3.class, new ColumnMetadata_v3.Serializer());
+ mapper.registerModule(module);
+ FSDataOutputStream os = fs.create(p);
+ mapper.writerWithDefaultPrettyPrinter().writeValue(os, parquetTableMetadata);
+ os.flush();
+ os.close();
+ }
+
+ private void writeFile(ParquetTableMetadataDirs parquetTableMetadataDirs, Path p, FileSystem fs) throws IOException {
+ JsonFactory jsonFactory = new JsonFactory();
+ jsonFactory.configure(Feature.AUTO_CLOSE_TARGET, false);
+ jsonFactory.configure(JsonParser.Feature.AUTO_CLOSE_SOURCE, false);
+ ObjectMapper mapper = new ObjectMapper(jsonFactory);
+ SimpleModule module = new SimpleModule();
+ mapper.registerModule(module);
+ FSDataOutputStream os = fs.create(p);
+ mapper.writerWithDefaultPrettyPrinter().writeValue(os, parquetTableMetadataDirs);
+ os.flush();
+ os.close();
+ }
+
+ /**
+ * Read the parquet metadata from a file
+ *
+ * @param path to metadata file
+ * @param dirsOnly true for {@link Metadata#METADATA_DIRECTORIES_FILENAME}
+ * or false for {@link Metadata#METADATA_FILENAME} files reading
+ * @param metaContext current metadata context
+ */
+ private void readBlockMeta(Path path, boolean dirsOnly, MetadataContext metaContext, FileSystem fs) {
+ Stopwatch timer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
+ Path metadataParentDir = Path.getPathWithoutSchemeAndAuthority(path.getParent());
+ String metadataParentDirPath = metadataParentDir.toUri().getPath();
+ ObjectMapper mapper = new ObjectMapper();
+
+ final SimpleModule serialModule = new SimpleModule();
+ serialModule.addDeserializer(SchemaPath.class, new SchemaPath.De());
+ serialModule.addKeyDeserializer(Metadata_V2.ColumnTypeMetadata_v2.Key.class, new Metadata_V2.ColumnTypeMetadata_v2.Key.DeSerializer());
+ serialModule.addKeyDeserializer(ColumnTypeMetadata_v3.Key.class, new ColumnTypeMetadata_v3.Key.DeSerializer());
+
+ AfterburnerModule module = new AfterburnerModule();
+ module.setUseOptimizedBeanDeserializer(true);
+
+ mapper.registerModule(serialModule);
+ mapper.registerModule(module);
+ mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+ try (FSDataInputStream is = fs.open(path)) {
+ boolean alreadyCheckedModification;
+ boolean newMetadata = false;
+ alreadyCheckedModification = metaContext.getStatus(metadataParentDirPath);
+
+ if (dirsOnly) {
+ parquetTableMetadataDirs = mapper.readValue(is, ParquetTableMetadataDirs.class);
+ if (timer != null) {
+ logger.debug("Took {} ms to read directories from directory cache file", timer.elapsed(TimeUnit.MILLISECONDS));
+ timer.stop();
+ }
+ parquetTableMetadataDirs.updateRelativePaths(metadataParentDirPath);
+ if (!alreadyCheckedModification && tableModified(parquetTableMetadataDirs.getDirectories(), path, metadataParentDir, metaContext, fs)) {
+ parquetTableMetadataDirs =
+ (createMetaFilesRecursively(Path.getPathWithoutSchemeAndAuthority(path.getParent()).toString(), fs)).getRight();
+ newMetadata = true;
+ }
+ } else {
+ parquetTableMetadata = mapper.readValue(is, ParquetTableMetadataBase.class);
+ if (timer != null) {
+ logger.debug("Took {} ms to read metadata from cache file", timer.elapsed(TimeUnit.MILLISECONDS));
+ timer.stop();
+ }
+ if (new MetadataVersion(parquetTableMetadata.getMetadataVersion()).compareTo(new MetadataVersion(3, 0)) >= 0) {
+ ((ParquetTableMetadata_v3) parquetTableMetadata).updateRelativePaths(metadataParentDirPath);
+ }
+ if (!alreadyCheckedModification && tableModified(parquetTableMetadata.getDirectories(), path, metadataParentDir, metaContext, fs)) {
+ parquetTableMetadata =
+ (createMetaFilesRecursively(Path.getPathWithoutSchemeAndAuthority(path.getParent()).toString(), fs)).getLeft();
+ newMetadata = true;
+ }
+
+ // DRILL-5009: Remove the RowGroup if it is empty
+ List<? extends ParquetFileMetadata> files = parquetTableMetadata.getFiles();
+ for (ParquetFileMetadata file : files) {
+ List<? extends RowGroupMetadata> rowGroups = file.getRowGroups();
+ for (Iterator<? extends RowGroupMetadata> iter = rowGroups.iterator(); iter.hasNext(); ) {
+ RowGroupMetadata r = iter.next();
+ if (r.getRowCount() == 0) {
+ iter.remove();
+ }
+ }
+ }
+
+ }
+ if (newMetadata) {
+ // if new metadata files were created, invalidate the existing metadata context
+ metaContext.clear();
+ }
+ } catch (IOException e) {
+ logger.error("Failed to read '{}' metadata file", path, e);
+ metaContext.setMetadataCacheCorrupted(true);
+ }
+ }
+
+ /**
+ * Check if the parquet metadata needs to be updated by comparing the modification time of the directories with
+ * the modification time of the metadata file
+ *
+ * @param directories List of directories
+ * @param metaFilePath path of parquet metadata cache file
+ * @return true if metadata needs to be updated, false otherwise
+ * @throws IOException if some resources are not accessible
+ */
+ private boolean tableModified(List<String> directories, Path metaFilePath, Path parentDir, MetadataContext metaContext, FileSystem fs) throws IOException {
+ Stopwatch timer = logger.isDebugEnabled() ? Stopwatch.createStarted() : null;
+ metaContext.setStatus(parentDir.toUri().getPath());
+ long metaFileModifyTime = fs.getFileStatus(metaFilePath).getModificationTime();
+ FileStatus directoryStatus = fs.getFileStatus(parentDir);
+ int numDirs = 1;
+ if (directoryStatus.getModificationTime() > metaFileModifyTime) {
+ if (timer != null) {
+ logger.debug("Directory {} was modified. Took {} ms to check modification time of {} directories",
+ directoryStatus.getPath().toString(), timer.elapsed(TimeUnit.MILLISECONDS), numDirs);
+ timer.stop();
+ }
+ return true;
+ }
+ for (String directory : directories) {
+ numDirs++;
+ metaContext.setStatus(directory);
+ directoryStatus = fs.getFileStatus(new Path(directory));
+ if (directoryStatus.getModificationTime() > metaFileModifyTime) {
+ if (timer != null) {
+ logger.debug("Directory {} was modified. Took {} ms to check modification time of {} directories",
+ directoryStatus.getPath().toString(), timer.elapsed(TimeUnit.MILLISECONDS), numDirs);
+ timer.stop();
+ }
+ return true;
+ }
+ }
+ if (timer != null) {
+ logger.debug("No directories were modified. Took {} ms to check modification time of {} directories",
+ timer.elapsed(TimeUnit.MILLISECONDS), numDirs);
+ timer.stop();
+ }
+ return false;
+ }
+
+}
+
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataBase.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataBase.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataBase.java
new file mode 100644
index 0000000..d7d56c3
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataBase.java
@@ -0,0 +1,142 @@
+/*
+ * 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.metadata;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonSubTypes;
+import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.drill.exec.store.parquet.metadata.MetadataVersion.Constants.V1;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataVersion.Constants.V2;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataVersion.Constants.V3;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataVersion.Constants.V3_1;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataVersion.Constants.V3_2;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataVersion.Constants.V3_3;
+
+public class MetadataBase {
+
+ /**
+ * Basic class for parquet metadata. Inheritors of this class are json serializable structures which contain
+ * different metadata versions for an entire parquet directory structure
+ * <p>
+ * If any new code changes affect on the metadata files content, please update metadata version in such manner:
+ * Bump up metadata major version if metadata structure is changed.
+ * Bump up metadata minor version if only metadata content is changed, but metadata structure is the same.
+ * <p>
+ * Note: keep metadata versions synchronized with {@link MetadataVersion.Constants}
+ */
+ @JsonTypeInfo(use = JsonTypeInfo.Id.NAME,
+ include = JsonTypeInfo.As.PROPERTY,
+ property = "metadata_version",
+ visible = true)
+ @JsonSubTypes({
+ @JsonSubTypes.Type(value = Metadata_V1.ParquetTableMetadata_v1.class, name = V1),
+ @JsonSubTypes.Type(value = Metadata_V2.ParquetTableMetadata_v2.class, name = V2),
+ @JsonSubTypes.Type(value = Metadata_V3.ParquetTableMetadata_v3.class, name = V3),
+ @JsonSubTypes.Type(value = Metadata_V3.ParquetTableMetadata_v3.class, name = V3_1),
+ @JsonSubTypes.Type(value = Metadata_V3.ParquetTableMetadata_v3.class, name = V3_2),
+ @JsonSubTypes.Type(value = Metadata_V3.ParquetTableMetadata_v3.class, name = V3_3)
+ })
+ public static abstract class ParquetTableMetadataBase {
+
+ @JsonIgnore
+ public abstract List<String> getDirectories();
+
+ @JsonIgnore public abstract List<? extends ParquetFileMetadata> getFiles();
+
+ @JsonIgnore public abstract void assignFiles(List<? extends ParquetFileMetadata> newFiles);
+
+ public abstract boolean hasColumnMetadata();
+
+ @JsonIgnore public abstract PrimitiveType.PrimitiveTypeName getPrimitiveType(String[] columnName);
+
+ @JsonIgnore public abstract OriginalType getOriginalType(String[] columnName);
+
+ @JsonIgnore public abstract Integer getRepetitionLevel(String[] columnName);
+
+ @JsonIgnore public abstract Integer getDefinitionLevel(String[] columnName);
+
+ @JsonIgnore public abstract boolean isRowGroupPrunable();
+
+ @JsonIgnore public abstract ParquetTableMetadataBase clone();
+
+ @JsonIgnore public abstract String getDrillVersion();
+
+ @JsonIgnore public abstract String getMetadataVersion();
+ }
+
+ public static abstract class ParquetFileMetadata {
+ @JsonIgnore public abstract String getPath();
+
+ @JsonIgnore public abstract Long getLength();
+
+ @JsonIgnore public abstract List<? extends RowGroupMetadata> getRowGroups();
+ }
+
+
+ public static abstract class RowGroupMetadata {
+ @JsonIgnore public abstract Long getStart();
+
+ @JsonIgnore public abstract Long getLength();
+
+ @JsonIgnore public abstract Long getRowCount();
+
+ @JsonIgnore public abstract Map<String, Float> getHostAffinity();
+
+ @JsonIgnore public abstract List<? extends ColumnMetadata> getColumns();
+ }
+
+
+ public static abstract class ColumnMetadata {
+ public abstract String[] getName();
+
+ public abstract Long getNulls();
+
+ public abstract boolean hasSingleValue(long rowCount);
+
+ public abstract Object getMinValue();
+
+ public abstract Object getMaxValue();
+
+ /**
+ * Set the max value recorded in the parquet metadata statistics.
+ *
+ * This object would just be immutable, but due to Drill-4203 we need to correct
+ * date values that had been corrupted by earlier versions of Drill.
+ */
+ public abstract void setMax(Object newMax);
+
+ /**
+ * Set the min value recorded in the parquet metadata statistics.
+ *
+ * This object would just be immutable, but due to Drill-4203 we need to correct
+ * date values that had been corrupted by earlier versions of Drill.
+ */
+ public abstract void setMin(Object newMax);
+
+ public abstract PrimitiveType.PrimitiveTypeName getPrimitiveType();
+
+ public abstract OriginalType getOriginalType();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataPathUtils.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataPathUtils.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataPathUtils.java
new file mode 100644
index 0000000..b9480e8
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataPathUtils.java
@@ -0,0 +1,122 @@
+/*
+ * 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.metadata;
+
+import com.google.common.collect.Lists;
+import org.apache.drill.common.util.DrillVersionInfo;
+import org.apache.hadoop.fs.Path;
+
+import java.util.List;
+
+import static org.apache.drill.exec.store.parquet.metadata.MetadataVersion.Constants.SUPPORTED_VERSIONS;
+import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ParquetFileMetadata_v3;
+import static org.apache.drill.exec.store.parquet.metadata.Metadata_V3.ParquetTableMetadata_v3;
+
+/**
+ * Util class that contains helper methods for converting paths in the table and directory metadata structures
+ */
+public class MetadataPathUtils {
+
+ /**
+ * Helper method that converts a list of relative paths to absolute ones
+ *
+ * @param paths list of relative paths
+ * @param baseDir base parent directory
+ * @return list of absolute paths
+ */
+ public static List<String> convertToAbsolutePaths(List<String> paths, String baseDir) {
+ if (!paths.isEmpty()) {
+ List<String> absolutePaths = Lists.newArrayList();
+ for (String relativePath : paths) {
+ String absolutePath = (new Path(relativePath).isAbsolute()) ? relativePath
+ : new Path(baseDir, relativePath).toUri().getPath();
+ absolutePaths.add(absolutePath);
+ }
+ return absolutePaths;
+ }
+ return paths;
+ }
+
+ /**
+ * Convert a list of files with relative paths to files with absolute ones
+ *
+ * @param files list of files with relative paths
+ * @param baseDir base parent directory
+ * @return list of files with absolute paths
+ */
+ public static List<ParquetFileMetadata_v3> convertToFilesWithAbsolutePaths(
+ List<ParquetFileMetadata_v3> files, String baseDir) {
+ if (!files.isEmpty()) {
+ List<ParquetFileMetadata_v3> filesWithAbsolutePaths = Lists.newArrayList();
+ for (ParquetFileMetadata_v3 file : files) {
+ Path relativePath = new Path(file.getPath());
+ // create a new file if old one contains a relative path, otherwise use an old file
+ ParquetFileMetadata_v3 fileWithAbsolutePath = (relativePath.isAbsolute()) ? file
+ : new ParquetFileMetadata_v3(new Path(baseDir, relativePath).toUri().getPath(), file.length, file.rowGroups);
+ filesWithAbsolutePaths.add(fileWithAbsolutePath);
+ }
+ return filesWithAbsolutePaths;
+ }
+ return files;
+ }
+
+ /**
+ * Creates a new parquet table metadata from the {@code tableMetadataWithAbsolutePaths} parquet table.
+ * A new parquet table will contain relative paths for the files and directories.
+ *
+ * @param tableMetadataWithAbsolutePaths parquet table metadata with absolute paths for the files and directories
+ * @param baseDir base parent directory
+ * @return parquet table metadata with relative paths for the files and directories
+ */
+ public static ParquetTableMetadata_v3 createMetadataWithRelativePaths(
+ ParquetTableMetadata_v3 tableMetadataWithAbsolutePaths, String baseDir) {
+ List<String> directoriesWithRelativePaths = Lists.newArrayList();
+ for (String directory : tableMetadataWithAbsolutePaths.getDirectories()) {
+ directoriesWithRelativePaths.add(relativize(baseDir, directory)) ;
+ }
+ List<ParquetFileMetadata_v3> filesWithRelativePaths = Lists.newArrayList();
+ for (ParquetFileMetadata_v3 file : tableMetadataWithAbsolutePaths.files) {
+ filesWithRelativePaths.add(new ParquetFileMetadata_v3(
+ relativize(baseDir, file.getPath()), file.length, file.rowGroups));
+ }
+ return new ParquetTableMetadata_v3(SUPPORTED_VERSIONS.last().toString(), tableMetadataWithAbsolutePaths,
+ filesWithRelativePaths, directoriesWithRelativePaths, DrillVersionInfo.getVersion());
+ }
+
+ /**
+ * Constructs relative path from child full path and base path. Or return child path if the last one is already relative
+ *
+ * @param childPath full absolute path
+ * @param baseDir base path (the part of the Path, which should be cut off from child path)
+ * @return relative path
+ */
+ public static String relativize(String baseDir, String childPath) {
+ Path fullPathWithoutSchemeAndAuthority = Path.getPathWithoutSchemeAndAuthority(new Path(childPath));
+ Path basePathWithoutSchemeAndAuthority = Path.getPathWithoutSchemeAndAuthority(new Path(baseDir));
+
+ // Since hadoop Path hasn't relativize() we use uri.relativize() to get relative path
+ Path relativeFilePath = new Path(basePathWithoutSchemeAndAuthority.toUri()
+ .relativize(fullPathWithoutSchemeAndAuthority.toUri()));
+ if (relativeFilePath.isAbsolute()) {
+ throw new IllegalStateException(String.format("Path %s is not a subpath of %s.",
+ basePathWithoutSchemeAndAuthority.toUri().getPath(), fullPathWithoutSchemeAndAuthority.toUri().getPath()));
+ }
+ return relativeFilePath.toUri().getPath();
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataVersion.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataVersion.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataVersion.java
new file mode 100644
index 0000000..15b4b9d
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/MetadataVersion.java
@@ -0,0 +1,169 @@
+/*
+ * 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.metadata;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.ComparisonChain;
+import com.google.common.collect.ImmutableSortedSet;
+import org.apache.drill.common.exceptions.DrillRuntimeException;
+
+import java.util.SortedSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+public class MetadataVersion implements Comparable<MetadataVersion> {
+
+ private static final String FORMAT = "v?((?!0)\\d+)(\\.(\\d+))?";
+ private static final Pattern PATTERN = Pattern.compile(FORMAT);
+
+ private final int major;
+ private final int minor;
+
+ public MetadataVersion(int major, int minor) {
+ this.major = major;
+ this.minor = minor;
+ }
+
+ public MetadataVersion(String metadataVersion) {
+ Matcher matcher = PATTERN.matcher(metadataVersion);
+ if (!matcher.matches()) {
+ DrillRuntimeException.format("Could not parse metadata version '%s' using format '%s'", metadataVersion, FORMAT);
+ }
+ this.major = Integer.parseInt(matcher.group(1));
+ this.minor = matcher.group(3) != null ? Integer.parseInt(matcher.group(3)) : 0;
+ }
+
+ public int getMajor() {
+ return major;
+ }
+
+ public int getMinor() {
+ return minor;
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (!(o instanceof MetadataVersion)) {
+ return false;
+ }
+ MetadataVersion that = (MetadataVersion) o;
+ return this.major == that.major
+ && this.minor == that.minor;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = major;
+ result = 31 * result + minor;
+ return result;
+ }
+
+ /**
+ * @return string representation of the metadata file version, for example: "1", "10", "4.13"
+ * <p>
+ * String metadata version consists of the following characters:<p>
+ * major metadata version (any number of digits, except a single zero digit),<p>
+ * optional "." delimiter (used if minor metadata version is specified),<p>
+ * minor metadata version (not specified for "0" minor version)
+ */
+ @Override
+ public String toString() {
+ StringBuilder builder = new StringBuilder();
+ builder.append(major);
+ if (minor != 0) {
+ builder.append(".").append(minor);
+ }
+ return builder.toString();
+ }
+
+ @Override
+ public int compareTo(MetadataVersion o) {
+ Preconditions.checkNotNull(o);
+ return ComparisonChain.start()
+ .compare(this.major, o.major)
+ .compare(this.minor, o.minor)
+ .result();
+ }
+
+/**
+ * Supported metadata versions.
+ * <p>
+ * Note: keep them synchronized with {@link Metadata.ParquetTableMetadataBase} versions
+ */
+ public static class Constants {
+ /**
+ * Version 1: Introduces parquet file metadata caching.<br>
+ * See DRILL-2743
+ */
+ public static final String V1 = "v1";
+ /**
+ * Version 2: Metadata cache file size is reduced.<br>
+ * See DRILL-4053
+ */
+ public static final String V2 = "v2";
+ /**
+ * Version 3: Difference between v3 and v2 : min/max, type_length, precision, scale, repetitionLevel, definitionLevel.<br>
+ * Filter pushdown for Parquet is implemented. <br>
+ * See DRILL-1950
+ */
+ public static final String V3 = "v3";
+ /**
+ * Version 3.1: Absolute paths of files and directories are replaced with relative ones. Metadata version value
+ * doesn't contain `v` letter<br>
+ * See DRILL-3867, DRILL-5660
+ */
+ public static final String V3_1 = "3.1";
+
+ /**
+ * Version 3.2: An array with the components of the field name in
+ * {@link Metadata.ColumnTypeMetadata_v3.Key} class is replaced by the SchemaPath.<br>
+ * See DRILL-4264
+ */
+ public static final String V3_2 = "3.2";
+
+ /**
+ * Version 3.3: Changed serialization of BINARY and FIXED_LEN_BYTE_ARRAY fields.<br>
+ * See DRILL-4139
+ */
+ public static final String V3_3 = "3.3";
+
+ /**
+ * All historical versions of the Drill metadata cache files. In case of introducing a new parquet metadata version
+ * please follow the {@link MetadataVersion#FORMAT}.
+ */
+ public static final SortedSet<MetadataVersion> SUPPORTED_VERSIONS = ImmutableSortedSet.of(
+ new MetadataVersion(V1),
+ new MetadataVersion(V2),
+ new MetadataVersion(V3),
+ new MetadataVersion(V3_1),
+ new MetadataVersion(V3_2),
+ new MetadataVersion(V3_3)
+ );
+
+ /**
+ * @param metadataVersion string representation of the parquet metadata version
+ * @return true if metadata version is supported, false otherwise
+ */
+ public static boolean isVersionSupported(String metadataVersion) {
+ return SUPPORTED_VERSIONS.contains(new MetadataVersion(metadataVersion));
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/c6549e58/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata_V1.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata_V1.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata_V1.java
new file mode 100644
index 0000000..92feb5f
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/metadata/Metadata_V1.java
@@ -0,0 +1,329 @@
+/*
+ * 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.metadata;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.parquet.io.api.Binary;
+import org.apache.parquet.schema.OriginalType;
+import org.apache.parquet.schema.PrimitiveType;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+import static org.apache.drill.exec.store.parquet.metadata.MetadataVersion.Constants.V1;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ColumnMetadata;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ParquetFileMetadata;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.ParquetTableMetadataBase;
+import static org.apache.drill.exec.store.parquet.metadata.MetadataBase.RowGroupMetadata;
+
+
+public class Metadata_V1 {
+
+ @JsonTypeName(V1)
+ public static class ParquetTableMetadata_v1 extends ParquetTableMetadataBase {
+ @JsonProperty(value = "metadata_version", access = JsonProperty.Access.WRITE_ONLY) private String metadataVersion;
+ @JsonProperty
+ List<ParquetFileMetadata_v1> files;
+ @JsonProperty List<String> directories;
+
+ public ParquetTableMetadata_v1() {
+ }
+
+ public ParquetTableMetadata_v1(String metadataVersion, List<ParquetFileMetadata_v1> files, List<String> directories) {
+ this.metadataVersion = metadataVersion;
+ this.files = files;
+ this.directories = directories;
+ }
+
+ @JsonIgnore
+ @Override public List<String> getDirectories() {
+ return directories;
+ }
+
+ @JsonIgnore @Override public List<? extends ParquetFileMetadata> getFiles() {
+ return files;
+ }
+
+ @JsonIgnore @Override public void assignFiles(List<? extends ParquetFileMetadata> newFiles) {
+ this.files = (List<ParquetFileMetadata_v1>) newFiles;
+ }
+
+ @Override public boolean hasColumnMetadata() {
+ return false;
+ }
+
+ @JsonIgnore @Override public PrimitiveType.PrimitiveTypeName getPrimitiveType(String[] columnName) {
+ return null;
+ }
+
+ @JsonIgnore @Override public OriginalType getOriginalType(String[] columnName) {
+ return null;
+ }
+
+ @JsonIgnore @Override
+ public Integer getRepetitionLevel(String[] columnName) {
+ return null;
+ }
+
+ @JsonIgnore @Override
+ public Integer getDefinitionLevel(String[] columnName) {
+ return null;
+ }
+
+ @JsonIgnore @Override
+ public boolean isRowGroupPrunable() {
+ return false;
+ }
+
+ @JsonIgnore @Override public MetadataBase.ParquetTableMetadataBase clone() {
+ return new ParquetTableMetadata_v1(metadataVersion, files, directories);
+ }
+
+ @JsonIgnore @Override
+ public String getDrillVersion() {
+ return null;
+ }
+
+ @JsonIgnore @Override public String getMetadataVersion() {
+ return metadataVersion;
+ }
+ }
+
+
+ /**
+ * Struct which contains the metadata for a single parquet file
+ */
+ public static class ParquetFileMetadata_v1 extends ParquetFileMetadata {
+ @JsonProperty
+ public String path;
+ @JsonProperty
+ public Long length;
+ @JsonProperty
+ public List<RowGroupMetadata_v1> rowGroups;
+
+ public ParquetFileMetadata_v1() {
+ }
+
+ public ParquetFileMetadata_v1(String path, Long length, List<RowGroupMetadata_v1> rowGroups) {
+ this.path = path;
+ this.length = length;
+ this.rowGroups = rowGroups;
+ }
+
+ @Override
+ public String toString() {
+ return String.format("path: %s rowGroups: %s", path, rowGroups);
+ }
+
+ @JsonIgnore @Override public String getPath() {
+ return path;
+ }
+
+ @JsonIgnore @Override public Long getLength() {
+ return length;
+ }
+
+ @JsonIgnore @Override public List<? extends RowGroupMetadata> getRowGroups() {
+ return rowGroups;
+ }
+ }
+
+
+ /**
+ * A struct that contains the metadata for a parquet row group
+ */
+ public static class RowGroupMetadata_v1 extends RowGroupMetadata {
+ @JsonProperty
+ public Long start;
+ @JsonProperty
+ public Long length;
+ @JsonProperty
+ public Long rowCount;
+ @JsonProperty
+ public Map<String, Float> hostAffinity;
+ @JsonProperty
+ public List<ColumnMetadata_v1> columns;
+
+ public RowGroupMetadata_v1() {
+ }
+
+ public RowGroupMetadata_v1(Long start, Long length, Long rowCount, Map<String, Float> hostAffinity,
+ List<ColumnMetadata_v1> columns) {
+ this.start = start;
+ this.length = length;
+ this.rowCount = rowCount;
+ this.hostAffinity = hostAffinity;
+ this.columns = columns;
+ }
+
+ @Override public Long getStart() {
+ return start;
+ }
+
+ @Override public Long getLength() {
+ return length;
+ }
+
+ @Override public Long getRowCount() {
+ return rowCount;
+ }
+
+ @Override public Map<String, Float> getHostAffinity() {
+ return hostAffinity;
+ }
+
+ @Override public List<? extends ColumnMetadata> getColumns() {
+ return columns;
+ }
+ }
+
+
+ /**
+ * A struct that contains the metadata for a column in a parquet file
+ */
+ public static class ColumnMetadata_v1 extends ColumnMetadata {
+ @JsonProperty
+ public SchemaPath name;
+ @JsonProperty
+ public PrimitiveType.PrimitiveTypeName primitiveType;
+ @JsonProperty
+ public OriginalType originalType;
+ @JsonProperty
+ public Long nulls;
+
+ // JsonProperty for these are associated with the getters and setters
+ public Object max;
+ public Object min;
+
+
+ public ColumnMetadata_v1() {
+ }
+
+ public ColumnMetadata_v1(SchemaPath name, PrimitiveType.PrimitiveTypeName primitiveType, OriginalType originalType,
+ Object max, Object min, Long nulls) {
+ this.name = name;
+ this.primitiveType = primitiveType;
+ this.originalType = originalType;
+ this.max = max;
+ this.min = min;
+ this.nulls = nulls;
+ }
+
+ @JsonProperty(value = "min")
+ public Object getMin() {
+ if (primitiveType == PrimitiveType.PrimitiveTypeName.BINARY && min != null) {
+ return new String(((Binary) min).getBytes());
+ }
+ return min;
+ }
+
+ @JsonProperty(value = "max")
+ public Object getMax() {
+ if (primitiveType == PrimitiveType.PrimitiveTypeName.BINARY && max != null) {
+ return new String(((Binary) max).getBytes());
+ }
+ return max;
+ }
+
+ @Override public PrimitiveType.PrimitiveTypeName getPrimitiveType() {
+ return primitiveType;
+ }
+
+ @Override public OriginalType getOriginalType() {
+ return originalType;
+ }
+
+ /**
+ * setter used during deserialization of the 'min' field of the metadata cache file.
+ *
+ * @param min
+ */
+ @JsonProperty(value = "min")
+ public void setMin(Object min) {
+ this.min = min;
+ }
+
+ /**
+ * setter used during deserialization of the 'max' field of the metadata cache file.
+ *
+ * @param max
+ */
+ @JsonProperty(value = "max")
+ public void setMax(Object max) {
+ this.max = max;
+ }
+
+ @Override public String[] getName() {
+ String[] s = new String[1];
+ String nameString = name.toString();
+ // Strip out the surrounding backticks.
+ s[0]=nameString.substring(1, nameString.length()-1);
+ return s;
+ }
+
+ @Override public Long getNulls() {
+ return nulls;
+ }
+
+ /**
+ * Checks that the column chunk has a single value.
+ * Returns {@code true} if {@code min} and {@code max} are the same but not null
+ * and nulls count is 0 or equal to the rows count.
+ * <p>
+ * Returns {@code true} if {@code min} and {@code max} are null and the number of null values
+ * in the column chunk is equal to the rows count.
+ * <p>
+ * Comparison of nulls and rows count is needed for the cases:
+ * <ul>
+ * <li>column with primitive type has single value and null values</li>
+ *
+ * <li>column <b>with primitive type</b> has only null values, min/max couldn't be null,
+ * but column has single value</li>
+ * </ul>
+ *
+ * @param rowCount rows count in column chunk
+ * @return true if column has single value
+ */
+ @Override
+ public boolean hasSingleValue(long rowCount) {
+ if (nulls != null) {
+ if (min != null) {
+ // Objects.deepEquals() is used here, since min and max may be byte arrays
+ return Objects.deepEquals(min, max) && (nulls == 0 || nulls == rowCount);
+ } else {
+ return nulls == rowCount && max == null;
+ }
+ }
+ return false;
+ }
+
+ @Override public Object getMinValue() {
+ return min;
+ }
+
+ @Override public Object getMaxValue() {
+ return max;
+ }
+
+ }
+
+}