DRILL-8353: Format plugin for Delta Lake (#2702)
diff --git a/contrib/format-deltalake/README.md b/contrib/format-deltalake/README.md
new file mode 100644
index 0000000..e4cbaef
--- /dev/null
+++ b/contrib/format-deltalake/README.md
@@ -0,0 +1,36 @@
+# Delta Lake format plugin
+
+This format plugin enables Drill to query Delta Lake tables.
+
+## Supported optimizations and features
+
+### Project pushdown
+
+This format plugin supports project and filter pushdown optimizations.
+
+For the case of project pushdown, only columns specified in the query will be read, even when they are nested columns.
+
+### Filter pushdown
+
+For the case of filter pushdown, all expressions supported by Delta Lake API will be pushed down, so only data that
+matches the filter expression will be read. Additionally, filtering logic for parquet files is enabled
+to allow pruning of parquet files that do not match the filter expression.
+
+## Configuration
+
+The format plugin has the following configuration options:
+
+- `type` - format plugin type, should be `'delta'`
+
+### Format config example:
+
+```json
+{
+ "type": "file",
+ "formats": {
+ "delta": {
+ "type": "delta"
+ }
+ }
+}
+```
diff --git a/contrib/format-deltalake/pom.xml b/contrib/format-deltalake/pom.xml
new file mode 100644
index 0000000..5f0e31f
--- /dev/null
+++ b/contrib/format-deltalake/pom.xml
@@ -0,0 +1,73 @@
+<?xml version="1.0"?>
+<!--
+
+ 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.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <artifactId>drill-contrib-parent</artifactId>
+ <groupId>org.apache.drill.contrib</groupId>
+ <version>2.0.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>drill-deltalake-format</artifactId>
+
+ <name>Drill : Contrib : Format : Delta Lake</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.drill.exec</groupId>
+ <artifactId>drill-java-exec</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-storage</artifactId>
+ <version>2.1.0</version>
+ </dependency>
+ <dependency>
+ <groupId>io.delta</groupId>
+ <artifactId>delta-standalone_2.13</artifactId>
+ <version>0.5.0</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.parquet</groupId>
+ <artifactId>parquet-avro</artifactId>
+ <version>${parquet.version}</version>
+ </dependency>
+
+ <!-- Test dependency -->
+ <dependency>
+ <groupId>org.apache.drill.exec</groupId>
+ <artifactId>drill-java-exec</artifactId>
+ <classifier>tests</classifier>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.drill</groupId>
+ <artifactId>drill-common</artifactId>
+ <classifier>tests</classifier>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+</project>
diff --git a/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/DeltaGroupScan.java b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/DeltaGroupScan.java
new file mode 100644
index 0000000..2130bea
--- /dev/null
+++ b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/DeltaGroupScan.java
@@ -0,0 +1,579 @@
+/*
+ * 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.delta;
+
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import io.delta.standalone.DeltaLog;
+import io.delta.standalone.DeltaScan;
+import io.delta.standalone.Snapshot;
+import io.delta.standalone.actions.AddFile;
+import io.delta.standalone.data.CloseableIterator;
+import io.delta.standalone.types.ArrayType;
+import io.delta.standalone.types.BinaryType;
+import io.delta.standalone.types.BooleanType;
+import io.delta.standalone.types.ByteType;
+import io.delta.standalone.types.DataType;
+import io.delta.standalone.types.DateType;
+import io.delta.standalone.types.DecimalType;
+import io.delta.standalone.types.DoubleType;
+import io.delta.standalone.types.FloatType;
+import io.delta.standalone.types.IntegerType;
+import io.delta.standalone.types.LongType;
+import io.delta.standalone.types.ShortType;
+import io.delta.standalone.types.StringType;
+import io.delta.standalone.types.StructField;
+import io.delta.standalone.types.StructType;
+import io.delta.standalone.types.TimestampType;
+import org.apache.calcite.util.Pair;
+import org.apache.drill.common.PlanStringBuilder;
+import org.apache.drill.common.exceptions.DrillRuntimeException;
+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.common.types.TypeProtos;
+import org.apache.drill.exec.expr.StatisticsProvider;
+import org.apache.drill.exec.expr.fn.impl.DateUtility;
+import org.apache.drill.exec.metastore.MetadataProviderManager;
+import org.apache.drill.exec.metastore.store.FileSystemMetadataProviderManager;
+import org.apache.drill.exec.physical.EndpointAffinity;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.record.metadata.ColumnMetadata;
+import org.apache.drill.exec.record.metadata.MetadataUtils;
+import org.apache.drill.exec.record.metadata.TupleBuilder;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.delta.format.DeltaFormatPlugin;
+import org.apache.drill.exec.store.delta.plan.DrillExprToDeltaTranslator;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.dfs.FileSelection;
+import org.apache.drill.exec.store.dfs.ReadEntryWithPath;
+import org.apache.drill.exec.store.parquet.AbstractParquetGroupScan;
+import org.apache.drill.exec.store.parquet.ParquetReaderConfig;
+import org.apache.drill.exec.store.parquet.RowGroupReadEntry;
+import org.apache.drill.exec.util.ImpersonationUtil;
+import org.apache.drill.metastore.metadata.LocationProvider;
+import org.apache.drill.metastore.metadata.Metadata;
+import org.apache.drill.metastore.statistics.ColumnStatistics;
+import org.apache.drill.metastore.statistics.TableStatisticsKind;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.net.URI;
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+@JsonTypeName("delta-scan")
+public class DeltaGroupScan extends AbstractParquetGroupScan {
+
+ private final DeltaFormatPlugin formatPlugin;
+
+ private final String path;
+
+ private final TupleMetadata schema;
+
+ private final LogicalExpression condition;
+
+ private final DrillFileSystem fs;
+
+ private List<AddFile> addFiles;
+
+ private List<EndpointAffinity> endpointAffinities;
+
+ private final Map<Path, Map<String, String>> partitionHolder;
+
+ @JsonCreator
+ public DeltaGroupScan(
+ @JsonProperty("userName") String userName,
+ @JsonProperty("entries") List<ReadEntryWithPath> entries,
+ @JsonProperty("storage") StoragePluginConfig storageConfig,
+ @JsonProperty("format") FormatPluginConfig formatConfig,
+ @JsonProperty("columns") List<SchemaPath> columns,
+ @JsonProperty("schema") TupleMetadata schema,
+ @JsonProperty("path") String path,
+ @JsonProperty("readerConfig") ParquetReaderConfig readerConfig,
+ @JsonProperty("condition") LogicalExpression condition,
+ @JsonProperty("limit") Integer limit,
+ @JsonProperty("partitionHolder") Map<Path, Map<String, String>> partitionHolder,
+ @JacksonInject StoragePluginRegistry pluginRegistry) throws IOException {
+ super(ImpersonationUtil.resolveUserName(userName), columns, entries, readerConfig, condition);
+ this.formatPlugin = pluginRegistry.resolveFormat(storageConfig, formatConfig, DeltaFormatPlugin.class);
+ this.columns = columns;
+ this.path = path;
+ this.schema = schema;
+ this.condition = condition;
+ this.limit = limit;
+ this.fs = ImpersonationUtil.createFileSystem(
+ ImpersonationUtil.resolveUserName(userName), formatPlugin.getFsConf());
+
+ DeltaParquetTableMetadataProvider metadataProvider =
+ defaultTableMetadataProviderBuilder(new FileSystemMetadataProviderManager())
+ .withEntries(entries)
+ .withFormatPlugin(formatPlugin)
+ .withReaderConfig(readerConfig)
+ .withSchema(schema)
+ .build();
+
+ this.metadataProvider = metadataProvider;
+ this.entries = metadataProvider.getEntries();
+ this.partitionHolder = partitionHolder;
+ this.fileSet = metadataProvider.getFileSet();
+
+ init();
+ }
+
+ private DeltaGroupScan(DeltaGroupScanBuilder builder) throws IOException {
+ super(ImpersonationUtil.resolveUserName(builder.userName), builder.columns,
+ builder.entries, builder.readerConfig, builder.condition);
+ this.formatPlugin = builder.formatPlugin;
+ this.columns = builder.columns;
+ this.path = builder.path;
+ this.schema = builder.schema;
+ this.condition = builder.condition;
+ this.limit = builder.limit;
+ this.fs = ImpersonationUtil.createFileSystem(
+ ImpersonationUtil.resolveUserName(userName), formatPlugin.getFsConf());
+
+ DeltaParquetTableMetadataProvider metadataProvider =
+ defaultTableMetadataProviderBuilder(new FileSystemMetadataProviderManager())
+ .withEntries(entries)
+ .withFormatPlugin(formatPlugin)
+ .withReaderConfig(readerConfig)
+ .withSchema(schema)
+ .build();
+
+ this.metadataProvider = metadataProvider;
+ this.entries = metadataProvider.getEntries();
+ this.partitionHolder = builder.partitionValues;
+ this.fileSet = metadataProvider.getFileSet();
+
+ init();
+ }
+
+ /**
+ * Private constructor, used for cloning.
+ *
+ * @param that The DeltaGroupScan to clone
+ */
+ private DeltaGroupScan(DeltaGroupScan that) {
+ super(that);
+ this.columns = that.columns;
+ this.formatPlugin = that.formatPlugin;
+ this.path = that.path;
+ this.condition = that.condition;
+ this.schema = that.schema;
+ this.mappings = that.mappings;
+ this.fs = that.fs;
+ this.limit = that.limit;
+ this.addFiles = that.addFiles;
+ this.endpointAffinities = that.endpointAffinities;
+ this.partitionHolder = that.partitionHolder;
+ }
+
+ @Override
+ protected DeltaParquetTableMetadataProvider.Builder tableMetadataProviderBuilder(MetadataProviderManager source) {
+ return defaultTableMetadataProviderBuilder(source);
+ }
+
+ @Override
+ protected DeltaParquetTableMetadataProvider.Builder defaultTableMetadataProviderBuilder(MetadataProviderManager source) {
+ return new DeltaParquetTableMetadataProvider.Builder(source);
+ }
+
+ public static DeltaGroupScanBuilder builder() {
+ return new DeltaGroupScanBuilder();
+ }
+
+ @Override
+ public DeltaGroupScan clone(List<SchemaPath> columns) {
+ try {
+ return toBuilder().columns(columns).build();
+ } catch (IOException e) {
+ throw new DrillRuntimeException(e);
+ }
+ }
+
+ @Override
+ public DeltaGroupScan applyLimit(int maxRecords) {
+ DeltaGroupScan clone = new DeltaGroupScan(this);
+ clone.limit = maxRecords;
+ return clone;
+ }
+
+ @Override
+ public SubScan getSpecificScan(int minorFragmentId) {
+ List<RowGroupReadEntry> readEntries = getReadEntries(minorFragmentId);
+ Map<Path, Map<String, String>> subPartitionHolder = new HashMap<>();
+ for (RowGroupReadEntry readEntry : readEntries) {
+ Map<String, String> values = partitionHolder.get(readEntry.getPath());
+ subPartitionHolder.put(readEntry.getPath(), values);
+ }
+ return new DeltaRowGroupScan(getUserName(), formatPlugin, readEntries, columns, subPartitionHolder,
+ readerConfig, filter, getTableMetadata().getSchema());
+ }
+
+ @Override
+ public DeltaGroupScan clone(FileSelection selection) throws IOException {
+ DeltaGroupScan newScan = new DeltaGroupScan(this);
+ newScan.modifyFileSelection(selection);
+ newScan.init();
+ return newScan;
+ }
+
+ @Override
+ protected RowGroupScanFilterer<?> getFilterer() {
+ return new DeltaParquetScanFilterer(this);
+ }
+
+ @Override
+ protected Collection<DrillbitEndpoint> getDrillbits() {
+ return formatPlugin.getContext().getBits();
+ }
+
+ @Override
+ protected AbstractParquetGroupScan cloneWithFileSelection(Collection<Path> filePaths) throws IOException {
+ FileSelection newSelection = new FileSelection(null, new ArrayList<>(filePaths), null, null, false);
+ return clone(newSelection);
+ }
+
+ @Override
+ protected boolean supportsFileImplicitColumns() {
+ // current group scan should populate directory partition values
+ return false;
+ }
+
+ @Override
+ protected List<String> getPartitionValues(LocationProvider locationProvider) {
+ return Collections.emptyList();
+ }
+
+ @Override
+ public PhysicalOperator getNewWithChildren(List<PhysicalOperator> children) {
+ Preconditions.checkArgument(children.isEmpty());
+ return new DeltaGroupScan(this);
+ }
+
+ @Override
+ public boolean supportsLimitPushdown() {
+ return false;
+ }
+
+ @JsonProperty("schema")
+ public TupleMetadata getSchema() {
+ return schema;
+ }
+
+ @JsonProperty("storage")
+ public StoragePluginConfig getStorageConfig() {
+ return formatPlugin.getStorageConfig();
+ }
+
+ @JsonProperty("format")
+ public FormatPluginConfig getFormatConfig() {
+ return formatPlugin.getConfig();
+ }
+
+ @JsonProperty("path")
+ public String getPath() {
+ return path;
+ }
+
+ @JsonProperty("condition")
+ public LogicalExpression getCondition() {
+ return condition;
+ }
+
+ @JsonProperty("partitionHolder")
+ public Map<Path, Map<String, String>> getPartitionHolder() {
+ return partitionHolder;
+ }
+
+ @Override
+ public String toString() {
+ return new PlanStringBuilder(this)
+ .field("path", path)
+ .field("entries", entries)
+ .field("schema", schema)
+ .field("columns", columns)
+ .field("addFiles", addFiles)
+ .field("limit", limit)
+ .field("numFiles", getEntries().size())
+ .toString();
+ }
+
+ public DeltaGroupScanBuilder toBuilder() {
+ return new DeltaGroupScanBuilder()
+ .userName(this.userName)
+ .formatPlugin(this.formatPlugin)
+ .schema(this.schema)
+ .path(this.path)
+ .condition(this.condition)
+ .columns(this.columns)
+ .limit(this.limit);
+ }
+
+ private static class DeltaParquetScanFilterer extends RowGroupScanFilterer<DeltaParquetScanFilterer> {
+
+ public DeltaParquetScanFilterer(DeltaGroupScan source) {
+ super(source);
+ }
+
+ @Override
+ protected AbstractParquetGroupScan getNewScan() {
+ return new DeltaGroupScan((DeltaGroupScan) source);
+ }
+
+ @Override
+ protected DeltaParquetScanFilterer self() {
+ return this;
+ }
+
+ @Override
+ protected <T extends Metadata> Map<SchemaPath, ColumnStatistics<?>> getImplicitColumnStatistics(
+ OptionManager optionManager, T metadata, Map<SchemaPath, ColumnStatistics<?>> columnsStatistics) {
+ if (metadata instanceof LocationProvider && optionManager != null) {
+ LocationProvider locationProvider = (LocationProvider) metadata;
+ columnsStatistics = new HashMap<>(columnsStatistics);
+ Map<String, String> partitions =
+ ((DeltaGroupScan) source).getPartitionHolder().get(locationProvider.getPath());
+ for (Map.Entry<String, String> partitionValue : partitions.entrySet()) {
+ TypeProtos.MinorType minorType =
+ tableSchema.column(partitionValue.getKey()).getType().getMinorType();
+ String value = partitionValue.getValue();
+ if (value != null) {
+ columnsStatistics.put(SchemaPath.getCompoundPath(partitionValue.getKey()),
+ StatisticsProvider.getConstantColumnStatistics(
+ castPartitionValue(value, minorType), minorType));
+ } else {
+ Long rowCount = TableStatisticsKind.ROW_COUNT.getValue(metadata);
+ columnsStatistics.put(SchemaPath.getCompoundPath(partitionValue.getKey()),
+ StatisticsProvider.getColumnStatistics(null, null, rowCount, minorType));
+ }
+ }
+ }
+
+ return columnsStatistics;
+ }
+
+ private Object castPartitionValue(String value, TypeProtos.MinorType type) {
+ switch (type) {
+ case BIT:
+ return Boolean.parseBoolean(value);
+ case TINYINT:
+ return Byte.parseByte(value);
+ case SMALLINT:
+ return Short.parseShort(value);
+ case INT:
+ return Integer.parseInt(value);
+ case BIGINT:
+ return Long.parseLong(value);
+ case FLOAT4:
+ return Float.parseFloat(value);
+ case FLOAT8:
+ return Double.parseDouble(value);
+ case DATE:
+ return DateUtility.parseLocalDate(value);
+ case TIME:
+ return DateUtility.parseLocalTime(value);
+ case TIMESTAMP:
+ return DateUtility.parseBest(value);
+ case VARCHAR:
+ return value;
+ case VARDECIMAL:
+ return new BigDecimal(value);
+ default:
+ throw new UnsupportedOperationException("Unsupported partition type: " + type);
+ }
+ }
+ }
+
+ public static class DeltaGroupScanBuilder {
+ private String userName;
+
+ private DeltaFormatPlugin formatPlugin;
+
+ private TupleMetadata schema;
+
+ private String path;
+
+ private LogicalExpression condition;
+
+ private List<SchemaPath> columns;
+
+ private int limit;
+
+ private List<ReadEntryWithPath> entries;
+
+ private ParquetReaderConfig readerConfig = ParquetReaderConfig.getDefaultInstance();
+
+ private Map<Path, Map<String, String>> partitionValues;
+
+ public DeltaGroupScanBuilder userName(String userName) {
+ this.userName = userName;
+ return this;
+ }
+
+ public DeltaGroupScanBuilder formatPlugin(DeltaFormatPlugin formatPlugin) {
+ this.formatPlugin = formatPlugin;
+ return this;
+ }
+
+ public DeltaGroupScanBuilder schema(TupleMetadata schema) {
+ this.schema = schema;
+ return this;
+ }
+
+ public DeltaGroupScanBuilder path(String path) {
+ this.path = path;
+ return this;
+ }
+
+ public DeltaGroupScanBuilder condition(LogicalExpression condition) {
+ this.condition = condition;
+ return this;
+ }
+
+ public DeltaGroupScanBuilder columns(List<SchemaPath> columns) {
+ this.columns = columns;
+ return this;
+ }
+
+ public DeltaGroupScanBuilder limit(int maxRecords) {
+ this.limit = maxRecords;
+ return this;
+ }
+
+ public DeltaGroupScanBuilder readerConfig(ParquetReaderConfig readerConfig) {
+ this.readerConfig = readerConfig;
+ return this;
+ }
+
+ public DeltaGroupScan build() throws IOException {
+ DeltaLog log = DeltaLog.forTable(formatPlugin.getFsConf(), path);
+ Snapshot snapshot = log.snapshot();
+ StructType structType = snapshot.getMetadata().getSchema();
+ schema = toSchema(structType);
+
+ DeltaScan scan = Optional.ofNullable(condition)
+ .map(c -> c.accept(new DrillExprToDeltaTranslator(structType), null))
+ .map(snapshot::scan)
+ .orElse(snapshot.scan());
+
+ try {
+ CloseableIterator<AddFile> files = scan.getFiles();
+ ArrayList<AddFile> addFiles = Lists.newArrayList(() -> files);
+ entries = addFiles.stream()
+ .map(addFile -> new ReadEntryWithPath(new Path(URI.create(path).getPath(), URI.create(addFile.getPath()).getPath())))
+ .collect(Collectors.toList());
+
+ partitionValues = addFiles.stream()
+ .collect(Collectors.toMap(
+ addFile -> new Path(URI.create(path).getPath(), URI.create(addFile.getPath()).getPath()),
+ addFile -> collectPartitionedValues(snapshot, addFile)));
+
+ files.close();
+ } catch (IOException e) {
+ throw new DrillRuntimeException(e);
+ }
+
+ return new DeltaGroupScan(this);
+ }
+
+ private Map<String, String> collectPartitionedValues(Snapshot snapshot, AddFile addFile) {
+ Map<String, String> partitionValues = new LinkedHashMap<>();
+ snapshot.getMetadata().getPartitionColumns().stream()
+ .map(col -> Pair.of(col, addFile.getPartitionValues().get(col)))
+ .forEach(pair -> partitionValues.put(pair.getKey(), pair.getValue()));
+ return partitionValues;
+ }
+
+ private TupleMetadata toSchema(StructType structType) {
+ TupleBuilder tupleBuilder = new TupleBuilder();
+ for (StructField field : structType.getFields()) {
+ tupleBuilder.addColumn(toColumnMetadata(field));
+ }
+
+ return tupleBuilder.schema();
+ }
+
+ private ColumnMetadata toColumnMetadata(StructField field) {
+ DataType dataType = field.getDataType();
+ if (dataType instanceof ArrayType) {
+ DataType elementType = ((ArrayType) dataType).getElementType();
+ if (elementType instanceof ArrayType) {
+ return MetadataUtils.newRepeatedList(field.getName(),
+ toColumnMetadata(new StructField(field.getName(), ((ArrayType) elementType).getElementType(), false)));
+ } else if (elementType instanceof StructType) {
+ return MetadataUtils.newMapArray(field.getName(), toSchema((StructType) elementType));
+ }
+ return MetadataUtils.newScalar(field.getName(), toMinorType(elementType), TypeProtos.DataMode.REPEATED);
+ } else if (dataType instanceof StructType) {
+ return MetadataUtils.newMap(field.getName(), toSchema((StructType) dataType));
+ } else {
+ return MetadataUtils.newScalar(field.getName(), toMinorType(field.getDataType()),
+ field.isNullable() ? TypeProtos.DataMode.OPTIONAL : TypeProtos.DataMode.REQUIRED);
+ }
+ }
+
+ private TypeProtos.MinorType toMinorType(DataType dataType) {
+ if (dataType instanceof BinaryType) {
+ return TypeProtos.MinorType.VARBINARY;
+ } else if (dataType instanceof BooleanType) {
+ return TypeProtos.MinorType.BIT;
+ } else if (dataType instanceof ByteType) {
+ return TypeProtos.MinorType.TINYINT;
+ } else if (dataType instanceof DateType) {
+ return TypeProtos.MinorType.DATE;
+ } else if (dataType instanceof DecimalType) {
+ return TypeProtos.MinorType.VARDECIMAL;
+ } else if (dataType instanceof DoubleType) {
+ return TypeProtos.MinorType.FLOAT8;
+ } else if (dataType instanceof FloatType) {
+ return TypeProtos.MinorType.FLOAT4;
+ } else if (dataType instanceof IntegerType) {
+ return TypeProtos.MinorType.INT;
+ } else if (dataType instanceof LongType) {
+ return TypeProtos.MinorType.BIGINT;
+ } else if (dataType instanceof ShortType) {
+ return TypeProtos.MinorType.SMALLINT;
+ } else if (dataType instanceof StringType) {
+ return TypeProtos.MinorType.VARCHAR;
+ } else if (dataType instanceof TimestampType) {
+ return TypeProtos.MinorType.TIMESTAMP;
+ } else {
+ throw new DrillRuntimeException("Unsupported data type: " + dataType);
+ }
+ }
+ }
+}
diff --git a/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/DeltaParquetTableMetadataProvider.java b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/DeltaParquetTableMetadataProvider.java
new file mode 100644
index 0000000..aeec5b0
--- /dev/null
+++ b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/DeltaParquetTableMetadataProvider.java
@@ -0,0 +1,81 @@
+/*
+ * 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.delta;
+
+import org.apache.drill.exec.metastore.MetadataProviderManager;
+import org.apache.drill.exec.store.delta.format.DeltaFormatPlugin;
+import org.apache.drill.exec.store.dfs.ReadEntryWithPath;
+import org.apache.drill.exec.store.parquet.BaseParquetMetadataProvider;
+import org.apache.drill.exec.store.parquet.metadata.Metadata;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.LinkedHashMap;
+import java.util.Map;
+
+/**
+ * This is Metadata provider for Delta tables, which are read by Drill native Parquet reader
+ */
+public class DeltaParquetTableMetadataProvider extends BaseParquetMetadataProvider {
+
+ private final DeltaFormatPlugin deltaFormatPlugin;
+
+ private DeltaParquetTableMetadataProvider(Builder builder) throws IOException {
+ super(builder);
+
+ this.deltaFormatPlugin = builder.formatPlugin;
+
+ init((BaseParquetMetadataProvider) builder.metadataProviderManager().getTableMetadataProvider());
+ }
+
+ @Override
+ protected void initInternal() throws IOException {
+ Map<FileStatus, FileSystem> fileStatusConfMap = new LinkedHashMap<>();
+ for (ReadEntryWithPath entry : entries) {
+ Path path = entry.getPath();
+ FileSystem fs = path.getFileSystem(deltaFormatPlugin.getFsConf());
+ fileStatusConfMap.put(fs.getFileStatus(Path.getPathWithoutSchemeAndAuthority(path)), fs);
+ }
+ parquetTableMetadata = Metadata.getParquetTableMetadata(fileStatusConfMap, readerConfig);
+ }
+
+ public static class Builder extends BaseParquetMetadataProvider.Builder<Builder> {
+ private DeltaFormatPlugin formatPlugin;
+
+ public Builder(MetadataProviderManager source) {
+ super(source);
+ }
+
+ protected Builder withFormatPlugin(DeltaFormatPlugin formatPlugin) {
+ this.formatPlugin = formatPlugin;
+ return self();
+ }
+
+ @Override
+ protected Builder self() {
+ return this;
+ }
+
+ @Override
+ public DeltaParquetTableMetadataProvider build() throws IOException {
+ return new DeltaParquetTableMetadataProvider(this);
+ }
+ }
+}
diff --git a/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/DeltaRowGroupScan.java b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/DeltaRowGroupScan.java
new file mode 100644
index 0000000..5645c3f
--- /dev/null
+++ b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/DeltaRowGroupScan.java
@@ -0,0 +1,141 @@
+/*
+ * 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.delta;
+
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+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.PhysicalOperator;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.delta.format.DeltaFormatPlugin;
+import org.apache.drill.exec.store.delta.format.DeltaFormatPluginConfig;
+import org.apache.drill.exec.store.parquet.AbstractParquetRowGroupScan;
+import org.apache.drill.exec.store.parquet.ParquetReaderConfig;
+import org.apache.drill.exec.store.parquet.RowGroupReadEntry;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+@JsonTypeName("delta-row-group-scan")
+public class DeltaRowGroupScan extends AbstractParquetRowGroupScan {
+
+ public static final String OPERATOR_TYPE = "DELTA_ROW_GROUP_SCAN";
+
+ private final DeltaFormatPlugin formatPlugin;
+ private final DeltaFormatPluginConfig formatPluginConfig;
+ private final Map<Path, Map<String, String>> partitions;
+
+ @JsonCreator
+ public DeltaRowGroupScan(@JacksonInject StoragePluginRegistry registry,
+ @JsonProperty("userName") String userName,
+ @JsonProperty("storage") StoragePluginConfig storageConfig,
+ @JsonProperty("formatPluginConfig") FormatPluginConfig formatPluginConfig,
+ @JsonProperty("rowGroupReadEntries") List<RowGroupReadEntry> rowGroupReadEntries,
+ @JsonProperty("columns") List<SchemaPath> columns,
+ @JsonProperty("partitions") Map<Path, Map<String, String>> partitions,
+ @JsonProperty("readerConfig") ParquetReaderConfig readerConfig,
+ @JsonProperty("filter") LogicalExpression filter,
+ @JsonProperty("schema") TupleMetadata schema) {
+ this(userName,
+ registry.resolveFormat(storageConfig, formatPluginConfig, DeltaFormatPlugin.class),
+ rowGroupReadEntries,
+ columns,
+ partitions,
+ readerConfig,
+ filter,
+ schema);
+ }
+
+ public DeltaRowGroupScan(String userName,
+ DeltaFormatPlugin formatPlugin,
+ List<RowGroupReadEntry> rowGroupReadEntries,
+ List<SchemaPath> columns,
+ Map<Path, Map<String, String>> partitions,
+ ParquetReaderConfig readerConfig,
+ LogicalExpression filter,
+ TupleMetadata schema) {
+ super(userName, rowGroupReadEntries, columns, readerConfig, filter,null, schema);
+ this.formatPlugin = formatPlugin;
+ this.formatPluginConfig = formatPlugin.getConfig();
+ this.partitions = partitions;
+ }
+
+ @JsonProperty
+ public DeltaFormatPluginConfig getFormatPluginConfig() {
+ return formatPluginConfig;
+ }
+
+ @JsonProperty
+ public Map<Path, Map<String, String>> getPartitions() {
+ return partitions;
+ }
+
+ @Override
+ public PhysicalOperator getNewWithChildren(List<PhysicalOperator> children) {
+ Preconditions.checkArgument(children.isEmpty());
+ return new DeltaRowGroupScan(getUserName(), formatPlugin, rowGroupReadEntries, columns, partitions,
+ readerConfig, filter, schema);
+ }
+
+ @Override
+ public String getOperatorType() {
+ return OPERATOR_TYPE;
+ }
+
+ @Override
+ public AbstractParquetRowGroupScan copy(List<SchemaPath> columns) {
+ return new DeltaRowGroupScan(getUserName(), formatPlugin, rowGroupReadEntries, columns, partitions,
+ readerConfig, filter, schema);
+ }
+
+ @Override
+ public Configuration getFsConf(RowGroupReadEntry rowGroupReadEntry) {
+ return formatPlugin.getFsConf();
+ }
+
+ @Override
+ public boolean supportsFileImplicitColumns() {
+ return true;
+ }
+
+ @Override
+ public List<String> getPartitionValues(RowGroupReadEntry rowGroupReadEntry) {
+ return Collections.emptyList();
+ }
+
+ public Map<String, String> getPartitions(RowGroupReadEntry rowGroupReadEntry) {
+ return partitions.get(rowGroupReadEntry.getPath());
+ }
+
+ @Override
+ public boolean isImplicitColumn(SchemaPath path, String partitionColumnLabel) {
+ return partitions.values().stream()
+ .anyMatch(map -> map.containsKey(path.getAsUnescapedPath()));
+ }
+}
+
diff --git a/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/format/DeltaFormatMatcher.java b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/format/DeltaFormatMatcher.java
new file mode 100644
index 0000000..502a74d
--- /dev/null
+++ b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/format/DeltaFormatMatcher.java
@@ -0,0 +1,68 @@
+/*
+ * 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.delta.format;
+
+import io.delta.standalone.DeltaLog;
+import org.apache.drill.exec.planner.logical.DrillTable;
+import org.apache.drill.exec.store.SchemaConfig;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.dfs.FileSelection;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.apache.drill.exec.store.dfs.FormatMatcher;
+import org.apache.drill.exec.store.dfs.FormatPlugin;
+import org.apache.drill.exec.store.dfs.FormatSelection;
+import org.apache.drill.exec.store.plan.rel.PluginDrillTable;
+import org.apache.hadoop.fs.FileStatus;
+
+public class DeltaFormatMatcher extends FormatMatcher {
+
+ private final DeltaFormatPlugin formatPlugin;
+
+ public DeltaFormatMatcher(DeltaFormatPlugin formatPlugin) {
+ this.formatPlugin = formatPlugin;
+ }
+
+ @Override
+ public boolean supportDirectoryReads() {
+ return true;
+ }
+
+ @Override
+ public DrillTable isReadable(DrillFileSystem fs, FileSelection selection, FileSystemPlugin fsPlugin,
+ String storageEngineName, SchemaConfig schemaConfig) {
+ if (DeltaLog.forTable(fsPlugin.getFsConf(), selection.getSelectionRoot()).tableExists()) {
+ FormatSelection formatSelection = new FormatSelection(formatPlugin.getConfig(), selection);
+ return new PluginDrillTable(fsPlugin, storageEngineName, schemaConfig.getUserName(), formatSelection, formatPlugin.getConvention());
+ }
+ return null;
+ }
+
+ @Override
+ public boolean isFileReadable(DrillFileSystem fs, FileStatus status) {
+ return false;
+ }
+
+ @Override
+ public FormatPlugin getFormatPlugin() {
+ return formatPlugin;
+ }
+
+ public int priority() {
+ return HIGH_PRIORITY;
+ }
+}
diff --git a/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/format/DeltaFormatPlugin.java b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/format/DeltaFormatPlugin.java
new file mode 100644
index 0000000..37e4803
--- /dev/null
+++ b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/format/DeltaFormatPlugin.java
@@ -0,0 +1,233 @@
+/*
+ * 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.delta.format;
+
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.metastore.MetadataProviderManager;
+import org.apache.drill.exec.physical.base.AbstractGroupScan;
+import org.apache.drill.exec.physical.base.AbstractWriter;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.PlannerPhase;
+import org.apache.drill.exec.planner.common.DrillStatsTable;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.record.metadata.schema.SchemaProvider;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.store.PluginRulesProviderImpl;
+import org.apache.drill.exec.store.StoragePluginRulesSupplier;
+import org.apache.drill.exec.store.dfs.FileSelection;
+import org.apache.drill.exec.store.dfs.FileSystemConfig;
+import org.apache.drill.exec.store.dfs.FormatMatcher;
+import org.apache.drill.exec.store.dfs.FormatPlugin;
+import org.apache.drill.exec.store.delta.DeltaGroupScan;
+import org.apache.drill.exec.store.delta.plan.DeltaPluginImplementor;
+import org.apache.drill.exec.store.parquet.ParquetReaderConfig;
+import org.apache.drill.exec.store.plan.rel.PluginRel;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
+
+public class DeltaFormatPlugin implements FormatPlugin {
+
+ private static final String DELTA_CONVENTION_PREFIX = "DELTA.";
+
+ /**
+ * Generator for format id values. Formats with the same name may be defined
+ * in multiple storage plugins, so using the unique id within the convention name
+ * to ensure the rule names will be unique for different plugin instances.
+ */
+ private static final AtomicInteger NEXT_ID = new AtomicInteger(0);
+
+ private final FileSystemConfig storageConfig;
+
+ private final DeltaFormatPluginConfig config;
+
+ private final Configuration fsConf;
+
+ private final DrillbitContext context;
+
+ private final String name;
+
+ private final DeltaFormatMatcher matcher;
+
+ private final StoragePluginRulesSupplier storagePluginRulesSupplier;
+
+ public DeltaFormatPlugin(
+ String name,
+ DrillbitContext context,
+ Configuration fsConf,
+ FileSystemConfig storageConfig,
+ DeltaFormatPluginConfig config) {
+ this.storageConfig = storageConfig;
+ this.config = config;
+ this.fsConf = fsConf;
+ this.context = context;
+ this.name = name;
+ this.matcher = new DeltaFormatMatcher(this);
+ this.storagePluginRulesSupplier = storagePluginRulesSupplier(name + NEXT_ID.getAndIncrement());
+ }
+
+ private static StoragePluginRulesSupplier storagePluginRulesSupplier(String name) {
+ Convention convention = new Convention.Impl(DELTA_CONVENTION_PREFIX + name, PluginRel.class);
+ return StoragePluginRulesSupplier.builder()
+ .rulesProvider(new PluginRulesProviderImpl(convention, DeltaPluginImplementor::new))
+ .supportsFilterPushdown(true)
+ .supportsProjectPushdown(true)
+ .supportsLimitPushdown(true)
+ .convention(convention)
+ .build();
+ }
+
+ @Override
+ public boolean supportsRead() {
+ return true;
+ }
+
+ @Override
+ public boolean supportsWrite() {
+ return false;
+ }
+
+ @Override
+ public boolean supportsAutoPartitioning() {
+ return false;
+ }
+
+ @Override
+ public FormatMatcher getMatcher() {
+ return matcher;
+ }
+
+ @Override
+ public AbstractWriter getWriter(PhysicalOperator child, String location, List<String> partitionColumns) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public Set<? extends RelOptRule> getOptimizerRules(PlannerPhase phase) {
+ switch (phase) {
+ case PHYSICAL:
+ case LOGICAL:
+ return storagePluginRulesSupplier.getOptimizerRules();
+ case LOGICAL_PRUNE_AND_JOIN:
+ case LOGICAL_PRUNE:
+ case PARTITION_PRUNING:
+ case JOIN_PLANNING:
+ default:
+ return Collections.emptySet();
+ }
+ }
+
+ @Override
+ public AbstractGroupScan getGroupScan(String userName, FileSelection selection, List<SchemaPath> columns) throws IOException {
+ return getGroupScan(userName, selection, columns, (OptionManager) null);
+ }
+
+ @Override
+ public AbstractGroupScan getGroupScan(String userName, FileSelection selection, List<SchemaPath> columns, OptionManager options) throws IOException {
+ return getGroupScan(userName, selection, columns, options, null);
+ }
+
+ @Override
+ public AbstractGroupScan getGroupScan(String userName, FileSelection selection,
+ List<SchemaPath> columns, OptionManager options, MetadataProviderManager metadataProviderManager) throws IOException {
+ ParquetReaderConfig readerConfig = ParquetReaderConfig.builder()
+ .withConf(fsConf)
+ .withOptions(options)
+ .build();
+ return DeltaGroupScan.builder()
+ .userName(userName)
+ .formatPlugin(this)
+ .readerConfig(readerConfig)
+ .path(selection.selectionRoot.toUri().getPath())
+ .columns(columns)
+ .limit(-1)
+ .build();
+ }
+
+ @Override
+ public AbstractGroupScan getGroupScan(String userName, FileSelection selection,
+ List<SchemaPath> columns, MetadataProviderManager metadataProviderManager) throws IOException {
+ SchemaProvider schemaProvider = metadataProviderManager.getSchemaProvider();
+ TupleMetadata schema = schemaProvider != null
+ ? schemaProvider.read().getSchema()
+ : null;
+ return DeltaGroupScan.builder()
+ .userName(userName)
+ .formatPlugin(this)
+ .readerConfig(ParquetReaderConfig.builder().withConf(fsConf).build())
+ .schema(schema)
+ .path(selection.selectionRoot.toUri().getPath())
+ .columns(columns)
+ .limit(-1)
+ .build();
+ }
+
+ @Override
+ public boolean supportsStatistics() {
+ return false;
+ }
+
+ @Override
+ public DrillStatsTable.TableStatistics readStatistics(FileSystem fs, Path statsTablePath) {
+ throw new UnsupportedOperationException("unimplemented");
+ }
+
+ @Override
+ public void writeStatistics(DrillStatsTable.TableStatistics statistics, FileSystem fs, Path statsTablePath) {
+ throw new UnsupportedOperationException("unimplemented");
+ }
+
+ @Override
+ public DeltaFormatPluginConfig getConfig() {
+ return config;
+ }
+
+ @Override
+ public FileSystemConfig getStorageConfig() {
+ return storageConfig;
+ }
+
+ @Override
+ public Configuration getFsConf() {
+ return fsConf;
+ }
+
+ @Override
+ public DrillbitContext getContext() {
+ return context;
+ }
+
+ @Override
+ public String getName() {
+ return name;
+ }
+
+ public Convention getConvention() {
+ return storagePluginRulesSupplier.convention();
+ }
+
+}
diff --git a/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/format/DeltaFormatPluginConfig.java b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/format/DeltaFormatPluginConfig.java
new file mode 100644
index 0000000..9fd7bb7
--- /dev/null
+++ b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/format/DeltaFormatPluginConfig.java
@@ -0,0 +1,32 @@
+/*
+ * 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.delta.format;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import org.apache.drill.common.logical.FormatPluginConfig;
+
+@JsonTypeName(DeltaFormatPluginConfig.NAME)
+public class DeltaFormatPluginConfig implements FormatPluginConfig {
+
+ public static final String NAME = "delta";
+
+ @JsonCreator
+ public DeltaFormatPluginConfig() {
+ }
+}
diff --git a/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/plan/DeltaPluginImplementor.java b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/plan/DeltaPluginImplementor.java
new file mode 100644
index 0000000..84be68c
--- /dev/null
+++ b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/plan/DeltaPluginImplementor.java
@@ -0,0 +1,227 @@
+/*
+ * 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.delta.plan;
+
+import org.apache.calcite.plan.volcano.RelSubset;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelShuttleImpl;
+import org.apache.calcite.rel.core.Filter;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.logical.LogicalFilter;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.util.Util;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.physical.base.GroupScan;
+import org.apache.drill.exec.planner.common.DrillLimitRelBase;
+import org.apache.drill.exec.planner.logical.DrillConstExecutor;
+import org.apache.drill.exec.planner.logical.DrillOptiq;
+import org.apache.drill.exec.planner.logical.DrillParseContext;
+import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.planner.physical.PrelUtil;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.delta.DeltaGroupScan;
+import org.apache.drill.exec.store.dfs.FileSystemPlugin;
+import org.apache.drill.exec.store.plan.AbstractPluginImplementor;
+import org.apache.drill.exec.store.plan.rel.PluginFilterRel;
+import org.apache.drill.exec.store.plan.rel.PluginLimitRel;
+import org.apache.drill.exec.store.plan.rel.PluginProjectRel;
+import org.apache.drill.exec.store.plan.rel.StoragePluginTableScan;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+
+public class DeltaPluginImplementor extends AbstractPluginImplementor {
+
+ private DeltaGroupScan groupScan;
+
+ @Override
+ public void implement(StoragePluginTableScan scan) {
+ groupScan = (DeltaGroupScan) scan.getGroupScan();
+ }
+
+ @Override
+ public void implement(PluginFilterRel filter) throws IOException {
+ visitChild(filter.getInput());
+
+ RexNode condition = filter.getCondition();
+ LogicalExpression expression = DrillOptiq.toDrill(
+ new DrillParseContext(PrelUtil.getPlannerSettings(filter.getCluster().getPlanner())),
+ filter.getInput(),
+ condition);
+
+ DrillConstExecutor executor = (DrillConstExecutor) filter.getCluster().getPlanner().getExecutor();
+ PlannerSettings plannerSettings = filter.getCluster().getPlanner().getContext().unwrap(PlannerSettings.class);
+ groupScan = Optional.ofNullable((DeltaGroupScan) groupScan.applyFilter(expression, executor.getUdfUtilities(),
+ plannerSettings.functionImplementationRegistry, plannerSettings.getOptions()))
+ .orElse(groupScan);
+ }
+
+ @Override
+ public void implement(PluginProjectRel project) throws IOException {
+ visitChild(project.getInput());
+
+ DrillParseContext context = new DrillParseContext(PrelUtil.getPlannerSettings(project.getCluster().getPlanner()));
+ RelNode input = project.getInput();
+
+ List<SchemaPath> projects = project.getProjects().stream()
+ .map(e -> (SchemaPath) DrillOptiq.toDrill(context, input, e))
+ .collect(Collectors.toList());
+ groupScan = groupScan.clone(projects);
+ }
+
+ @Override
+ public boolean canImplement(Filter filter) {
+ FilterFinder filterFinder = new FilterFinder();
+ filter.getInput().accept(filterFinder);
+ return filterFinder.getFilter() == null;
+ }
+
+ @Override
+ public void implement(PluginLimitRel limit) throws IOException {
+ visitChild(limit.getInput());
+ int maxRecords = getArtificialLimit(limit);
+ if (maxRecords >= 0) {
+ groupScan = groupScan.applyLimit(maxRecords);
+ }
+ }
+
+ @Override
+ public boolean canImplement(DrillLimitRelBase limit) {
+ if (hasPluginGroupScan(limit)) {
+ FirstLimitFinder finder = new FirstLimitFinder();
+ limit.getInput().accept(finder);
+ int oldLimit = getArtificialLimit(finder.getFetch(), finder.getOffset());
+ int newLimit = getArtificialLimit(limit);
+ return newLimit >= 0 && (oldLimit < 0 || newLimit < oldLimit);
+ }
+ return false;
+ }
+
+ @Override
+ public boolean artificialLimit() {
+ return true;
+ }
+
+ @Override
+ public boolean artificialFilter() {
+ return true;
+ }
+
+ @Override
+ protected Class<? extends StoragePlugin> supportedPlugin() {
+ return FileSystemPlugin.class;
+ }
+
+ @Override
+ public boolean splitProject(Project project) {
+ return true;
+ }
+
+ @Override
+ public boolean canImplement(Project project) {
+ return hasPluginGroupScan(project);
+ }
+
+ @Override
+ public GroupScan getPhysicalOperator() {
+ return groupScan;
+ }
+
+ @Override
+ protected boolean hasPluginGroupScan(RelNode node) {
+ return findGroupScan(node) instanceof DeltaGroupScan;
+ }
+
+ private int rexLiteralIntValue(RexLiteral offset) {
+ return ((BigDecimal) offset.getValue()).intValue();
+ }
+
+ private int getArtificialLimit(DrillLimitRelBase limit) {
+ return getArtificialLimit(limit.getFetch(), limit.getOffset());
+ }
+
+ private int getArtificialLimit(RexNode fetch, RexNode offset) {
+ int maxRows = -1;
+ if (fetch != null) {
+ maxRows = rexLiteralIntValue((RexLiteral) fetch);
+ if (offset != null) {
+ maxRows += rexLiteralIntValue((RexLiteral) offset);
+ }
+ }
+ return maxRows;
+ }
+
+ private static class FilterFinder extends RelShuttleImpl {
+ private RelNode filter;
+
+ @Override
+ public RelNode visit(LogicalFilter filter) {
+ this.filter = filter;
+ return filter;
+ }
+
+ @Override
+ public RelNode visit(RelNode other) {
+ if (other instanceof Filter) {
+ this.filter = other;
+ return other;
+ } else if (other instanceof RelSubset) {
+ RelSubset relSubset = (RelSubset) other;
+ Util.first(relSubset.getBest(), relSubset.getOriginal()).accept(this);
+ }
+ return super.visit(other);
+ }
+
+ public RelNode getFilter() {
+ return filter;
+ }
+ }
+
+ private static class FirstLimitFinder extends RelShuttleImpl {
+ private RexNode fetch;
+
+ private RexNode offset;
+
+ @Override
+ public RelNode visit(RelNode other) {
+ if (other instanceof DrillLimitRelBase) {
+ DrillLimitRelBase limitRelBase = (DrillLimitRelBase) other;
+ fetch = limitRelBase.getFetch();
+ offset = limitRelBase.getOffset();
+ return other;
+ } else if (other instanceof RelSubset) {
+ RelSubset relSubset = (RelSubset) other;
+ Util.first(relSubset.getBest(), relSubset.getOriginal()).accept(this);
+ }
+ return super.visit(other);
+ }
+
+ public RexNode getFetch() {
+ return fetch;
+ }
+
+ public RexNode getOffset() {
+ return offset;
+ }
+ }
+}
diff --git a/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/plan/DrillExprToDeltaTranslator.java b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/plan/DrillExprToDeltaTranslator.java
new file mode 100644
index 0000000..bd95bde
--- /dev/null
+++ b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/plan/DrillExprToDeltaTranslator.java
@@ -0,0 +1,246 @@
+/*
+ * 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.delta.plan;
+
+import io.delta.standalone.expressions.And;
+import io.delta.standalone.expressions.EqualTo;
+import io.delta.standalone.expressions.Expression;
+import io.delta.standalone.expressions.GreaterThan;
+import io.delta.standalone.expressions.GreaterThanOrEqual;
+import io.delta.standalone.expressions.IsNotNull;
+import io.delta.standalone.expressions.IsNull;
+import io.delta.standalone.expressions.LessThan;
+import io.delta.standalone.expressions.LessThanOrEqual;
+import io.delta.standalone.expressions.Literal;
+import io.delta.standalone.expressions.Not;
+import io.delta.standalone.expressions.Or;
+import io.delta.standalone.expressions.Predicate;
+import io.delta.standalone.types.StructType;
+import org.apache.drill.common.FunctionNames;
+import org.apache.drill.common.expression.FunctionCall;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.common.expression.PathSegment;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.expression.ValueExpressions;
+import org.apache.drill.common.expression.visitors.AbstractExprVisitor;
+
+public class DrillExprToDeltaTranslator extends AbstractExprVisitor<Expression, Void, RuntimeException> {
+
+ private final StructType structType;
+
+ public DrillExprToDeltaTranslator(StructType structType) {
+ this.structType = structType;
+ }
+
+ @Override
+ public Expression visitFunctionCall(FunctionCall call, Void value) {
+ try {
+ return visitFunctionCall(call);
+ } catch (Exception e) {
+ return null;
+ }
+ }
+
+ private Predicate visitFunctionCall(FunctionCall call) {
+ switch (call.getName()) {
+ case FunctionNames.AND: {
+ Expression left = call.arg(0).accept(this, null);
+ Expression right = call.arg(1).accept(this, null);
+ if (left != null && right != null) {
+ return new And(left, right);
+ }
+ return null;
+ }
+ case FunctionNames.OR: {
+ Expression left = call.arg(0).accept(this, null);
+ Expression right = call.arg(1).accept(this, null);
+ if (left != null && right != null) {
+ return new Or(left, right);
+ }
+ return null;
+ }
+ case FunctionNames.NOT: {
+ Expression expression = call.arg(0).accept(this, null);
+ if (expression != null) {
+ return new Not(expression);
+ }
+ return null;
+ }
+ case FunctionNames.IS_NULL: {
+ LogicalExpression arg = call.arg(0);
+ if (arg instanceof SchemaPath) {
+ String name = getPath((SchemaPath) arg);
+ return new IsNull(structType.column(name));
+ }
+ return null;
+ }
+ case FunctionNames.IS_NOT_NULL: {
+ LogicalExpression arg = call.arg(0);
+ if (arg instanceof SchemaPath) {
+ String name = getPath((SchemaPath) arg);
+ return new IsNotNull(structType.column(name));
+ }
+ return null;
+ }
+ case FunctionNames.LT: {
+ LogicalExpression nameRef = call.arg(0);
+ Expression expression = call.arg(1).accept(this, null);
+ if (nameRef instanceof SchemaPath) {
+ String name = getPath((SchemaPath) nameRef);
+ return new LessThan(structType.column(name), expression);
+ }
+ return null;
+ }
+ case FunctionNames.LE: {
+ LogicalExpression nameRef = call.arg(0);
+ Expression expression = call.arg(1).accept(this, null);
+ if (nameRef instanceof SchemaPath) {
+ String name = getPath((SchemaPath) nameRef);
+ return new LessThanOrEqual(structType.column(name), expression);
+ }
+ return null;
+ }
+ case FunctionNames.GT: {
+ LogicalExpression nameRef = call.args().get(0);
+ Expression expression = call.args().get(1).accept(this, null);
+ if (nameRef instanceof SchemaPath) {
+ String name = getPath((SchemaPath) nameRef);
+ return new GreaterThan(structType.column(name), expression);
+ }
+ return null;
+ }
+ case FunctionNames.GE: {
+ LogicalExpression nameRef = call.args().get(0);
+ Expression expression = call.args().get(0).accept(this, null);
+ if (nameRef instanceof SchemaPath) {
+ String name = getPath((SchemaPath) nameRef);
+ return new GreaterThanOrEqual(structType.column(name), expression);
+ }
+ return null;
+ }
+ case FunctionNames.EQ: {
+ LogicalExpression nameRef = call.args().get(0);
+ Expression expression = call.args().get(1).accept(this, null);
+ if (nameRef instanceof SchemaPath) {
+ String name = getPath((SchemaPath) nameRef);
+ return new EqualTo(structType.column(name), expression);
+ }
+ return null;
+ }
+ case FunctionNames.NE: {
+ LogicalExpression nameRef = call.args().get(0);
+ Expression expression = call.args().get(1).accept(this, null);
+ if (nameRef instanceof SchemaPath) {
+ String name = getPath((SchemaPath) nameRef);
+ return new Not(new EqualTo(structType.column(name), expression));
+ }
+ return null;
+ }
+ }
+ return null;
+ }
+
+ @Override
+ public Expression visitFloatConstant(ValueExpressions.FloatExpression fExpr, Void value) {
+ return Literal.of(fExpr.getFloat());
+ }
+
+ @Override
+ public Expression visitIntConstant(ValueExpressions.IntExpression intExpr, Void value) {
+ return Literal.of(intExpr.getInt());
+ }
+
+ @Override
+ public Expression visitLongConstant(ValueExpressions.LongExpression longExpr, Void value) {
+ return Literal.of(longExpr.getLong());
+ }
+
+ @Override
+ public Expression visitDecimal9Constant(ValueExpressions.Decimal9Expression decExpr, Void value) {
+ return Literal.of(decExpr.getIntFromDecimal());
+ }
+
+ @Override
+ public Expression visitDecimal18Constant(ValueExpressions.Decimal18Expression decExpr, Void value) {
+ return Literal.of(decExpr.getLongFromDecimal());
+ }
+
+ @Override
+ public Expression visitDecimal28Constant(ValueExpressions.Decimal28Expression decExpr, Void value) {
+ return Literal.of(decExpr.getBigDecimal());
+ }
+
+ @Override
+ public Expression visitDecimal38Constant(ValueExpressions.Decimal38Expression decExpr, Void value) {
+ return Literal.of(decExpr.getBigDecimal());
+ }
+
+ @Override
+ public Expression visitVarDecimalConstant(ValueExpressions.VarDecimalExpression decExpr, Void value) {
+ return Literal.of(decExpr.getBigDecimal());
+ }
+
+ @Override
+ public Expression visitDateConstant(ValueExpressions.DateExpression dateExpr, Void value) {
+ return Literal.of(dateExpr.getDate());
+ }
+
+ @Override
+ public Expression visitTimeConstant(ValueExpressions.TimeExpression timeExpr, Void value) {
+ return Literal.of(timeExpr.getTime());
+ }
+
+ @Override
+ public Expression visitTimeStampConstant(ValueExpressions.TimeStampExpression timestampExpr, Void value) {
+ return Literal.of(timestampExpr.getTimeStamp());
+ }
+
+ @Override
+ public Expression visitDoubleConstant(ValueExpressions.DoubleExpression dExpr, Void value) {
+ return Literal.of(dExpr.getDouble());
+ }
+
+ @Override
+ public Expression visitBooleanConstant(ValueExpressions.BooleanExpression e, Void value) {
+ return Literal.of(e.getBoolean());
+ }
+
+ @Override
+ public Expression visitQuotedStringConstant(ValueExpressions.QuotedString e, Void value) {
+ return Literal.of(e.getString());
+ }
+
+ @Override
+ public Expression visitUnknown(LogicalExpression e, Void value) {
+ return null;
+ }
+
+ private static String getPath(SchemaPath schemaPath) {
+ StringBuilder sb = new StringBuilder();
+ PathSegment segment = schemaPath.getRootSegment();
+ sb.append(segment.getNameSegment().getPath());
+
+ while ((segment = segment.getChild()) != null) {
+ sb.append('.')
+ .append(segment.isNamed()
+ ? segment.getNameSegment().getPath()
+ : "element");
+ }
+ return sb.toString();
+ }
+}
diff --git a/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/read/DeltaScanBatchCreator.java b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/read/DeltaScanBatchCreator.java
new file mode 100644
index 0000000..78cac37
--- /dev/null
+++ b/contrib/format-deltalake/src/main/java/org/apache/drill/exec/store/delta/read/DeltaScanBatchCreator.java
@@ -0,0 +1,65 @@
+/*
+ * 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.delta.read;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.BatchCreator;
+import org.apache.drill.exec.record.CloseableRecordBatch;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.store.ColumnExplorer;
+import org.apache.drill.exec.store.delta.DeltaRowGroupScan;
+import org.apache.drill.exec.store.dfs.DrillFileSystem;
+import org.apache.drill.exec.store.parquet.AbstractParquetRowGroupScan;
+import org.apache.drill.exec.store.parquet.AbstractParquetScanBatchCreator;
+import org.apache.drill.exec.store.parquet.ParquetScanBatchCreator;
+import org.apache.drill.exec.store.parquet.RowGroupReadEntry;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+
+import java.util.List;
+import java.util.Map;
+
+@SuppressWarnings("unused")
+public class DeltaScanBatchCreator extends AbstractParquetScanBatchCreator
+ implements BatchCreator<DeltaRowGroupScan> {
+
+ @Override
+ public CloseableRecordBatch getBatch(ExecutorFragmentContext context, DeltaRowGroupScan rowGroupScan,
+ List<RecordBatch> children) throws ExecutionSetupException {
+ Preconditions.checkArgument(children.isEmpty());
+ OperatorContext oContext = context.newOperatorContext(rowGroupScan);
+ return getBatch(context, rowGroupScan, oContext);
+ }
+
+ @Override
+ protected AbstractDrillFileSystemManager getDrillFileSystemCreator(
+ OperatorContext operatorContext, OptionManager optionManager) {
+ return new ParquetScanBatchCreator.ParquetDrillFileSystemManager(operatorContext,
+ optionManager.getOption(ExecConstants.PARQUET_PAGEREADER_ASYNC).bool_val);
+ }
+
+ @Override
+ protected Map<String, String> getImplicitValues(AbstractParquetRowGroupScan rowGroupScan,
+ ColumnExplorer columnExplorer, RowGroupReadEntry rowGroup, DrillFileSystem fs) {
+ return ((DeltaRowGroupScan) rowGroupScan).getPartitions(rowGroup);
+ }
+
+}
diff --git a/contrib/format-deltalake/src/main/resources/bootstrap-format-plugins.json b/contrib/format-deltalake/src/main/resources/bootstrap-format-plugins.json
new file mode 100644
index 0000000..41d6f2f
--- /dev/null
+++ b/contrib/format-deltalake/src/main/resources/bootstrap-format-plugins.json
@@ -0,0 +1,20 @@
+{
+ "storage":{
+ "dfs": {
+ "type": "file",
+ "formats": {
+ "delta": {
+ "type": "delta"
+ }
+ }
+ },
+ "s3": {
+ "type": "file",
+ "formats": {
+ "delta": {
+ "type": "delta"
+ }
+ }
+ }
+ }
+}
diff --git a/contrib/format-deltalake/src/main/resources/drill-module.conf b/contrib/format-deltalake/src/main/resources/drill-module.conf
new file mode 100644
index 0000000..d4f6639
--- /dev/null
+++ b/contrib/format-deltalake/src/main/resources/drill-module.conf
@@ -0,0 +1,24 @@
+#
+# 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.
+#
+
+# This file tells Drill to consider this module when class path scanning.
+# This file can also include any supplementary configuration information.
+# This file is in HOCON format, see https://github.com/typesafehub/config/blob/master/HOCON.md for more information.
+drill.classpath.scanning: {
+ packages += "org.apache.drill.exec.store.delta"
+}
diff --git a/contrib/format-deltalake/src/test/java/org/apache/drill/exec/store/delta/DeltaQueriesTest.java b/contrib/format-deltalake/src/test/java/org/apache/drill/exec/store/delta/DeltaQueriesTest.java
new file mode 100644
index 0000000..e74b7d3
--- /dev/null
+++ b/contrib/format-deltalake/src/test/java/org/apache/drill/exec/store/delta/DeltaQueriesTest.java
@@ -0,0 +1,196 @@
+/*
+ * 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.delta;
+
+import org.apache.drill.common.logical.FormatPluginConfig;
+import org.apache.drill.common.logical.security.PlainCredentialsProvider;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.delta.format.DeltaFormatPluginConfig;
+import org.apache.drill.exec.store.dfs.FileSystemConfig;
+import org.apache.drill.test.ClusterFixture;
+import org.apache.drill.test.ClusterTest;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.math.BigDecimal;
+import java.nio.file.Paths;
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.apache.drill.exec.util.StoragePluginTestUtils.DFS_PLUGIN_NAME;
+import static org.junit.Assert.assertEquals;
+
+public class DeltaQueriesTest extends ClusterTest {
+
+ @BeforeClass
+ public static void setUpBeforeClass() throws Exception {
+ startCluster(ClusterFixture.builder(dirTestWatcher));
+
+ StoragePluginRegistry pluginRegistry = cluster.drillbit().getContext().getStorage();
+ FileSystemConfig pluginConfig = (FileSystemConfig) pluginRegistry.getPlugin(DFS_PLUGIN_NAME).getConfig();
+ Map<String, FormatPluginConfig> formats = new HashMap<>(pluginConfig.getFormats());
+ formats.put("delta", new DeltaFormatPluginConfig());
+ FileSystemConfig newPluginConfig = new FileSystemConfig(
+ pluginConfig.getConnection(),
+ pluginConfig.getConfig(),
+ pluginConfig.getWorkspaces(),
+ formats,
+ PlainCredentialsProvider.EMPTY_CREDENTIALS_PROVIDER);
+ newPluginConfig.setEnabled(pluginConfig.isEnabled());
+ pluginRegistry.put(DFS_PLUGIN_NAME, newPluginConfig);
+
+ dirTestWatcher.copyResourceToRoot(Paths.get("data-reader-primitives"));
+ dirTestWatcher.copyResourceToRoot(Paths.get("data-reader-partition-values"));
+ dirTestWatcher.copyResourceToRoot(Paths.get("data-reader-nested-struct"));
+ }
+
+ @Test
+ public void testSerDe() throws Exception {
+ String plan = queryBuilder().sql("select * from dfs.`data-reader-partition-values`").explainJson();
+ long count = queryBuilder().physical(plan).run().recordCount();
+ assertEquals(3, count);
+ }
+
+ @Test
+ public void testAllPrimitives() throws Exception {
+ testBuilder()
+ .sqlQuery("select * from dfs.`data-reader-primitives`")
+ .ordered()
+ .baselineColumns("as_int", "as_long", "as_byte", "as_short", "as_boolean", "as_float",
+ "as_double", "as_string", "as_binary", "as_big_decimal")
+ .baselineValues(null, null, null, null, null, null, null, null, null, null)
+ .baselineValues(0, 0L, 0, 0, true, 0.0f, 0.0, "0", new byte[]{0, 0}, BigDecimal.valueOf(0))
+ .baselineValues(1, 1L, 1, 1, false, 1.0f, 1.0, "1", new byte[]{1, 1}, BigDecimal.valueOf(1))
+ .baselineValues(2, 2L, 2, 2, true, 2.0f, 2.0, "2", new byte[]{2, 2}, BigDecimal.valueOf(2))
+ .baselineValues(3, 3L, 3, 3, false, 3.0f, 3.0, "3", new byte[]{3, 3}, BigDecimal.valueOf(3))
+ .baselineValues(4, 4L, 4, 4, true, 4.0f, 4.0, "4", new byte[]{4, 4}, BigDecimal.valueOf(4))
+ .baselineValues(5, 5L, 5, 5, false, 5.0f, 5.0, "5", new byte[]{5, 5}, BigDecimal.valueOf(5))
+ .baselineValues(6, 6L, 6, 6, true, 6.0f, 6.0, "6", new byte[]{6, 6}, BigDecimal.valueOf(6))
+ .baselineValues(7, 7L, 7, 7, false, 7.0f, 7.0, "7", new byte[]{7, 7}, BigDecimal.valueOf(7))
+ .baselineValues(8, 8L, 8, 8, true, 8.0f, 8.0, "8", new byte[]{8, 8}, BigDecimal.valueOf(8))
+ .baselineValues(9, 9L, 9, 9, false, 9.0f, 9.0, "9", new byte[]{9, 9}, BigDecimal.valueOf(9))
+ .go();
+ }
+
+ @Test
+ public void testProjectingColumns() throws Exception {
+
+ String query = "select as_int, as_string from dfs.`data-reader-primitives`";
+
+ queryBuilder()
+ .sql(query)
+ .planMatcher()
+ .include("columns=\\[`as_int`, `as_string`\\]")
+ .match();
+
+ testBuilder()
+ .sqlQuery(query)
+ .unOrdered()
+ .baselineColumns("as_int", "as_string")
+ .baselineValues(null, null)
+ .baselineValues(0, "0")
+ .baselineValues(1, "1")
+ .baselineValues(2, "2")
+ .baselineValues(3, "3")
+ .baselineValues(4, "4")
+ .baselineValues(5, "5")
+ .baselineValues(6, "6")
+ .baselineValues(7, "7")
+ .baselineValues(8, "8")
+ .baselineValues(9, "9")
+ .go();
+ }
+
+ @Test
+ public void testProjectNestedColumn() throws Exception {
+ String query = "select t.a.ac.acb as acb, b from dfs.`data-reader-nested-struct` t";
+
+ queryBuilder()
+ .sql(query)
+ .planMatcher()
+ .include("columns=\\[`a`.`ac`.`acb`, `b`\\]")
+ .match();
+
+ testBuilder()
+ .sqlQuery(query)
+ .unOrdered()
+ .baselineColumns("acb", "b")
+ .baselineValues(0L, 0)
+ .baselineValues(1L, 1)
+ .baselineValues(2L, 2)
+ .baselineValues(3L, 3)
+ .baselineValues(4L, 4)
+ .baselineValues(5L, 5)
+ .baselineValues(6L, 6)
+ .baselineValues(7L, 7)
+ .baselineValues(8L, 8)
+ .baselineValues(9L, 9)
+ .go();
+ }
+
+ @Test
+ public void testPartitionPruning() throws Exception {
+ String query = "select as_int, as_string from dfs.`data-reader-partition-values` where as_long = 1";
+
+ queryBuilder()
+ .sql(query)
+ .planMatcher()
+ .include("numFiles\\=1")
+ .match();
+
+ testBuilder()
+ .sqlQuery(query)
+ .ordered()
+ .baselineColumns("as_int", "as_string")
+ .baselineValues("1", "1")
+ .go();
+ }
+
+ @Test
+ public void testEmptyResults() throws Exception {
+ String query = "select as_int, as_string from dfs.`data-reader-partition-values` where as_long = 101";
+
+ queryBuilder()
+ .sql(query)
+ .planMatcher()
+ .include("numFiles\\=1")
+ .match();
+
+ testBuilder()
+ .sqlQuery(query)
+ .ordered()
+ .expectsEmptyResultSet()
+ .go();
+ }
+
+ @Test
+ public void testLimit() throws Exception {
+ String query = "select as_int, as_string from dfs.`data-reader-partition-values` limit 1";
+
+ // Note that both of the following two limits are expected because this format plugin supports an "artificial" limit.
+ queryBuilder()
+ .sql(query)
+ .planMatcher()
+ .include("Limit\\(fetch\\=\\[1\\]\\)")
+ .include("limit\\=1")
+ .match();
+
+ long count = queryBuilder().sql(query).run().recordCount();
+ assertEquals(1, count);
+ }
+}
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/.part-00000-f2547b28-9219-4628-8462-cc9c56edfebb-c000.snappy.parquet.crc b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/.part-00000-f2547b28-9219-4628-8462-cc9c56edfebb-c000.snappy.parquet.crc
new file mode 100644
index 0000000..0dacf51
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/.part-00000-f2547b28-9219-4628-8462-cc9c56edfebb-c000.snappy.parquet.crc
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/.part-00001-0f755735-3b5b-449a-8f93-92a40d9f065d-c000.snappy.parquet.crc b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/.part-00001-0f755735-3b5b-449a-8f93-92a40d9f065d-c000.snappy.parquet.crc
new file mode 100644
index 0000000..976c62f
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/.part-00001-0f755735-3b5b-449a-8f93-92a40d9f065d-c000.snappy.parquet.crc
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/_delta_log/00000000000000000000.json b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/_delta_log/00000000000000000000.json
new file mode 100644
index 0000000..4046b14
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/_delta_log/00000000000000000000.json
@@ -0,0 +1,5 @@
+{"commitInfo":{"timestamp":1603724040818,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[]"},"isBlindAppend":true}}
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
+{"metaData":{"id":"975ef365-8dec-4bbf-ab88-264c10987001","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"a\",\"type\":{\"type\":\"struct\",\"fields\":[{\"name\":\"aa\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"ab\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"ac\",\"type\":{\"type\":\"struct\",\"fields\":[{\"name\":\"aca\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"acb\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]},\"nullable\":true,\"metadata\":{}}]},\"nullable\":true,\"metadata\":{}},{\"name\":\"b\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1603724040747}}
+{"add":{"path":"part-00000-f2547b28-9219-4628-8462-cc9c56edfebb-c000.snappy.parquet","partitionValues":{},"size":1432,"modificationTime":1603724040000,"dataChange":true}}
+{"add":{"path":"part-00001-0f755735-3b5b-449a-8f93-92a40d9f065d-c000.snappy.parquet","partitionValues":{},"size":1439,"modificationTime":1603724040000,"dataChange":true}}
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/part-00000-f2547b28-9219-4628-8462-cc9c56edfebb-c000.snappy.parquet b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/part-00000-f2547b28-9219-4628-8462-cc9c56edfebb-c000.snappy.parquet
new file mode 100644
index 0000000..d1b8614
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/part-00000-f2547b28-9219-4628-8462-cc9c56edfebb-c000.snappy.parquet
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/part-00001-0f755735-3b5b-449a-8f93-92a40d9f065d-c000.snappy.parquet b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/part-00001-0f755735-3b5b-449a-8f93-92a40d9f065d-c000.snappy.parquet
new file mode 100644
index 0000000..b2114ea
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-nested-struct/part-00001-0f755735-3b5b-449a-8f93-92a40d9f065d-c000.snappy.parquet
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-partition-values/_delta_log/00000000000000000000.json b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/_delta_log/00000000000000000000.json
new file mode 100644
index 0000000..c0cc5a3
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/_delta_log/00000000000000000000.json
@@ -0,0 +1,6 @@
+{"commitInfo":{"timestamp":1636147668568,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[\"as_int\",\"as_long\",\"as_byte\",\"as_short\",\"as_boolean\",\"as_float\",\"as_double\",\"as_string\",\"as_string_lit_null\",\"as_date\",\"as_timestamp\",\"as_big_decimal\"]"},"isBlindAppend":true,"operationMetrics":{"numFiles":"3","numOutputBytes":"5832","numOutputRows":"3"}}}
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
+{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"as_int\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_long\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_byte\",\"type\":\"byte\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_short\",\"type\":\"short\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_boolean\",\"type\":\"boolean\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_float\",\"type\":\"float\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_double\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_string\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_string_lit_null\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_date\",\"type\":\"date\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_timestamp\",\"type\":\"timestamp\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_big_decimal\",\"type\":\"decimal(1,0)\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_list_of_records\",\"type\":{\"type\":\"array\",\"elementType\":{\"type\":\"struct\",\"fields\":[{\"name\":\"val\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}}]},\"containsNull\":true},\"nullable\":true,\"metadata\":{}},{\"name\":\"as_nested_struct\",\"type\":{\"type\":\"struct\",\"fields\":[{\"name\":\"aa\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"ab\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"ac\",\"type\":{\"type\":\"struct\",\"fields\":[{\"name\":\"aca\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"acb\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}}]},\"nullable\":true,\"metadata\":{}}]},\"nullable\":true,\"metadata\":{}},{\"name\":\"value\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":["as_int","as_long","as_byte","as_short","as_boolean","as_float","as_double","as_string","as_string_lit_null","as_date","as_timestamp","as_big_decimal"],"configuration":{},"createdTime":1636147666386}}
+{"add":{"path":"as_int=0/as_long=0/as_byte=0/as_short=0/as_boolean=true/as_float=0.0/as_double=0.0/as_string=0/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08%2011%253A11%253A11/as_big_decimal=0/part-00000-b9dc86ae-0134-4363-bd87-19cfb3403e9a.c000.snappy.parquet","partitionValues":{"as_big_decimal":"0","as_int":"0","as_byte":"0","as_long":"0","as_date":"2021-09-08","as_string":"0","as_timestamp":"2021-09-08 11:11:11","as_float":"0.0","as_short":"0","as_boolean":"true","as_string_lit_null":"null","as_double":"0.0"},"size":1944,"modificationTime":1636147668000,"dataChange":true}}
+{"add":{"path":"as_int=__HIVE_DEFAULT_PARTITION__/as_long=__HIVE_DEFAULT_PARTITION__/as_byte=__HIVE_DEFAULT_PARTITION__/as_short=__HIVE_DEFAULT_PARTITION__/as_boolean=__HIVE_DEFAULT_PARTITION__/as_float=__HIVE_DEFAULT_PARTITION__/as_double=__HIVE_DEFAULT_PARTITION__/as_string=__HIVE_DEFAULT_PARTITION__/as_string_lit_null=__HIVE_DEFAULT_PARTITION__/as_date=__HIVE_DEFAULT_PARTITION__/as_timestamp=__HIVE_DEFAULT_PARTITION__/as_big_decimal=__HIVE_DEFAULT_PARTITION__/part-00001-9ee474eb-385b-43cf-9acb-0fbed63e011c.c000.snappy.parquet","partitionValues":{"as_big_decimal":null,"as_int":null,"as_byte":null,"as_long":null,"as_date":null,"as_string":null,"as_timestamp":null,"as_float":null,"as_short":null,"as_boolean":null,"as_string_lit_null":null,"as_double":null},"size":1944,"modificationTime":1636147668000,"dataChange":true}}
+{"add":{"path":"as_int=1/as_long=1/as_byte=1/as_short=1/as_boolean=false/as_float=1.0/as_double=1.0/as_string=1/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08%2011%253A11%253A11/as_big_decimal=1/part-00001-cb007d48-a9f5-40e7-adbe-60920680770f.c000.snappy.parquet","partitionValues":{"as_big_decimal":"1","as_int":"1","as_byte":"1","as_long":"1","as_date":"2021-09-08","as_string":"1","as_timestamp":"2021-09-08 11:11:11","as_float":"1.0","as_short":"1","as_boolean":"false","as_string_lit_null":"null","as_double":"1.0"},"size":1944,"modificationTime":1636147668000,"dataChange":true}}
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=0/as_long=0/as_byte=0/as_short=0/as_boolean=true/as_float=0.0/as_double=0.0/as_string=0/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=0/.part-00000-b9dc86ae-0134-4363-bd87-19cfb3403e9a.c000.snappy.parquet.crc b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=0/as_long=0/as_byte=0/as_short=0/as_boolean=true/as_float=0.0/as_double=0.0/as_string=0/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=0/.part-00000-b9dc86ae-0134-4363-bd87-19cfb3403e9a.c000.snappy.parquet.crc
new file mode 100644
index 0000000..0191061
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=0/as_long=0/as_byte=0/as_short=0/as_boolean=true/as_float=0.0/as_double=0.0/as_string=0/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=0/.part-00000-b9dc86ae-0134-4363-bd87-19cfb3403e9a.c000.snappy.parquet.crc
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=0/as_long=0/as_byte=0/as_short=0/as_boolean=true/as_float=0.0/as_double=0.0/as_string=0/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=0/part-00000-b9dc86ae-0134-4363-bd87-19cfb3403e9a.c000.snappy.parquet b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=0/as_long=0/as_byte=0/as_short=0/as_boolean=true/as_float=0.0/as_double=0.0/as_string=0/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=0/part-00000-b9dc86ae-0134-4363-bd87-19cfb3403e9a.c000.snappy.parquet
new file mode 100644
index 0000000..e4919ae
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=0/as_long=0/as_byte=0/as_short=0/as_boolean=true/as_float=0.0/as_double=0.0/as_string=0/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=0/part-00000-b9dc86ae-0134-4363-bd87-19cfb3403e9a.c000.snappy.parquet
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=1/as_long=1/as_byte=1/as_short=1/as_boolean=false/as_float=1.0/as_double=1.0/as_string=1/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=1/.part-00001-cb007d48-a9f5-40e7-adbe-60920680770f.c000.snappy.parquet.crc b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=1/as_long=1/as_byte=1/as_short=1/as_boolean=false/as_float=1.0/as_double=1.0/as_string=1/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=1/.part-00001-cb007d48-a9f5-40e7-adbe-60920680770f.c000.snappy.parquet.crc
new file mode 100644
index 0000000..b79ff09
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=1/as_long=1/as_byte=1/as_short=1/as_boolean=false/as_float=1.0/as_double=1.0/as_string=1/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=1/.part-00001-cb007d48-a9f5-40e7-adbe-60920680770f.c000.snappy.parquet.crc
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=1/as_long=1/as_byte=1/as_short=1/as_boolean=false/as_float=1.0/as_double=1.0/as_string=1/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=1/part-00001-cb007d48-a9f5-40e7-adbe-60920680770f.c000.snappy.parquet b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=1/as_long=1/as_byte=1/as_short=1/as_boolean=false/as_float=1.0/as_double=1.0/as_string=1/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=1/part-00001-cb007d48-a9f5-40e7-adbe-60920680770f.c000.snappy.parquet
new file mode 100644
index 0000000..b67fcb7
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=1/as_long=1/as_byte=1/as_short=1/as_boolean=false/as_float=1.0/as_double=1.0/as_string=1/as_string_lit_null=null/as_date=2021-09-08/as_timestamp=2021-09-08 11%3A11%3A11/as_big_decimal=1/part-00001-cb007d48-a9f5-40e7-adbe-60920680770f.c000.snappy.parquet
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=__HIVE_DEFAULT_PARTITION__/as_long=__HIVE_DEFAULT_PARTITION__/as_byte=__HIVE_DEFAULT_PARTITION__/as_short=__HIVE_DEFAULT_PARTITION__/as_boolean=__HIVE_DEFAULT_PARTITION__/as_float=__HIVE_DEFAULT_PARTITION__/as_double=__HIVE_DEFAULT_PARTITION__/as_string=__HIVE_DEFAULT_PARTITION__/as_string_lit_null=__HIVE_DEFAULT_PARTITION__/as_date=__HIVE_DEFAULT_PARTITION__/as_timestamp=__HIVE_DEFAULT_PARTITION__/as_big_decimal=__HIVE_DEFAULT_PARTITION__/.part-00001-9ee474eb-385b-43cf-9acb-0fbed63e011c.c000.snappy.parquet.crc b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=__HIVE_DEFAULT_PARTITION__/as_long=__HIVE_DEFAULT_PARTITION__/as_byte=__HIVE_DEFAULT_PARTITION__/as_short=__HIVE_DEFAULT_PARTITION__/as_boolean=__HIVE_DEFAULT_PARTITION__/as_float=__HIVE_DEFAULT_PARTITION__/as_double=__HIVE_DEFAULT_PARTITION__/as_string=__HIVE_DEFAULT_PARTITION__/as_string_lit_null=__HIVE_DEFAULT_PARTITION__/as_date=__HIVE_DEFAULT_PARTITION__/as_timestamp=__HIVE_DEFAULT_PARTITION__/as_big_decimal=__HIVE_DEFAULT_PARTITION__/.part-00001-9ee474eb-385b-43cf-9acb-0fbed63e011c.c000.snappy.parquet.crc
new file mode 100644
index 0000000..bf418f6
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=__HIVE_DEFAULT_PARTITION__/as_long=__HIVE_DEFAULT_PARTITION__/as_byte=__HIVE_DEFAULT_PARTITION__/as_short=__HIVE_DEFAULT_PARTITION__/as_boolean=__HIVE_DEFAULT_PARTITION__/as_float=__HIVE_DEFAULT_PARTITION__/as_double=__HIVE_DEFAULT_PARTITION__/as_string=__HIVE_DEFAULT_PARTITION__/as_string_lit_null=__HIVE_DEFAULT_PARTITION__/as_date=__HIVE_DEFAULT_PARTITION__/as_timestamp=__HIVE_DEFAULT_PARTITION__/as_big_decimal=__HIVE_DEFAULT_PARTITION__/.part-00001-9ee474eb-385b-43cf-9acb-0fbed63e011c.c000.snappy.parquet.crc
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=__HIVE_DEFAULT_PARTITION__/as_long=__HIVE_DEFAULT_PARTITION__/as_byte=__HIVE_DEFAULT_PARTITION__/as_short=__HIVE_DEFAULT_PARTITION__/as_boolean=__HIVE_DEFAULT_PARTITION__/as_float=__HIVE_DEFAULT_PARTITION__/as_double=__HIVE_DEFAULT_PARTITION__/as_string=__HIVE_DEFAULT_PARTITION__/as_string_lit_null=__HIVE_DEFAULT_PARTITION__/as_date=__HIVE_DEFAULT_PARTITION__/as_timestamp=__HIVE_DEFAULT_PARTITION__/as_big_decimal=__HIVE_DEFAULT_PARTITION__/part-00001-9ee474eb-385b-43cf-9acb-0fbed63e011c.c000.snappy.parquet b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=__HIVE_DEFAULT_PARTITION__/as_long=__HIVE_DEFAULT_PARTITION__/as_byte=__HIVE_DEFAULT_PARTITION__/as_short=__HIVE_DEFAULT_PARTITION__/as_boolean=__HIVE_DEFAULT_PARTITION__/as_float=__HIVE_DEFAULT_PARTITION__/as_double=__HIVE_DEFAULT_PARTITION__/as_string=__HIVE_DEFAULT_PARTITION__/as_string_lit_null=__HIVE_DEFAULT_PARTITION__/as_date=__HIVE_DEFAULT_PARTITION__/as_timestamp=__HIVE_DEFAULT_PARTITION__/as_big_decimal=__HIVE_DEFAULT_PARTITION__/part-00001-9ee474eb-385b-43cf-9acb-0fbed63e011c.c000.snappy.parquet
new file mode 100644
index 0000000..4387e52
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-partition-values/as_int=__HIVE_DEFAULT_PARTITION__/as_long=__HIVE_DEFAULT_PARTITION__/as_byte=__HIVE_DEFAULT_PARTITION__/as_short=__HIVE_DEFAULT_PARTITION__/as_boolean=__HIVE_DEFAULT_PARTITION__/as_float=__HIVE_DEFAULT_PARTITION__/as_double=__HIVE_DEFAULT_PARTITION__/as_string=__HIVE_DEFAULT_PARTITION__/as_string_lit_null=__HIVE_DEFAULT_PARTITION__/as_date=__HIVE_DEFAULT_PARTITION__/as_timestamp=__HIVE_DEFAULT_PARTITION__/as_big_decimal=__HIVE_DEFAULT_PARTITION__/part-00001-9ee474eb-385b-43cf-9acb-0fbed63e011c.c000.snappy.parquet
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-primitives/.part-00000-4f2f0b9f-50b3-4e7b-96a1-e2bb0f246b06-c000.snappy.parquet.crc b/contrib/format-deltalake/src/test/resources/data-reader-primitives/.part-00000-4f2f0b9f-50b3-4e7b-96a1-e2bb0f246b06-c000.snappy.parquet.crc
new file mode 100644
index 0000000..11f8928
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-primitives/.part-00000-4f2f0b9f-50b3-4e7b-96a1-e2bb0f246b06-c000.snappy.parquet.crc
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-primitives/.part-00001-09e47b80-36c2-4475-a810-fbd8e7994971-c000.snappy.parquet.crc b/contrib/format-deltalake/src/test/resources/data-reader-primitives/.part-00001-09e47b80-36c2-4475-a810-fbd8e7994971-c000.snappy.parquet.crc
new file mode 100644
index 0000000..852ffc4
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-primitives/.part-00001-09e47b80-36c2-4475-a810-fbd8e7994971-c000.snappy.parquet.crc
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-primitives/_delta_log/00000000000000000000.json b/contrib/format-deltalake/src/test/resources/data-reader-primitives/_delta_log/00000000000000000000.json
new file mode 100644
index 0000000..9c9a0d1
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-primitives/_delta_log/00000000000000000000.json
@@ -0,0 +1,5 @@
+{"commitInfo":{"timestamp":1607520163636,"operation":"WRITE","operationParameters":{"mode":"ErrorIfExists","partitionBy":"[]"},"isBlindAppend":true,"operationMetrics":{"numFiles":"2","numOutputBytes":"5050","numOutputRows":"11"}}}
+{"protocol":{"minReaderVersion":1,"minWriterVersion":2}}
+{"metaData":{"id":"testId","format":{"provider":"parquet","options":{}},"schemaString":"{\"type\":\"struct\",\"fields\":[{\"name\":\"as_int\",\"type\":\"integer\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_long\",\"type\":\"long\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_byte\",\"type\":\"byte\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_short\",\"type\":\"short\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_boolean\",\"type\":\"boolean\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_float\",\"type\":\"float\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_double\",\"type\":\"double\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_string\",\"type\":\"string\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_binary\",\"type\":\"binary\",\"nullable\":true,\"metadata\":{}},{\"name\":\"as_big_decimal\",\"type\":\"decimal(1,0)\",\"nullable\":true,\"metadata\":{}}]}","partitionColumns":[],"configuration":{},"createdTime":1607520161353}}
+{"add":{"path":"part-00000-4f2f0b9f-50b3-4e7b-96a1-e2bb0f246b06-c000.snappy.parquet","partitionValues":{},"size":2482,"modificationTime":1607520163000,"dataChange":true}}
+{"add":{"path":"part-00001-09e47b80-36c2-4475-a810-fbd8e7994971-c000.snappy.parquet","partitionValues":{},"size":2568,"modificationTime":1607520163000,"dataChange":true}}
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-primitives/part-00000-4f2f0b9f-50b3-4e7b-96a1-e2bb0f246b06-c000.snappy.parquet b/contrib/format-deltalake/src/test/resources/data-reader-primitives/part-00000-4f2f0b9f-50b3-4e7b-96a1-e2bb0f246b06-c000.snappy.parquet
new file mode 100644
index 0000000..b0442b0
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-primitives/part-00000-4f2f0b9f-50b3-4e7b-96a1-e2bb0f246b06-c000.snappy.parquet
Binary files differ
diff --git a/contrib/format-deltalake/src/test/resources/data-reader-primitives/part-00001-09e47b80-36c2-4475-a810-fbd8e7994971-c000.snappy.parquet b/contrib/format-deltalake/src/test/resources/data-reader-primitives/part-00001-09e47b80-36c2-4475-a810-fbd8e7994971-c000.snappy.parquet
new file mode 100644
index 0000000..745394c
--- /dev/null
+++ b/contrib/format-deltalake/src/test/resources/data-reader-primitives/part-00001-09e47b80-36c2-4475-a810-fbd8e7994971-c000.snappy.parquet
Binary files differ
diff --git a/contrib/pom.xml b/contrib/pom.xml
index 6fd6925..e728da9 100644
--- a/contrib/pom.xml
+++ b/contrib/pom.xml
@@ -56,6 +56,8 @@
<module>format-xml</module>
<module>format-image</module>
<module>format-pcapng</module>
+ <module>format-iceberg</module>
+ <module>format-deltalake</module>
<module>storage-phoenix</module>
<module>storage-googlesheets</module>
<module>storage-hive</module>
@@ -69,7 +71,6 @@
<module>storage-druid</module>
<module>storage-elasticsearch</module>
<module>storage-cassandra</module>
- <module>format-iceberg</module>
</modules>
</project>
diff --git a/distribution/pom.xml b/distribution/pom.xml
index ce39807..f0fb953 100644
--- a/distribution/pom.xml
+++ b/distribution/pom.xml
@@ -484,6 +484,11 @@
<artifactId>drill-iceberg-format</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.drill.contrib</groupId>
+ <artifactId>drill-deltalake-format</artifactId>
+ <version>${project.version}</version>
+ </dependency>
</dependencies>
</profile>
diff --git a/distribution/src/assemble/component.xml b/distribution/src/assemble/component.xml
index 38d79f0..7708be4 100644
--- a/distribution/src/assemble/component.xml
+++ b/distribution/src/assemble/component.xml
@@ -69,6 +69,7 @@
<include>org.apache.drill.contrib:drill-udfs:jar</include>
<include>org.apache.drill.contrib:drill-druid-storage:jar</include>
<include>org.apache.drill.contrib:drill-iceberg-format:jar</include>
+ <include>org.apache.drill.contrib:drill-deltalake-format:jar</include>
</includes>
<outputDirectory>jars</outputDirectory>
<useProjectArtifact>false</useProjectArtifact>
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/StatisticsProvider.java b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/StatisticsProvider.java
index 5379760..93bb04b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/expr/StatisticsProvider.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/expr/StatisticsProvider.java
@@ -31,11 +31,17 @@
import org.apache.drill.exec.expr.fn.DrillSimpleFuncHolder;
import org.apache.drill.exec.expr.fn.interpreter.InterpreterEvaluator;
import org.apache.drill.exec.expr.holders.BigIntHolder;
+import org.apache.drill.exec.expr.holders.BitHolder;
+import org.apache.drill.exec.expr.holders.DateHolder;
import org.apache.drill.exec.expr.holders.Float4Holder;
import org.apache.drill.exec.expr.holders.Float8Holder;
import org.apache.drill.exec.expr.holders.IntHolder;
+import org.apache.drill.exec.expr.holders.TimeHolder;
import org.apache.drill.exec.expr.holders.TimeStampHolder;
import org.apache.drill.exec.expr.holders.ValueHolder;
+import org.apache.drill.exec.expr.holders.VarDecimalHolder;
+import org.apache.drill.exec.ops.UdfUtilities;
+import org.apache.drill.exec.util.DecimalUtility;
import org.apache.drill.exec.vector.ValueHolderHelper;
import org.apache.drill.metastore.statistics.ColumnStatistics;
import org.apache.drill.metastore.statistics.ColumnStatisticsKind;
@@ -52,10 +58,13 @@
private final Map<SchemaPath, ColumnStatistics<?>> columnStatMap;
private final long rowCount;
+ private final UdfUtilities udfUtilities;
- public StatisticsProvider(Map<SchemaPath, ColumnStatistics<?>> columnStatMap, long rowCount) {
+ public StatisticsProvider(Map<SchemaPath, ColumnStatistics<?>> columnStatMap, long rowCount,
+ UdfUtilities udfUtilities) {
this.columnStatMap = columnStatMap;
this.rowCount = rowCount;
+ this.udfUtilities = udfUtilities;
}
public long getRowCount() {
@@ -154,12 +163,13 @@
private ColumnStatistics<?> evalCastFunc(FunctionHolderExpression holderExpr, ColumnStatistics<T> input) {
try {
- DrillSimpleFuncHolder funcHolder = (DrillSimpleFuncHolder) holderExpr.getHolder();
- DrillSimpleFunc interpreter = funcHolder.createInterpreter();
-
- ValueHolder minHolder;
- ValueHolder maxHolder;
+ T minValue = ComparisonPredicate.getMinValue(input);
+ T maxValue = ComparisonPredicate.getMaxValue(input);
+ if (minValue == null && maxValue == null) {
+ // no need to evaluate cast for null arguments
+ return input;
+ }
TypeProtos.MinorType srcType = holderExpr.args.get(0).getMajorType().getMinorType();
TypeProtos.MinorType destType = holderExpr.getMajorType().getMinorType();
@@ -171,26 +181,33 @@
return null; // cast func between srcType and destType is NOT allowed.
}
+ ValueHolder minHolder;
+ ValueHolder maxHolder;
+
switch (srcType) {
case INT :
- minHolder = ValueHolderHelper.getIntHolder((Integer) ComparisonPredicate.getMinValue(input));
- maxHolder = ValueHolderHelper.getIntHolder((Integer) ComparisonPredicate.getMaxValue(input));
+ minHolder = ValueHolderHelper.getIntHolder((Integer) minValue);
+ maxHolder = ValueHolderHelper.getIntHolder((Integer) maxValue);
break;
case BIGINT:
- minHolder = ValueHolderHelper.getBigIntHolder((Long) ComparisonPredicate.getMinValue(input));
- maxHolder = ValueHolderHelper.getBigIntHolder((Long) ComparisonPredicate.getMaxValue(input));
+ minHolder = ValueHolderHelper.getBigIntHolder((Long) minValue);
+ maxHolder = ValueHolderHelper.getBigIntHolder((Long) maxValue);
break;
case FLOAT4:
- minHolder = ValueHolderHelper.getFloat4Holder((Float) ComparisonPredicate.getMinValue(input));
- maxHolder = ValueHolderHelper.getFloat4Holder((Float) ComparisonPredicate.getMaxValue(input));
+ minHolder = ValueHolderHelper.getFloat4Holder((Float) minValue);
+ maxHolder = ValueHolderHelper.getFloat4Holder((Float) maxValue);
break;
case FLOAT8:
- minHolder = ValueHolderHelper.getFloat8Holder((Double) ComparisonPredicate.getMinValue(input));
- maxHolder = ValueHolderHelper.getFloat8Holder((Double) ComparisonPredicate.getMaxValue(input));
+ minHolder = ValueHolderHelper.getFloat8Holder((Double) minValue);
+ maxHolder = ValueHolderHelper.getFloat8Holder((Double) maxValue);
break;
case DATE:
- minHolder = ValueHolderHelper.getDateHolder((Long) ComparisonPredicate.getMinValue(input));
- maxHolder = ValueHolderHelper.getDateHolder((Long) ComparisonPredicate.getMaxValue(input));
+ minHolder = ValueHolderHelper.getDateHolder((Long) minValue);
+ maxHolder = ValueHolderHelper.getDateHolder((Long) maxValue);
+ break;
+ case VARCHAR:
+ minHolder = ValueHolderHelper.getVarCharHolder(udfUtilities.getManagedBuffer(), (String) minValue);
+ maxHolder = ValueHolderHelper.getVarCharHolder(udfUtilities.getManagedBuffer(), (String) maxValue);
break;
default:
return null;
@@ -199,10 +216,20 @@
ValueHolder[] args1 = {minHolder};
ValueHolder[] args2 = {maxHolder};
+ DrillSimpleFuncHolder funcHolder = (DrillSimpleFuncHolder) holderExpr.getHolder();
+
+ DrillSimpleFunc interpreter = funcHolder.createInterpreter();
+
ValueHolder minFuncHolder = InterpreterEvaluator.evaluateFunction(interpreter, args1, holderExpr.getName());
ValueHolder maxFuncHolder = InterpreterEvaluator.evaluateFunction(interpreter, args2, holderExpr.getName());
switch (destType) {
+ case BIT:
+ return StatisticsProvider.getColumnStatistics(
+ ((BitHolder) minFuncHolder).value,
+ ((BitHolder) maxFuncHolder).value,
+ ColumnStatisticsKind.NULLS_COUNT.getFrom(input),
+ destType);
case INT:
return StatisticsProvider.getColumnStatistics(
((IntHolder) minFuncHolder).value,
@@ -227,12 +254,32 @@
((Float8Holder) maxFuncHolder).value,
ColumnStatisticsKind.NULLS_COUNT.getFrom(input),
destType);
+ case DATE:
+ return StatisticsProvider.getColumnStatistics(
+ ((DateHolder) minFuncHolder).value,
+ ((DateHolder) maxFuncHolder).value,
+ ColumnStatisticsKind.NULLS_COUNT.getFrom(input),
+ destType);
+ case TIME:
+ return StatisticsProvider.getColumnStatistics(
+ ((TimeHolder) minFuncHolder).value,
+ ((TimeHolder) maxFuncHolder).value,
+ ColumnStatisticsKind.NULLS_COUNT.getFrom(input),
+ destType);
case TIMESTAMP:
return StatisticsProvider.getColumnStatistics(
((TimeStampHolder) minFuncHolder).value,
((TimeStampHolder) maxFuncHolder).value,
ColumnStatisticsKind.NULLS_COUNT.getFrom(input),
destType);
+ case VARDECIMAL:
+ VarDecimalHolder minVarDecimalHolder = (VarDecimalHolder) minFuncHolder;
+ VarDecimalHolder maxVarDecimalHolder = (VarDecimalHolder) maxFuncHolder;
+ return StatisticsProvider.getColumnStatistics(
+ DecimalUtility.getBigDecimalFromDrillBuf(minVarDecimalHolder.buffer, minVarDecimalHolder.start, minVarDecimalHolder.scale, minVarDecimalHolder.precision),
+ DecimalUtility.getBigDecimalFromDrillBuf(maxVarDecimalHolder.buffer, maxVarDecimalHolder.start, maxVarDecimalHolder.scale, maxVarDecimalHolder.precision),
+ ColumnStatisticsKind.NULLS_COUNT.getFrom(input),
+ destType);
default:
return null;
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractGroupScanWithMetadata.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractGroupScanWithMetadata.java
index a11f834..2a701ce 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractGroupScanWithMetadata.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/base/AbstractGroupScanWithMetadata.java
@@ -41,6 +41,7 @@
import org.apache.drill.exec.metastore.analyze.FileMetadataInfoCollector;
import org.apache.drill.exec.ops.OptimizerRulesContext;
import org.apache.drill.exec.ops.UdfUtilities;
+import org.apache.drill.exec.physical.impl.scan.v3.FixedReceiver;
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.record.metadata.ColumnMetadata;
@@ -1076,25 +1077,21 @@
for (T metadata : metadataList) {
TupleMetadata schema = metadata.getSchema();
if (schema != null && !tableSchema.isEquivalent(schema)) {
+ schema = FixedReceiver.Builder.mergeSchemas(schema, tableSchema);
filterPredicate = getFilterPredicate(filterExpression, udfUtilities,
context, optionManager, true, true, schema);
}
Map<SchemaPath, ColumnStatistics<?>> columnsStatistics = metadata.getColumnsStatistics();
// adds partition (dir) column statistics if it may be used during filter evaluation
- if (metadata instanceof LocationProvider && optionManager != null) {
- LocationProvider locationProvider = (LocationProvider) metadata;
- columnsStatistics = ParquetTableMetadataUtils.addImplicitColumnsStatistics(columnsStatistics,
- source.columns, source.getPartitionValues(locationProvider), optionManager,
- locationProvider.getPath(), source.supportsFileImplicitColumns());
- }
+ columnsStatistics = getImplicitColumnStatistics(optionManager, metadata, columnsStatistics);
if (source.getNonInterestingColumnsMetadata() != null) {
columnsStatistics.putAll(source.getNonInterestingColumnsMetadata().getColumnsStatistics());
}
RowsMatch match = FilterEvaluatorUtils.matches(filterPredicate,
columnsStatistics, TableStatisticsKind.ROW_COUNT.getValue(metadata),
- schema, schemaPathsInExpr);
+ schema, schemaPathsInExpr, udfUtilities);
if (match == RowsMatch.NONE) {
continue; // No file comply to the filter => drop the file
}
@@ -1109,6 +1106,17 @@
return qualifiedMetadata;
}
+ protected <T extends Metadata> Map<SchemaPath, ColumnStatistics<?>> getImplicitColumnStatistics(
+ OptionManager optionManager, T metadata, Map<SchemaPath, ColumnStatistics<?>> columnsStatistics) {
+ if (metadata instanceof LocationProvider && optionManager != null) {
+ LocationProvider locationProvider = (LocationProvider) metadata;
+ columnsStatistics = ParquetTableMetadataUtils.addImplicitColumnsStatistics(columnsStatistics,
+ source.columns, source.getPartitionValues(locationProvider), optionManager,
+ locationProvider.getPath(), source.supportsFileImplicitColumns());
+ }
+ return columnsStatistics;
+ }
+
protected abstract B self();
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillConstExecutor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillConstExecutor.java
index 09045f5..cefd727 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillConstExecutor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillConstExecutor.java
@@ -127,6 +127,10 @@
this.plannerSettings = plannerSettings;
}
+ public UdfUtilities getUdfUtilities() {
+ return udfUtilities;
+ }
+
@Override
@SuppressWarnings("deprecation")
public void reduce(RexBuilder rexBuilder, List<RexNode> constExps, List<RexNode> reducedValues) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetGroupScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetGroupScan.java
index 3a3800d..cf1fd96 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetGroupScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetGroupScan.java
@@ -663,7 +663,7 @@
AbstractParquetGroupScan abstractParquetGroupScan = (AbstractParquetGroupScan) source;
Map<Path, FileMetadata> filesToFilter = new HashMap<>(prunedFiles);
- if (!abstractParquetGroupScan.rowGroups.isEmpty()) {
+ if (!abstractParquetGroupScan.getRowGroupsMetadata().isEmpty()) {
prunedFiles.forEach((path, fileMetadata) -> {
if (abstractParquetGroupScan.rowGroups.get(path).size() == 1) {
omittedFiles.put(path, fileMetadata);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetRowGroupScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetRowGroupScan.java
index 460fa6c..52a5e4a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetRowGroupScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetRowGroupScan.java
@@ -111,6 +111,10 @@
@JsonProperty
public TupleMetadata getSchema() { return schema; }
+ public boolean isImplicitColumn(SchemaPath path, String partitionColumnLabel) {
+ return path.toString().matches(partitionColumnLabel + "\\d+");
+ }
+
public abstract AbstractParquetRowGroupScan copy(List<SchemaPath> columns);
@JsonIgnore
public abstract Configuration getFsConf(RowGroupReadEntry rowGroupReadEntry) throws IOException;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetScanBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetScanBatchCreator.java
index ffba7f2..f5c252c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetScanBatchCreator.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/AbstractParquetScanBatchCreator.java
@@ -123,7 +123,7 @@
String partitionColumnLabel = context.getOptions().getOption(ExecConstants.FILESYSTEM_PARTITION_COLUMN_LABEL).string_val;
for (SchemaPath path : schemaPathsInExpr) {
if (rowGroupScan.supportsFileImplicitColumns() &&
- path.toString().matches(partitionColumnLabel+"\\d+")) {
+ rowGroupScan.isImplicitColumn(path, partitionColumnLabel)) {
continue; // skip implicit columns like dir0, dir1
}
columnsInExpr.add(SchemaPath.getSimplePath(path.getRootSegmentPath()));
@@ -210,7 +210,7 @@
rowGroupSchema);
}
- matchResult = FilterEvaluatorUtils.matches(filterPredicate, columnsStatistics, footerRowCount, rowGroupSchema, schemaPathsInExpr);
+ matchResult = FilterEvaluatorUtils.matches(filterPredicate, columnsStatistics, footerRowCount, rowGroupSchema, schemaPathsInExpr, context);
// collect logging info
long timeToRead = pruneTimer.elapsed(TimeUnit.MICROSECONDS);
@@ -343,10 +343,7 @@
reader.getClass().getSimpleName());
readers.add(reader);
- List<String> partitionValues = rowGroupScan.getPartitionValues(rowGroup);
- Map<String, String> implicitValues =
- columnExplorer.populateColumns(rowGroup.getPath(), partitionValues,
- rowGroupScan.supportsFileImplicitColumns(), fs, rowGroup.getRowGroupIndex(), rowGroup.getStart(), rowGroup.getLength());
+ Map<String, String> implicitValues = getImplicitValues(rowGroupScan, columnExplorer, rowGroup, fs);
implicitColumns.add(implicitValues);
if (implicitValues.size() > mapWithMaxColumns.size()) {
mapWithMaxColumns = implicitValues;
@@ -354,6 +351,12 @@
return mapWithMaxColumns;
}
+ protected Map<String, String> getImplicitValues(AbstractParquetRowGroupScan rowGroupScan, ColumnExplorer columnExplorer, RowGroupReadEntry rowGroup, DrillFileSystem fs) {
+ List<String> partitionValues = rowGroupScan.getPartitionValues(rowGroup);
+ return columnExplorer.populateColumns(rowGroup.getPath(), partitionValues,
+ rowGroupScan.supportsFileImplicitColumns(), fs, rowGroup.getRowGroupIndex(), rowGroup.getStart(), rowGroup.getLength());
+ }
+
protected abstract AbstractDrillFileSystemManager getDrillFileSystemCreator(OperatorContext operatorContext, OptionManager optionManager);
private ParquetMetadata readFooter(Configuration conf, Path path, ParquetReaderConfig readerConfig) throws IOException {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/FilterEvaluatorUtils.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/FilterEvaluatorUtils.java
index 8d35b1d..2c0942d 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/FilterEvaluatorUtils.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/parquet/FilterEvaluatorUtils.java
@@ -98,7 +98,7 @@
FilterPredicate<?> parquetPredicate = FilterBuilder.buildFilterPredicate(
materializedFilter, constantBoundaries, udfUtilities, true);
- return matches(parquetPredicate, columnsStatistics, rowCount, schema, schemaPathsInExpr);
+ return matches(parquetPredicate, columnsStatistics, rowCount, schema, schemaPathsInExpr, udfUtilities);
}
@SuppressWarnings("unchecked")
@@ -106,12 +106,13 @@
Map<SchemaPath, ColumnStatistics<?>> columnsStatistics,
long rowCount,
TupleMetadata fileMetadata,
- Set<SchemaPath> schemaPathsInExpr) {
+ Set<SchemaPath> schemaPathsInExpr,
+ UdfUtilities udfUtilities) {
if (parquetPredicate == null) {
return RowsMatch.SOME;
}
@SuppressWarnings("rawtypes")
- StatisticsProvider<T> rangeExprEvaluator = new StatisticsProvider(columnsStatistics, rowCount);
+ StatisticsProvider<T> rangeExprEvaluator = new StatisticsProvider(columnsStatistics, rowCount, udfUtilities);
RowsMatch rowsMatch = parquetPredicate.matches(rangeExprEvaluator);
if (rowsMatch == RowsMatch.ALL && isMetaNotApplicable(schemaPathsInExpr, fileMetadata)) {
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 8c91200..d6fade8 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
@@ -51,12 +51,12 @@
/**
* Creates file system only if it was not created before, otherwise returns already created instance.
*/
- private class ParquetDrillFileSystemManager extends AbstractDrillFileSystemManager {
+ public static class ParquetDrillFileSystemManager extends AbstractDrillFileSystemManager {
private final boolean useAsyncPageReader;
private DrillFileSystem fs;
- ParquetDrillFileSystemManager(OperatorContext operatorContext, boolean useAsyncPageReader) {
+ public ParquetDrillFileSystemManager(OperatorContext operatorContext, boolean useAsyncPageReader) {
super(operatorContext);
this.useAsyncPageReader = useAsyncPageReader;
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/AbstractPluginImplementor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/AbstractPluginImplementor.java
index 06384e5..17572f5 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/AbstractPluginImplementor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/AbstractPluginImplementor.java
@@ -143,6 +143,11 @@
return false;
}
+ @Override
+ public boolean artificialFilter() {
+ return false;
+ }
+
private UserException getUnsupported(String rel) {
return UserException.unsupportedError()
.message("Plugin implementor doesn't support push down for %s", rel)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/PluginImplementor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/PluginImplementor.java
index f4589a6..7b54a6f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/PluginImplementor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/PluginImplementor.java
@@ -93,4 +93,12 @@
* to ensure returning the correct rows number.
*/
boolean artificialLimit();
+
+ /**
+ * If the plugin doesn't support native filter pushdown,
+ * but the reader can prune the set of rows to read.
+ * In this case filter operator on top of the scan should be preserved
+ * to ensure returning the correct subset of rows.
+ */
+ boolean artificialFilter();
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginFilterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginFilterRule.java
index e5e02e7..8a2cf7c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginFilterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginFilterRule.java
@@ -24,6 +24,8 @@
import org.apache.drill.exec.store.plan.PluginImplementor;
import org.apache.drill.exec.store.plan.rel.PluginFilterRel;
+import java.util.Collections;
+
/**
* The rule that converts provided filter operator to plugin-specific implementation.
*/
@@ -36,11 +38,15 @@
@Override
public RelNode convert(RelNode rel) {
Filter filter = (Filter) rel;
- return new PluginFilterRel(
- getOutConvention(),
- rel.getCluster(),
- filter.getTraitSet().replace(getOutConvention()),
- convert(filter.getInput(), filter.getTraitSet().replace(getOutConvention())),
- filter.getCondition());
+ PluginFilterRel pluginFilterRel = new PluginFilterRel(
+ getOutConvention(),
+ rel.getCluster(),
+ filter.getTraitSet().replace(getOutConvention()),
+ convert(filter.getInput(), filter.getTraitSet().replace(getOutConvention())),
+ filter.getCondition());
+ if (getPluginImplementor().artificialFilter()) {
+ return filter.copy(filter.getTraitSet(), Collections.singletonList(pluginFilterRel));
+ }
+ return pluginFilterRel;
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginJoinRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginJoinRule.java
index a71060b..9df71ae 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginJoinRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginJoinRule.java
@@ -31,7 +31,7 @@
public class PluginJoinRule extends PluginConverterRule {
public PluginJoinRule(RelTrait in, Convention out, PluginImplementor pluginImplementor) {
- super(Join.class, in, out, "PluginProjectRule", pluginImplementor);
+ super(Join.class, in, out, "PluginJoinRule", pluginImplementor);
}
@Override
