You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by me...@apache.org on 2015/08/02 16:55:26 UTC
[4/4] drill git commit: DRILL-3121: Add support for interpreter based
partition pruning for Hive tables. Remove the old partition pruning logic.
DRILL-3121: Add support for interpreter based partition pruning for Hive tables.
Remove the old partition pruning logic.
Project: http://git-wip-us.apache.org/repos/asf/drill/repo
Commit: http://git-wip-us.apache.org/repos/asf/drill/commit/6d5d7cc2
Tree: http://git-wip-us.apache.org/repos/asf/drill/tree/6d5d7cc2
Diff: http://git-wip-us.apache.org/repos/asf/drill/diff/6d5d7cc2
Branch: refs/heads/master
Commit: 6d5d7cc27140963291e76eac72218b862aa6af45
Parents: 95d576d
Author: Mehant Baid <me...@gmail.com>
Authored: Sat Jul 25 00:28:55 2015 -0700
Committer: Mehant Baid <me...@gmail.com>
Committed: Thu Jul 30 20:23:52 2015 -0700
----------------------------------------------------------------------
.../planner/sql/HivePartitionDescriptor.java | 171 ++++++++++--
.../exec/planner/sql/HivePartitionLocation.java | 54 ++++
.../HivePushPartitionFilterIntoScan.java | 150 ++++-------
.../exec/store/hive/HiveDataTypeUtility.java | 122 +++++++++
.../drill/exec/store/hive/HiveRecordReader.java | 48 +---
.../exec/store/hive/HiveStoragePlugin.java | 2 +-
.../apache/drill/exec/store/hive/HiveTable.java | 22 +-
.../drill/exec/TestHivePartitionPruning.java | 28 +-
.../exec/planner/logical/DirPathBuilder.java | 265 -------------------
.../DrillPushPartitionFilterIntoScan.java | 139 ----------
.../planner/logical/PartitionPruningUtil.java | 78 ------
11 files changed, 424 insertions(+), 655 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
index 93bfadd..d323db9 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionDescriptor.java
@@ -17,29 +17,54 @@
*/
package org.apache.drill.exec.planner.sql;
+import org.apache.calcite.util.BitSets;
+import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.exec.expr.fn.impl.DateUtility;
import org.apache.drill.exec.physical.base.GroupScan;
import org.apache.drill.exec.planner.PartitionDescriptor;
import org.apache.drill.exec.planner.PartitionLocation;
+import org.apache.drill.exec.planner.logical.DrillScanRel;
import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.store.hive.HiveDataTypeUtility;
+import org.apache.drill.exec.store.hive.HiveReadEntry;
+import org.apache.drill.exec.store.hive.HiveScan;
import org.apache.drill.exec.store.hive.HiveTable;
+import org.apache.drill.exec.vector.NullableBigIntVector;
+import org.apache.drill.exec.vector.NullableDateVector;
+import org.apache.drill.exec.vector.NullableFloat4Vector;
+import org.apache.drill.exec.vector.NullableFloat8Vector;
+import org.apache.drill.exec.vector.NullableIntVector;
+import org.apache.drill.exec.vector.NullableTimeStampVector;
+import org.apache.drill.exec.vector.NullableVarCharVector;
import org.apache.drill.exec.vector.ValueVector;
+import org.apache.hadoop.hive.metastore.api.Partition;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+import org.joda.time.format.DateTimeFormatter;
+import java.sql.Timestamp;
import java.util.BitSet;
import java.util.HashMap;
+import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+
// Partition descriptor for hive tables
public class HivePartitionDescriptor implements PartitionDescriptor {
private final Map<String, Integer> partitionMap = new HashMap<>();
private final int MAX_NESTED_SUBDIRS;
+ private final DrillScanRel scanRel;
- public HivePartitionDescriptor(List<HiveTable.FieldSchemaWrapper> partitionName) {
+ public HivePartitionDescriptor(PlannerSettings settings, DrillScanRel scanRel) {
int i = 0;
- for (HiveTable.FieldSchemaWrapper wrapper : partitionName) {
+ this.scanRel = scanRel;
+ for (HiveTable.FieldSchemaWrapper wrapper : ((HiveScan) scanRel.getGroupScan()).hiveReadEntry.table.partitionKeys) {
partitionMap.put(wrapper.name, i);
i++;
}
@@ -61,37 +86,145 @@ public class HivePartitionDescriptor implements PartitionDescriptor {
return MAX_NESTED_SUBDIRS;
}
- @Override
- public Integer getIdIfValid(String name) {
- return partitionMap.get(name);
- }
-
- /*
- * Following method stubs are just added to satisfy the interface implementation.
- * Actual implementation will be added when hive partition pruning is plugged in
- * as part of DRILL-3121
- */
- private String getBaseTableLocation() {
- return null;
+ public String getBaseTableLocation() {
+ HiveReadEntry origEntry = ((HiveScan) scanRel.getGroupScan()).hiveReadEntry;
+ return origEntry.table.getTable().getSd().getLocation();
}
@Override
- public GroupScan createNewGroupScan(List<String> newFiles) throws Exception {
- return null;
+ public GroupScan createNewGroupScan(List<String> newFiles) throws ExecutionSetupException {
+ HiveScan hiveScan = (HiveScan) scanRel.getGroupScan();
+ HiveReadEntry origReadEntry = hiveScan.hiveReadEntry;
+ List<HiveTable.HivePartition> oldPartitions = origReadEntry.partitions;
+ List<HiveTable.HivePartition> newPartitions = new LinkedList<>();
+
+ for (HiveTable.HivePartition part: oldPartitions) {
+ String partitionLocation = part.getPartition().getSd().getLocation();
+ for (String newPartitionLocation: newFiles) {
+ if (partitionLocation.equals(newPartitionLocation)) {
+ newPartitions.add(part);
+ }
+ }
+ }
+
+ HiveReadEntry newReadEntry = new HiveReadEntry(origReadEntry.table, newPartitions, origReadEntry.hiveConfigOverride);
+ HiveScan newScan = new HiveScan(hiveScan.getUserName(), newReadEntry, hiveScan.storagePlugin, hiveScan.columns);
+ return newScan;
}
@Override
public List<PartitionLocation> getPartitions() {
- return null;
+ List<PartitionLocation> partitions = new LinkedList<>();
+ HiveReadEntry origEntry = ((HiveScan) scanRel.getGroupScan()).hiveReadEntry;
+ List<String> allFileLocations = new LinkedList<>();
+ for (Partition partition: origEntry.getPartitions()) {
+ allFileLocations.add(partition.getSd().getLocation());
+ }
+ for (String file: allFileLocations) {
+ partitions.add(new HivePartitionLocation(MAX_NESTED_SUBDIRS, getBaseTableLocation(),file));
+ }
+ return partitions;
}
@Override
- public void populatePartitionVectors(ValueVector[] vectors, List<PartitionLocation> partitions, BitSet partitionColumnBitSet, Map<Integer, String> fieldNameMap) {
+ public void populatePartitionVectors(ValueVector[] vectors, List<PartitionLocation> partitions,
+ BitSet partitionColumnBitSet, Map<Integer, String> fieldNameMap) {
+ int record = 0;
+ for(PartitionLocation partitionLocation: partitions) {
+ for(int partitionColumnIndex : BitSets.toIter(partitionColumnBitSet)){
+ populateVector(vectors[partitionColumnIndex], partitionLocation.getPartitionValue(partitionColumnIndex),
+ record);
+ }
+ record++;
+ }
+ for(ValueVector v : vectors) {
+ if(v == null){
+ continue;
+ }
+ v.getMutator().setValueCount(partitions.size());
+ }
}
@Override
public TypeProtos.MajorType getVectorType(SchemaPath column, PlannerSettings plannerSettings) {
- return null;
+ HiveScan hiveScan = (HiveScan) scanRel.getGroupScan();
+ String partitionName = column.getAsNamePart().getName();
+ Map<String, String> partitionNameTypeMap = hiveScan.hiveReadEntry.table.getPartitionNameTypeMap();
+ String hiveType = partitionNameTypeMap.get(partitionName);
+ PrimitiveTypeInfo primitiveTypeInfo = (PrimitiveTypeInfo) TypeInfoUtils.getTypeInfoFromTypeString(hiveType);
+
+ TypeProtos.MinorType partitionType = HiveDataTypeUtility.getMinorTypeFromHivePrimitiveTypeInfo(primitiveTypeInfo,
+ plannerSettings.getOptions());
+ return TypeProtos.MajorType.newBuilder().setMode(TypeProtos.DataMode.OPTIONAL).setMinorType(partitionType).build();
+ }
+
+ @Override
+ public Integer getIdIfValid(String name) {
+ return partitionMap.get(name);
+ }
+
+ public static void populateVector(ValueVector vector, String value, int record) {
+ TypeProtos.MinorType type = vector.getField().getType().getMinorType();
+
+ switch (type) {
+ case TINYINT:
+ case SMALLINT:
+ case INT:
+ if (value == null) {
+ ((NullableIntVector) vector).getMutator().setNull(record);
+ } else {
+ ((NullableIntVector) vector).getMutator().setSafe(record, Integer.parseInt(value));
+ }
+ break;
+
+ case BIGINT:
+ if (value == null) {
+ ((NullableBigIntVector) vector).getMutator().setNull(record);
+ } else {
+ ((NullableBigIntVector) vector).getMutator().setSafe(record, Long.parseLong(value));
+ }
+ break;
+ case FLOAT4:
+ if (value == null) {
+ ((NullableFloat4Vector) vector).getMutator().setNull(record);
+ } else {
+ ((NullableFloat4Vector) vector).getMutator().setSafe(record, Float.parseFloat(value));
+ }
+ break;
+ case FLOAT8:
+ if (value == null) {
+ ((NullableFloat8Vector) vector).getMutator().setNull(record);
+ } else {
+ ((NullableFloat8Vector) vector).getMutator().setSafe(record, Double.parseDouble(value));
+ }
+ break;
+ case TIMESTAMP:
+ if (value == null) {
+ ((NullableTimeStampVector) vector).getMutator().setNull(record);
+ } else {
+ DateTimeFormatter f = DateUtility.getDateTimeFormatter();
+ value = value.replace("%3A", ":");
+ long ts = DateTime.parse(value, f).withZoneRetainFields(DateTimeZone.UTC).getMillis();
+ ((NullableTimeStampVector) vector).getMutator().set(record, ts);
+ }
+ break;
+ case DATE:
+ if (value == null) {
+ ((NullableDateVector) vector).getMutator().setNull(record);
+ } else {
+ DateTimeFormatter f = DateUtility.formatDate;
+ long ts = DateTime.parse(value, f).withZoneRetainFields(DateTimeZone.UTC).getMillis();
+ ((NullableDateVector) vector).getMutator().set(record, ts);
+ }
+ break;
+ case VARCHAR:
+ if (value == null) {
+ ((NullableVarCharVector) vector).getMutator().setNull(record);
+ } else {
+ ((NullableVarCharVector) vector).getMutator().set(record, value.getBytes());
+ }
+ break;
+ }
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionLocation.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionLocation.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionLocation.java
new file mode 100644
index 0000000..e3066a4
--- /dev/null
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/HivePartitionLocation.java
@@ -0,0 +1,54 @@
+/**
+ * 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.planner.sql;
+
+import org.apache.drill.exec.planner.PartitionLocation;
+
+public class HivePartitionLocation implements PartitionLocation {
+ private final String partitionLocation;
+ private final String[] partitionValue;
+ private final String fileSeparator = System.getProperty("file.separator");
+ public HivePartitionLocation(int max, String baseTableLocation, String entireLocation) {
+ this.partitionLocation = entireLocation;
+ partitionValue = new String[max];
+ int start = partitionLocation.indexOf(baseTableLocation) + baseTableLocation.length();
+ String postPath = entireLocation.substring(start);
+ if (postPath.length() == 0) {
+ return;
+ }
+ if (postPath.startsWith(fileSeparator)) {
+ postPath = postPath.substring(postPath.indexOf(fileSeparator) + 1);
+ }
+ String[] mostDirs = postPath.split(fileSeparator);
+ assert mostDirs.length <= max;
+ for (int i = 0; i < mostDirs.length; i++) {
+ this.partitionValue[i] = mostDirs[i].substring(mostDirs[i].indexOf("=") + 1);
+ }
+
+ }
+ @Override
+ public String getPartitionValue(int index) {
+ assert index < partitionValue.length;
+ return partitionValue[index];
+ }
+
+ @Override
+ public String getEntirePartitionLocation() {
+ return partitionLocation;
+ }
+}
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/HivePushPartitionFilterIntoScan.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/HivePushPartitionFilterIntoScan.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/HivePushPartitionFilterIntoScan.java
index 6ab1a78..90b0c5f 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/HivePushPartitionFilterIntoScan.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/planner/sql/logical/HivePushPartitionFilterIntoScan.java
@@ -18,125 +18,75 @@
package org.apache.drill.exec.planner.sql.logical;
-import java.util.LinkedList;
-import java.util.List;
-
-import org.apache.drill.common.exceptions.DrillRuntimeException;
-import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.exec.ops.OptimizerRulesContext;
import org.apache.drill.exec.physical.base.GroupScan;
-import org.apache.drill.exec.planner.logical.DirPathBuilder;
+import org.apache.drill.exec.planner.PartitionDescriptor;
import org.apache.drill.exec.planner.logical.DrillFilterRel;
import org.apache.drill.exec.planner.logical.DrillProjectRel;
-import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillScanRel;
-import org.apache.drill.exec.planner.logical.PartitionPruningUtil;
import org.apache.drill.exec.planner.logical.RelOptHelper;
+import org.apache.drill.exec.planner.logical.partition.PruneScanRule;
+import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.planner.sql.HivePartitionDescriptor;
import org.apache.drill.exec.store.StoragePluginOptimizerRule;
-import org.apache.drill.exec.store.hive.HiveReadEntry;
import org.apache.drill.exec.store.hive.HiveScan;
-import org.apache.drill.exec.store.hive.HiveTable;
-import org.apache.drill.exec.store.hive.HiveTable.HivePartition;
import org.apache.calcite.plan.RelOptRuleCall;
-import org.apache.calcite.plan.RelOptRuleOperand;
-
-import com.google.common.collect.Lists;
-
-public abstract class HivePushPartitionFilterIntoScan extends StoragePluginOptimizerRule {
-
- public static final StoragePluginOptimizerRule HIVE_FILTER_ON_PROJECT =
- new HivePushPartitionFilterIntoScan(
- RelOptHelper.some(DrillFilterRel.class, RelOptHelper.some(DrillProjectRel.class, RelOptHelper.any(DrillScanRel.class))),
- "HivePushPartitionFilterIntoScan:Filter_On_Project") {
- @Override
- public boolean matches(RelOptRuleCall call) {
- final DrillScanRel scan = (DrillScanRel) call.rel(2);
- GroupScan groupScan = scan.getGroupScan();
- return groupScan instanceof HiveScan && groupScan.supportsPartitionFilterPushdown();
- }
+public abstract class HivePushPartitionFilterIntoScan {
- @Override
- public void onMatch(RelOptRuleCall call) {
- final DrillFilterRel filterRel = (DrillFilterRel) call.rel(0);
- final DrillProjectRel projectRel = (DrillProjectRel) call.rel(1);
- final DrillScanRel scanRel = (DrillScanRel) call.rel(2);
- doOnMatch(call, filterRel, projectRel, scanRel);
- }
- };
+ public static final StoragePluginOptimizerRule getFilterOnProject(OptimizerRulesContext optimizerRulesContext) {
+ return new PruneScanRule(
+ RelOptHelper.some(DrillFilterRel.class, RelOptHelper.some(DrillProjectRel.class, RelOptHelper.any(DrillScanRel.class))),
+ "HivePushPartitionFilterIntoScan:Filter_On_Project_Hive",
+ optimizerRulesContext) {
- public static final StoragePluginOptimizerRule HIVE_FILTER_ON_SCAN =
- new HivePushPartitionFilterIntoScan(
- RelOptHelper.some(DrillFilterRel.class, RelOptHelper.any(DrillScanRel.class)),
- "HivePushPartitionFilterIntoScan:Filter_On_Scan") {
-
- @Override
- public boolean matches(RelOptRuleCall call) {
- final DrillScanRel scan = (DrillScanRel) call.rel(1);
- GroupScan groupScan = scan.getGroupScan();
- return groupScan instanceof HiveScan && groupScan.supportsPartitionFilterPushdown();
- }
-
- @Override
- public void onMatch(RelOptRuleCall call) {
- final DrillFilterRel filterRel = (DrillFilterRel) call.rel(0);
- final DrillScanRel scanRel = (DrillScanRel) call.rel(1);
- doOnMatch(call, filterRel, null, scanRel);
- }
- };
-
- private HivePushPartitionFilterIntoScan(
- RelOptRuleOperand operand,
- String id) {
- super(operand, id);
- }
-
- private HiveReadEntry splitFilter(HiveReadEntry origReadEntry, DirPathBuilder builder) {
- HiveTable table = origReadEntry.table;
- List<HivePartition> partitions = origReadEntry.partitions;
- List<HivePartition> newPartitions = new LinkedList<>();
- String pathPrefix = PartitionPruningUtil.truncatePrefixFromPath(table.getTable().getSd().getLocation());
- List<String> newFiles = Lists.newArrayList();
- List<String> dirPathList = builder.getDirPath();
-
- for (String dirPath : dirPathList) {
- String fullPath = pathPrefix + dirPath;
- // check containment of this path in the list of files
- for (HivePartition part: partitions) {
- String origFilePath = origReadEntry.getPartitionLocation(part);
- String origFileName = PartitionPruningUtil.truncatePrefixFromPath(origFilePath);
+ @Override
+ public PartitionDescriptor getPartitionDescriptor(PlannerSettings settings, DrillScanRel scanRel) {
+ return new HivePartitionDescriptor(settings, scanRel);
+ }
- if (origFileName.startsWith(fullPath)) {
- newFiles.add(origFileName);
- newPartitions.add(part);
- }
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ final DrillScanRel scan = (DrillScanRel) call.rel(2);
+ GroupScan groupScan = scan.getGroupScan();
+ // this rule is applicable only for Hive based partition pruning
+ return groupScan instanceof HiveScan && groupScan.supportsPartitionFilterPushdown();
}
- }
- if (newFiles.size() > 0) {
- HiveReadEntry newReadEntry = new HiveReadEntry(table, newPartitions, origReadEntry.hiveConfigOverride);
- return newReadEntry;
- }
- return origReadEntry;
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final DrillFilterRel filterRel = (DrillFilterRel) call.rel(0);
+ final DrillProjectRel projectRel = (DrillProjectRel) call.rel(1);
+ final DrillScanRel scanRel = (DrillScanRel) call.rel(2);
+ doOnMatch(call, filterRel, projectRel, scanRel);
+ }
+ };
}
- protected void doOnMatch(RelOptRuleCall call, DrillFilterRel filterRel, DrillProjectRel projectRel, DrillScanRel scanRel) {
- DrillRel inputRel = projectRel != null ? projectRel : scanRel;
- HiveReadEntry origReadEntry = ((HiveScan)scanRel.getGroupScan()).hiveReadEntry;
- DirPathBuilder builder = new DirPathBuilder(filterRel, inputRel, filterRel.getCluster().getRexBuilder(), new HivePartitionDescriptor(origReadEntry.table.partitionKeys));
- HiveReadEntry newReadEntry = splitFilter(origReadEntry, builder);
+ public static final StoragePluginOptimizerRule getFilterOnScan(OptimizerRulesContext optimizerRulesContext) {
+ return new PruneScanRule(
+ RelOptHelper.some(DrillFilterRel.class, RelOptHelper.any(DrillScanRel.class)),
+ "HivePushPartitionFilterIntoScan:Filter_On_Scan_Hive", optimizerRulesContext) {
- if (origReadEntry == newReadEntry) {
- return; // no directory filter was pushed down
- }
+ @Override
+ public PartitionDescriptor getPartitionDescriptor(PlannerSettings settings, DrillScanRel scanRel) {
+ return new HivePartitionDescriptor(settings, scanRel);
+ }
- try {
- HiveScan oldScan = (HiveScan) scanRel.getGroupScan();
- HiveScan hiveScan = new HiveScan(oldScan.getUserName(), newReadEntry, oldScan.storagePlugin, oldScan.columns);
- PartitionPruningUtil.rewritePlan(call, filterRel, projectRel, scanRel, hiveScan, builder);
- } catch (ExecutionSetupException e) {
- throw new DrillRuntimeException(e);
- }
+ @Override
+ public boolean matches(RelOptRuleCall call) {
+ final DrillScanRel scan = (DrillScanRel) call.rel(1);
+ GroupScan groupScan = scan.getGroupScan();
+ // this rule is applicable only for Hive based partition pruning
+ return groupScan instanceof HiveScan && groupScan.supportsPartitionFilterPushdown();
+ }
+ @Override
+ public void onMatch(RelOptRuleCall call) {
+ final DrillFilterRel filterRel = (DrillFilterRel) call.rel(0);
+ final DrillScanRel scanRel = (DrillScanRel) call.rel(1);
+ doOnMatch(call, filterRel, null, scanRel);
+ }
+ };
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDataTypeUtility.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDataTypeUtility.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDataTypeUtility.java
new file mode 100644
index 0000000..84d8790
--- /dev/null
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveDataTypeUtility.java
@@ -0,0 +1,122 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.hive;
+
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.exec.expr.holders.Decimal18Holder;
+import org.apache.drill.exec.expr.holders.Decimal28SparseHolder;
+import org.apache.drill.exec.expr.holders.Decimal38SparseHolder;
+import org.apache.drill.exec.expr.holders.Decimal9Holder;
+import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.exec.util.DecimalUtility;
+import org.apache.drill.exec.vector.AllocationHelper;
+import org.apache.drill.exec.vector.BigIntVector;
+import org.apache.drill.exec.vector.BitVector;
+import org.apache.drill.exec.vector.DateVector;
+import org.apache.drill.exec.vector.Decimal18Vector;
+import org.apache.drill.exec.vector.Decimal28SparseVector;
+import org.apache.drill.exec.vector.Decimal38SparseVector;
+import org.apache.drill.exec.vector.Decimal9Vector;
+import org.apache.drill.exec.vector.Float4Vector;
+import org.apache.drill.exec.vector.Float8Vector;
+import org.apache.drill.exec.vector.IntVector;
+import org.apache.drill.exec.vector.TimeStampVector;
+import org.apache.drill.exec.vector.ValueVector;
+import org.apache.drill.exec.vector.VarBinaryVector;
+import org.apache.drill.exec.vector.VarCharVector;
+import org.apache.drill.exec.work.ExecErrorConstants;
+
+import org.apache.hadoop.hive.serde2.ColumnProjectionUtils;
+import org.apache.hadoop.hive.serde2.SerDe;
+import org.apache.hadoop.hive.serde2.SerDeException;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector;
+import org.apache.hadoop.hive.serde2.objectinspector.ObjectInspector.Category;
+import org.apache.hadoop.hive.serde2.objectinspector.PrimitiveObjectInspector.PrimitiveCategory;
+import org.apache.hadoop.hive.serde2.objectinspector.StructObjectInspector;
+import org.apache.hadoop.hive.serde2.typeinfo.DecimalTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.HiveDecimalUtils;
+import org.apache.hadoop.hive.serde2.typeinfo.PrimitiveTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.StructTypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfo;
+import org.apache.hadoop.hive.serde2.typeinfo.TypeInfoUtils;
+import org.joda.time.DateTime;
+import org.joda.time.DateTimeZone;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Timestamp;
+
+public class HiveDataTypeUtility {
+ static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(HiveDataTypeUtility.class);
+
+ public static TypeProtos.MinorType getMinorTypeFromHivePrimitiveTypeInfo(PrimitiveTypeInfo primitiveTypeInfo,
+ OptionManager options) {
+ switch(primitiveTypeInfo.getPrimitiveCategory()) {
+ case BINARY:
+ return TypeProtos.MinorType.VARBINARY;
+ case BOOLEAN:
+ return TypeProtos.MinorType.BIT;
+ case DECIMAL: {
+
+ if (options.getOption(PlannerSettings.ENABLE_DECIMAL_DATA_TYPE_KEY).bool_val == false) {
+ throw UserException.unsupportedError()
+ .message(ExecErrorConstants.DECIMAL_DISABLE_ERR_MSG)
+ .build(logger);
+ }
+ DecimalTypeInfo decimalTypeInfo = (DecimalTypeInfo) primitiveTypeInfo;
+ return DecimalUtility.getDecimalDataType(decimalTypeInfo.precision());
+ }
+ case DOUBLE:
+ return TypeProtos.MinorType.FLOAT8;
+ case FLOAT:
+ return TypeProtos.MinorType.FLOAT4;
+ // TODO (DRILL-2470)
+ // Byte and short (tinyint and smallint in SQL types) are currently read as integers
+ // as these smaller integer types are not fully supported in Drill today.
+ case SHORT:
+ case BYTE:
+ case INT:
+ return TypeProtos.MinorType.INT;
+ case LONG:
+ return TypeProtos.MinorType.BIGINT;
+ case STRING:
+ case VARCHAR:
+ return TypeProtos.MinorType.VARCHAR;
+ case TIMESTAMP:
+ return TypeProtos.MinorType.TIMESTAMP;
+ case DATE:
+ return TypeProtos.MinorType.DATE;
+ }
+ throwUnsupportedHiveDataTypeError(primitiveTypeInfo.getPrimitiveCategory().toString());
+ return null;
+ }
+
+ public static void throwUnsupportedHiveDataTypeError(String unsupportedType) {
+ StringBuilder errMsg = new StringBuilder();
+ errMsg.append(String.format("Unsupported Hive data type %s. ", unsupportedType));
+ errMsg.append(System.getProperty("line.separator"));
+ errMsg.append("Following Hive data types are supported in Drill for querying: ");
+ errMsg.append(
+ "BOOLEAN, BYTE, SHORT, INT, LONG, FLOAT, DOUBLE, DATE, TIMESTAMP, BINARY, DECIMAL, STRING, and VARCHAR");
+
+ throw new RuntimeException(errMsg.toString());
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveRecordReader.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveRecordReader.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveRecordReader.java
index 088fb74..1a66ad9 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveRecordReader.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveRecordReader.java
@@ -126,7 +126,7 @@ public class HiveRecordReader extends AbstractRecordReader {
protected static final int FIELD_SIZE = 50;
public HiveRecordReader(Table table, Partition partition, InputSplit inputSplit, List<SchemaPath> projectedColumns,
- FragmentContext context, Map<String, String> hiveConfigOverride) throws ExecutionSetupException {
+ FragmentContext context, Map<String, String> hiveConfigOverride) throws ExecutionSetupException {
this.table = table;
this.partition = partition;
this.inputSplit = inputSplit;
@@ -353,53 +353,11 @@ public class HiveRecordReader extends AbstractRecordReader {
}
}
- private MinorType getMinorTypeFromHivePrimitiveTypeInfo(PrimitiveTypeInfo primitiveTypeInfo) {
- switch(primitiveTypeInfo.getPrimitiveCategory()) {
- case BINARY:
- return TypeProtos.MinorType.VARBINARY;
- case BOOLEAN:
- return MinorType.BIT;
- case DECIMAL: {
-
- if (context.getOptions().getOption(PlannerSettings.ENABLE_DECIMAL_DATA_TYPE_KEY).bool_val == false) {
- throw UserException.unsupportedError()
- .message(ExecErrorConstants.DECIMAL_DISABLE_ERR_MSG)
- .build(logger);
- }
- DecimalTypeInfo decimalTypeInfo = (DecimalTypeInfo) primitiveTypeInfo;
- return DecimalUtility.getDecimalDataType(decimalTypeInfo.precision());
- }
- case DOUBLE:
- return MinorType.FLOAT8;
- case FLOAT:
- return MinorType.FLOAT4;
- // TODO (DRILL-2470)
- // Byte and short (tinyint and smallint in SQL types) are currently read as integers
- // as these smaller integer types are not fully supported in Drill today.
- case SHORT:
- case BYTE:
- case INT:
- return MinorType.INT;
- case LONG:
- return MinorType.BIGINT;
- case STRING:
- case VARCHAR:
- return MinorType.VARCHAR;
- case TIMESTAMP:
- return MinorType.TIMESTAMP;
- case DATE:
- return MinorType.DATE;
- }
-
- throwUnsupportedHiveDataTypeError(primitiveTypeInfo.getPrimitiveCategory().toString());
- return null;
- }
-
public MajorType getMajorTypeFromHiveTypeInfo(TypeInfo typeInfo, boolean nullable) {
switch (typeInfo.getCategory()) {
case PRIMITIVE: {
PrimitiveTypeInfo primitiveTypeInfo = (PrimitiveTypeInfo) typeInfo;
- MinorType minorType = getMinorTypeFromHivePrimitiveTypeInfo(primitiveTypeInfo);
+ MinorType minorType = HiveDataTypeUtility.getMinorTypeFromHivePrimitiveTypeInfo(primitiveTypeInfo, context.getOptions());
MajorType.Builder typeBuilder = MajorType.newBuilder().setMinorType(minorType)
.setMode((nullable ? DataMode.OPTIONAL : DataMode.REQUIRED));
@@ -526,7 +484,7 @@ public class HiveRecordReader extends AbstractRecordReader {
}
private void populateDecimalPartitionVector(DecimalTypeInfo typeInfo, ValueVector vector, BigDecimal bigDecimal,
- int recordCount) {
+ int recordCount) {
int precision = typeInfo.precision();
int scale = typeInfo.scale();
if (precision <= 9) {
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveStoragePlugin.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveStoragePlugin.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveStoragePlugin.java
index 4813c5e..22552b7 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveStoragePlugin.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveStoragePlugin.java
@@ -89,7 +89,7 @@ public class HiveStoragePlugin extends AbstractStoragePlugin {
@Override
public Set<StoragePluginOptimizerRule> getOptimizerRules(OptimizerRulesContext optimizerRulesContext) {
- return ImmutableSet.of(HivePushPartitionFilterIntoScan.HIVE_FILTER_ON_PROJECT, HivePushPartitionFilterIntoScan.HIVE_FILTER_ON_SCAN);
+ return ImmutableSet.of(HivePushPartitionFilterIntoScan.getFilterOnProject(optimizerRulesContext), HivePushPartitionFilterIntoScan.getFilterOnScan(optimizerRulesContext));
}
}
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTable.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTable.java b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTable.java
index 99101cc..88fe8c3 100644
--- a/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTable.java
+++ b/contrib/storage-hive/core/src/main/java/org/apache/drill/exec/store/hive/HiveTable.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.store.hive;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
@@ -64,12 +65,15 @@ public class HiveTable {
@JsonProperty
public String tableType;
+ @JsonIgnore
+ public final Map<String, String> partitionNameTypeMap = new HashMap();
+
@JsonCreator
public HiveTable(@JsonProperty("tableName") String tableName, @JsonProperty("dbName") String dbName, @JsonProperty("owner") String owner, @JsonProperty("createTime") int createTime,
@JsonProperty("lastAccessTime") int lastAccessTime, @JsonProperty("retention") int retention, @JsonProperty("sd") StorageDescriptorWrapper sd,
@JsonProperty("partitionKeys") List<FieldSchemaWrapper> partitionKeys, @JsonProperty("parameters") Map<String, String> parameters,
@JsonProperty("viewOriginalText") String viewOriginalText, @JsonProperty("viewExpandedText") String viewExpandedText, @JsonProperty("tableType") String tableType
- ) {
+ ) {
this.tableName = tableName;
this.dbName = dbName;
this.owner = owner;
@@ -86,10 +90,11 @@ public class HiveTable {
List<FieldSchema> partitionKeysUnwrapped = Lists.newArrayList();
for (FieldSchemaWrapper w : partitionKeys) {
partitionKeysUnwrapped.add(w.getFieldSchema());
+ partitionNameTypeMap.put(w.name, w.type);
}
StorageDescriptor sdUnwrapped = sd.getSd();
this.table = new Table(tableName, dbName, owner, createTime, lastAccessTime, retention, sdUnwrapped, partitionKeysUnwrapped,
- parameters, viewOriginalText, viewExpandedText, tableType);
+ parameters, viewOriginalText, viewExpandedText, tableType);
}
public HiveTable(Table table) {
@@ -107,6 +112,7 @@ public class HiveTable {
this.partitionKeys = Lists.newArrayList();
for (FieldSchema f : table.getPartitionKeys()) {
this.partitionKeys.add(new FieldSchemaWrapper(f));
+ partitionNameTypeMap.put(f.getName(), f.getType());
}
this.parameters = table.getParameters();
this.viewOriginalText = table.getViewOriginalText();
@@ -156,8 +162,8 @@ public class HiveTable {
@JsonCreator
public HivePartition(@JsonProperty("values") List<String> values, @JsonProperty("tableName") String tableName, @JsonProperty("dbName") String dbName, @JsonProperty("createTime") int createTime,
- @JsonProperty("lastAccessTime") int lastAccessTime, @JsonProperty("sd") StorageDescriptorWrapper sd,
- @JsonProperty("parameters") Map<String, String> parameters
+ @JsonProperty("lastAccessTime") int lastAccessTime, @JsonProperty("sd") StorageDescriptorWrapper sd,
+ @JsonProperty("parameters") Map<String, String> parameters
) {
this.values = values;
this.tableName = tableName;
@@ -217,7 +223,7 @@ public class HiveTable {
public int numBuckets;
@JsonProperty
public SerDeInfoWrapper serDeInfo;
-// @JsonProperty
+ // @JsonProperty
// public List<String> bucketCols;
@JsonProperty
public List<OrderWrapper> sortCols;
@@ -251,7 +257,7 @@ public class HiveTable {
// this.sd = new StorageDescriptor(colsUnwrapped, location, inputFormat, outputFormat, compressed, numBuckets, serDeInfoUnwrapped,
// bucketCols, sortColsUnwrapped, parameters);
this.sd = new StorageDescriptor(colsUnwrapped, location, inputFormat, outputFormat, compressed, numBuckets, serDeInfoUnwrapped,
- null, sortColsUnwrapped, parameters);
+ null, sortColsUnwrapped, parameters);
}
public StorageDescriptorWrapper(StorageDescriptor sd) {
@@ -369,4 +375,8 @@ public class HiveTable {
}
}
+ public Map<String, String> getPartitionNameTypeMap() {
+ return partitionNameTypeMap;
+ }
+
}
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java
----------------------------------------------------------------------
diff --git a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java
index c846328..0ea9d53 100644
--- a/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java
+++ b/contrib/storage-hive/core/src/test/java/org/apache/drill/exec/TestHivePartitionPruning.java
@@ -37,7 +37,7 @@ public class TestHivePartitionPruning extends HiveTestBase {
/* Partition pruning is not supported for disjuncts that do not meet pruning criteria.
* Will be enabled when we can do wild card comparison for partition pruning
*/
- @Ignore
+ @Test
public void testDisjunctsPartitionFilter() throws Exception {
final String query = "explain plan for select * from hive.`default`.partition_pruning_test where (c = 1) or (d = 1)";
final String plan = getPlanInString(query, OPTIQ_FORMAT);
@@ -55,7 +55,7 @@ public class TestHivePartitionPruning extends HiveTestBase {
assertFalse(plan.contains("Filter"));
}
- @Ignore("DRILL-1571")
+ @Test
public void testComplexFilter() throws Exception {
final String query = "explain plan for select * from hive.`default`.partition_pruning_test where (c = 1 and d = 1) or (c = 2 and d = 3)";
final String plan = getPlanInString(query, OPTIQ_FORMAT);
@@ -63,4 +63,28 @@ public class TestHivePartitionPruning extends HiveTestBase {
// Check and make sure that Filter is not present in the plan
assertFalse(plan.contains("Filter"));
}
+
+ @Test
+ public void testRangeFilter() throws Exception {
+ final String query = "explain plan for " +
+ "select * from hive.`default`.partition_pruning_test where " +
+ "c > 1 and d > 1";
+
+ final String plan = getPlanInString(query, OPTIQ_FORMAT);
+
+ // Check and make sure that Filter is not present in the plan
+ assertFalse(plan.contains("Filter"));
+ }
+
+ @Test
+ public void testRangeFilterWithDisjunct() throws Exception {
+ final String query = "explain plan for " +
+ "select * from hive.`default`.partition_pruning_test where " +
+ "(c > 1 and d > 1) or (c < 2 and d < 2)";
+
+ final String plan = getPlanInString(query, OPTIQ_FORMAT);
+
+ // Check and make sure that Filter is not present in the plan
+ assertFalse(plan.contains("Filter"));
+ }
}
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DirPathBuilder.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DirPathBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DirPathBuilder.java
deleted file mode 100644
index 892e8cb..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DirPathBuilder.java
+++ /dev/null
@@ -1,265 +0,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.
- */
-package org.apache.drill.exec.planner.logical;
-
-import java.util.ArrayList;
-import java.util.List;
-
-import org.apache.drill.common.expression.FieldReference;
-import org.apache.drill.common.expression.SchemaPath;
-import org.apache.drill.exec.planner.PartitionDescriptor;
-import org.apache.calcite.plan.RelOptUtil;
-import org.apache.calcite.rel.type.RelDataTypeField;
-import org.apache.calcite.rex.RexBuilder;
-import org.apache.calcite.rex.RexCall;
-import org.apache.calcite.rex.RexCorrelVariable;
-import org.apache.calcite.rex.RexDynamicParam;
-import org.apache.calcite.rex.RexFieldAccess;
-import org.apache.calcite.rex.RexInputRef;
-import org.apache.calcite.rex.RexLiteral;
-import org.apache.calcite.rex.RexLocalRef;
-import org.apache.calcite.rex.RexNode;
-import org.apache.calcite.rex.RexOver;
-import org.apache.calcite.rex.RexRangeRef;
-import org.apache.calcite.rex.RexUtil;
-import org.apache.calcite.rex.RexVisitorImpl;
-import org.apache.calcite.sql.SqlKind;
-import org.apache.calcite.sql.SqlSyntax;
-
-import com.google.common.collect.Lists;
-
-public class DirPathBuilder extends RexVisitorImpl <SchemaPath> {
- static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(DirPathBuilder.class);
-
- static final String EMPTY_STRING = "";
-
- final private DrillFilterRel filterRel;
- final private DrillRel inputRel;
- final private RexBuilder builder;
- final private PartitionDescriptor partitionDescriptor;
-
- private List<String> dirNameList;
- private List<RexNode> conjunctList;
- private List<String> dirPathList = Lists.newArrayList();
- private final static List<String> emptyDirPathList = new ArrayList<>(0);
- private RexNode currentConjunct = null; // the current conjunct are we evaluating during visitor traversal
- private RexNode finalCondition = null; // placeholder for the final filter condition
- private boolean dirMatch = false;
-
- public DirPathBuilder(DrillFilterRel filterRel, DrillRel inputRel, RexBuilder builder, PartitionDescriptor partitionDescriptor) {
- super(true);
- this.filterRel = filterRel;
- this.inputRel = inputRel;
- this.builder = builder;
- this.finalCondition = filterRel.getCondition();
- this.partitionDescriptor = partitionDescriptor;
- }
-
- private void initPathComponents() {
- int maxHierarchy = partitionDescriptor.getMaxHierarchyLevel();
- dirNameList = Lists.newArrayListWithExpectedSize(maxHierarchy);
- conjunctList = Lists.newArrayListWithExpectedSize(maxHierarchy);
- for (int i=0; i < maxHierarchy; i++) {
- dirNameList.add(EMPTY_STRING);
- conjunctList.add(null);
- }
- }
-
- /**
- * Build a directory path string for filter conditions containing directory filters.
- * For example, suppose we have directory hierarchy:
- * {orders/2012/Jan...Dec, orders/2013/Jan...Dec, orders/2014/Jan...Dec}
- * path will be built for following types of filters (More types of filters will be added in the future):
- * 1. SELECT * FROM <path>/orders WHERE o_custkey = 5 AND dir0 = '2014' AND dir1 = 'June'
- * 2. SELECT * FROM <path>/orders WHERE (dir0 = '2013' AND dir1 = 'June') OR (dir0 = '2014' AND dir1 = 'June')
- * For (1) dirPath = <path>/orders/2014/June
- * For (2) there are 2 dirPaths: {<path>/orders/2013/June, <path>/orders/2014/June}
- * @return The list of strings corresponding to directory paths
- */
- public List<String> getDirPath() {
- List<RexNode> disjuncts = RelOptUtil.disjunctions(filterRel.getCondition());
- boolean buildDisjunction = false;
- List<RexNode> newDisjunctList = Lists.newArrayList();
-
- for (RexNode d : disjuncts) { // process the top-level disjuncts
- List<RexNode> conjuncts = RelOptUtil.conjunctions(d);
- String dirPath = EMPTY_STRING;
- initPathComponents();
-
- boolean buildConjunction = false;
-
- // go through the conjuncts to identify the directory filters
- for (RexNode c : conjuncts) {
- currentConjunct = c;
- SchemaPath expr = c.accept(this);
-
- if (expr != null) {
- logger.debug("Found directory filter: " + expr.getRootSegment().getPath());
- }
- }
-
- String prevPath = dirNameList.get(0);
-
- // compose the final path string
- for (int i = 0; i < dirNameList.size(); i++) {
- String path = dirNameList.get(i);
- if (i > 0) {
- prevPath = dirNameList.get(i-1);
- }
- // Check if both the current path and the previous path are non-empty; currently
- // we will only push a dir<N> filter if dir<N-1> filter is also specified
- if (!path.equals(EMPTY_STRING) && !prevPath.equals(EMPTY_STRING)) {
- dirPath += "/" + path;
-
- // since we are pushing this directory filter we should remove it from the
- // list of conjuncts
- RexNode thisConjunct = conjunctList.get(i);
- conjuncts.remove(thisConjunct);
- buildConjunction = true;
- }
- }
- if (!dirPath.equals(EMPTY_STRING)) {
- dirPathList.add(dirPath);
- } else {
- // If one of the disjuncts do not satisfy our criteria then we shouldn't apply any optimization
- return emptyDirPathList;
- }
-
- if (buildConjunction) {
- RexNode newConjunct = RexUtil.composeConjunction(builder, conjuncts, false);
- newDisjunctList.add(newConjunct);
- buildDisjunction = true;
- }
-
- } // for (disjuncts)
-
- if (buildDisjunction) {
- this.finalCondition = RexUtil.composeDisjunction(builder, newDisjunctList, false);
- }
- return dirPathList;
- }
-
- public RexNode getFinalCondition() {
- return finalCondition;
- }
-
- @Override
- public SchemaPath visitInputRef(RexInputRef inputRef) {
- final int index = inputRef.getIndex();
- final RelDataTypeField field = inputRel.getRowType().getFieldList().get(index);
- if (partitionDescriptor.isPartitionName(field.getName())) {
- dirMatch = true;
- }
- return FieldReference.getWithQuotedRef(field.getName());
- }
-
- @Override
- public SchemaPath visitCall(RexCall call) {
-// logger.debug("RexCall {}, {}", call);
- final SqlSyntax syntax = call.getOperator().getSyntax();
- switch (syntax) {
- case BINARY:
- if (call.getKind() == SqlKind.EQUALS) {
- dirMatch = false;
- // TODO: an assumption here is that the binary predicate is of the form <column> = <value>.
- // In order to handle predicates of the form '<value> = <column>' we would need to canonicalize
- // the predicate first before calling this function.
- SchemaPath e1 = call.getOperands().get(0).accept(this);
-
- if (dirMatch && e1 != null) {
- // get the index for the 'dir<N>' filter
- String dirName = e1.getRootSegment().getPath();
- int hierarychyIndex = partitionDescriptor.getPartitionHierarchyIndex(dirName);
-
- if (hierarychyIndex >= partitionDescriptor.getMaxHierarchyLevel()) {
- return null;
- }
-
- // SchemaPath e2 = call.getOperands().get(1).accept(this);
- if (call.getOperands().get(1).getKind() == SqlKind.LITERAL) {
- String e2 = ((RexLiteral)call.getOperands().get(1)).getValue2().toString();
- dirNameList.set(hierarychyIndex, e2);
- // dirNameList.set(suffixIndex, e2.getRootSegment().getPath());
- conjunctList.set(hierarychyIndex, currentConjunct);
- return e1;
- }
- }
- }
-
- return null;
-
- case SPECIAL:
- switch(call.getKind()) {
- case CAST:
- return getInputFromCast(call);
-
- default:
-
- }
- // fall through
- default:
- // throw new AssertionError("Unexpected expression");
-
- }
- return null;
- }
-
- private SchemaPath getInputFromCast(RexCall call){
- SchemaPath arg = call.getOperands().get(0).accept(this);
- if (dirMatch) {
- return arg;
- }
- return null;
- }
-
- @Override
- public SchemaPath visitLocalRef(RexLocalRef localRef) {
- return null;
- }
-
- @Override
- public SchemaPath visitOver(RexOver over) {
- return null;
- }
-
- @Override
- public SchemaPath visitCorrelVariable(RexCorrelVariable correlVariable) {
- return null;
- }
-
- @Override
- public SchemaPath visitDynamicParam(RexDynamicParam dynamicParam) {
- return null;
- }
-
- @Override
- public SchemaPath visitRangeRef(RexRangeRef rangeRef) {
- return null;
- }
-
- @Override
- public SchemaPath visitFieldAccess(RexFieldAccess fieldAccess) {
- return super.visitFieldAccess(fieldAccess);
- }
-
- @Override
- public SchemaPath visitLiteral(RexLiteral literal) {
- return FieldReference.getWithQuotedRef(literal.getValue2().toString());
- }
-
-}
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushPartitionFilterIntoScan.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushPartitionFilterIntoScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushPartitionFilterIntoScan.java
deleted file mode 100644
index 811eef1..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushPartitionFilterIntoScan.java
+++ /dev/null
@@ -1,139 +0,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.
- */
-
-package org.apache.drill.exec.planner.logical;
-
-import java.io.IOException;
-import java.util.List;
-
-import org.apache.drill.common.exceptions.DrillRuntimeException;
-import org.apache.drill.exec.physical.base.FileGroupScan;
-import org.apache.drill.exec.physical.base.GroupScan;
-import org.apache.drill.exec.planner.FileSystemPartitionDescriptor;
-import org.apache.drill.exec.planner.physical.PlannerSettings;
-import org.apache.drill.exec.planner.physical.PrelUtil;
-import org.apache.drill.exec.store.dfs.FileSelection;
-import org.apache.drill.exec.store.dfs.FormatSelection;
-import org.apache.calcite.plan.RelOptRule;
-import org.apache.calcite.plan.RelOptRuleCall;
-import org.apache.calcite.plan.RelOptRuleOperand;
-
-import com.google.common.collect.Lists;
-
-public abstract class DrillPushPartitionFilterIntoScan extends RelOptRule {
-
- public static final RelOptRule FILTER_ON_PROJECT =
- new DrillPushPartitionFilterIntoScan(
- RelOptHelper.some(DrillFilterRel.class, RelOptHelper.some(DrillProjectRel.class, RelOptHelper.any(DrillScanRel.class))),
- "DrillPushPartitionFilterIntoScan:Filter_On_Project") {
-
- @Override
- public boolean matches(RelOptRuleCall call) {
- final DrillScanRel scan = (DrillScanRel) call.rel(2);
- GroupScan groupScan = scan.getGroupScan();
- // this rule is applicable only for dfs based partition pruning
- return groupScan instanceof FileGroupScan && groupScan.supportsPartitionFilterPushdown();
- }
-
- @Override
- public void onMatch(RelOptRuleCall call) {
- final DrillFilterRel filterRel = (DrillFilterRel) call.rel(0);
- final DrillProjectRel projectRel = (DrillProjectRel) call.rel(1);
- final DrillScanRel scanRel = (DrillScanRel) call.rel(2);
- doOnMatch(call, filterRel, projectRel, scanRel);
- }
- };
-
- public static final RelOptRule FILTER_ON_SCAN =
- new DrillPushPartitionFilterIntoScan(
- RelOptHelper.some(DrillFilterRel.class, RelOptHelper.any(DrillScanRel.class)),
- "DrillPushPartitionFilterIntoScan:Filter_On_Scan") {
-
- @Override
- public boolean matches(RelOptRuleCall call) {
- final DrillScanRel scan = (DrillScanRel) call.rel(1);
- GroupScan groupScan = scan.getGroupScan();
- // this rule is applicable only for dfs based partition pruning
- return groupScan instanceof FileGroupScan && groupScan.supportsPartitionFilterPushdown();
- }
-
- @Override
- public void onMatch(RelOptRuleCall call) {
- final DrillFilterRel filterRel = (DrillFilterRel) call.rel(0);
- final DrillScanRel scanRel = (DrillScanRel) call.rel(1);
- doOnMatch(call, filterRel, null, scanRel);
- }
- };
-
- private DrillPushPartitionFilterIntoScan(
- RelOptRuleOperand operand,
- String id) {
- super(operand, id);
- }
-
- private FormatSelection splitFilter(FormatSelection origSelection, DirPathBuilder builder) {
-
- List<String> origFiles = origSelection.getAsFiles();
- String pathPrefix = origSelection.getSelection().selectionRoot;
-
- List<String> newFiles = Lists.newArrayList();
-
- List<String> dirPathList = builder.getDirPath();
-
- for (String dirPath : dirPathList) {
- String fullPath = pathPrefix + dirPath;
- // check containment of this path in the list of files
- for (String origFilePath : origFiles) {
- String origFileName = PartitionPruningUtil.truncatePrefixFromPath(origFilePath);
-
- if (origFileName.startsWith(fullPath)) {
- newFiles.add(origFileName);
- }
- }
- }
-
- if (newFiles.size() > 0) {
- FileSelection newFileSelection = new FileSelection(newFiles, origSelection.getSelection().selectionRoot, true);
- FormatSelection newFormatSelection = new FormatSelection(origSelection.getFormat(), newFileSelection);
- return newFormatSelection;
- }
-
- return origSelection;
- }
-
- protected void doOnMatch(RelOptRuleCall call, DrillFilterRel filterRel, DrillProjectRel projectRel, DrillScanRel scanRel) {
- DrillRel inputRel = projectRel != null ? projectRel : scanRel;
-
- PlannerSettings settings = PrelUtil.getPlannerSettings(call.getPlanner());
- DirPathBuilder builder = new DirPathBuilder(filterRel, inputRel, filterRel.getCluster().getRexBuilder(), new FileSystemPartitionDescriptor(settings, scanRel));
-
- FormatSelection origSelection = (FormatSelection)scanRel.getDrillTable().getSelection();
- FormatSelection newSelection = splitFilter(origSelection, builder);
-
- if (origSelection == newSelection) {
- return; // no directory filter was pushed down
- }
-
- try {
- FileGroupScan fgscan = ((FileGroupScan)scanRel.getGroupScan()).clone(newSelection.getSelection());
- PartitionPruningUtil.rewritePlan(call, filterRel, projectRel, scanRel, fgscan, builder);
- } catch (IOException e) {
- throw new DrillRuntimeException(e) ;
- }
- }
-}
http://git-wip-us.apache.org/repos/asf/drill/blob/6d5d7cc2/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/PartitionPruningUtil.java
----------------------------------------------------------------------
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/PartitionPruningUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/PartitionPruningUtil.java
deleted file mode 100644
index 05ccfb9..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/PartitionPruningUtil.java
+++ /dev/null
@@ -1,78 +0,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.
- */
-package org.apache.drill.exec.planner.logical;
-
-import org.apache.drill.exec.physical.base.GroupScan;
-import org.apache.calcite.plan.RelOptRuleCall;
-import org.apache.calcite.rex.RexNode;
-
-public class PartitionPruningUtil {
- public static void rewritePlan(RelOptRuleCall call, DrillFilterRel filterRel, DrillProjectRel projectRel, DrillScanRel scanRel, GroupScan newScan, DirPathBuilder builder) {
- RexNode origFilterCondition = filterRel.getCondition();
- RexNode newFilterCondition = builder.getFinalCondition();
-
- if (newFilterCondition.isAlwaysTrue()) {
-
- final DrillScanRel newScanRel =
- new DrillScanRel(scanRel.getCluster(),
- scanRel.getTraitSet().plus(DrillRel.DRILL_LOGICAL),
- scanRel.getTable(),
- newScan,
- scanRel.getRowType(),
- scanRel.getColumns());
-
- if (projectRel != null) {
- DrillProjectRel newProjectRel = new DrillProjectRel(projectRel.getCluster(), projectRel.getTraitSet(),
- newScanRel, projectRel.getProjects(), filterRel.getRowType());
-
- call.transformTo(newProjectRel);
- } else {
- call.transformTo(newScanRel);
- }
- } else {
- DrillRel inputRel;
- final DrillScanRel newScanRel =
- new DrillScanRel(scanRel.getCluster(),
- scanRel.getTraitSet().plus(DrillRel.DRILL_LOGICAL),
- scanRel.getTable(),
- newScan,
- scanRel.getRowType(),
- scanRel.getColumns());
- if (projectRel != null) {
- DrillProjectRel newProjectRel = new DrillProjectRel(projectRel.getCluster(), projectRel.getTraitSet(),
- newScanRel, projectRel.getProjects(), projectRel.getRowType());
- inputRel = newProjectRel;
- } else {
- inputRel = newScanRel;
- }
- final DrillFilterRel newFilterRel = new DrillFilterRel(filterRel.getCluster(), filterRel.getTraitSet(),
- inputRel, origFilterCondition /* for now keep the original condition until we add more test coverage */);
-
- call.transformTo(newFilterRel);
- }
- }
-
- public static String truncatePrefixFromPath(String fileName) {
- String pathPrefixComponent[] = fileName.split(":", 2);
- if (pathPrefixComponent.length == 1) {
- return pathPrefixComponent[0];
- } else {
- return pathPrefixComponent[1];
- }
- }
-}