You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by bo...@apache.org on 2019/05/09 01:58:50 UTC
[drill] branch master updated (ba9837e -> e5e9b35)
This is an automated email from the ASF dual-hosted git repository.
boaz pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git.
from ba9837e DRILL-7187: Improve selectivity estimation of BETWEEN predicates and arbitrary combination of range predicates.
new edbfd64 DRILL-7030: Make format plugins fully pluggable
new c1ca512 DRILL-7148: Use improved join cardinality and ndv estimation with statistics
new 4a0d2d0 DRILL-7164: KafkaFilterPushdownTest is sometimes failing to pattern match correctly
new 3603f24 DRILL-7238: Fixed ConvertCountToDirectScan to handle non-existent columns
new e5e9b35 DRILL-7240: Catch runtime pruning filter-match exceptions and do not prune these rowgroups
The 5 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../main/resources/bootstrap-format-plugins.json | 26 +++
.../main/resources/bootstrap-format-plugins.json | 20 +++
.../main/resources/bootstrap-format-plugins.json | 26 +++
.../exec/store/kafka/KafkaFilterPushdownTest.java | 74 ++++++---
.../java/org/apache/drill/exec/ExecConstants.java | 1 +
.../exec/planner/common/DrillJoinRelBase.java | 34 ++--
.../drill/exec/planner/common/DrillRelOptUtil.java | 62 +++++++
.../drill/exec/planner/common/DrillStatsTable.java | 2 +-
.../planner/cost/DrillRelMdDistinctRowCount.java | 181 +++++++++++++++++----
.../exec/planner/cost/DrillRelMdRowCount.java | 41 +++--
.../exec/planner/cost/DrillRelMdSelectivity.java | 40 +++--
.../logical/ConvertCountToDirectScanRule.java | 6 +-
.../exec/planner/physical/PlannerSettings.java | 6 +
.../exec/server/options/SystemOptionManager.java | 1 +
.../exec/store/StoragePluginRegistryImpl.java | 88 ++++++++--
.../drill/exec/store/dfs/FileSystemConfig.java | 3 +-
.../parquet/AbstractParquetScanBatchCreator.java | 37 +++--
.../java/org/apache/drill/exec/util/Utilities.java | 2 +-
.../java-exec/src/main/resources/drill-module.conf | 1 +
.../test/java/org/apache/drill/PlanTestBase.java | 48 ++++--
.../logical/TestConvertCountToDirectScan.java | 37 +++++
.../org/apache/drill/exec/sql/TestAnalyze.java | 12 +-
.../drill/metastore/ColumnStatisticsKind.java | 2 +-
23 files changed, 590 insertions(+), 160 deletions(-)
create mode 100644 contrib/format-ltsv/src/main/resources/bootstrap-format-plugins.json
create mode 100644 contrib/format-maprdb/src/main/resources/bootstrap-format-plugins.json
create mode 100644 contrib/format-syslog/src/main/resources/bootstrap-format-plugins.json
[drill] 01/05: DRILL-7030: Make format plugins fully pluggable
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boaz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit edbfd64c4b7d8b7af406a507e959e5856185eaea
Author: Anton Gozhiy <an...@gmail.com>
AuthorDate: Mon Apr 15 19:24:11 2019 +0300
DRILL-7030: Make format plugins fully pluggable
- Bootstrap files for format plugins were introduced and added to the existing plugins in contrib.
- Formats from these files are being added dynamically to the corresponding storage plugins.
closes #1780
---
.../main/resources/bootstrap-format-plugins.json | 26 +++++++
.../main/resources/bootstrap-format-plugins.json | 20 +++++
.../main/resources/bootstrap-format-plugins.json | 26 +++++++
.../java/org/apache/drill/exec/ExecConstants.java | 1 +
.../exec/store/StoragePluginRegistryImpl.java | 88 ++++++++++++++++++----
.../drill/exec/store/dfs/FileSystemConfig.java | 3 +-
6 files changed, 148 insertions(+), 16 deletions(-)
diff --git a/contrib/format-ltsv/src/main/resources/bootstrap-format-plugins.json b/contrib/format-ltsv/src/main/resources/bootstrap-format-plugins.json
new file mode 100644
index 0000000..3dda8cf
--- /dev/null
+++ b/contrib/format-ltsv/src/main/resources/bootstrap-format-plugins.json
@@ -0,0 +1,26 @@
+{
+ "storage":{
+ "dfs": {
+ "type": "file",
+ "formats": {
+ "ltsv": {
+ "type": "ltsv",
+ "extensions": [
+ "ltsv"
+ ]
+ }
+ }
+ },
+ "s3": {
+ "type": "file",
+ "formats": {
+ "ltsv": {
+ "type": "ltsv",
+ "extensions": [
+ "ltsv"
+ ]
+ }
+ }
+ }
+ }
+}
diff --git a/contrib/format-maprdb/src/main/resources/bootstrap-format-plugins.json b/contrib/format-maprdb/src/main/resources/bootstrap-format-plugins.json
new file mode 100644
index 0000000..a126709
--- /dev/null
+++ b/contrib/format-maprdb/src/main/resources/bootstrap-format-plugins.json
@@ -0,0 +1,20 @@
+{
+ "storage":{
+ "dfs": {
+ "type": "file",
+ "formats": {
+ "maprdb": {
+ "type": "maprdb"
+ }
+ }
+ },
+ "s3": {
+ "type": "file",
+ "formats": {
+ "maprdb": {
+ "type": "maprdb"
+ }
+ }
+ }
+ }
+}
diff --git a/contrib/format-syslog/src/main/resources/bootstrap-format-plugins.json b/contrib/format-syslog/src/main/resources/bootstrap-format-plugins.json
new file mode 100644
index 0000000..ee5a396
--- /dev/null
+++ b/contrib/format-syslog/src/main/resources/bootstrap-format-plugins.json
@@ -0,0 +1,26 @@
+{
+ "storage":{
+ "dfs": {
+ "type": "file",
+ "formats": {
+ "syslog": {
+ "type": "syslog",
+ "extensions": [
+ "syslog"
+ ]
+ }
+ }
+ },
+ "s3": {
+ "type": "file",
+ "formats": {
+ "syslog": {
+ "type": "syslog",
+ "extensions": [
+ "syslog"
+ ]
+ }
+ }
+ }
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
index 3503507..93c9902 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/ExecConstants.java
@@ -726,6 +726,7 @@ public final class ExecConstants {
new OptionDescription("Min width for text readers, mostly for testing."));
public static final String BOOTSTRAP_STORAGE_PLUGINS_FILE = "bootstrap-storage-plugins.json";
+ public static final String BOOTSTRAP_FORMAT_PLUGINS_FILE = "bootstrap-format-plugins.json";
public static final String SKIP_RUNTIME_ROWGROUP_PRUNING_KEY = "exec.storage.skip_runtime_rowgroup_pruning";
public static final OptionValidator SKIP_RUNTIME_ROWGROUP_PRUNING = new BooleanValidator(SKIP_RUNTIME_ROWGROUP_PRUNING_KEY,
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistryImpl.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistryImpl.java
index c5554f8..4718a20 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistryImpl.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/StoragePluginRegistryImpl.java
@@ -36,6 +36,7 @@ import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import org.apache.drill.exec.store.dfs.FileSystemConfig;
import org.apache.drill.shaded.guava.com.google.common.annotations.VisibleForTesting;
import org.apache.calcite.schema.SchemaPlus;
import org.apache.drill.common.config.LogicalPlanPersistence;
@@ -300,7 +301,8 @@ public class StoragePluginRegistryImpl implements StoragePluginRegistry {
}
/**
- * Read bootstrap storage plugins {@link ExecConstants#BOOTSTRAP_STORAGE_PLUGINS_FILE} files for the first fresh
+ * Read bootstrap storage plugins {@link ExecConstants#BOOTSTRAP_STORAGE_PLUGINS_FILE}
+ * and format plugins {@link ExecConstants#BOOTSTRAP_FORMAT_PLUGINS_FILE} files for the first fresh
* instantiating of Drill
*
* @param lpPersistence deserialization mapper provider
@@ -310,22 +312,19 @@ public class StoragePluginRegistryImpl implements StoragePluginRegistry {
private StoragePlugins loadBootstrapPlugins(LogicalPlanPersistence lpPersistence) throws IOException {
// bootstrap load the config since no plugins are stored.
logger.info("No storage plugin instances configured in persistent store, loading bootstrap configuration.");
- Set<URL> urls = ClassPathScanner.forResource(ExecConstants.BOOTSTRAP_STORAGE_PLUGINS_FILE, false);
- if (urls != null && !urls.isEmpty()) {
- logger.info("Loading the storage plugin configs from URLs {}.", urls);
+ Set<URL> storageUrls = ClassPathScanner.forResource(ExecConstants.BOOTSTRAP_STORAGE_PLUGINS_FILE, false);
+ Set<URL> formatUrls = ClassPathScanner.forResource(ExecConstants.BOOTSTRAP_FORMAT_PLUGINS_FILE, false);
+ if (storageUrls != null && !storageUrls.isEmpty()) {
+ logger.info("Loading the storage plugin configs from URLs {}.", storageUrls);
StoragePlugins bootstrapPlugins = new StoragePlugins(new HashMap<>());
Map<String, URL> pluginURLMap = new HashMap<>();
- for (URL url : urls) {
- String pluginsData = Resources.toString(url, Charsets.UTF_8);
- StoragePlugins plugins = lpPersistence.getMapper().readValue(pluginsData, StoragePlugins.class);
- for (Entry<String, StoragePluginConfig> plugin : plugins) {
- StoragePluginConfig oldPluginConfig = bootstrapPlugins.putIfAbsent(plugin.getKey(), plugin.getValue());
- if (oldPluginConfig != null) {
- logger.warn("Duplicate plugin instance '{}' defined in [{}, {}], ignoring the later one.",
- plugin.getKey(), pluginURLMap.get(plugin.getKey()), url);
- } else {
- pluginURLMap.put(plugin.getKey(), url);
- }
+ for (URL url : storageUrls) {
+ loadStoragePlugins(url, bootstrapPlugins, pluginURLMap, lpPersistence);
+ }
+ if (formatUrls != null && !formatUrls.isEmpty()) {
+ logger.info("Loading the format plugin configs from URLs {}.", formatUrls);
+ for (URL url : formatUrls) {
+ loadFormatPlugins(url, bootstrapPlugins, pluginURLMap, lpPersistence);
}
}
return bootstrapPlugins;
@@ -335,6 +334,65 @@ public class StoragePluginRegistryImpl implements StoragePluginRegistry {
}
/**
+ * Loads storage plugins from the given URL
+ *
+ * @param url URL to the storage plugins bootstrap file
+ * @param bootstrapPlugins a collection where the plugins should be loaded to
+ * @param pluginURLMap a map to store correspondence between storage plugins and bootstrap files in which they are defined. Used for logging
+ * @param lpPersistence need to get an object mapper for the bootstrap files
+ * @throws IOException if failed to retrieve a plugin from a bootstrap file
+ */
+ private void loadStoragePlugins(URL url, StoragePlugins bootstrapPlugins, Map<String, URL> pluginURLMap, LogicalPlanPersistence lpPersistence) throws IOException {
+ StoragePlugins plugins = getPluginsFromResource(url, lpPersistence);
+ plugins.forEach(plugin -> {
+ StoragePluginConfig oldPluginConfig = bootstrapPlugins.putIfAbsent(plugin.getKey(), plugin.getValue());
+ if (oldPluginConfig != null) {
+ logger.warn("Duplicate plugin instance '[{}]' defined in [{}, {}], ignoring the later one.",
+ plugin.getKey(), pluginURLMap.get(plugin.getKey()), url);
+ } else {
+ pluginURLMap.put(plugin.getKey(), url);
+ }
+ });
+ }
+
+ /**
+ * Loads format plugins from the given URL and adds the formats to the specified storage plugins
+ *
+ * @param url URL to the format plugins bootstrap file
+ * @param bootstrapPlugins a collection with loaded storage plugins. New formats will be added to them
+ * @param pluginURLMap a map to store correspondence between storage plugins and bootstrap files in which they are defined. Used for logging
+ * @param lpPersistence need to get an object mapper for the bootstrap files
+ * @throws IOException if failed to retrieve a plugin from a bootstrap file
+ */
+ private void loadFormatPlugins(URL url, StoragePlugins bootstrapPlugins, Map<String, URL> pluginURLMap, LogicalPlanPersistence lpPersistence) throws IOException {
+ StoragePlugins plugins = getPluginsFromResource(url, lpPersistence);
+ plugins.forEach(formatPlugin -> {
+ String targetStoragePluginName = formatPlugin.getKey();
+ StoragePluginConfig storagePlugin = bootstrapPlugins.getConfig(targetStoragePluginName);
+ StoragePluginConfig formatPluginValue = formatPlugin.getValue();
+ if (storagePlugin == null) {
+ logger.warn("No storage plugins with the given name are registered: '[{}]'", targetStoragePluginName);
+ } else if (storagePlugin instanceof FileSystemConfig && formatPluginValue instanceof FileSystemConfig) {
+ FileSystemConfig targetPlugin = (FileSystemConfig) storagePlugin;
+ ((FileSystemConfig) formatPluginValue).getFormats().forEach((formatName, formatValue) -> {
+ FormatPluginConfig oldPluginConfig = targetPlugin.getFormats().putIfAbsent(formatName, formatValue);
+ if (oldPluginConfig != null) {
+ logger.warn("Duplicate format instance '[{}]' defined in [{}, {}], ignoring the later one.",
+ formatName, pluginURLMap.get(targetStoragePluginName), url);
+ }
+ });
+ } else {
+ logger.warn("Formats are only supported by File System plugin type: '[{}]'", targetStoragePluginName);
+ }
+ });
+ }
+
+ private StoragePlugins getPluginsFromResource(URL resource, LogicalPlanPersistence lpPersistence) throws IOException {
+ String pluginsData = Resources.toString(resource, Charsets.UTF_8);
+ return lpPersistence.getMapper().readValue(pluginsData, StoragePlugins.class);
+ }
+
+ /**
* Dynamically loads system plugins annotated with {@link SystemPlugin}.
* Will skip plugin initialization if no matching constructor, incorrect class implementation, name absence are detected.
*
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/FileSystemConfig.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/FileSystemConfig.java
index 4b75e33..58f69a4 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/FileSystemConfig.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/dfs/FileSystemConfig.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.store.dfs;
+import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Optional;
@@ -48,7 +49,7 @@ public class FileSystemConfig extends StoragePluginConfig {
Map<String, WorkspaceConfig> caseInsensitiveWorkspaces = CaseInsensitiveMap.newHashMap();
Optional.ofNullable(workspaces).ifPresent(caseInsensitiveWorkspaces::putAll);
this.workspaces = caseInsensitiveWorkspaces;
- this.formats = formats;
+ this.formats = formats != null ? formats : new LinkedHashMap<>();
}
@JsonProperty
[drill] 04/05: DRILL-7238: Fixed ConvertCountToDirectScan to handle
non-existent columns
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boaz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 3603f240922389939ca0498face0eb255448ea53
Author: Venkata Jyothsna Donapati <jy...@gmail.com>
AuthorDate: Fri May 3 15:21:59 2019 -0700
DRILL-7238: Fixed ConvertCountToDirectScan to handle non-existent columns
closes #1781
---
.../logical/ConvertCountToDirectScanRule.java | 6 +++-
.../logical/TestConvertCountToDirectScan.java | 37 ++++++++++++++++++++++
2 files changed, 42 insertions(+), 1 deletion(-)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/ConvertCountToDirectScanRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/ConvertCountToDirectScanRule.java
index 7375499..3a4e6ab 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/ConvertCountToDirectScanRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/ConvertCountToDirectScanRule.java
@@ -228,6 +228,7 @@ public class ConvertCountToDirectScanRule extends RelOptRule {
* 2. For COUNT(*) and COUNT(<non null column>) and COUNT(<implicit column>), the count = total row count
* 3. For COUNT(nullable column), count = (total row count - column's null count)
* 4. Also count can not be calculated for parition columns.
+ * 5. For the columns that are not present in the Summary(Non-existent columns), the count = 0
*
* @param settings planner options
* @param metadataSummary metadata summary containing row counts and column counts
@@ -288,7 +289,10 @@ public class ConvertCountToDirectScanRule extends RelOptRule {
Metadata_V4.ColumnTypeMetadata_v4 columnMetadata = metadataSummary.getColumnTypeInfo(new Metadata_V4.ColumnTypeMetadata_v4.Key(simplePath));
- if (columnMetadata == null || columnMetadata.totalNullCount == Statistic.NO_COLUMN_STATS) {
+ if (columnMetadata == null) {
+ // If the column doesn't exist in the table, row count is set to 0
+ cnt = 0;
+ } else if (columnMetadata.totalNullCount == Statistic.NO_COLUMN_STATS) {
// if column stats is not available don't apply this rule, return empty counts
return ImmutableMap.of();
} else {
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/planner/logical/TestConvertCountToDirectScan.java b/exec/java-exec/src/test/java/org/apache/drill/exec/planner/logical/TestConvertCountToDirectScan.java
index eaf9257..c35ab2d 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/planner/logical/TestConvertCountToDirectScan.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/planner/logical/TestConvertCountToDirectScan.java
@@ -340,4 +340,41 @@ public class TestConvertCountToDirectScan extends PlanTestBase {
test("drop table if exists %s", tableName);
}
}
+
+ @Test
+ public void testCountsWithNonExColumn() throws Exception {
+ test("use dfs.tmp");
+ String tableName = "parquet_table_counts_nonex";
+
+ try {
+ test(String.format("create table `%s/1` as select * from cp.`parquet/alltypes_optional.parquet`", tableName));
+ test(String.format("create table `%s/2` as select * from cp.`parquet/alltypes_optional.parquet`", tableName));
+ test(String.format("create table `%s/3` as select * from cp.`parquet/alltypes_optional.parquet`", tableName));
+ test(String.format("create table `%s/4` as select * from cp.`parquet/alltypes_optional.parquet`", tableName));
+
+ test("refresh table metadata %s", tableName);
+
+ String sql = String.format("select\n" +
+ "count(*) as star_count,\n" +
+ "count(col_int) as int_column_count,\n" +
+ "count(col_vrchr) as vrchr_column_count,\n" +
+ "count(non_existent) as non_existent\n" +
+ "from %s", tableName);
+
+ String usedMetaSummaryPattern = "usedMetadataSummaryFile = true";
+ String recordReaderPattern = "DynamicPojoRecordReader";
+
+ testPlanMatchingPatterns(sql, new String[]{usedMetaSummaryPattern, recordReaderPattern});
+
+ testBuilder()
+ .sqlQuery(sql)
+ .unOrdered()
+ .baselineColumns("star_count", "int_column_count", "vrchr_column_count", "non_existent" )
+ .baselineValues(24L, 8L, 12L, 0L)
+ .go();
+
+ } finally {
+ test("drop table if exists %s", tableName);
+ }
+ }
}
[drill] 03/05: DRILL-7164: KafkaFilterPushdownTest is sometimes
failing to pattern match correctly
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boaz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 4a0d2d01a702f20df9204da39f55f3da4d6a7143
Author: Sorabh Hamirwasia <so...@apache.org>
AuthorDate: Thu Apr 18 14:50:38 2019 -0700
DRILL-7164: KafkaFilterPushdownTest is sometimes failing to pattern match correctly
closes #1760
---
.../exec/store/kafka/KafkaFilterPushdownTest.java | 74 ++++++++++++++--------
.../test/java/org/apache/drill/PlanTestBase.java | 48 +++++++++-----
2 files changed, 79 insertions(+), 43 deletions(-)
diff --git a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaFilterPushdownTest.java b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaFilterPushdownTest.java
index 2a5cddd..8e8319b 100644
--- a/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaFilterPushdownTest.java
+++ b/contrib/storage-kafka/src/test/java/org/apache/drill/exec/store/kafka/KafkaFilterPushdownTest.java
@@ -17,10 +17,10 @@
*/
package org.apache.drill.exec.store.kafka;
+import org.apache.drill.PlanTestBase;
import org.apache.drill.categories.KafkaStorageTest;
import org.apache.drill.categories.SlowTest;
import org.apache.kafka.common.serialization.StringSerializer;
-
import org.junit.BeforeClass;
import org.junit.Test;
import org.junit.experimental.categories.Category;
@@ -32,10 +32,8 @@ import static org.junit.Assert.assertTrue;
@Category({KafkaStorageTest.class, SlowTest.class})
public class KafkaFilterPushdownTest extends KafkaTestBase {
private static final int NUM_PARTITIONS = 5;
- private static final String expectedSubStr = " \"kafkaScanSpec\" : {\n" +
- " \"topicName\" : \"drill-pushdown-topic\"\n" +
- " },\n" +
- " \"cost\"";
+ private static final String EXPECTED_PATTERN = "kafkaScanSpec.*\\n.*\"topicName\" : \"drill-pushdown-topic\"\\n(" +
+ ".*\\n)?(.*\\n)?(.*\\n)?.*cost\"(.*\\n)(.*\\n).*outputRowCount\" : (%s.0)";
@BeforeClass
public static void setup() throws Exception {
@@ -63,7 +61,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate1, predicate2);
runKafkaSQLVerifyCount(queryString, expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -79,7 +78,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate);
runKafkaSQLVerifyCount(queryString, expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -95,7 +95,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -112,7 +113,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowInPlan));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowInPlan)}, null);
}
/**
@@ -128,7 +130,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -144,7 +147,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -161,7 +165,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate1, predicate2);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -178,42 +183,48 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, "kafkaMsgOffset = 10");
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
//"equal" such that value < startOffset
queryString = String.format(TestQueryConstants.QUERY_TEMPLATE_BASIC,
TestQueryConstants.JSON_PUSHDOWN_TOPIC, "kafkaMsgOffset = -1");
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
//"greater_than" such that value = endOffset-1
queryString = String.format(TestQueryConstants.QUERY_TEMPLATE_BASIC,
TestQueryConstants.JSON_PUSHDOWN_TOPIC, "kafkaMsgOffset > 9");
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
//"greater_than_or_equal" such that value = endOffset
queryString = String.format(TestQueryConstants.QUERY_TEMPLATE_BASIC,
TestQueryConstants.JSON_PUSHDOWN_TOPIC, "kafkaMsgOffset >= 10");
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
//"less_than" such that value = startOffset
queryString = String.format(TestQueryConstants.QUERY_TEMPLATE_BASIC,
TestQueryConstants.JSON_PUSHDOWN_TOPIC, "kafkaMsgOffset < 0");
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
//"less_than_or_equal" such that value < startOffset
queryString = String.format(TestQueryConstants.QUERY_TEMPLATE_BASIC,
TestQueryConstants.JSON_PUSHDOWN_TOPIC, "kafkaMsgOffset <= -1");
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -230,21 +241,24 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, "kafkaMsgOffset = 9");
runKafkaSQLVerifyCount(queryString, expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
//"greater_than" such that value = endOffset-2
queryString = String.format(TestQueryConstants.QUERY_TEMPLATE_BASIC,
TestQueryConstants.JSON_PUSHDOWN_TOPIC, "kafkaMsgOffset > 8");
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
//"greater_than_or_equal" such that value = endOffset-1
queryString = String.format(TestQueryConstants.QUERY_TEMPLATE_BASIC,
TestQueryConstants.JSON_PUSHDOWN_TOPIC, "kafkaMsgOffset >= 9");
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -262,7 +276,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate1, predicate2);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -280,7 +295,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate1, predicate2);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowInPlan));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowInPlan)}, null);
}
/**
@@ -299,7 +315,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate1, predicate2, predicate3);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCount));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCount)}, null);
}
/**
@@ -319,7 +336,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate1, predicate2, predicate3, predicate4);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCountInPlan));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCountInPlan)}, null);
}
/**
@@ -338,7 +356,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate1, predicate2);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCountInPlan));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCountInPlan)}, null);
}
/**
@@ -358,7 +377,8 @@ public class KafkaFilterPushdownTest extends KafkaTestBase {
TestQueryConstants.JSON_PUSHDOWN_TOPIC, predicate1, predicate2, predicate3);
runKafkaSQLVerifyCount(queryString,expectedRowCount);
- testPhysicalPlan(queryString, String.format(expectedSubStr, expectedRowCountInPlan));
+ PlanTestBase.testPlanMatchingPatterns(queryString, JSON_FORMAT,
+ new String[] {String.format(EXPECTED_PATTERN, expectedRowCountInPlan)}, null);
}
}
diff --git a/exec/java-exec/src/test/java/org/apache/drill/PlanTestBase.java b/exec/java-exec/src/test/java/org/apache/drill/PlanTestBase.java
index fae593f..a777276 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/PlanTestBase.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/PlanTestBase.java
@@ -17,18 +17,8 @@
*/
package org.apache.drill;
-import java.nio.file.Paths;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
-import java.io.File;
-import java.io.IOException;
-import java.nio.file.Files;
-import java.util.List;
-import java.util.Stack;
-import java.util.regex.Matcher;
-import java.util.regex.Pattern;
-
+import org.apache.calcite.sql.SqlExplain.Depth;
+import org.apache.calcite.sql.SqlExplainLevel;
import org.apache.commons.io.FileUtils;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.exec.record.RecordBatchLoader;
@@ -37,13 +27,23 @@ import org.apache.drill.exec.rpc.user.QueryDataBatch;
import org.apache.drill.exec.store.parquet.metadata.Metadata;
import org.apache.drill.exec.vector.NullableVarCharVector;
import org.apache.drill.exec.vector.ValueVector;
-import org.apache.calcite.sql.SqlExplain.Depth;
-import org.apache.calcite.sql.SqlExplainLevel;
-
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
import org.apache.drill.shaded.guava.com.google.common.base.Strings;
import org.apache.drill.test.BaseTestQuery;
import org.apache.drill.test.QueryTestUtil;
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Paths;
+import java.util.List;
+import java.util.Stack;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
public class PlanTestBase extends BaseTestQuery {
static final org.slf4j.Logger logger = org.slf4j.LoggerFactory.getLogger(PlanTestBase.class);
@@ -92,7 +92,23 @@ public class PlanTestBase extends BaseTestQuery {
public static void testPlanMatchingPatterns(String query, Pattern[] expectedPatterns, Pattern[] excludedPatterns)
throws Exception {
- final String plan = getPlanInString("EXPLAIN PLAN for " + QueryTestUtil.normalizeQuery(query), OPTIQ_FORMAT);
+ testPlanMatchingPatterns(query, OPTIQ_FORMAT, expectedPatterns, excludedPatterns);
+ }
+
+ public static void testPlanMatchingPatterns(String query, String planFormat,
+ String[] expectedPatterns, String[] excludedPatterns)
+ throws Exception {
+ Preconditions.checkArgument((planFormat.equals(OPTIQ_FORMAT) || planFormat.equals(JSON_FORMAT)), "Unsupported " +
+ "plan format %s is provided for explain plan query. Supported formats are: %s, %s", planFormat, OPTIQ_FORMAT,
+ JSON_FORMAT);
+ testPlanMatchingPatterns(query, planFormat, stringsToPatterns(expectedPatterns),
+ stringsToPatterns(excludedPatterns));
+ }
+
+ private static void testPlanMatchingPatterns(String query, String planFormat,
+ Pattern[] expectedPatterns, Pattern[] excludedPatterns)
+ throws Exception {
+ final String plan = getPlanInString("EXPLAIN PLAN for " + QueryTestUtil.normalizeQuery(query), planFormat);
// Check and make sure all expected patterns are in the plan
if (expectedPatterns != null) {
[drill] 05/05: DRILL-7240: Catch runtime pruning filter-match
exceptions and do not prune these rowgroups
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boaz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit e5e9b35588c06de6ba5b0b072567031daa50e00c
Author: Ben-Zvi <bb...@mapr.com>
AuthorDate: Tue May 7 13:30:21 2019 -0700
DRILL-7240: Catch runtime pruning filter-match exceptions and do not prune these rowgroups
closes #1783
---
.../parquet/AbstractParquetScanBatchCreator.java | 37 +++++++++++++++-------
1 file changed, 26 insertions(+), 11 deletions(-)
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 41f52d3..d95ee6d 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
@@ -107,6 +107,8 @@ public abstract class AbstractParquetScanBatchCreator {
long totalPruneTime = 0;
long totalRowgroups = rowGroupScan.getRowGroupReadEntries().size();
Stopwatch pruneTimer = Stopwatch.createUnstarted();
+ int countMatchClassCastExceptions = 0; // in case match() hits CCE, count and report these
+ String matchCastErrorMessage = ""; // report the error too (Java insists on initializing this ....)
// If pruning - Prepare the predicate and the columns before the FOR LOOP
if ( doRuntimePruning ) {
@@ -182,22 +184,31 @@ public abstract class AbstractParquetScanBatchCreator {
Map<SchemaPath, ColumnStatistics> columnsStatistics = ParquetTableMetadataUtils.getRowGroupColumnStatistics(tableMetadataV4, rowGroupMetadata);
//
- // Perform the Run-Time Pruning - i.e. Skip this rowgroup if the match fails
+ // Perform the Run-Time Pruning - i.e. Skip/prune this rowgroup if the match fails
//
- RowsMatch match = FilterEvaluatorUtils.matches(filterPredicate, columnsStatistics, footerRowCount);
-
- // collect logging info
- long timeToRead = pruneTimer.elapsed(TimeUnit.MICROSECONDS);
+ RowsMatch matchResult = RowsMatch.ALL; // default (in case of exception) - do not prune this rowgroup
+ try {
+ matchResult = FilterEvaluatorUtils.matches(filterPredicate, columnsStatistics, footerRowCount);
+
+ // collect logging info
+ long timeToRead = pruneTimer.elapsed(TimeUnit.MICROSECONDS);
+ totalPruneTime += timeToRead;
+ logger.trace("Run-time pruning: {} row-group {} (RG index: {} row count: {}), took {} usec", // trace each single rowgroup
+ matchResult == RowsMatch.NONE ? "Excluded" : "Included", rowGroup.getPath(), rowGroupIndex, footerRowCount, timeToRead);
+ } catch (ClassCastException cce) {
+ countMatchClassCastExceptions++; // one more CCE occured
+ matchCastErrorMessage = cce.getMessage(); // report the (last) error message
+ } catch (Exception e) {
+ // in case some unexpected exception is raised
+ logger.warn("Run-time pruning check failed - {}. Skip pruning rowgroup - {}", e.getMessage(), rowGroup.getPath());
+ }
pruneTimer.stop();
pruneTimer.reset();
- totalPruneTime += timeToRead;
- logger.trace("Run-time pruning: {} row-group {} (RG index: {} row count: {}), took {} usec", // trace each single rowgroup
- match == RowsMatch.NONE ? "Excluded" : "Included", rowGroup.getPath(), rowGroupIndex, footerRowCount, timeToRead);
- // If this rowgroup failed the match - skip it
- if (match == RowsMatch.NONE) {
+ // If this rowgroup failed the match - skip it (i.e., no reader for this rowgroup)
+ if (matchResult == RowsMatch.NONE) {
rowgroupsPruned++; // one more RG was pruned
- if (firstRowGroup == null) { // keep first RG, to be used in case all row groups are pruned
+ if (firstRowGroup == null) { // keep the first RG, to be used in case all row groups are pruned
firstRowGroup = rowGroup;
firstFooter = footer;
}
@@ -214,10 +225,14 @@ public abstract class AbstractParquetScanBatchCreator {
mapWithMaxColumns = createReaderAndImplicitColumns(context, rowGroupScan, oContext, columnExplorer, readers, implicitColumns, mapWithMaxColumns, firstRowGroup, fs,
firstFooter, true);
}
+ // do some logging, if relevant
if ( totalPruneTime > 0 ) {
logger.info("Finished parquet_runtime_pruning in {} usec. Out of given {} rowgroups, {} were pruned. {}", totalPruneTime, totalRowgroups, rowgroupsPruned,
totalRowgroups == rowgroupsPruned ? "ALL_PRUNED !!" : "");
}
+ if ( countMatchClassCastExceptions > 0 ) {
+ logger.info("Run-time pruning skipped for {} out of {} rowgroups due to: {}",countMatchClassCastExceptions, totalRowgroups, matchCastErrorMessage);
+ }
// Update stats (same in every reader - the others would just overwrite the stats)
for (CommonParquetRecordReader rr : readers ) {
[drill] 02/05: DRILL-7148: Use improved join cardinality and ndv
estimation with statistics
Posted by bo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
boaz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
commit c1ca512ea54cc7fe1b7ee581f21aa84a95aab14f
Author: Gautam Parai <gp...@maprtech.com>
AuthorDate: Tue Mar 19 12:04:13 2019 -0700
DRILL-7148: Use improved join cardinality and ndv estimation with statistics
closes #1744
---
.../exec/planner/common/DrillJoinRelBase.java | 34 ++--
.../drill/exec/planner/common/DrillRelOptUtil.java | 62 +++++++
.../drill/exec/planner/common/DrillStatsTable.java | 2 +-
.../planner/cost/DrillRelMdDistinctRowCount.java | 181 +++++++++++++++++----
.../exec/planner/cost/DrillRelMdRowCount.java | 41 +++--
.../exec/planner/cost/DrillRelMdSelectivity.java | 40 +++--
.../exec/planner/physical/PlannerSettings.java | 6 +
.../exec/server/options/SystemOptionManager.java | 1 +
.../java/org/apache/drill/exec/util/Utilities.java | 2 +-
.../java-exec/src/main/resources/drill-module.conf | 1 +
.../org/apache/drill/exec/sql/TestAnalyze.java | 12 +-
.../drill/metastore/ColumnStatisticsKind.java | 2 +-
12 files changed, 295 insertions(+), 89 deletions(-)
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillJoinRelBase.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillJoinRelBase.java
index a6b6f4a..6150bf3 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillJoinRelBase.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillJoinRelBase.java
@@ -17,6 +17,7 @@
*/
package org.apache.drill.exec.planner.common;
+import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
@@ -33,6 +34,7 @@ import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.util.ImmutableBitSet;
+import org.apache.calcite.util.Pair;
import org.apache.drill.exec.ExecConstants;
import org.apache.drill.exec.expr.holders.IntHolder;
import org.apache.drill.exec.physical.impl.join.JoinUtils;
@@ -102,26 +104,32 @@ public abstract class DrillJoinRelBase extends Join implements DrillJoin {
return joinRowFactor * this.getLeft().estimateRowCount(mq) * this.getRight().estimateRowCount(mq);
}
- int[] joinFields = new int[2];
-
LogicalJoin jr = LogicalJoin.create(this.getLeft(), this.getRight(), this.getCondition(),
this.getVariablesSet(), this.getJoinType());
if (!DrillRelOptUtil.guessRows(this) //Statistics present for left and right side of the join
- && jr.getJoinType() == JoinRelType.INNER
- && DrillRelOptUtil.analyzeSimpleEquiJoin((Join)jr, joinFields)) {
- ImmutableBitSet leq = ImmutableBitSet.of(joinFields[0]);
- ImmutableBitSet req = ImmutableBitSet.of(joinFields[1]);
+ && jr.getJoinType() == JoinRelType.INNER) {
+ List<Pair<Integer, Integer>> joinConditions = DrillRelOptUtil.analyzeSimpleEquiJoin((Join)jr);
+ if (joinConditions.size() > 0) {
+ List<Integer> leftSide = new ArrayList<>();
+ List<Integer> rightSide = new ArrayList<>();
+ for (Pair<Integer, Integer> condition : joinConditions) {
+ leftSide.add(condition.left);
+ rightSide.add(condition.right);
+ }
+ ImmutableBitSet leq = ImmutableBitSet.of(leftSide);
+ ImmutableBitSet req = ImmutableBitSet.of(rightSide);
- Double ldrc = mq.getDistinctRowCount(this.getLeft(), leq, null);
- Double rdrc = mq.getDistinctRowCount(this.getRight(), req, null);
+ Double ldrc = mq.getDistinctRowCount(this.getLeft(), leq, null);
+ Double rdrc = mq.getDistinctRowCount(this.getRight(), req, null);
- Double lrc = mq.getRowCount(this.getLeft());
- Double rrc = mq.getRowCount(this.getRight());
+ Double lrc = mq.getRowCount(this.getLeft());
+ Double rrc = mq.getRowCount(this.getRight());
- if (ldrc != null && rdrc != null && lrc != null && rrc != null) {
- // Join cardinality = (lrc * rrc) / Math.max(ldrc, rdrc). Avoid overflow by dividing earlier
- return (lrc / Math.max(ldrc, rdrc)) * rrc;
+ if (ldrc != null && rdrc != null && lrc != null && rrc != null) {
+ // Join cardinality = (lrc * rrc) / Math.max(ldrc, rdrc). Avoid overflow by dividing earlier
+ return (lrc / Math.max(ldrc, rdrc)) * rrc;
+ }
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillRelOptUtil.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillRelOptUtil.java
index 66499d6..3838bf9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillRelOptUtil.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillRelOptUtil.java
@@ -667,4 +667,66 @@ public abstract class DrillRelOptUtil {
}
return drillTable;
}
+
+ public static List<Pair<Integer, Integer>> analyzeSimpleEquiJoin(Join join) {
+ List<Pair<Integer, Integer>> joinConditions = new ArrayList<>();
+ try {
+ RexVisitor<Void> visitor =
+ new RexVisitorImpl<Void>(true) {
+ public Void visitCall(RexCall call) {
+ if (call.getKind() == SqlKind.AND || call.getKind() == SqlKind.OR) {
+ super.visitCall(call);
+ } else {
+ if (call.getKind() == SqlKind.EQUALS) {
+ int leftFieldCount = join.getLeft().getRowType().getFieldCount();
+ int rightFieldCount = join.getRight().getRowType().getFieldCount();
+ RexNode leftComparand = call.operands.get(0);
+ RexNode rightComparand = call.operands.get(1);
+ RexInputRef leftFieldAccess = (RexInputRef) leftComparand;
+ RexInputRef rightFieldAccess = (RexInputRef) rightComparand;
+ if (leftFieldAccess.getIndex() >= leftFieldCount + rightFieldCount ||
+ rightFieldAccess.getIndex() >= leftFieldCount + rightFieldCount) {
+ joinConditions.clear();
+ throw new Util.FoundOne(call);
+ }
+ /* Both columns reference same table */
+ if ((leftFieldAccess.getIndex() >= leftFieldCount &&
+ rightFieldAccess.getIndex() >= leftFieldCount) ||
+ (leftFieldAccess.getIndex() < leftFieldCount &&
+ rightFieldAccess.getIndex() < leftFieldCount)) {
+ joinConditions.clear();
+ throw new Util.FoundOne(call);
+ } else {
+ if (leftFieldAccess.getIndex() < leftFieldCount) {
+ joinConditions.add(Pair.of(leftFieldAccess.getIndex(),
+ rightFieldAccess.getIndex() - leftFieldCount));
+ } else {
+ joinConditions.add(Pair.of(rightFieldAccess.getIndex(),
+ leftFieldAccess.getIndex() - leftFieldCount));
+ }
+ }
+ }
+ }
+ return null;
+ }
+ };
+ join.getCondition().accept(visitor);
+ } catch (Util.FoundOne ex) {
+ Util.swallow(ex, null);
+ }
+ return joinConditions;
+ }
+
+ public static List<RexInputRef> findAllRexInputRefs(final RexNode node) {
+ List<RexInputRef> rexRefs = new ArrayList<>();
+ RexVisitor<Void> visitor =
+ new RexVisitorImpl<Void>(true) {
+ public Void visitInputRef(RexInputRef inputRef) {
+ rexRefs.add(inputRef);
+ return super.visitInputRef(inputRef);
+ }
+ };
+ node.accept(visitor);
+ return rexRefs;
+ }
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillStatsTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillStatsTable.java
index 78e87d0..7565abf 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillStatsTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/common/DrillStatsTable.java
@@ -483,7 +483,7 @@ public class DrillStatsTable {
Map<StatisticsKind, Object> statisticsValues = new HashMap<>();
Double ndv = statsProvider.getNdv(fieldName);
if (ndv != null) {
- statisticsValues.put(ColumnStatisticsKind.NVD, ndv);
+ statisticsValues.put(ColumnStatisticsKind.NDV, ndv);
}
Double nonNullCount = statsProvider.getNNRowCount(fieldName);
if (nonNullCount != null) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdDistinctRowCount.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdDistinctRowCount.java
index 37cd55a..8b11a9a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdDistinctRowCount.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdDistinctRowCount.java
@@ -17,14 +17,18 @@
*/
package org.apache.drill.exec.planner.cost;
+import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
+import java.util.Set;
import org.apache.calcite.plan.RelOptUtil;
import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.SingleRel;
import org.apache.calcite.rel.core.Join;
import org.apache.calcite.rel.core.JoinRelType;
+import org.apache.calcite.rel.core.TableScan;
import org.apache.calcite.rel.core.Window;
import org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider;
import org.apache.calcite.rel.metadata.RelMdDistinctRowCount;
@@ -33,23 +37,25 @@ import org.apache.calcite.rel.metadata.RelMetadataProvider;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexInputRef;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.rex.RexUtil;
+import org.apache.calcite.sql.SqlKind;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.exec.planner.common.DrillJoinRelBase;
import org.apache.drill.exec.planner.common.DrillRelOptUtil;
-import org.apache.drill.exec.planner.common.DrillScanRelBase;
import org.apache.drill.exec.planner.logical.DrillScanRel;
import org.apache.drill.exec.planner.logical.DrillTable;
+import org.apache.drill.exec.planner.physical.PlannerSettings;
+import org.apache.drill.exec.planner.physical.PrelUtil;
+import org.apache.drill.exec.util.Utilities;
import org.apache.drill.metastore.ColumnStatistics;
import org.apache.drill.metastore.ColumnStatisticsKind;
import org.apache.drill.metastore.TableMetadata;
-import java.io.IOException;
-import org.apache.drill.exec.util.Utilities;
-
public class DrillRelMdDistinctRowCount extends RelMdDistinctRowCount{
private static final DrillRelMdDistinctRowCount INSTANCE =
new DrillRelMdDistinctRowCount();
@@ -80,10 +86,10 @@ public class DrillRelMdDistinctRowCount extends RelMdDistinctRowCount{
@Override
public Double getDistinctRowCount(RelNode rel, RelMetadataQuery mq, ImmutableBitSet groupKey, RexNode predicate) {
- if (rel instanceof DrillScanRelBase) { // Applies to both Drill Logical and Physical Rels
+ if (rel instanceof TableScan) { // Applies to Calcite/Drill logical and Drill physical rels
if (!DrillRelOptUtil.guessRows(rel)) {
DrillTable table = Utilities.getDrillTable(rel.getTable());
- return getDistinctRowCountInternal(((DrillScanRelBase) rel), mq, table, groupKey, rel.getRowType(), predicate);
+ return getDistinctRowCountInternal(((TableScan) rel), mq, table, groupKey, rel.getRowType(), predicate);
} else {
/* If we are not using statistics OR there is no table or metadata (stats) table associated with scan,
* estimate the distinct row count. Consistent with the estimation of Aggregate row count in
@@ -124,7 +130,7 @@ public class DrillRelMdDistinctRowCount extends RelMdDistinctRowCount{
* set of columns indicated by groupKey.
* column").
*/
- private Double getDistinctRowCountInternal(DrillScanRelBase scan, RelMetadataQuery mq, DrillTable table,
+ private Double getDistinctRowCountInternal(TableScan scan, RelMetadataQuery mq, DrillTable table,
ImmutableBitSet groupKey, RelDataType type, RexNode predicate) {
double selectivity, rowCount;
/* If predicate is present, determine its selectivity to estimate filtered rows.
@@ -136,7 +142,6 @@ public class DrillRelMdDistinctRowCount extends RelMdDistinctRowCount{
if (groupKey.length() == 0) {
return selectivity * rowCount;
}
-
/* If predicate is present, determine its selectivity to estimate filtered rows. Thereafter,
* compute the number of distinct rows
*/
@@ -148,29 +153,39 @@ public class DrillRelMdDistinctRowCount extends RelMdDistinctRowCount{
// Statistics cannot be obtained, use default behaviour
return scan.estimateRowCount(mq) * 0.1;
}
- double s = 1.0;
+ double s = 1.0;
+ boolean allCols = true;
for (int i = 0; i < groupKey.length(); i++) {
final String colName = type.getFieldNames().get(i);
// Skip NDV, if not available
if (!groupKey.get(i)) {
- continue;
+ allCols = false;
+ break;
}
- ColumnStatistics columnStatistics = tableMetadata != null ? tableMetadata.getColumnStatistics(SchemaPath.getSimplePath(colName)) : null;
- Double ndv = columnStatistics != null ? (Double) columnStatistics.getStatistic(ColumnStatisticsKind.NVD) : null;
+ ColumnStatistics columnStatistics = tableMetadata != null ?
+ tableMetadata.getColumnStatistics(SchemaPath.getSimplePath(colName)) : null;
+ Double ndv = columnStatistics != null ? (Double) columnStatistics.getStatistic(ColumnStatisticsKind.NDV) : null;
if (ndv == null) {
continue;
}
- s *= 1 - ndv / rowCount;
+ s *= ndv;
+ selectivity = getPredSelectivityContainingInputRef(predicate, i, mq, scan);
+ /* If predicate is on group-by column, scale down the NDV by selectivity. Consider the query
+ * select a, b from t where a = 10 group by a, b. Here, NDV(a) will be scaled down by SEL(a)
+ * whereas NDV(b) will not.
+ */
+ if (selectivity > 0) {
+ s *= selectivity;
+ }
}
- if (s > 0 && s < 1.0) {
- return (1 - s) * selectivity * rowCount;
- } else if (s == 1.0) {
+ s = Math.min(s, rowCount);
+ if (!allCols) {
// Could not get any NDV estimate from stats - probably stats not present for GBY cols. So Guess!
return scan.estimateRowCount(mq) * 0.1;
} else {
- /* rowCount maybe less than NDV(different source), sanity check OR NDV not used at all */
- return selectivity * rowCount;
+ /* rowCount maybe less than NDV(different source), sanity check OR NDV not used at all */
+ return s;
}
}
@@ -195,7 +210,7 @@ public class DrillRelMdDistinctRowCount extends RelMdDistinctRowCount{
if (predicate != null) {
List<RexNode> leftFilters = new ArrayList<>();
List<RexNode> rightFilters = new ArrayList<>();
- List<RexNode> joinFilters = new ArrayList<>();
+ List<RexNode> joinFilters = new ArrayList();
List<RexNode> predList = RelOptUtil.conjunctions(predicate);
RelOptUtil.classifyFilters(joinRel, predList, joinType, joinType == JoinRelType.INNER,
!joinType.generatesNullsOnLeft(), !joinType.generatesNullsOnRight(), joinFilters,
@@ -205,28 +220,122 @@ public class DrillRelMdDistinctRowCount extends RelMdDistinctRowCount{
rightPred = RexUtil.composeConjunction(rexBuilder, rightFilters, true);
}
- Double leftDistRowCount = null;
- Double rightDistRowCount = null;
double distRowCount = 1;
- ImmutableBitSet lmb = leftMask.build();
- ImmutableBitSet rmb = rightMask.build();
- // Get NDV estimates for the left and right side predicates, if applicable
- if (lmb.length() > 0) {
- leftDistRowCount = mq.getDistinctRowCount(left, lmb, leftPred);
- if (leftDistRowCount != null) {
- distRowCount = leftDistRowCount;
+ int gbyCols = 0;
+ PlannerSettings plannerSettings = PrelUtil.getPlannerSettings(joinRel.getCluster().getPlanner());
+ /*
+ * The NDV for a multi-column GBY key past a join is determined as follows:
+ * GBY(s1, s2, s3) = CNDV(s1)*CNDV(s2)*CNDV(s3)
+ * where CNDV is determined as follows:
+ * A) If sX is present as a join column (sX = tX) CNDV(sX) = MIN(NDV(sX), NDV(tX)) where X =1, 2, 3, etc
+ * B) Otherwise, based on independence assumption CNDV(sX) = NDV(sX)
+ */
+ Set<ImmutableBitSet> joinFiltersSet = new HashSet<>();
+ for (RexNode filter : RelOptUtil.conjunctions(joinRel.getCondition())) {
+ final RelOptUtil.InputFinder inputFinder = RelOptUtil.InputFinder.analyze(filter);
+ joinFiltersSet.add(inputFinder.inputBitSet.build());
+ }
+ for (int idx = 0; idx < groupKey.length(); idx++) {
+ if (groupKey.get(idx)) {
+ // GBY key is present in some filter - now try options A) and B) as described above
+ double ndvSGby = Double.MAX_VALUE;
+ boolean presentInFilter = false;
+ ImmutableBitSet sGby = getSingleGbyKey(groupKey, idx);
+ if (sGby != null) {
+ for (ImmutableBitSet jFilter : joinFiltersSet) {
+ if (jFilter.contains(sGby)) {
+ presentInFilter = true;
+ // Found join condition containing this GBY key. Pick min NDV across all columns in this join
+ for (int fidx : jFilter) {
+ if (fidx < left.getRowType().getFieldCount()) {
+ ndvSGby = Math.min(ndvSGby, mq.getDistinctRowCount(left, ImmutableBitSet.of(fidx), leftPred));
+ } else {
+ ndvSGby = Math.min(ndvSGby, mq.getDistinctRowCount(right, ImmutableBitSet.of(fidx-left.getRowType().getFieldCount()), rightPred));
+ }
+ }
+ break;
+ }
+ }
+ // Did not find it in any join condition(s)
+ if (!presentInFilter) {
+ for (int sidx : sGby) {
+ if (sidx < left.getRowType().getFieldCount()) {
+ ndvSGby = mq.getDistinctRowCount(left, ImmutableBitSet.of(sidx), leftPred);
+ } else {
+ ndvSGby = mq.getDistinctRowCount(right, ImmutableBitSet.of(sidx-left.getRowType().getFieldCount()), rightPred);
+ }
+ }
+ }
+ ++gbyCols;
+ // Multiply NDV(s) of different GBY cols to determine the overall NDV
+ distRowCount *= ndvSGby;
+ }
}
}
- if (rmb.length() > 0) {
- rightDistRowCount = mq.getDistinctRowCount(right, rmb, rightPred);
- if (rightDistRowCount != null) {
- distRowCount = rightDistRowCount;
+ if (gbyCols > 1) { // Scale with multi-col NDV factor if more than one GBY cols were found
+ distRowCount *= plannerSettings.getStatisticsMultiColNdvAdjustmentFactor();
+ }
+ double joinRowCount = mq.getRowCount(joinRel);
+ // Cap NDV to join row count
+ distRowCount = Math.min(distRowCount, joinRowCount);
+ return RelMdUtil.numDistinctVals(distRowCount, joinRowCount);
+ }
+
+ private ImmutableBitSet getSingleGbyKey(ImmutableBitSet groupKey, int idx) {
+ if (groupKey.get(idx)) {
+ return ImmutableBitSet.builder().set(idx, idx+1).build();
+ } else {
+ return null;
+ }
+ }
+
+ private double getPredSelectivityContainingInputRef(RexNode predicate, int inputRef,
+ RelMetadataQuery mq, TableScan scan) {
+ if (predicate instanceof RexCall) {
+ if (predicate.getKind() == SqlKind.AND) {
+ double sel, andSel = 1.0;
+ for (RexNode op : ((RexCall) predicate).getOperands()) {
+ sel = getPredSelectivityContainingInputRef(op, inputRef, mq, scan);
+ if (sel > 0) {
+ andSel *= sel;
+ }
+ }
+ return andSel;
+ } else if (predicate.getKind() == SqlKind.OR) {
+ double sel, orSel = 0.0;
+ for (RexNode op : ((RexCall) predicate).getOperands()) {
+ sel = getPredSelectivityContainingInputRef(op, inputRef, mq, scan);
+ if (sel > 0) {
+ orSel += sel;
+ }
+ }
+ return orSel;
+ } else {
+ for (RexNode op : ((RexCall) predicate).getOperands()) {
+ if (op instanceof RexInputRef && inputRef != ((RexInputRef) op).getIndex()) {
+ return -1.0;
+ }
+ }
+ return mq.getSelectivity(scan, predicate);
}
+ } else {
+ return -1.0;
}
- // Use max of NDVs from both sides of the join, if applicable
- if (leftDistRowCount != null && rightDistRowCount != null) {
- distRowCount = Math.max(leftDistRowCount, rightDistRowCount);
+ }
+
+ @Override
+ public Double getDistinctRowCount(RelSubset rel, RelMetadataQuery mq,
+ ImmutableBitSet groupKey, RexNode predicate) {
+ if (!DrillRelOptUtil.guessRows(rel)) {
+ final RelNode best = rel.getBest();
+ if (best != null) {
+ return mq.getDistinctRowCount(best, groupKey, predicate);
+ }
+ final RelNode original = rel.getOriginal();
+ if (original != null) {
+ return mq.getDistinctRowCount(original, groupKey, predicate);
+ }
}
- return RelMdUtil.numDistinctVals(distRowCount, mq.getRowCount(joinRel));
+ return super.getDistinctRowCount(rel, mq, groupKey, predicate);
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdRowCount.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdRowCount.java
index 814a96d..b65c582 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdRowCount.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdRowCount.java
@@ -18,7 +18,6 @@
package org.apache.drill.exec.planner.cost;
import java.io.IOException;
-import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.SingleRel;
import org.apache.calcite.rel.core.Aggregate;
import org.apache.calcite.rel.core.Filter;
@@ -35,7 +34,9 @@ import org.apache.calcite.util.BuiltInMethod;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.drill.exec.planner.common.DrillLimitRelBase;
import org.apache.drill.exec.planner.common.DrillRelOptUtil;
+import org.apache.drill.exec.planner.common.DrillScanRelBase;
import org.apache.drill.exec.planner.logical.DrillTable;
+import org.apache.drill.exec.planner.physical.AggPrelBase;
import org.apache.drill.exec.planner.physical.PlannerSettings;
import org.apache.drill.exec.planner.physical.PrelUtil;
import org.apache.drill.exec.util.Utilities;
@@ -53,7 +54,21 @@ public class DrillRelMdRowCount extends RelMdRowCount{
if (groupKey.isEmpty()) {
return 1.0;
- } else {
+ } else if (!DrillRelOptUtil.guessRows(rel) &&
+ rel instanceof AggPrelBase &&
+ ((AggPrelBase) rel).getOperatorPhase() == AggPrelBase.OperatorPhase.PHASE_1of2) {
+ // Phase 1 Aggregate would return rows in the range [NDV, input_rows]. Hence, use the
+ // existing estimate of 1/10 * input_rows
+ Double rowCount = mq.getRowCount(rel.getInput()) / 10;
+ Double ndv = mq.getDistinctRowCount(rel.getInput(), groupKey, null);
+ // Use max of NDV and input_rows/10
+ if (ndv != null) {
+ rowCount = Math.max(ndv, rowCount);
+ }
+ // Grouping sets multiply
+ rowCount *= rel.getGroupSets().size();
+ return rowCount;
+ } else {
return super.getRowCount(rel, mq);
}
}
@@ -88,20 +103,13 @@ public class DrillRelMdRowCount extends RelMdRowCount{
}
@Override
- public Double getRowCount(RelNode rel, RelMetadataQuery mq) {
- if (rel instanceof TableScan) {
- return getRowCountInternal((TableScan)rel, mq);
- }
- return super.getRowCount(rel, mq);
- }
-
- @Override
public Double getRowCount(Filter rel, RelMetadataQuery mq) {
// Need capped selectivity estimates. See the Filter getRows() method
return rel.getRows();
}
- private Double getRowCountInternal(TableScan rel, RelMetadataQuery mq) {
+ @Override
+ public Double getRowCount(TableScan rel, RelMetadataQuery mq) {
DrillTable table = Utilities.getDrillTable(rel.getTable());
PlannerSettings settings = PrelUtil.getSettings(rel.getCluster());
// If guessing, return selectivity from RelMDRowCount
@@ -112,14 +120,19 @@ public class DrillRelMdRowCount extends RelMdRowCount{
try {
if (table != null
&& table.getGroupScan().getTableMetadata() != null
- && (boolean) TableStatisticsKind.HAS_STATISTICS.getValue(table.getGroupScan().getTableMetadata())
+ && (boolean) TableStatisticsKind.HAS_STATISTICS.getValue(table.getGroupScan().getTableMetadata())) {
/* For GroupScan rely on accurate count from the scan, if available, instead of
* statistics since partition pruning/filter pushdown might have occurred.
* e.g. ParquetGroupScan returns accurate rowcount. The other way would be to
* iterate over the rowgroups present in the GroupScan to compute the rowcount.
*/
- && !(table.getGroupScan().getScanStats(settings).getGroupScanProperty().hasExactRowCount())) {
- return (Double) TableStatisticsKind.EST_ROW_COUNT.getValue(table.getGroupScan().getTableMetadata());
+ if (!table.getGroupScan().getScanStats(settings).getGroupScanProperty().hasExactRowCount()) {
+ return (Double) TableStatisticsKind.EST_ROW_COUNT.getValue(table.getGroupScan().getTableMetadata());
+ } else {
+ if (!(rel instanceof DrillScanRelBase)) {
+ return table.getGroupScan().getScanStats(settings).getRecordCount();
+ }
+ }
}
} catch (IOException ex) {
return super.getRowCount(rel, mq);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdSelectivity.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdSelectivity.java
index d04b19e..533cd1e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdSelectivity.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdSelectivity.java
@@ -19,12 +19,12 @@ package org.apache.drill.exec.planner.cost;
import java.io.IOException;
import java.util.ArrayList;
+import java.util.HashSet;
import java.util.List;
import java.util.EnumSet;
import java.util.Set;
import java.util.Map;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.stream.Collectors;
@@ -180,7 +180,26 @@ public class DrillRelMdSelectivity extends RelMdSelectivity {
double orSel = 0;
for (RexNode orPred : RelOptUtil.disjunctions(pred)) {
if (isMultiColumnPredicate(orPred) && !combinedRangePredicates.contains(orPred)) {
- orSel += RelMdUtil.guessSelectivity(orPred); //CALCITE guess
+ Set uniqueRefs = new HashSet<>();
+ uniqueRefs.add(DrillRelOptUtil.findAllRexInputRefs(orPred));
+ // If equality predicate involving single column - selectivity is 1.0
+ if (uniqueRefs.size() == 1) {
+ try {
+ RexVisitor<Void> visitor =
+ new RexVisitorImpl<Void>(true) {
+ public Void visitCall(RexCall call) {
+ if (call.getKind() != SqlKind.EQUALS) {
+ throw new Util.FoundOne(call);
+ }
+ return super.visitCall(call);
+ }
+ };
+ pred.accept(visitor);
+ orSel += 1.0;
+ } catch (Util.FoundOne e) {
+ orSel += RelMdUtil.guessSelectivity(orPred); //CALCITE guess
+ }
+ }
} else if (orPred.isA(SqlKind.EQUALS)) {
orSel += computeEqualsSelectivity(tableMetadata, orPred, fieldNames);
} else if (orPred.isA(RANGE_PREDICATE) || combinedRangePredicates.contains(orPred)) {
@@ -276,7 +295,7 @@ public class DrillRelMdSelectivity extends RelMdSelectivity {
SchemaPath col = getColumn(orPred, fieldNames);
if (col != null) {
ColumnStatistics columnStatistics = tableMetadata != null ? tableMetadata.getColumnStatistics(col) : null;
- Double ndv = columnStatistics != null ? (Double) columnStatistics.getStatistic(ColumnStatisticsKind.NVD) : null;
+ Double ndv = columnStatistics != null ? (Double) columnStatistics.getStatistic(ColumnStatisticsKind.NDV) : null;
if (ndv != null) {
return 1.00 / ndv;
}
@@ -423,19 +442,6 @@ public class DrillRelMdSelectivity extends RelMdSelectivity {
}
private boolean isMultiColumnPredicate(final RexNode node) {
- return findAllRexInputRefs(node).size() > 1;
- }
-
- private static List<RexInputRef> findAllRexInputRefs(final RexNode node) {
- List<RexInputRef> rexRefs = new ArrayList<>();
- RexVisitor<Void> visitor =
- new RexVisitorImpl<Void>(true) {
- public Void visitInputRef(RexInputRef inputRef) {
- rexRefs.add(inputRef);
- return super.visitInputRef(inputRef);
- }
- };
- node.accept(visitor);
- return rexRefs;
+ return DrillRelOptUtil.findAllRexInputRefs(node).size() > 1;
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
index a5506d4..ee10a15 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/PlannerSettings.java
@@ -231,6 +231,8 @@ public class PlannerSettings implements Context{
public static final BooleanValidator STATISTICS_USE = new BooleanValidator("planner.statistics.use", null);
+ public static final RangeDoubleValidator STATISTICS_MULTICOL_NDV_ADJUST_FACTOR = new RangeDoubleValidator("planner.statistics.multicol_ndv_adjustment_factor", 0.0, 1.0, null);
+
public OptionManager options = null;
public FunctionImplementationRegistry functionImplementationRegistry = null;
@@ -475,6 +477,10 @@ public class PlannerSettings implements Context{
return options.getOption(STATISTICS_USE);
}
+ public double getStatisticsMultiColNdvAdjustmentFactor() {
+ return options.getOption(STATISTICS_MULTICOL_NDV_ADJUST_FACTOR);
+ }
+
@Override
public <T> T unwrap(Class<T> clazz) {
if(clazz == PlannerSettings.class){
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
index 61fefe7..4c28887 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/server/options/SystemOptionManager.java
@@ -121,6 +121,7 @@ public class SystemOptionManager extends BaseOptionManager implements AutoClosea
new OptionDefinition(PlannerSettings.ENABLE_UNNEST_LATERAL),
new OptionDefinition(PlannerSettings.FORCE_2PHASE_AGGR), // for testing
new OptionDefinition(PlannerSettings.STATISTICS_USE),
+ new OptionDefinition(PlannerSettings.STATISTICS_MULTICOL_NDV_ADJUST_FACTOR),
new OptionDefinition(ExecConstants.HASHJOIN_NUM_PARTITIONS_VALIDATOR),
new OptionDefinition(ExecConstants.HASHJOIN_MAX_MEMORY_VALIDATOR, new OptionMetaData(OptionValue.AccessibleScopes.SYSTEM, true, true)),
new OptionDefinition(ExecConstants.HASHJOIN_NUM_ROWS_IN_BATCH_VALIDATOR),
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java b/exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java
index 2446ba7..f440cf1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java
@@ -105,7 +105,7 @@ public class Utilities {
*/
public static DrillTable getDrillTable(RelOptTable table) {
DrillTable drillTable = table.unwrap(DrillTable.class);
- if (drillTable == null) {
+ if (drillTable == null && table.unwrap(DrillTranslatableTable.class) != null) {
drillTable = table.unwrap(DrillTranslatableTable.class).getDrillTable();
}
return drillTable;
diff --git a/exec/java-exec/src/main/resources/drill-module.conf b/exec/java-exec/src/main/resources/drill-module.conf
index 5096680..ea254a5 100644
--- a/exec/java-exec/src/main/resources/drill-module.conf
+++ b/exec/java-exec/src/main/resources/drill-module.conf
@@ -591,6 +591,7 @@ drill.exec.options: {
planner.producer_consumer_queue_size: 10,
planner.slice_target: 100000,
planner.statistics.use: false,
+ planner.statistics.multicol_ndv_adjustment_factor: 1.0,
planner.store.parquet.rowgroup.filter.pushdown.enabled: true,
planner.store.parquet.rowgroup.filter.pushdown.threshold: 10000,
# Max per node should always be configured as zero and
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/sql/TestAnalyze.java b/exec/java-exec/src/test/java/org/apache/drill/exec/sql/TestAnalyze.java
index 94583de..1d404e1 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/sql/TestAnalyze.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/sql/TestAnalyze.java
@@ -258,14 +258,14 @@ public class TestAnalyze extends BaseTestQuery {
query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
+ " on emp.department_id = dept.department_id";
- String[] expectedPlan4 = {"HashJoin\\(condition.*\\).*rowcount = 1154.9999999999995,.*",
+ String[] expectedPlan4 = {"HashJoin\\(condition.*\\).*rowcount = 1155.0,.*",
"Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
"Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan4, new String[]{});
query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
+ " on emp.department_id = dept.department_id where dept.department_id = 5";
- String[] expectedPlan5 = {"HashJoin\\(condition.*\\).*rowcount = 96.24999999999997,.*",
+ String[] expectedPlan5 = {"HashJoin\\(condition.*\\).*rowcount = 96.25,.*",
"Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
"Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan5, new String[]{});
@@ -290,8 +290,8 @@ public class TestAnalyze extends BaseTestQuery {
query = " select emp.employee_id from dfs.tmp.employeeUseStat emp join dfs.tmp.departmentUseStat dept"
+ " on emp.department_id = dept.department_id "
+ " group by emp.employee_id";
- String[] expectedPlan8 = {"HashAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 730.0992454469839,.*",
- "HashJoin\\(condition.*\\).*rowcount = 1154.9999999999995,.*",
+ String[] expectedPlan8 = {"HashAgg\\(group=\\[\\{0\\}\\]\\).*rowcount = 115.49475630811243,.*",
+ "HashJoin\\(condition.*\\).*rowcount = 1155.0,.*",
"Scan.*columns=\\[`department_id`, `employee_id`\\].*rowcount = 1155.0.*",
"Scan.*columns=\\[`department_id`\\].*rowcount = 12.0.*"};
PlanTestBase.testPlanWithAttributesMatchingPatterns(query, expectedPlan8, new String[]{});
@@ -301,8 +301,8 @@ public class TestAnalyze extends BaseTestQuery {
+ " on emp.department_id = dept.department_id "
+ " group by emp.employee_id, emp.store_id, dept.department_description "
+ " having dept.department_description = 'FINANCE'";
- String[] expectedPlan9 = {"HashAgg\\(group=\\[\\{0, 1, 2\\}\\]\\).*rowcount = 92.3487011031316.*",
- "HashJoin\\(condition.*\\).*rowcount = 96.24999999999997,.*",
+ String[] expectedPlan9 = {"HashAgg\\(group=\\[\\{0, 1, 2\\}\\]\\).*rowcount = 60.84160378724867.*",
+ "HashJoin\\(condition.*\\).*rowcount = 96.25,.*",
"Scan.*columns=\\[`department_id`, `employee_id`, `store_id`\\].*rowcount = 1155.0.*",
"Filter\\(condition=\\[=\\(\\$1, 'FINANCE'\\)\\]\\).*rowcount = 1.0,.*",
"Scan.*columns=\\[`department_id`, `department_description`\\].*rowcount = 12.0.*"};
diff --git a/metastore/metastore-api/src/main/java/org/apache/drill/metastore/ColumnStatisticsKind.java b/metastore/metastore-api/src/main/java/org/apache/drill/metastore/ColumnStatisticsKind.java
index f1c8196..51195f4 100644
--- a/metastore/metastore-api/src/main/java/org/apache/drill/metastore/ColumnStatisticsKind.java
+++ b/metastore/metastore-api/src/main/java/org/apache/drill/metastore/ColumnStatisticsKind.java
@@ -145,7 +145,7 @@ public enum ColumnStatisticsKind implements CollectableColumnStatisticsKind {
/**
* Column statistics kind which represents number of distinct values for the specific column.
*/
- NVD(Statistic.NDV) {
+ NDV(Statistic.NDV) {
@Override
public Object mergeStatistics(List<? extends ColumnStatistics> statisticsList) {
throw new UnsupportedOperationException("Cannot merge statistics for NDV");