You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by vo...@apache.org on 2021/01/14 21:23:03 UTC
[drill] branch master updated: DRILL-3637: ElasticSearch storage
plugin implementation based on Calcite adapter (#2135)
This is an automated email from the ASF dual-hosted git repository.
volodymyr pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/drill.git
The following commit(s) were added to refs/heads/master by this push:
new 7edbe0f DRILL-3637: ElasticSearch storage plugin implementation based on Calcite adapter (#2135)
7edbe0f is described below
commit 7edbe0f38eadcdf1a590e0e1e4730564ee58a498
Author: Volodymyr Vysotskyi <vv...@gmail.com>
AuthorDate: Thu Jan 14 23:22:48 2021 +0200
DRILL-3637: ElasticSearch storage plugin implementation based on Calcite adapter (#2135)
---
contrib/pom.xml | 1 +
.../src/test/resources/druid/docker-compose.yaml | 13 +-
contrib/storage-elasticsearch/README.md | 51 ++
contrib/storage-elasticsearch/pom.xml | 121 +++++
.../adapter/elasticsearch/CalciteUtils.java | 84 +++
.../DrillElasticsearchTableScan.java} | 25 +-
.../elasticsearch/ElasticsearchStorageConfig.java | 96 ++++
.../elasticsearch/ElasticsearchStoragePlugin.java | 74 +++
.../elasticsearch/plan/ElasticPlanTransformer.java | 172 ++++++
.../plan/ElasticSearchEnumerablePrelContext.java | 79 +++
.../plan/ElasticsearchFilterRule.java | 65 +++
.../plan/ElasticsearchProjectRule.java | 88 +++
.../store/elasticsearch/plan/NodeTypeFinder.java} | 32 +-
.../schema/ElasticsearchDrillSchema.java | 70 +++
.../schema/ElasticsearchDrillSchemaFactory.java | 44 ++
.../schema/ElasticsearchDynamicTable.java | 78 +++
.../main/resources/bootstrap-storage-plugins.json | 10 +
.../src/main/resources/drill-module.conf | 24 +
.../elasticsearch/ElasticComplexTypesTest.java | 152 ++++++
.../store/elasticsearch/ElasticInfoSchemaTest.java | 122 +++++
.../store/elasticsearch/ElasticSearchPlanTest.java | 141 +++++
.../elasticsearch/ElasticSearchQueryTest.java | 597 +++++++++++++++++++++
.../drill/exec/store/jdbc/DrillJdbcConvention.java | 3 +-
.../jdbc/JdbcIntermediatePrelConverterRule.java | 3 +-
.../org/apache/drill/exec/store/jdbc/JdbcPrel.java | 18 +-
distribution/pom.xml | 5 +
distribution/src/assemble/component.xml | 1 +
.../apache/drill/exec/compile/ClassBuilder.java | 20 +-
.../impl/scan/convert/ConvertDoubleToDecimal.java | 30 +-
.../impl/scan/convert/ConvertFloatToDecimal.java | 30 +-
.../impl/scan/convert/ConvertIntToDecimal.java | 30 +-
.../impl/scan/convert/ConvertLongToDecimal.java | 30 +-
.../impl/scan/convert/DirectConverter.java | 5 +
.../impl/scan/convert/StandardConversions.java | 21 +-
.../physical/resultSet/impl/ProjectionFilter.java | 20 +-
.../drill/exec/planner/common/DrillRelOptUtil.java | 18 +-
.../logical/DrillPushProjectIntoScanRule.java | 14 +-
.../drill/exec/planner/logical/DrillTable.java | 4 -
.../drill/exec/planner/physical/LeafPrel.java | 26 +-
.../drill/exec/planner/physical/ScanPrel.java | 13 +-
.../drill/exec/planner/physical/UnnestPrel.java | 14 +-
.../planner/physical/visitor/BasePrelVisitor.java | 5 +
.../exec/planner/physical/visitor/PrelVisitor.java | 2 +
.../physical/visitor/PrelVisualizerVisitor.java | 7 +
.../physical/visitor/StarColumnConverter.java | 10 +-
.../drill/exec/planner/sql/SchemaUtilites.java | 2 +-
.../planner/sql/handlers/AnalyzeTableHandler.java | 11 -
.../sql/handlers/MetastoreAnalyzeTableHandler.java | 3 +-
.../apache/drill/exec/record/ColumnConverter.java | 277 ++++++++++
.../drill/exec/record/ColumnConverterFactory.java | 149 +++++
.../org/apache/drill/exec/store/SubsetRemover.java | 32 +-
.../drill/exec/store/avro/AvroBatchReader.java | 14 +-
.../store/avro/AvroColumnConverterFactory.java | 215 ++++++++
.../drill/exec/store/avro/ColumnConverter.java | 172 ------
.../exec/store/avro/ColumnConverterFactory.java | 258 ---------
.../exec/store/enumerable/DrillDataContext.java | 58 ++
.../store/enumerable/EnumerableBatchCreator.java | 70 +++
.../exec/store/enumerable/EnumerableGroupScan.java | 128 +++++
.../store/enumerable/EnumerableRecordReader.java | 162 ++++++
.../exec/store/enumerable/EnumerableSubScan.java | 78 +++
.../plan/EnumerableIntermediatePrel.java | 76 +++
.../EnumerableIntermediatePrelConverterRule.java | 23 +-
.../exec/store/enumerable/plan/EnumerablePrel.java | 116 ++++
.../enumerable/plan/EnumerablePrelContext.java | 35 +-
.../exec/store/enumerable/plan/VertexDrel.java | 20 +-
.../enumerable/plan/VertexDrelConverterRule.java | 15 +-
.../table/function/WithOptionsTableMacro.java | 4 +-
.../physical/impl/common/HashPartitionTest.java | 12 +-
.../impl/scan/convert/TestDirectConverter.java | 12 +-
.../drill/exec/physical/rowSet/TestRowSet.java | 2 +-
.../exec/physical/rowSet/TestScalarAccessors.java | 16 +-
.../easy/text/compliant/TestCsvWithSchema.java | 4 +-
.../apache/drill/test/rowSet/RowSetComparison.java | 8 +
.../apache/drill/test/rowSet/RowSetUtilities.java | 5 +
.../src/main/codegen/data/ValueVectorTypes.tdd | 2 +-
.../main/codegen/templates/ColumnAccessors.java | 40 +-
.../record/metadata/PrimitiveColumnMetadata.java | 2 +-
.../drill/exec/vector/accessor/ScalarReader.java | 1 +
.../drill/exec/vector/accessor/ValueType.java | 5 +
.../drill/exec/vector/accessor/ValueWriter.java | 1 +
.../accessor/reader/AbstractScalarReader.java | 9 +
.../accessor/writer/AbstractFixedWidthWriter.java | 11 +
.../accessor/writer/AbstractScalarWriter.java | 2 +-
.../accessor/writer/AbstractTupleWriter.java | 36 +-
.../vector/accessor/writer/BaseScalarWriter.java | 5 +
.../accessor/writer/NullableScalarWriter.java | 7 +
.../accessor/writer/dummy/DummyScalarWriter.java | 3 +
pom.xml | 4 +-
88 files changed, 3825 insertions(+), 748 deletions(-)
diff --git a/contrib/pom.xml b/contrib/pom.xml
index d9ea07a..a6f823d 100644
--- a/contrib/pom.xml
+++ b/contrib/pom.xml
@@ -59,6 +59,7 @@
<module>storage-opentsdb</module>
<module>storage-http</module>
<module>storage-druid</module>
+ <module>storage-elasticsearch</module>
</modules>
</project>
diff --git a/contrib/storage-druid/src/test/resources/druid/docker-compose.yaml b/contrib/storage-druid/src/test/resources/druid/docker-compose.yaml
index ab81225..6efb68e 100644
--- a/contrib/storage-druid/src/test/resources/druid/docker-compose.yaml
+++ b/contrib/storage-druid/src/test/resources/druid/docker-compose.yaml
@@ -7,14 +7,13 @@
# "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
+# 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.
+# 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.
#
version: "2.2"
diff --git a/contrib/storage-elasticsearch/README.md b/contrib/storage-elasticsearch/README.md
new file mode 100644
index 0000000..2ef77ad
--- /dev/null
+++ b/contrib/storage-elasticsearch/README.md
@@ -0,0 +1,51 @@
+# Drill ElasticSearch Plugin
+
+Drill ElasticSearch storage plugin allows you to perform SQL queries against ElasticSearch indices.
+This storage plugin implementation is based on [Apache Calcite adapter for ElasticSearch](https://calcite.apache.org/docs/elasticsearch_adapter.html).
+
+For more details about supported versions please refer to [Supported versions](https://calcite.apache.org/docs/elasticsearch_adapter.html#supported-versions) page.
+
+### Supported optimizations and features
+
+This storage plugin supports the following optimizations:
+
+- Project pushdown
+- Filter pushdown (only expressions supported by Calcite adapter for ElasticSearch. Filter with unsupported expressions
+ wouldn't be pushed to ElasticSearch but will be produced by Drill)
+- Limit pushdown
+- Aggregation pushdown
+- Sort pushdown
+
+Besides these optimizations, ElasticSearch storage plugin supports the schema provisioning feature.
+For more details please refer to [Specifying the Schema as Table Function Parameter](https://drill.apache.org/docs/plugin-configuration-basics/#specifying-the-schema-as-table-function-parameter).
+
+### Plugin registration
+
+The plugin can be registered in Apache Drill using the drill web interface by navigating to the `storage` page.
+Following is the default registration configuration.
+
+```json
+{
+ "type": "elastic",
+ "hosts": [
+ "http://localhost:9200"
+ ],
+ "username": null,
+ "password": null,
+ "enabled": false
+}
+```
+
+### Developer notes
+
+Most of the common classes required for creating storage plugins based on Calcite adapters are placed in the
+`java-exec` module, so they can be reused in future plugin implementations.
+
+Here is the list of the classes that may be useful:
+
+- `VertexDrelConverterRule` with `VertexDrel` - used to hold plugin-specific part of the plan at the end of the
+ `LOGICAL` planning phase.
+- `EnumerableIntermediatePrelConverterRule` with `EnumerableIntermediatePrel` - the same as above, but for the
+ `PHYSICAL` planning phase.
+- `EnumerablePrel` - responsible for generating java code that will be executed to query the storage plugin data source.
+- `EnumerableRecordReader` - executes java code generated in `EnumerablePrel` and transforms obtained results to Drill internal representation.
diff --git a/contrib/storage-elasticsearch/pom.xml b/contrib/storage-elasticsearch/pom.xml
new file mode 100644
index 0000000..0e1f7a6
--- /dev/null
+++ b/contrib/storage-elasticsearch/pom.xml
@@ -0,0 +1,121 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing, software
+ distributed under the License is distributed on an "AS IS" BASIS,
+ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ See the License for the specific language governing permissions and
+ limitations under the License.
+
+-->
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+ xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+ xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <properties>
+ <test.elasticsearch.version>7.10.1</test.elasticsearch.version>
+ </properties>
+ <parent>
+ <artifactId>drill-contrib-parent</artifactId>
+ <groupId>org.apache.drill.contrib</groupId>
+ <version>1.19.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>drill-storage-elasticsearch</artifactId>
+
+ <name>contrib/elasticsearch-storage-plugin</name>
+
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.drill.exec</groupId>
+ <artifactId>drill-java-exec</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>${calcite.groupId}</groupId>
+ <artifactId>calcite-elasticsearch</artifactId>
+ <version>${calcite.version}</version>
+ <exclusions>
+ <exclusion>
+ <groupId>commons-logging</groupId>
+ <artifactId>commons-logging</artifactId>
+ </exclusion></exclusions>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.drill.exec</groupId>
+ <artifactId>drill-java-exec</artifactId>
+ <classifier>tests</classifier>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.drill</groupId>
+ <artifactId>drill-common</artifactId>
+ <classifier>tests</classifier>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.elasticsearch.client</groupId>
+ <artifactId>elasticsearch-rest-high-level-client</artifactId>
+ <version>7.0.1</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>net.hydromatic</groupId>
+ <artifactId>foodmart-data-json</artifactId>
+ <version>0.4</version>
+ <scope>test</scope>
+ </dependency>
+ </dependencies>
+
+ <build>
+ <plugins>
+ <plugin>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <configuration>
+ <forkCount combine.self="override">1</forkCount>
+ </configuration>
+ </plugin>
+ <plugin>
+ <groupId>com.github.alexcojocaru</groupId>
+ <artifactId>elasticsearch-maven-plugin</artifactId>
+ <version>6.19</version>
+ <configuration>
+ <version>${test.elasticsearch.version}</version>
+ <clusterName>test</clusterName>
+ <transportPort>9300</transportPort>
+ <httpPort>9200</httpPort>
+ <skip>${skipTests}</skip>
+ </configuration>
+ <executions>
+ <execution>
+ <id>start-elasticsearch</id>
+ <phase>process-test-classes</phase>
+ <goals>
+ <goal>runforked</goal>
+ </goals>
+ </execution>
+ <execution>
+ <id>stop-elasticsearch</id>
+ <phase>post-integration-test</phase>
+ <goals>
+ <goal>stop</goal>
+ </goals>
+ </execution>
+ </executions>
+ </plugin>
+ </plugins>
+ </build>
+
+</project>
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/CalciteUtils.java b/contrib/storage-elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/CalciteUtils.java
new file mode 100644
index 0000000..4c3d6ec
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/CalciteUtils.java
@@ -0,0 +1,84 @@
+/*
+ * 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.calcite.adapter.elasticsearch;
+
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexNode;
+import org.apache.drill.exec.store.elasticsearch.ElasticsearchStorageConfig;
+import org.apache.drill.exec.store.elasticsearch.plan.ElasticSearchEnumerablePrelContext;
+import org.apache.drill.exec.store.elasticsearch.plan.ElasticsearchFilterRule;
+import org.apache.drill.exec.store.elasticsearch.plan.ElasticsearchProjectRule;
+import org.apache.drill.exec.store.enumerable.plan.EnumerableIntermediatePrelConverterRule;
+import org.apache.drill.exec.store.enumerable.plan.VertexDrelConverterRule;
+
+import java.util.Arrays;
+import java.util.List;
+import java.util.Set;
+import java.util.function.Predicate;
+import java.util.stream.Collectors;
+
+public class CalciteUtils {
+
+ private static final List<String> BANNED_RULES =
+ Arrays.asList("ElasticsearchProjectRule", "ElasticsearchFilterRule");
+
+ public static final Predicate<RelOptRule> RULE_PREDICATE =
+ relOptRule -> BANNED_RULES.stream()
+ .noneMatch(banned -> relOptRule.toString().startsWith(banned));
+
+ public static final VertexDrelConverterRule ELASTIC_DREL_CONVERTER_RULE =
+ new VertexDrelConverterRule(ElasticsearchRel.CONVENTION);
+
+ public static final EnumerableIntermediatePrelConverterRule ENUMERABLE_INTERMEDIATE_PREL_CONVERTER_RULE =
+ new EnumerableIntermediatePrelConverterRule(
+ new ElasticSearchEnumerablePrelContext(ElasticsearchStorageConfig.NAME));
+
+ public static Set<RelOptRule> elasticSearchRules() {
+ // filters Calcite implementations of some rules and adds alternative versions specific for Drill
+ Set<RelOptRule> rules = Arrays.stream(ElasticsearchRules.RULES)
+ .filter(RULE_PREDICATE)
+ .collect(Collectors.toSet());
+ rules.add(ENUMERABLE_INTERMEDIATE_PREL_CONVERTER_RULE);
+ rules.add(ELASTIC_DREL_CONVERTER_RULE);
+ rules.add(ElasticsearchProjectRule.INSTANCE);
+ rules.add(ElasticsearchFilterRule.INSTANCE);
+ return rules;
+ }
+
+ public static ConverterRule getElasticsearchToEnumerableConverterRule() {
+ return ElasticsearchToEnumerableConverterRule.INSTANCE;
+ }
+
+ public static ElasticsearchProject createProject(RelTraitSet traitSet, RelNode input,
+ List<? extends RexNode> projects, RelDataType rowType) {
+ return new ElasticsearchProject(input.getCluster(), traitSet, input, projects, rowType);
+ }
+
+ public static ElasticsearchFilter createFilter(RelTraitSet traitSet, RelNode input,
+ RexNode condition) {
+ return new ElasticsearchFilter(input.getCluster(), traitSet, input, condition);
+ }
+
+ public static void analyzePredicate(RexNode condition) throws PredicateAnalyzer.ExpressionNotAnalyzableException {
+ PredicateAnalyzer.analyze(condition);
+ }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/contrib/storage-elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/DrillElasticsearchTableScan.java
similarity index 51%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
copy to contrib/storage-elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/DrillElasticsearchTableScan.java
index 7107e9f..6b12d94 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/calcite/adapter/elasticsearch/DrillElasticsearchTableScan.java
@@ -15,25 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
+package org.apache.calcite.adapter.elasticsearch;
-import java.util.function.Predicate;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptPlanner;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.type.RelDataType;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.convert.ConverterRule;
-import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRelFactories;
+public class DrillElasticsearchTableScan extends ElasticsearchTableScan {
-class JdbcDrelConverterRule extends ConverterRule {
-
- JdbcDrelConverterRule(DrillJdbcConvention in) {
- super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
+ public DrillElasticsearchTableScan(RelOptCluster cluster, RelTraitSet traitSet, RelOptTable table, ElasticsearchTable elasticsearchTable, RelDataType projectRowType) {
+ super(cluster, traitSet, table, elasticsearchTable, projectRowType);
}
@Override
- public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
- convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
+ public void register(RelOptPlanner planner) {
+ // noop
}
}
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/ElasticsearchStorageConfig.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/ElasticsearchStorageConfig.java
new file mode 100644
index 0000000..a5a5be4
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/ElasticsearchStorageConfig.java
@@ -0,0 +1,96 @@
+/*
+ * 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.elasticsearch;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonIgnore;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.annotation.JsonTypeName;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.ObjectWriter;
+import org.apache.drill.common.logical.StoragePluginConfig;
+import org.apache.drill.shaded.guava.com.google.common.collect.ImmutableMap;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+@JsonTypeName(ElasticsearchStorageConfig.NAME)
+public class ElasticsearchStorageConfig extends StoragePluginConfig {
+ public static final String NAME = "elastic";
+
+ private static final ObjectWriter OBJECT_WRITER = new ObjectMapper().writerFor(List.class);
+
+ private final List<String> hosts;
+ private final String username;
+ private final String password;
+
+ @JsonCreator
+ public ElasticsearchStorageConfig(
+ @JsonProperty("hosts") List<String> hosts,
+ @JsonProperty("username") String username,
+ @JsonProperty("password") String password) {
+ this.hosts = hosts;
+ this.username = username;
+ this.password = password;
+ }
+
+ public List<String> getHosts() {
+ return hosts;
+ }
+
+ public String getUsername() {
+ return username;
+ }
+
+ public String getPassword() {
+ return password;
+ }
+
+ @JsonIgnore
+ public Map<String, Object> toConfigMap()
+ throws JsonProcessingException {
+ ImmutableMap.Builder<String, Object> builder = ImmutableMap.builder();
+ builder.put("hosts", OBJECT_WRITER.writeValueAsString(hosts));
+ if (username != null) {
+ builder.put("username", username)
+ .put("password", password);
+ }
+ return builder.build();
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ ElasticsearchStorageConfig that = (ElasticsearchStorageConfig) o;
+ return Objects.equals(hosts, that.hosts)
+ && Objects.equals(username, that.username)
+ && Objects.equals(password, that.password);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(hosts, username, password);
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/ElasticsearchStoragePlugin.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/ElasticsearchStoragePlugin.java
new file mode 100644
index 0000000..95cdbee
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/ElasticsearchStoragePlugin.java
@@ -0,0 +1,74 @@
+/*
+ * 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.elasticsearch;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.calcite.adapter.elasticsearch.CalciteUtils;
+import org.apache.calcite.plan.RelOptRule;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.drill.exec.ops.OptimizerRulesContext;
+import org.apache.drill.exec.planner.PlannerPhase;
+import org.apache.drill.exec.server.DrillbitContext;
+import org.apache.drill.exec.store.AbstractStoragePlugin;
+import org.apache.drill.exec.store.SchemaConfig;
+import org.apache.drill.exec.store.elasticsearch.schema.ElasticsearchDrillSchemaFactory;
+
+import java.util.Collections;
+import java.util.Set;
+
+public class ElasticsearchStoragePlugin extends AbstractStoragePlugin {
+ private final ElasticsearchStorageConfig config;
+ private final ElasticsearchDrillSchemaFactory schemaFactory;
+
+ public ElasticsearchStoragePlugin(
+ ElasticsearchStorageConfig config, DrillbitContext context, String name) {
+ super(context, name);
+ this.config = config;
+ this.schemaFactory = new ElasticsearchDrillSchemaFactory(name, this);
+ }
+
+ @Override
+ public void registerSchemas(SchemaConfig schemaConfig, SchemaPlus parent) throws JsonProcessingException {
+ schemaFactory.registerSchemas(schemaConfig, parent);
+ }
+
+ @Override
+ public ElasticsearchStorageConfig getConfig() {
+ return config;
+ }
+
+ @Override
+ public boolean supportsRead() {
+ return true;
+ }
+
+ @Override
+ public Set<? extends RelOptRule> getOptimizerRules(OptimizerRulesContext optimizerContext, PlannerPhase phase) {
+ switch (phase) {
+ case PHYSICAL:
+ case LOGICAL:
+ return CalciteUtils.elasticSearchRules();
+ case LOGICAL_PRUNE_AND_JOIN:
+ case LOGICAL_PRUNE:
+ case PARTITION_PRUNING:
+ case JOIN_PLANNING:
+ default:
+ return Collections.emptySet();
+ }
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticPlanTransformer.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticPlanTransformer.java
new file mode 100644
index 0000000..1063462
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticPlanTransformer.java
@@ -0,0 +1,172 @@
+/*
+ * 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.elasticsearch.plan;
+
+import org.apache.calcite.adapter.elasticsearch.CalciteUtils;
+import org.apache.calcite.adapter.elasticsearch.DrillElasticsearchTableScan;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchAggregate;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchFilter;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchProject;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchSort;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchTable;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.prepare.RelOptTableImpl;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelShuttleImpl;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rel.type.RelRecordType;
+import org.apache.calcite.rel.type.StructKind;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Objects;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+/**
+ * Implementation of RelShuttleImpl that transforms plan to fit Calcite ElasticSearch rel implementor.
+ */
+public class ElasticPlanTransformer extends RelShuttleImpl {
+
+ private boolean hasProject = false;
+
+ private RelDataTypeField mapField;
+
+ /**
+ * Replaces rowType of RelOptTable by rowType obtained from ElasticsearchTable.
+ */
+ @Override
+ public RelNode visit(TableScan other) {
+ RelOptTableImpl table = (RelOptTableImpl) other.getTable();
+ ElasticsearchTable elasticsearchTable = Objects.requireNonNull(
+ table.unwrap(ElasticsearchTable.class), "ElasticSearch table cannot be null");
+ RelDataType rowType = elasticsearchTable.getRowType(other.getCluster().getTypeFactory());
+ mapField = rowType.getFieldList().get(0);
+ return new DrillElasticsearchTableScan(other.getCluster(), other.getTraitSet(), table.copy(rowType), elasticsearchTable, rowType);
+ }
+
+ @Override
+ public RelNode visit(RelNode other) {
+ // replaces project expressions with ITEM calls, since Calcite returns single map column `_MAP`
+ // with actual table fields
+ if (other instanceof ElasticsearchProject) {
+ ElasticsearchProject project = (ElasticsearchProject) other;
+ RelNode input = project.getInput().accept(this);
+ List<RexNode> convertedExpressions = project.getProjects();
+ // project closest to the scan should be rewritten only
+ if (!this.hasProject) {
+ ElasticExpressionMapper expressionMapper =
+ new ElasticExpressionMapper(project.getCluster().getRexBuilder(),
+ project.getInput().getRowType(), mapField);
+ convertedExpressions = convertedExpressions.stream()
+ .map(expression -> expression.accept(expressionMapper))
+ .collect(Collectors.toList());
+
+ RelRecordType relDataType = getRelRecordType(other.getRowType());
+ this.hasProject = true;
+ return CalciteUtils.createProject(project.getTraitSet(), input,
+ convertedExpressions, relDataType);
+ } else {
+ return input;
+ }
+ } else if (other instanceof ElasticsearchFilter) {
+ ElasticsearchFilter filter = (ElasticsearchFilter) other;
+ RexNode convertedCondition = filter.getCondition().accept(
+ new ElasticExpressionMapper(other.getCluster().getRexBuilder(), filter.getInput().getRowType(), mapField));
+ return filter.copy(other.getTraitSet(), filter.getInput().accept(this), convertedCondition);
+ } else if (other instanceof ElasticsearchSort) {
+ ElasticsearchSort sort = (ElasticsearchSort) other;
+ RelNode input = getMappedInput(sort.getInput());
+ return sort.copy(other.getTraitSet(), input, sort.getCollation(), sort.offset, sort.fetch);
+ } else if (other instanceof ElasticsearchAggregate) {
+ ElasticsearchAggregate aggregate = (ElasticsearchAggregate) other;
+ RelNode input = getMappedInput(aggregate.getInput());
+ return aggregate.copy(other.getTraitSet(), input, aggregate.getGroupSet(),
+ aggregate.getGroupSets(), aggregate.getAggCallList());
+ }
+
+ return super.visit(other);
+ }
+
+ /**
+ * Generates project with mapped expressions above specified rel node
+ * if there is no other project in the tree.
+ */
+ private RelNode getMappedInput(RelNode relNode) {
+ boolean hasProject = this.hasProject;
+ this.hasProject = false;
+ RelNode input = relNode.accept(this);
+ if (!this.hasProject) {
+ this.hasProject = hasProject;
+ RelOptCluster cluster = relNode.getCluster();
+ List<RexNode> projections = IntStream.range(0, relNode.getRowType().getFieldCount())
+ .mapToObj(i -> cluster.getRexBuilder().makeInputRef(relNode, i))
+ .collect(Collectors.toList());
+
+ return CalciteUtils.createProject(relNode.getTraitSet(), relNode,
+ projections, relNode.getRowType()).accept(this);
+ } else {
+ return input;
+ }
+ }
+
+ private RelRecordType getRelRecordType(RelDataType rowType) {
+ List<RelDataTypeField> fields = new ArrayList<>();
+ for (RelDataTypeField relDataTypeField : rowType.getFieldList()) {
+ if (relDataTypeField.isDynamicStar()) {
+ fields.add(mapField);
+ } else {
+ fields.add(relDataTypeField);
+ }
+ }
+
+ return new RelRecordType(StructKind.FULLY_QUALIFIED, fields, false);
+ }
+
+ /**
+ * Implementation of RexShuttle that replaces RexInputRef expressions with ITEM calls to _MAP field.
+ */
+ public static class ElasticExpressionMapper extends RexShuttle {
+ private final RexBuilder rexBuilder;
+ private final RelDataType relDataType;
+ private final RelDataTypeField mapField;
+
+ public ElasticExpressionMapper(RexBuilder rexBuilder, RelDataType relDataType, RelDataTypeField mapField) {
+ this.rexBuilder = rexBuilder;
+ this.relDataType = relDataType;
+ this.mapField = mapField;
+ }
+
+ @Override
+ public RexNode visitInputRef(RexInputRef inputRef) {
+ if (inputRef.getType().getSqlTypeName() == SqlTypeName.DYNAMIC_STAR) {
+ return rexBuilder.makeInputRef(mapField.getType(), 0);
+ }
+ return rexBuilder.makeCall(SqlStdOperatorTable.ITEM, rexBuilder.makeInputRef(relDataType, 0),
+ rexBuilder.makeLiteral(relDataType.getFieldNames().get(inputRef.getIndex())));
+ }
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticSearchEnumerablePrelContext.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticSearchEnumerablePrelContext.java
new file mode 100644
index 0000000..29db1ca
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticSearchEnumerablePrelContext.java
@@ -0,0 +1,79 @@
+/*
+ * 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.elasticsearch.plan;
+
+import org.apache.calcite.adapter.elasticsearch.CalciteUtils;
+import org.apache.calcite.adapter.enumerable.EnumerableRel;
+import org.apache.calcite.adapter.enumerable.EnumerableRelImplementor;
+import org.apache.calcite.linq4j.tree.ClassDeclaration;
+import org.apache.calcite.linq4j.tree.Expressions;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.store.SubsetRemover;
+import org.apache.drill.exec.store.enumerable.plan.EnumerablePrelContext;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+public class ElasticSearchEnumerablePrelContext implements EnumerablePrelContext {
+
+ private final String planPrefix;
+
+ public ElasticSearchEnumerablePrelContext(String planPrefix) {
+ this.planPrefix = planPrefix;
+ }
+
+ @Override
+ public String generateCode(RelOptCluster cluster, RelNode elasticNode) {
+ RelNode enumerableRel =
+ CalciteUtils.getElasticsearchToEnumerableConverterRule().convert(elasticNode);
+
+ ClassDeclaration classDeclaration = new EnumerableRelImplementor(cluster.getRexBuilder(), Collections.emptyMap())
+ .implementRoot((EnumerableRel) enumerableRel, EnumerableRel.Prefer.ARRAY);
+ return Expressions.toString(Collections.singletonList(classDeclaration), "\n", false);
+ }
+
+ @Override
+ public RelNode transformNode(RelNode input) {
+ return input.accept(SubsetRemover.INSTANCE).accept(new ElasticPlanTransformer());
+ }
+
+ @Override
+ public Map<String, Integer> getFieldsMap(RelNode transformedNode) {
+ return transformedNode.getRowType().getFieldList().stream()
+ .collect(Collectors.toMap(
+ relDataTypeField -> relDataTypeField.getName().equals("_MAP")
+ ? SchemaPath.DYNAMIC_STAR
+ : relDataTypeField.getName(),
+ RelDataTypeField::getIndex
+ ));
+ }
+
+ @Override
+ public String getPlanPrefix() {
+ return planPrefix;
+ }
+
+ @Override
+ public String getTablePath(RelNode input) {
+ return null;
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticsearchFilterRule.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticsearchFilterRule.java
new file mode 100644
index 0000000..794f2e3
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticsearchFilterRule.java
@@ -0,0 +1,65 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.elasticsearch.plan;
+
+import org.apache.calcite.adapter.elasticsearch.CalciteUtils;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchFilter;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchRel;
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.rel.core.Filter;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ElasticsearchFilterRule extends ConverterRule {
+ private static final Logger logger = LoggerFactory.getLogger(ElasticsearchFilterRule.class);
+
+ public static final ElasticsearchFilterRule INSTANCE = new ElasticsearchFilterRule();
+
+ private final Convention out;
+
+ private ElasticsearchFilterRule() {
+ super(Filter.class, Convention.NONE, ElasticsearchRel.CONVENTION,
+ "DrillElasticsearchFilterRule");
+ this.out = ElasticsearchRel.CONVENTION;
+ }
+
+ @Override
+ public RelNode convert(RelNode relNode) {
+ Filter filter = (Filter) relNode;
+ NodeTypeFinder filterFinder = new NodeTypeFinder(ElasticsearchFilter.class);
+ filter.getInput().accept(filterFinder);
+ if (filterFinder.containsNode) {
+ return null;
+ }
+ RelTraitSet traitSet = filter.getTraitSet().replace(out);
+
+ try {
+ CalciteUtils.analyzePredicate(filter.getCondition());
+ } catch (Exception e) {
+ logger.info("Unable to push filter into ElasticSearch :{}", e.getMessage(), e);
+ return null;
+ }
+
+ return CalciteUtils.createFilter(traitSet,
+ convert(filter.getInput(), out), filter.getCondition());
+ }
+
+}
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticsearchProjectRule.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticsearchProjectRule.java
new file mode 100644
index 0000000..1995262
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/ElasticsearchProjectRule.java
@@ -0,0 +1,88 @@
+/*
+ * 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.elasticsearch.plan;
+
+import org.apache.calcite.adapter.elasticsearch.CalciteUtils;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchProject;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchRel;
+import org.apache.calcite.plan.Convention;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.convert.ConverterRule;
+import org.apache.calcite.rel.core.Project;
+import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rel.type.RelDataTypeField;
+import org.apache.calcite.rex.RexInputRef;
+import org.apache.calcite.rex.RexNode;
+import org.apache.drill.exec.planner.common.DrillRelOptUtil;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Rule for converting Drill project to ElasticSearch project.
+ * This rule contains a logic to split the project if it would have expressions
+ * and convert only a simple project to ElasticSearch project.
+ */
+public class ElasticsearchProjectRule extends ConverterRule {
+ private final Convention out;
+
+ public static final ElasticsearchProjectRule INSTANCE = new ElasticsearchProjectRule();
+
+ private ElasticsearchProjectRule() {
+ super(Project.class, Convention.NONE, ElasticsearchRel.CONVENTION,
+ "DrillElasticsearchProjectRule");
+ this.out = ElasticsearchRel.CONVENTION;
+ }
+
+ @Override
+ public RelNode convert(RelNode relNode) {
+ Project project = (Project) relNode;
+ NodeTypeFinder projectFinder = new NodeTypeFinder(ElasticsearchProject.class);
+ project.getInput().accept(projectFinder);
+ if (projectFinder.containsNode) {
+ // Calcite adapter allows only a single Elasticsearch project per tree
+ return null;
+ }
+ RelTraitSet traitSet = project.getTraitSet().replace(out);
+ List<RexNode> innerProjections = new ArrayList<>();
+ RelDataType rowType = project.getInput().getRowType();
+
+ // check for literals only without input exprs
+ DrillRelOptUtil.InputRefVisitor collectRefs = new DrillRelOptUtil.InputRefVisitor();
+ project.getChildExps().forEach(exp -> exp.accept(collectRefs));
+
+ if (!collectRefs.getInputRefs().isEmpty()) {
+ for (RelDataTypeField relDataTypeField : rowType.getFieldList()) {
+ innerProjections.add(project.getCluster().getRexBuilder().makeInputRef(project.getInput(), relDataTypeField.getIndex()));
+ }
+ }
+
+ boolean allExprsInputRefs = project.getChildExps().stream().allMatch(rexNode -> rexNode instanceof RexInputRef);
+ if (collectRefs.getInputRefs().isEmpty() || allExprsInputRefs) {
+ return CalciteUtils.createProject(traitSet,
+ convert(project.getInput(), out), project.getProjects(), project.getRowType());
+ } else {
+ Project elasticsearchProject = CalciteUtils.createProject(traitSet,
+ convert(project.getInput(), out), innerProjections, project.getInput().getRowType());
+ return project.copy(project.getTraitSet(), elasticsearchProject,
+ project.getProjects(), project.getRowType());
+ }
+ }
+
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/NodeTypeFinder.java
similarity index 54%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
copy to contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/NodeTypeFinder.java
index 7107e9f..68125fa 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/plan/NodeTypeFinder.java
@@ -15,25 +15,31 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
-
-import java.util.function.Predicate;
+package org.apache.drill.exec.store.elasticsearch.plan;
+import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.convert.ConverterRule;
-import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRelFactories;
+import org.apache.calcite.rel.RelShuttleImpl;
+import org.apache.calcite.util.Util;
+
+public class NodeTypeFinder extends RelShuttleImpl {
+ public boolean containsNode = false;
-class JdbcDrelConverterRule extends ConverterRule {
+ private final Class<?> clazz;
- JdbcDrelConverterRule(DrillJdbcConvention in) {
- super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
+ public NodeTypeFinder(Class<?> clazz) {
+ this.clazz = clazz;
}
@Override
- public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
- convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
+ public RelNode visit(RelNode other) {
+ if (other.getClass().isAssignableFrom(clazz)) {
+ this.containsNode = true;
+ return other;
+ } else if (other instanceof RelSubset) {
+ RelSubset relSubset = (RelSubset) other;
+ return Util.first(relSubset.getBest(), relSubset.getOriginal()).accept(this);
+ }
+ return super.visit(other);
}
}
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/schema/ElasticsearchDrillSchema.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/schema/ElasticsearchDrillSchema.java
new file mode 100644
index 0000000..ca1be74
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/schema/ElasticsearchDrillSchema.java
@@ -0,0 +1,70 @@
+/*
+ * 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.elasticsearch.schema;
+
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.schema.Schema;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
+import org.apache.drill.exec.planner.logical.DrillTable;
+import org.apache.drill.exec.store.AbstractSchema;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.elasticsearch.ElasticsearchStorageConfig;
+
+import java.util.Collections;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+public class ElasticsearchDrillSchema extends AbstractSchema {
+ private final Schema delegatingSchema;
+ private final StoragePlugin plugin;
+ private final Map<String, Table> tables = new ConcurrentHashMap<>();
+
+ public ElasticsearchDrillSchema(String name, StoragePlugin plugin, Schema delegatingSchema) {
+ super(Collections.emptyList(), name);
+ this.plugin = plugin;
+ this.delegatingSchema = delegatingSchema;
+ }
+
+ @Override
+ public String getTypeName() {
+ return ElasticsearchStorageConfig.NAME;
+ }
+
+ @Override
+ public Table getTable(String tableName) {
+ return tables.computeIfAbsent(tableName, this::getDrillTable);
+ }
+
+ private DrillTable getDrillTable(String tableName) {
+ Table table = delegatingSchema.getTable(tableName);
+ return table == null ? null
+ : new ElasticsearchDynamicTable(plugin, tableName, null, table);
+ }
+
+ @Override
+ public Set<String> getTableNames() {
+ return delegatingSchema.getTableNames();
+ }
+
+ @Override
+ public Expression getExpression(SchemaPlus parentSchema, String name) {
+ return delegatingSchema.getExpression(parentSchema, name);
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/schema/ElasticsearchDrillSchemaFactory.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/schema/ElasticsearchDrillSchemaFactory.java
new file mode 100644
index 0000000..5430743
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/schema/ElasticsearchDrillSchemaFactory.java
@@ -0,0 +1,44 @@
+/*
+ * 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.elasticsearch.schema;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchSchemaFactory;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.drill.exec.store.AbstractSchemaFactory;
+import org.apache.drill.exec.store.SchemaConfig;
+import org.apache.drill.exec.store.elasticsearch.ElasticsearchStoragePlugin;
+
+public class ElasticsearchDrillSchemaFactory extends AbstractSchemaFactory {
+
+ private final ElasticsearchStoragePlugin plugin;
+ private final ElasticsearchSchemaFactory delegate;
+
+ public ElasticsearchDrillSchemaFactory(String name, ElasticsearchStoragePlugin plugin) {
+ super(name);
+ this.plugin = plugin;
+ this.delegate = new ElasticsearchSchemaFactory();
+ }
+
+ @Override
+ public void registerSchemas(SchemaConfig schemaConfig, SchemaPlus parent) throws JsonProcessingException {
+ ElasticsearchDrillSchema schema = new ElasticsearchDrillSchema(getName(), plugin,
+ delegate.create(parent, getName(), plugin.getConfig().toConfigMap()));
+ parent.add(getName(), schema);
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/schema/ElasticsearchDynamicTable.java b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/schema/ElasticsearchDynamicTable.java
new file mode 100644
index 0000000..a1919f5
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/java/org/apache/drill/exec/store/elasticsearch/schema/ElasticsearchDynamicTable.java
@@ -0,0 +1,78 @@
+/*
+ * 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.elasticsearch.schema;
+
+import org.apache.calcite.adapter.elasticsearch.DrillElasticsearchTableScan;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchRel;
+import org.apache.calcite.adapter.elasticsearch.ElasticsearchTable;
+import org.apache.calcite.linq4j.QueryProvider;
+import org.apache.calcite.linq4j.Queryable;
+import org.apache.calcite.linq4j.tree.Expression;
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptTable;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.schema.QueryableTable;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
+import org.apache.calcite.schema.TranslatableTable;
+import org.apache.calcite.schema.Wrapper;
+import org.apache.drill.exec.planner.logical.DynamicDrillTable;
+import org.apache.drill.exec.store.StoragePlugin;
+
+import java.lang.reflect.Type;
+
+public class ElasticsearchDynamicTable extends DynamicDrillTable implements TranslatableTable, QueryableTable, Wrapper {
+
+ private final ElasticsearchTable table;
+
+ public ElasticsearchDynamicTable(StoragePlugin plugin, String storageEngineName, Object selection, Table table) {
+ super(plugin, storageEngineName, selection);
+ this.table = (ElasticsearchTable) table;
+ }
+
+ @Override
+ public RelNode toRel(RelOptTable.ToRelContext context, RelOptTable relOptTable) {
+ RelOptCluster cluster = context.getCluster();
+ return new DrillElasticsearchTableScan(cluster,
+ cluster.traitSetOf(ElasticsearchRel.CONVENTION), relOptTable, table, relOptTable.getRowType());
+ }
+
+ @Override
+ public <T> Queryable<T> asQueryable(QueryProvider queryProvider, SchemaPlus schema, String tableName) {
+ return table.asQueryable(queryProvider, schema, tableName);
+ }
+
+ @Override
+ public Type getElementType() {
+ return table.getElementType();
+ }
+
+ public <C> C unwrap(Class<C> aClass) {
+ if (aClass.isInstance(this)) {
+ return aClass.cast(this);
+ } else if (aClass.isInstance(table)) {
+ return aClass.cast(table);
+ }
+ return null;
+ }
+
+ @Override
+ public Expression getExpression(SchemaPlus schema, String tableName, Class clazz) {
+ return table.getExpression(schema, tableName, clazz);
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/main/resources/bootstrap-storage-plugins.json b/contrib/storage-elasticsearch/src/main/resources/bootstrap-storage-plugins.json
new file mode 100644
index 0000000..ded7efa
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/resources/bootstrap-storage-plugins.json
@@ -0,0 +1,10 @@
+{
+ "storage":{
+ "elastic" : {
+ "type" : "elastic",
+ "hosts": ["http://localhost:9200"],
+ "username" : null,
+ "password" : null
+ }
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/main/resources/drill-module.conf b/contrib/storage-elasticsearch/src/main/resources/drill-module.conf
new file mode 100755
index 0000000..f0836fe
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/main/resources/drill-module.conf
@@ -0,0 +1,24 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+# This file tells Drill to consider this module when class path scanning.
+# This file can also include any supplementary configuration information.
+# This file is in HOCON format, see https://github.com/typesafehub/config/blob/master/HOCON.md for more information.
+drill.classpath.scanning: {
+ packages += "org.apache.drill.exec.store.elasticsearch"
+}
diff --git a/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticComplexTypesTest.java b/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticComplexTypesTest.java
new file mode 100644
index 0000000..75192e5
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticComplexTypesTest.java
@@ -0,0 +1,152 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.drill.shaded.guava.com.google.common.collect.ImmutableMap;
+import org.apache.drill.test.ClusterFixture;
+import org.apache.drill.test.ClusterTest;
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.client.indices.CreateIndexRequest;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+
+import static org.apache.drill.test.TestBuilder.listOf;
+import static org.apache.drill.test.TestBuilder.mapOf;
+
+public class ElasticComplexTypesTest extends ClusterTest {
+
+ private static final String HOST = "http://localhost:9200";
+
+ private static final List<String> indexNames = new ArrayList<>();
+
+ public static RestHighLevelClient restHighLevelClient;
+
+ @BeforeClass
+ public static void init() throws Exception {
+ startCluster(ClusterFixture.builder(dirTestWatcher));
+
+ ElasticsearchStorageConfig config = new ElasticsearchStorageConfig(
+ Collections.singletonList(HOST), null, null);
+ config.setEnabled(true);
+ cluster.defineStoragePlugin("elastic", config);
+
+ prepareData();
+ }
+
+ @AfterClass
+ public static void cleanUp() throws IOException {
+ for (String indexName : indexNames) {
+ restHighLevelClient.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
+ }
+ }
+
+ private static void prepareData() throws IOException {
+ restHighLevelClient = new RestHighLevelClient(RestClient.builder(HttpHost.create(HOST)));
+
+ String indexName = "arr";
+ indexNames.add(indexName);
+ CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
+
+ restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
+
+ XContentBuilder builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("string_arr", Arrays.asList("a", "b", "c", "d"));
+ builder.field("int_arr", Arrays.asList(1, 2, 3, 4, 0));
+ builder.field("nest_int_arr", Arrays.asList(Arrays.asList(1, 2), Arrays.asList(3, 4, 0)));
+ builder.endObject();
+ IndexRequest indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+ restHighLevelClient.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
+
+ indexName = "map";
+ indexNames.add(indexName);
+ createIndexRequest = new CreateIndexRequest(indexName);
+
+ restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("prim_field", 321);
+ builder.field("nest_field", ImmutableMap.of("a", 123, "b", "abc"));
+ builder.field("more_nest_field", ImmutableMap.of("a", 123, "b", ImmutableMap.of("c", "abc")));
+ builder.field("map_arr", Collections.singletonList(ImmutableMap.of("a", 123, "b", ImmutableMap.of("c", "abc"))));
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+ restHighLevelClient.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
+ }
+
+ @Test
+ public void testSelectStarWithArray() throws Exception {
+ testBuilder()
+ .sqlQuery("select * from elastic.arr")
+ .unOrdered()
+ .baselineColumns("string_arr", "int_arr", "nest_int_arr")
+ .baselineValues(listOf("a", "b", "c", "d"), listOf(1, 2, 3, 4, 0),
+ listOf(listOf(1, 2), listOf(3, 4, 0)))
+ .go();
+ }
+
+ @Test
+ public void testSelectArrayElem() throws Exception {
+ testBuilder()
+ .sqlQuery("select string_arr[0] c1, int_arr[1] c2, nest_int_arr[0][1] c3 from elastic.arr")
+ .unOrdered()
+ .baselineColumns("c1", "c2", "c3")
+ .baselineValues("a", 2, 2)
+ .go();
+ }
+
+ @Test
+ public void testSelectStarWithJson() throws Exception {
+ testBuilder()
+ .sqlQuery("select * from elastic.map")
+ .unOrdered()
+ .baselineColumns("prim_field", "nest_field", "more_nest_field", "map_arr")
+ .baselineValues(321, mapOf("a", 123, "b", "abc"),
+ mapOf("a", 123, "b", mapOf("c", "abc")),
+ listOf(mapOf("a", 123, "b", mapOf("c", "abc"))))
+ .go();
+ }
+
+ @Test
+ public void testSelectNestedFields() throws Exception {
+ testBuilder()
+ .sqlQuery("select m.nest_field.a a, m.nest_field.b b, m.more_nest_field.b.c c, map_arr[0].b.c d from elastic.map m")
+ .unOrdered()
+ .baselineColumns("a", "b", "c", "d")
+ .baselineValues(123, "abc", "abc", "abc")
+ .go();
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticInfoSchemaTest.java b/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticInfoSchemaTest.java
new file mode 100644
index 0000000..2bafc11
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticInfoSchemaTest.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.elasticsearch;
+
+import org.apache.drill.test.ClusterFixture;
+import org.apache.drill.test.ClusterTest;
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.client.indices.CreateIndexRequest;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+
+public class ElasticInfoSchemaTest extends ClusterTest {
+
+ private static final String HOST = "http://localhost:9200";
+
+ private static final List<String> indexNames = new ArrayList<>();
+
+ public static RestHighLevelClient restHighLevelClient;
+
+ @BeforeClass
+ public static void init() throws Exception {
+ startCluster(ClusterFixture.builder(dirTestWatcher));
+
+ ElasticsearchStorageConfig config = new ElasticsearchStorageConfig(
+ Collections.singletonList(HOST), null, null);
+ config.setEnabled(true);
+ cluster.defineStoragePlugin("elastic", config);
+
+ prepareData();
+ }
+
+ @AfterClass
+ public static void cleanUp() throws IOException {
+ for (String indexName : indexNames) {
+ restHighLevelClient.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
+ }
+ }
+
+ private static void prepareData() throws IOException {
+ restHighLevelClient = new RestHighLevelClient(RestClient.builder(HttpHost.create(HOST)));
+
+ String indexName = "t1";
+ indexNames.add(indexName);
+ CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
+
+ restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
+
+ XContentBuilder builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("string_field", "a");
+ builder.field("int_field", 123);
+ builder.endObject();
+ IndexRequest indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+ restHighLevelClient.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
+
+ indexName = "t2";
+ indexNames.add(indexName);
+ createIndexRequest = new CreateIndexRequest(indexName);
+
+ restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("another_int_field", 321);
+ builder.field("another_string_field", "b");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+ restHighLevelClient.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
+ }
+
+ @Test
+ public void testShowTables() throws Exception {
+ testBuilder()
+ .sqlQuery("show tables in elastic")
+ .unOrdered()
+ .baselineColumns("TABLE_SCHEMA", "TABLE_NAME")
+ .baselineValues("elastic", "t1")
+ .baselineValues("elastic", "t2")
+ .go();
+ }
+
+ @Test
+ public void testShowTablesLike() throws Exception {
+ testBuilder()
+ .sqlQuery("show tables in elastic like '%2%'")
+ .unOrdered()
+ .baselineColumns("TABLE_SCHEMA", "TABLE_NAME")
+ .baselineValues("elastic", "t2")
+ .go();
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticSearchPlanTest.java b/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticSearchPlanTest.java
new file mode 100644
index 0000000..89c0410
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticSearchPlanTest.java
@@ -0,0 +1,141 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.elasticsearch;
+
+import org.apache.drill.test.ClusterFixture;
+import org.apache.drill.test.ClusterTest;
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.client.indices.CreateIndexRequest;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.Collections;
+
+public class ElasticSearchPlanTest extends ClusterTest {
+
+ private static final String HOST = "http://localhost:9200";
+
+ public static RestHighLevelClient restHighLevelClient;
+
+ private static String indexName;
+
+ @BeforeClass
+ public static void init() throws Exception {
+ startCluster(ClusterFixture.builder(dirTestWatcher));
+
+ ElasticsearchStorageConfig config = new ElasticsearchStorageConfig(
+ Collections.singletonList(HOST), null, null);
+ config.setEnabled(true);
+ cluster.defineStoragePlugin("elastic", config);
+
+ prepareData();
+ }
+
+ @AfterClass
+ public static void cleanUp() throws IOException {
+ restHighLevelClient.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
+ }
+
+ private static void prepareData() throws IOException {
+ restHighLevelClient = new RestHighLevelClient(RestClient.builder(HttpHost.create(HOST)));
+
+ indexName = "nation";
+ CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
+
+ restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
+
+ XContentBuilder builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("n_nationkey", 0);
+ builder.field("n_name", "ALGERIA");
+ builder.field("n_regionkey", 1);
+ builder.endObject();
+ IndexRequest indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ restHighLevelClient.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
+ }
+
+ @Test
+ public void testProjectPushDown() throws Exception {
+ queryBuilder()
+ .sql("select n_name, n_nationkey from elastic.`nation`")
+ .planMatcher()
+ .include("ElasticsearchProject.*n_name.*n_nationkey")
+ .exclude("\\*\\*")
+ .match();
+ }
+
+ @Test
+ public void testFilterPushDown() throws Exception {
+ queryBuilder()
+ .sql("select n_name, n_nationkey from elastic.`nation` where n_nationkey = 0")
+ .planMatcher()
+ .include("ElasticsearchFilter")
+ .match();
+ }
+
+ @Test
+ public void testFilterPushDownWithJoin() throws Exception {
+ String query = "select * from elastic.`nation` e\n" +
+ "join elastic.`nation` s on e.n_nationkey = s.n_nationkey where e.n_name = 'algeria'";
+
+ queryBuilder()
+ .sql(query)
+ .planMatcher()
+ .include("ElasticsearchFilter")
+ .match();
+ }
+
+ @Test
+ public void testAggregationPushDown() throws Exception {
+ queryBuilder()
+ .sql("select count(*) from elastic.`nation`")
+ .planMatcher()
+ .include("ElasticsearchAggregate.*COUNT")
+ .match();
+ }
+
+ @Test
+ public void testLimitWithSortPushDown() throws Exception {
+ queryBuilder()
+ .sql("select n_nationkey from elastic.`nation` order by n_name limit 3")
+ .planMatcher()
+ .include("ElasticsearchSort.*sort.*fetch")
+ .match();
+ }
+
+ @Test
+ public void testAggregationWithGroupByPushDown() throws Exception {
+ queryBuilder()
+ .sql("select sum(n_nationkey) from elastic.`nation` group by n_regionkey")
+ .planMatcher()
+ .include("ElasticsearchAggregate.*SUM")
+ .match();
+ }
+}
diff --git a/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticSearchQueryTest.java b/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticSearchQueryTest.java
new file mode 100644
index 0000000..f7b1738
--- /dev/null
+++ b/contrib/storage-elasticsearch/src/test/java/org/apache/drill/exec/store/elasticsearch/ElasticSearchQueryTest.java
@@ -0,0 +1,597 @@
+/*
+ * 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.elasticsearch;
+
+import org.apache.drill.common.exceptions.UserRemoteException;
+import org.apache.drill.test.ClusterFixture;
+import org.apache.drill.test.ClusterTest;
+import org.apache.http.HttpHost;
+import org.elasticsearch.action.admin.indices.delete.DeleteIndexRequest;
+import org.elasticsearch.action.admin.indices.refresh.RefreshRequest;
+import org.elasticsearch.action.index.IndexRequest;
+import org.elasticsearch.client.RequestOptions;
+import org.elasticsearch.client.RestClient;
+import org.elasticsearch.client.RestHighLevelClient;
+import org.elasticsearch.client.indices.CreateIndexRequest;
+import org.elasticsearch.common.xcontent.XContentBuilder;
+import org.elasticsearch.common.xcontent.XContentFactory;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.time.LocalDate;
+import java.util.Collections;
+
+import static org.hamcrest.CoreMatchers.containsString;
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.junit.Assert.fail;
+
+public class ElasticSearchQueryTest extends ClusterTest {
+
+ private static final String HOST = "http://localhost:9200";
+
+ public static RestHighLevelClient restHighLevelClient;
+
+ private static String indexName;
+
+ @BeforeClass
+ public static void init() throws Exception {
+ startCluster(ClusterFixture.builder(dirTestWatcher));
+
+ ElasticsearchStorageConfig config = new ElasticsearchStorageConfig(
+ Collections.singletonList(HOST), null, null);
+ config.setEnabled(true);
+ cluster.defineStoragePlugin("elastic", config);
+
+ prepareData();
+ }
+
+ @AfterClass
+ public static void cleanUp() throws IOException {
+ restHighLevelClient.indices().delete(new DeleteIndexRequest(indexName), RequestOptions.DEFAULT);
+ }
+
+ private static void prepareData() throws IOException {
+ restHighLevelClient = new RestHighLevelClient(RestClient.builder(HttpHost.create(HOST)));
+
+ indexName = "employee";
+ CreateIndexRequest createIndexRequest = new CreateIndexRequest(indexName);
+
+ restHighLevelClient.indices().create(createIndexRequest, RequestOptions.DEFAULT);
+
+ XContentBuilder builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 1);
+ builder.field("full_name", "Sheri Nowmer");
+ builder.field("first_name", "Sheri");
+ builder.field("last_name", "Nowmer");
+ builder.field("position_id", 1);
+ builder.field("position_title", "President");
+ builder.field("store_id", 0);
+ builder.field("department_id", 1);
+ builder.field("birth_date", "1961-08-26");
+ builder.field("hire_date", "1994-12-01 00:00:00.0");
+ builder.field("salary", 80000.0);
+ builder.field("supervisor_id", 0);
+ builder.field("education_level", "Graduate Degree");
+ builder.field("marital_status", "S");
+ builder.field("gender", "F");
+ builder.field("management_role", "Senior Management");
+ builder.endObject();
+ IndexRequest indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 2);
+ builder.field("full_name", "Derrick Whelply");
+ builder.field("first_name", "Derrick");
+ builder.field("last_name", "Whelply");
+ builder.field("position_id", 2);
+ builder.field("position_title", "VP Country Manager");
+ builder.field("store_id", 0);
+ builder.field("department_id", 1);
+ builder.field("birth_date", "1915-07-03");
+ builder.field("hire_date", "1994-12-01 00:00:00.0");
+ builder.field("salary", 40000.0);
+ builder.field("supervisor_id", 1);
+ builder.field("education_level", "Graduate Degree");
+ builder.field("marital_status", "M");
+ builder.field("gender", "M");
+ builder.field("management_role", "Senior Management");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 4);
+ builder.field("full_name", "Michael Spence");
+ builder.field("first_name", "Michael");
+ builder.field("last_name", "Spence");
+ builder.field("position_id", 2);
+ builder.field("position_title", "VP Country Manager");
+ builder.field("store_id", 0);
+ builder.field("department_id", 1);
+ builder.field("birth_date", "1969-06-20");
+ builder.field("hire_date", "1998-01-01 00:00:00.0");
+ builder.field("salary", 40000.0);
+ builder.field("supervisor_id", 1);
+ builder.field("education_level", "Graduate Degree");
+ builder.field("marital_status", "S");
+ builder.field("gender", "M");
+ builder.field("management_role", "Senior Management");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 5);
+ builder.field("full_name", "Maya Gutierrez");
+ builder.field("first_name", "Maya");
+ builder.field("last_name", "Gutierrez");
+ builder.field("position_id", 2);
+ builder.field("position_title", "VP Country Manager");
+ builder.field("store_id", 0);
+ builder.field("department_id", 1);
+ builder.field("birth_date", "1951-05-10");
+ builder.field("hire_date", "1998-01-01 00:00:00.0");
+ builder.field("salary", 35000.0);
+ builder.field("supervisor_id", 1);
+ builder.field("education_level", "Bachelors Degree");
+ builder.field("marital_status", "M");
+ builder.field("gender", "F");
+ builder.field("management_role", "Senior Management");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 6);
+ builder.field("full_name", "Roberta Damstra");
+ builder.field("first_name", "Roberta");
+ builder.field("last_name", "Damstra");
+ builder.field("position_id", 3);
+ builder.field("position_title", "VP Information Systems");
+ builder.field("store_id", 0);
+ builder.field("department_id", 2);
+ builder.field("birth_date", "1942-10-08");
+ builder.field("hire_date", "1994-12-01 00:00:00.0");
+ builder.field("salary", 25000.0);
+ builder.field("supervisor_id", 1);
+ builder.field("education_level", "Bachelors Degree");
+ builder.field("marital_status", "M");
+ builder.field("gender", "F");
+ builder.field("management_role", "Senior Management");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 7);
+ builder.field("full_name", "Rebecca Kanagaki");
+ builder.field("first_name", "Rebecca");
+ builder.field("last_name", "Kanagaki");
+ builder.field("position_id", 4);
+ builder.field("position_title", "VP Human Resources");
+ builder.field("store_id", 0);
+ builder.field("department_id", 3);
+ builder.field("birth_date", "1949-03-27");
+ builder.field("hire_date", "1994-12-01 00:00:00.0");
+ builder.field("salary", 15000.0);
+ builder.field("supervisor_id", 1);
+ builder.field("education_level", "Bachelors Degree");
+ builder.field("marital_status", "M");
+ builder.field("gender", "F");
+ builder.field("management_role", "Senior Management");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 8);
+ builder.field("full_name", "Kim Brunner");
+ builder.field("first_name", "Kim");
+ builder.field("last_name", "Brunner");
+ builder.field("position_id", 11);
+ builder.field("position_title", "Store Manager");
+ builder.field("store_id", 9);
+ builder.field("department_id", 11);
+ builder.field("birth_date", "1922-08-10");
+ builder.field("hire_date", "1998-01-01 00:00:00.0");
+ builder.field("salary", 10000.0);
+ builder.field("supervisor_id", 5);
+ builder.field("education_level", "Bachelors Degree");
+ builder.field("marital_status", "S");
+ builder.field("gender", "F");
+ builder.field("management_role", "Store Management");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 9);
+ builder.field("full_name", "Brenda Blumberg");
+ builder.field("first_name", "Brenda");
+ builder.field("last_name", "Blumberg");
+ builder.field("position_id", 11);
+ builder.field("position_title", "Store Manager");
+ builder.field("store_id", 21);
+ builder.field("department_id", 11);
+ builder.field("birth_date", "1979-06-23");
+ builder.field("hire_date", "1998-01-01 00:00:00.0");
+ builder.field("salary", 17000.0);
+ builder.field("supervisor_id", 5);
+ builder.field("education_level", "Graduate Degree");
+ builder.field("marital_status", "M");
+ builder.field("gender", "F");
+ builder.field("management_role", "Store Management");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 10);
+ builder.field("full_name", "Darren Stanz");
+ builder.field("first_name", "Darren");
+ builder.field("last_name", "Stanz");
+ builder.field("position_id", 5);
+ builder.field("position_title", "VP Finance");
+ builder.field("store_id", 0);
+ builder.field("department_id", 5);
+ builder.field("birth_date", "1949-08-26");
+ builder.field("hire_date", "1994-12-01 00:00:00.0");
+ builder.field("salary", 50000.0);
+ builder.field("supervisor_id", 1);
+ builder.field("education_level", "Partial College");
+ builder.field("marital_status", "M");
+ builder.field("gender", "M");
+ builder.field("management_role", "Senior Management");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ builder = XContentFactory.jsonBuilder();
+ builder.startObject();
+ builder.field("employee_id", 11);
+ builder.field("full_name", "Jonathan Murraiin");
+ builder.field("first_name", "Jonathan");
+ builder.field("last_name", "Murraiin");
+ builder.field("position_id", 11);
+ builder.field("position_title", "Store Manager");
+ builder.field("store_id", 1);
+ builder.field("department_id", 11);
+ builder.field("birth_date", "1967-06-20");
+ builder.field("hire_date", "1998-01-01 00:00:00.0");
+ builder.field("salary", 15000.0);
+ builder.field("supervisor_id", 5);
+ builder.field("education_level", "Graduate Degree");
+ builder.field("marital_status", "S");
+ builder.field("gender", "M");
+ builder.field("management_role", "Store Management");
+ builder.endObject();
+ indexRequest = new IndexRequest(indexName).source(builder);
+ restHighLevelClient.index(indexRequest, RequestOptions.DEFAULT);
+
+ restHighLevelClient.indices().refresh(new RefreshRequest(indexName), RequestOptions.DEFAULT);
+ }
+
+ @Test
+ public void testSelectAll() throws Exception {
+ testBuilder()
+ .sqlQuery("select * from elastic.`employee`")
+ .unOrdered()
+ .baselineColumns("employee_id", "full_name", "first_name", "last_name", "position_id",
+ "position_title", "store_id", "department_id", "birth_date", "hire_date", "salary",
+ "supervisor_id", "education_level", "marital_status", "gender", "management_role")
+ .baselineValues(1, "Sheri Nowmer", "Sheri", "Nowmer", 1, "President", 0, 1, "1961-08-26", "1994-12-01 00:00:00.0", 80000.0, 0, "Graduate Degree", "S", "F", "Senior Management")
+ .baselineValues(2, "Derrick Whelply", "Derrick", "Whelply", 2, "VP Country Manager", 0, 1, "1915-07-03", "1994-12-01 00:00:00.0", 40000.0, 1, "Graduate Degree", "M", "M", "Senior Management")
+ .baselineValues(4, "Michael Spence", "Michael", "Spence", 2, "VP Country Manager", 0, 1, "1969-06-20", "1998-01-01 00:00:00.0", 40000.0, 1, "Graduate Degree", "S", "M", "Senior Management")
+ .baselineValues(5, "Maya Gutierrez", "Maya", "Gutierrez", 2, "VP Country Manager", 0, 1, "1951-05-10", "1998-01-01 00:00:00.0", 35000.0, 1, "Bachelors Degree", "M", "F", "Senior Management")
+ .baselineValues(6, "Roberta Damstra", "Roberta", "Damstra", 3, "VP Information Systems", 0, 2, "1942-10-08", "1994-12-01 00:00:00.0", 25000.0, 1, "Bachelors Degree", "M", "F", "Senior Management")
+ .baselineValues(7, "Rebecca Kanagaki", "Rebecca", "Kanagaki", 4, "VP Human Resources", 0, 3, "1949-03-27", "1994-12-01 00:00:00.0", 15000.0, 1, "Bachelors Degree", "M", "F", "Senior Management")
+ .baselineValues(8, "Kim Brunner", "Kim", "Brunner", 11, "Store Manager", 9, 11, "1922-08-10", "1998-01-01 00:00:00.0", 10000.0, 5, "Bachelors Degree", "S", "F", "Store Management")
+ .baselineValues(9, "Brenda Blumberg", "Brenda", "Blumberg", 11, "Store Manager", 21, 11, "1979-06-23", "1998-01-01 00:00:00.0", 17000.0, 5, "Graduate Degree", "M", "F", "Store Management")
+ .baselineValues(10, "Darren Stanz", "Darren", "Stanz", 5, "VP Finance", 0, 5, "1949-08-26", "1994-12-01 00:00:00.0", 50000.0, 1, "Partial College", "M", "M", "Senior Management")
+ .baselineValues(11, "Jonathan Murraiin", "Jonathan", "Murraiin", 11, "Store Manager", 1, 11, "1967-06-20", "1998-01-01 00:00:00.0", 15000.0, 5, "Graduate Degree", "S", "M", "Store Management")
+ .go();
+ }
+
+ @Test
+ public void testSelectColumns() throws Exception {
+ testBuilder()
+ .sqlQuery("select full_name, birth_date from elastic.`employee`")
+ .unOrdered()
+ .baselineColumns("full_name", "birth_date")
+ .baselineValues("Sheri Nowmer", "1961-08-26")
+ .baselineValues("Derrick Whelply", "1915-07-03")
+ .baselineValues("Michael Spence", "1969-06-20")
+ .baselineValues("Maya Gutierrez", "1951-05-10")
+ .baselineValues("Roberta Damstra", "1942-10-08")
+ .baselineValues("Rebecca Kanagaki", "1949-03-27")
+ .baselineValues("Kim Brunner", "1922-08-10")
+ .baselineValues("Brenda Blumberg", "1979-06-23")
+ .baselineValues("Darren Stanz", "1949-08-26")
+ .baselineValues("Jonathan Murraiin", "1967-06-20")
+ .go();
+ }
+
+ @Test
+ public void testSelectAllFiltered() throws Exception {
+ testBuilder()
+ .sqlQuery("select * from elastic.`employee` where employee_id = 1 and first_name = 'sheri'")
+ .unOrdered()
+ .baselineColumns("employee_id", "full_name", "first_name", "last_name", "position_id",
+ "position_title", "store_id", "department_id", "birth_date", "hire_date", "salary",
+ "supervisor_id", "education_level", "marital_status", "gender", "management_role")
+ .baselineValues(1, "Sheri Nowmer", "Sheri", "Nowmer", 1, "President", 0, 1, "1961-08-26",
+ "1994-12-01 00:00:00.0", 80000.0, 0, "Graduate Degree", "S", "F", "Senior Management")
+ .go();
+ }
+
+ @Test
+ public void testSelectColumnsFiltered() throws Exception {
+ testBuilder()
+ .sqlQuery("select full_name, birth_date from elastic.`employee` where employee_id = 1 and first_name = 'sheri'")
+ .unOrdered()
+ .baselineColumns("full_name", "birth_date")
+ .baselineValues("Sheri Nowmer", "1961-08-26")
+ .go();
+ }
+
+ @Test
+ public void testSelectColumnsUnsupportedFilter() throws Exception {
+ // Calcite doesn't support LIKE and other functions yet, so ensure Drill filter is used there
+ testBuilder()
+ .sqlQuery("select full_name, birth_date from elastic.`employee` where first_name like 'Sh%'")
+ .unOrdered()
+ .baselineColumns("full_name", "birth_date")
+ .baselineValues("Sheri Nowmer", "1961-08-26")
+ .go();
+ }
+
+ @Test
+ public void testSelectColumnsUnsupportedProject() throws Exception {
+ // Calcite doesn't support LIKE and other functions yet, so ensure Drill project is used there
+ testBuilder()
+ .sqlQuery("select first_name like 'Sh%' as c, birth_date from elastic.`employee` where first_name like 'Sh%'")
+ .unOrdered()
+ .baselineColumns("c", "birth_date")
+ .baselineValues(true, "1961-08-26")
+ .go();
+ }
+
+ @Test
+ public void testSelectColumnsUnsupportedAggregate() throws Exception {
+ // Calcite doesn't support stddev_samp and other functions yet, so ensure Drill agg is used there
+ testBuilder()
+ .sqlQuery("select stddev_samp(salary) as standard_deviation from elastic.`employee`")
+ .unOrdered()
+ .baselineColumns("standard_deviation")
+ .baselineValues(21333.593748410563)
+ .go();
+ }
+
+ @Test
+ public void testLimitWithSort() throws Exception {
+ testBuilder()
+ .sqlQuery("select full_name, birth_date from elastic.`employee` order by employee_id limit 3")
+ .ordered()
+ .baselineColumns("full_name", "birth_date")
+ .baselineValues("Sheri Nowmer", "1961-08-26")
+ .baselineValues("Derrick Whelply", "1915-07-03")
+ .baselineValues("Michael Spence", "1969-06-20")
+ .go();
+ }
+
+ @Test
+ public void testSelectAllWithLimitAndSort() throws Exception {
+ testBuilder()
+ .sqlQuery("select * from elastic.`employee` order by employee_id limit 3")
+ .ordered()
+ .baselineColumns("employee_id", "full_name", "first_name", "last_name", "position_id",
+ "position_title", "store_id", "department_id", "birth_date", "hire_date", "salary",
+ "supervisor_id", "education_level", "marital_status", "gender", "management_role")
+ .baselineValues(1, "Sheri Nowmer", "Sheri", "Nowmer", 1, "President", 0, 1, "1961-08-26", "1994-12-01 00:00:00.0", 80000.0, 0, "Graduate Degree", "S", "F", "Senior Management")
+ .baselineValues(2, "Derrick Whelply", "Derrick", "Whelply", 2, "VP Country Manager", 0, 1, "1915-07-03", "1994-12-01 00:00:00.0", 40000.0, 1, "Graduate Degree", "M", "M", "Senior Management")
+ .baselineValues(4, "Michael Spence", "Michael", "Spence", 2, "VP Country Manager", 0, 1, "1969-06-20", "1998-01-01 00:00:00.0", 40000.0, 1, "Graduate Degree", "S", "M", "Senior Management")
+ .go();
+ }
+
+ @Test
+ public void testSingleColumn() throws Exception {
+ testBuilder()
+ .sqlQuery("select full_name from elastic.`employee` order by employee_id limit 3")
+ .ordered()
+ .baselineColumns("full_name")
+ .baselineValues("Sheri Nowmer")
+ .baselineValues("Derrick Whelply")
+ .baselineValues("Michael Spence")
+ .go();
+ }
+
+ @Test
+ public void testJoin() throws Exception {
+ testBuilder()
+ .sqlQuery("select t1.full_name, t2.birth_date from elastic.`employee` t1 join elastic.`employee` t2 on t1.employee_id = t2.employee_id")
+ .unOrdered()
+ .baselineColumns("full_name", "birth_date")
+ .baselineValues("Sheri Nowmer", "1961-08-26")
+ .baselineValues("Derrick Whelply", "1915-07-03")
+ .baselineValues("Michael Spence", "1969-06-20")
+ .baselineValues("Maya Gutierrez", "1951-05-10")
+ .baselineValues("Roberta Damstra", "1942-10-08")
+ .baselineValues("Rebecca Kanagaki", "1949-03-27")
+ .baselineValues("Kim Brunner", "1922-08-10")
+ .baselineValues("Brenda Blumberg", "1979-06-23")
+ .baselineValues("Darren Stanz", "1949-08-26")
+ .baselineValues("Jonathan Murraiin", "1967-06-20")
+ .go();
+ }
+
+ @Test
+ public void testJoinWithFileTable() throws Exception {
+ testBuilder()
+ .sqlQuery("select t1.full_name, t2.birth_date from elastic.`employee` t1 join cp.`employee.json` t2 on t1.employee_id = t2.employee_id")
+ .unOrdered()
+ .baselineColumns("full_name", "birth_date")
+ .baselineValues("Sheri Nowmer", "1961-08-26")
+ .baselineValues("Derrick Whelply", "1915-07-03")
+ .baselineValues("Michael Spence", "1969-06-20")
+ .baselineValues("Maya Gutierrez", "1951-05-10")
+ .baselineValues("Roberta Damstra", "1942-10-08")
+ .baselineValues("Rebecca Kanagaki", "1949-03-27")
+ .baselineValues("Kim Brunner", "1922-08-10")
+ .baselineValues("Brenda Blumberg", "1979-06-23")
+ .baselineValues("Darren Stanz", "1949-08-26")
+ .baselineValues("Jonathan Murraiin", "1967-06-20")
+ .go();
+ }
+
+ @Test
+ public void testAggregate() throws Exception {
+ testBuilder()
+ .sqlQuery("select count(*) c from elastic.`employee`")
+ .ordered()
+ .baselineColumns("c")
+ .baselineValues(10L)
+ .go();
+ }
+
+ @Test
+ public void testAggregateWithGroupBy() throws Exception {
+ testBuilder()
+ .sqlQuery("select sum(`salary`) sal, department_id from elastic.`employee` e group by e.`department_id`")
+ .unOrdered()
+ .baselineColumns("sal", "department_id")
+ .baselineValues(195000.0, 1)
+ .baselineValues(42000.0, 11)
+ .baselineValues(25000.0, 2)
+ .baselineValues(15000.0, 3)
+ .baselineValues(50000.0, 5)
+ .go();
+ }
+
+ @Test
+ public void testSelectNonExistingColumn() throws Exception {
+ testBuilder()
+ .sqlQuery("select full_name123 from elastic.`employee` order by employee_id limit 3")
+ .unOrdered()
+ .baselineColumns("full_name123")
+ .baselineValuesForSingleColumn(null, null, null)
+ .go();
+ }
+
+ @Test
+ public void testSelectLiterals() throws Exception {
+ testBuilder()
+ .sqlQuery("select 'abc' as full_name, 123 as id from elastic.`employee` limit 3")
+ .unOrdered()
+ .baselineColumns("full_name", "id")
+ .baselineValues("abc", 123)
+ .baselineValues("abc", 123)
+ .baselineValues("abc", 123)
+ .go();
+ }
+
+ @Test
+ public void testSelectIntLiterals() throws Exception {
+ testBuilder()
+ .sqlQuery("select 333 as full_name, 123 as id from elastic.`employee` limit 3")
+ .unOrdered()
+ .baselineColumns("full_name", "id")
+ .baselineValues(333, 123)
+ .baselineValues(333, 123)
+ .baselineValues(333, 123)
+ .go();
+ }
+
+ @Test
+ public void testSelectLiteralWithStar() throws Exception {
+ testBuilder()
+ .sqlQuery("select *, 123 as full_name from elastic.`employee` order by employee_id limit 3")
+ .unOrdered()
+ .baselineColumns("full_name")
+ .baselineColumns("employee_id", "full_name", "first_name", "last_name", "position_id",
+ "position_title", "store_id", "department_id", "birth_date", "hire_date", "salary",
+ "supervisor_id", "education_level", "marital_status", "gender", "management_role", "full_name0")
+ .baselineValues(1, "Sheri Nowmer", "Sheri", "Nowmer", 1, "President", 0, 1, "1961-08-26", "1994-12-01 00:00:00.0", 80000.0, 0, "Graduate Degree", "S", "F", "Senior Management", 123)
+ .baselineValues(2, "Derrick Whelply", "Derrick", "Whelply", 2, "VP Country Manager", 0, 1, "1915-07-03", "1994-12-01 00:00:00.0", 40000.0, 1, "Graduate Degree", "M", "M", "Senior Management", 123)
+ .baselineValues(4, "Michael Spence", "Michael", "Spence", 2, "VP Country Manager", 0, 1, "1969-06-20", "1998-01-01 00:00:00.0", 40000.0, 1, "Graduate Degree", "S", "M", "Senior Management", 123)
+ .go();
+ }
+
+ @Test
+ public void testLiteralWithAggregateAndGroupBy() throws Exception {
+ testBuilder()
+ .sqlQuery("select sum(`salary`) sal, 1 as department_id from elastic.`employee` e group by e.`department_id`")
+ .unOrdered()
+ .baselineColumns("sal", "department_id")
+ .baselineValues(195000.0, 1)
+ .baselineValues(42000.0, 1)
+ .baselineValues(25000.0, 1)
+ .baselineValues(15000.0, 1)
+ .baselineValues(50000.0, 1)
+ .go();
+ }
+
+ @Test
+ public void testSelectNonExistingTable() throws Exception {
+ try {
+ queryBuilder().sql("select full_name from elastic.`non-existing`").run();
+ fail("Query didn't fail");
+ } catch (UserRemoteException e) {
+ assertThat(e.getMessage(), containsString("VALIDATION ERROR"));
+ assertThat(e.getMessage(), containsString("Object 'non-existing' not found within 'elastic'"));
+ }
+ }
+
+ @Test
+ public void testSelectNonExistingSubSchema() throws Exception {
+ try {
+ queryBuilder().sql("select full_name from elastic.`non-existing`.employee").run();
+ fail("Query didn't fail");
+ } catch (UserRemoteException e) {
+ assertThat(e.getMessage(), containsString("VALIDATION ERROR"));
+ assertThat(e.getMessage(), containsString("Schema [[elastic, non-existing]] is not valid with respect to either root schema or current default schema"));
+ }
+ }
+
+ @Test
+ public void testWithProvidedSchema() throws Exception {
+ testBuilder()
+ .sqlQuery("select * from " +
+ "table(elastic.`employee`(schema=>'inline=(birth_date date not null, salary decimal(10, 2))')) " +
+ "where employee_id = 1")
+ .ordered()
+ .baselineColumns("employee_id", "full_name", "first_name", "last_name", "position_id",
+ "position_title", "store_id", "department_id", "birth_date", "hire_date", "salary",
+ "supervisor_id", "education_level", "marital_status", "gender", "management_role")
+ .baselineValues(1, "Sheri Nowmer", "Sheri", "Nowmer", 1, "President", 0, 1, LocalDate.parse("1961-08-26"),
+ "1994-12-01 00:00:00.0", new BigDecimal("80000.00"), 0, "Graduate Degree", "S", "F", "Senior Management")
+ .go();
+ }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/DrillJdbcConvention.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/DrillJdbcConvention.java
index 0f416c9..401cf51 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/DrillJdbcConvention.java
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/DrillJdbcConvention.java
@@ -35,6 +35,7 @@ import org.apache.calcite.sql.SqlDialect;
import org.apache.drill.exec.planner.RuleInstance;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillRelFactories;
+import org.apache.drill.exec.store.enumerable.plan.VertexDrelConverterRule;
import org.apache.drill.shaded.guava.com.google.common.collect.ImmutableSet;
/**
@@ -60,7 +61,7 @@ class DrillJdbcConvention extends JdbcConvention {
this.rules = ImmutableSet.<RelOptRule>builder()
.addAll(calciteJdbcRules)
.add(JdbcIntermediatePrelConverterRule.INSTANCE)
- .add(new JdbcDrelConverterRule(this))
+ .add(new VertexDrelConverterRule(this))
.add(new DrillJdbcRuleBase.DrillJdbcProjectRule(Convention.NONE, this))
.add(new DrillJdbcRuleBase.DrillJdbcProjectRule(DrillRel.DRILL_LOGICAL, this))
.add(new DrillJdbcRuleBase.DrillJdbcFilterRule(Convention.NONE, this))
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcIntermediatePrelConverterRule.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcIntermediatePrelConverterRule.java
index 1fc58b4..5147cfe 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcIntermediatePrelConverterRule.java
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcIntermediatePrelConverterRule.java
@@ -24,13 +24,14 @@ import org.apache.calcite.rel.convert.ConverterRule;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillRelFactories;
import org.apache.drill.exec.planner.physical.Prel;
+import org.apache.drill.exec.store.enumerable.plan.VertexDrel;
final class JdbcIntermediatePrelConverterRule extends ConverterRule {
static final JdbcIntermediatePrelConverterRule INSTANCE = new JdbcIntermediatePrelConverterRule();
private JdbcIntermediatePrelConverterRule() {
- super(JdbcDrel.class, (Predicate<RelNode>) input -> true, DrillRel.DRILL_LOGICAL,
+ super(VertexDrel.class, (Predicate<RelNode>) input -> true, DrillRel.DRILL_LOGICAL,
Prel.DRILL_PHYSICAL, DrillRelFactories.LOGICAL_BUILDER, "JDBC_PREL_Converter");
}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcPrel.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcPrel.java
index d30feb0..815e43f 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcPrel.java
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcPrel.java
@@ -27,10 +27,8 @@ import org.apache.calcite.adapter.jdbc.JdbcImplementor;
import org.apache.calcite.plan.ConventionTraitDef;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
-import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.AbstractRelNode;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.RelShuttleImpl;
import org.apache.calcite.rel.RelWriter;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.calcite.sql.SqlDialect;
@@ -40,6 +38,7 @@ import org.apache.drill.exec.planner.physical.PhysicalPlanCreator;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
+import org.apache.drill.exec.store.SubsetRemover;
/**
* Represents a JDBC Plan once the children nodes have been rewritten into SQL.
@@ -61,7 +60,7 @@ public class JdbcPrel extends AbstractRelNode implements Prel {
dialect,
(JavaTypeFactory) getCluster().getTypeFactory());
final JdbcImplementor.Result result =
- jdbcImplementor.visitChild(0, input.accept(new SubsetRemover()));
+ jdbcImplementor.visitChild(0, input.accept(SubsetRemover.INSTANCE));
sql = result.asStatement().toSqlString(dialect).getSql();
rowType = input.getRowType();
}
@@ -78,19 +77,6 @@ public class JdbcPrel extends AbstractRelNode implements Prel {
return strippedSqlTextBldr.toString();
}
- private static class SubsetRemover extends RelShuttleImpl {
-
- @Override
- public RelNode visit(RelNode other) {
- if (other instanceof RelSubset) {
- return ((RelSubset) other).getBest().accept(this);
- } else {
- return super.visit(other);
- }
- }
-
- }
-
@Override
public PhysicalOperator getPhysicalOperator(PhysicalPlanCreator creator) {
List<SchemaPath> columns = new ArrayList<>();
diff --git a/distribution/pom.xml b/distribution/pom.xml
index 8490919..fe4b001 100644
--- a/distribution/pom.xml
+++ b/distribution/pom.xml
@@ -334,6 +334,11 @@
</dependency>
<dependency>
<groupId>org.apache.drill.contrib</groupId>
+ <artifactId>drill-storage-elasticsearch</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.drill.contrib</groupId>
<artifactId>drill-udfs</artifactId>
<version>${project.version}</version>
</dependency>
diff --git a/distribution/src/assemble/component.xml b/distribution/src/assemble/component.xml
index a7670e3..e8e7931 100644
--- a/distribution/src/assemble/component.xml
+++ b/distribution/src/assemble/component.xml
@@ -53,6 +53,7 @@
<include>org.apache.drill.contrib:drill-jdbc-storage:jar</include>
<include>org.apache.drill.contrib:drill-kudu-storage:jar</include>
<include>org.apache.drill.contrib:drill-storage-kafka:jar</include>
+ <include>org.apache.drill.contrib:drill-storage-elasticsearch:jar</include>
<include>org.apache.drill.contrib:drill-storage-http:jar</include>
<include>org.apache.drill.contrib:drill-opentsdb-storage:jar</include>
<include>org.apache.drill.contrib:drill-udfs:jar</include>
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java
index 2c580da..fc78f51 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/compile/ClassBuilder.java
@@ -23,7 +23,6 @@ import java.io.IOException;
import java.util.Map;
import org.apache.drill.common.config.DrillConfig;
-import org.apache.drill.common.util.DrillStringUtils;
import org.apache.drill.exec.compile.ClassTransformer.ClassNames;
import org.apache.drill.exec.exception.ClassTransformationException;
import org.apache.drill.exec.expr.CodeGenerator;
@@ -144,21 +143,22 @@ public class ClassBuilder {
saveCode(code, name);
}
+ Class<?> compiledClass = getCompiledClass(code, className, config, options);
+ logger.debug("Compiled {}: time = {} ms.",
+ className,
+ (System.nanoTime() - t1 + 500_000) / 1_000_000);
+ return compiledClass;
+ }
+
+ public static Class<?> getCompiledClass(String code, String className,
+ DrillConfig config, OptionSet options) throws CompileException, ClassNotFoundException, ClassTransformationException, IOException {
// Compile the code and load it into a class loader.
CachedClassLoader classLoader = new CachedClassLoader();
ClassCompilerSelector compilerSelector = new ClassCompilerSelector(classLoader, config, options);
+ ClassNames name = new ClassNames(className);
Map<String,byte[]> results = compilerSelector.compile(name, code);
classLoader.addClasses(results);
- long totalBytecodeSize = 0;
- for (byte[] clazz : results.values()) {
- totalBytecodeSize += clazz.length;
- }
- logger.debug("Compiled {}: bytecode size = {}, time = {} ms.",
- cg.getClassName(),
- DrillStringUtils.readable(totalBytecodeSize),
- (System.nanoTime() - t1 + 500_000) / 1_000_000);
-
// Get the class from the class loader.
try {
return classLoader.findClass(className);
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertDoubleToDecimal.java
similarity index 51%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
copy to exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertDoubleToDecimal.java
index 7107e9f..6d4289f 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertDoubleToDecimal.java
@@ -15,25 +15,29 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
+package org.apache.drill.exec.physical.impl.scan.convert;
-import java.util.function.Predicate;
+import org.apache.drill.exec.vector.accessor.ScalarWriter;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.convert.ConverterRule;
-import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRelFactories;
+import java.math.BigDecimal;
-class JdbcDrelConverterRule extends ConverterRule {
+public class ConvertDoubleToDecimal extends DirectConverter {
- JdbcDrelConverterRule(DrillJdbcConvention in) {
- super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
+ public ConvertDoubleToDecimal(ScalarWriter baseWriter) {
+ super(baseWriter);
}
@Override
- public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
- convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
+ public void setDouble(double value) {
+ baseWriter.setDecimal(BigDecimal.valueOf(value));
+ }
+
+ @Override
+ public void setValue(Object value) {
+ if (value == null) {
+ setNull();
+ } else {
+ setDouble((double) value);
+ }
}
}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertFloatToDecimal.java
similarity index 51%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
copy to exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertFloatToDecimal.java
index 7107e9f..411bb65 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertFloatToDecimal.java
@@ -15,25 +15,29 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
+package org.apache.drill.exec.physical.impl.scan.convert;
-import java.util.function.Predicate;
+import org.apache.drill.exec.vector.accessor.ScalarWriter;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.convert.ConverterRule;
-import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRelFactories;
+import java.math.BigDecimal;
-class JdbcDrelConverterRule extends ConverterRule {
+public class ConvertFloatToDecimal extends DirectConverter {
- JdbcDrelConverterRule(DrillJdbcConvention in) {
- super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
+ public ConvertFloatToDecimal(ScalarWriter baseWriter) {
+ super(baseWriter);
}
@Override
- public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
- convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
+ public void setFloat(float value) {
+ baseWriter.setDecimal(BigDecimal.valueOf(value));
+ }
+
+ @Override
+ public void setValue(Object value) {
+ if (value == null) {
+ setNull();
+ } else {
+ setDouble((float) value);
+ }
}
}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertIntToDecimal.java
similarity index 51%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
copy to exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertIntToDecimal.java
index 7107e9f..c90748c 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertIntToDecimal.java
@@ -15,25 +15,29 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
+package org.apache.drill.exec.physical.impl.scan.convert;
-import java.util.function.Predicate;
+import org.apache.drill.exec.vector.accessor.ScalarWriter;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.convert.ConverterRule;
-import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRelFactories;
+import java.math.BigDecimal;
-class JdbcDrelConverterRule extends ConverterRule {
+public class ConvertIntToDecimal extends DirectConverter {
- JdbcDrelConverterRule(DrillJdbcConvention in) {
- super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
+ public ConvertIntToDecimal(ScalarWriter baseWriter) {
+ super(baseWriter);
}
@Override
- public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
- convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
+ public void setInt(int value) {
+ baseWriter.setDecimal(BigDecimal.valueOf(value));
+ }
+
+ @Override
+ public void setValue(Object value) {
+ if (value == null) {
+ setNull();
+ } else {
+ setDouble((int) value);
+ }
}
}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertLongToDecimal.java
similarity index 51%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
copy to exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertLongToDecimal.java
index 7107e9f..92650ff 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/ConvertLongToDecimal.java
@@ -15,25 +15,29 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
+package org.apache.drill.exec.physical.impl.scan.convert;
-import java.util.function.Predicate;
+import org.apache.drill.exec.vector.accessor.ScalarWriter;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.convert.ConverterRule;
-import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRelFactories;
+import java.math.BigDecimal;
-class JdbcDrelConverterRule extends ConverterRule {
+public class ConvertLongToDecimal extends DirectConverter {
- JdbcDrelConverterRule(DrillJdbcConvention in) {
- super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
+ public ConvertLongToDecimal(ScalarWriter baseWriter) {
+ super(baseWriter);
}
@Override
- public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
- convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
+ public void setLong(long value) {
+ baseWriter.setDecimal(BigDecimal.valueOf(value));
+ }
+
+ @Override
+ public void setValue(Object value) {
+ if (value == null) {
+ setNull();
+ } else {
+ setDouble((long) value);
+ }
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/DirectConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/DirectConverter.java
index 5222926..344184c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/DirectConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/DirectConverter.java
@@ -64,6 +64,11 @@ public class DirectConverter extends ColumnConverter implements ValueWriter {
}
@Override
+ public void setFloat(float value) {
+ baseWriter.setDouble(value);
+ }
+
+ @Override
public void setDouble(double value) {
baseWriter.setDouble(value);
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/StandardConversions.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/StandardConversions.java
index 396e741..8deb0f1 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/StandardConversions.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/impl/scan/convert/StandardConversions.java
@@ -203,9 +203,9 @@ public class StandardConversions {
* <p>
* Does not support any of the "legacy" decimal types.
*
- * @param inputDefn the column schema for the input column which the
+ * @param inputSchema the column schema for the input column which the
* client code (e.g. reader) wants to produce
- * @param outputDefn the column schema for the output vector to be produced
+ * @param outputSchema the column schema for the output vector to be produced
* by this operator
* @return a description of the conversion needed (if any), along with the
* standard conversion class, if available
@@ -241,8 +241,9 @@ public class StandardConversions {
case BIGINT:
case FLOAT4:
case FLOAT8:
- case VARDECIMAL:
return IMPLICIT;
+ case VARDECIMAL:
+ return new ConversionDefn(ConvertIntToDecimal.class);
case VARCHAR:
return new ConversionDefn(ConvertIntToString.class);
default:
@@ -257,8 +258,9 @@ public class StandardConversions {
case BIGINT:
case FLOAT4:
case FLOAT8:
- case VARDECIMAL:
return IMPLICIT;
+ case VARDECIMAL:
+ return new ConversionDefn(ConvertIntToDecimal.class);
case VARCHAR:
return new ConversionDefn(ConvertIntToString.class);
default:
@@ -273,9 +275,10 @@ public class StandardConversions {
case BIGINT:
case FLOAT4:
case FLOAT8:
- case VARDECIMAL:
case TIME:
return IMPLICIT;
+ case VARDECIMAL:
+ return new ConversionDefn(ConvertIntToDecimal.class);
case VARCHAR:
return new ConversionDefn(ConvertIntToString.class);
default:
@@ -290,10 +293,11 @@ public class StandardConversions {
return IMPLICIT_UNSAFE;
case FLOAT4:
case FLOAT8:
- case VARDECIMAL:
case DATE:
case TIMESTAMP:
return IMPLICIT;
+ case VARDECIMAL:
+ return new ConversionDefn(ConvertLongToDecimal.class);
case VARCHAR:
return new ConversionDefn(ConvertLongToString.class);
default:
@@ -308,8 +312,9 @@ public class StandardConversions {
case BIGINT:
return IMPLICIT_UNSAFE;
case FLOAT8:
- case VARDECIMAL:
return IMPLICIT;
+ case VARDECIMAL:
+ return new ConversionDefn(ConvertFloatToDecimal.class);
case VARCHAR:
return new ConversionDefn(ConvertDoubleToString.class);
default:
@@ -325,7 +330,7 @@ public class StandardConversions {
case FLOAT4:
return IMPLICIT_UNSAFE;
case VARDECIMAL:
- return IMPLICIT;
+ return new ConversionDefn(ConvertDoubleToDecimal.class);
case VARCHAR:
return new ConversionDefn(ConvertDoubleToString.class);
default:
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/resultSet/impl/ProjectionFilter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/resultSet/impl/ProjectionFilter.java
index 0d97429..9bcbe5b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/physical/resultSet/impl/ProjectionFilter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/physical/resultSet/impl/ProjectionFilter.java
@@ -44,7 +44,7 @@ public interface ProjectionFilter {
ProjResult NOT_PROJECTED = new ProjResult(false, null, PROJECT_NONE);
ProjResult PROJECTED = new ProjResult(true, null, PROJECT_ALL);
- public static class ProjResult {
+ class ProjResult {
public final boolean isProjected;
public final ColumnMetadata projection;
public final ProjectionFilter mapFilter;
@@ -70,7 +70,7 @@ public interface ProjectionFilter {
boolean isEmpty();
- public static ProjectionFilter projectionFilter(RequestedTuple tupleProj,
+ static ProjectionFilter projectionFilter(RequestedTuple tupleProj,
CustomErrorContext errorContext) {
switch (tupleProj.type()) {
case ALL:
@@ -82,7 +82,7 @@ public interface ProjectionFilter {
}
}
- public static ProjectionFilter providedSchemaFilter(RequestedTuple tupleProj,
+ static ProjectionFilter providedSchemaFilter(RequestedTuple tupleProj,
TupleMetadata providedSchema, CustomErrorContext errorContext) {
if (tupleProj.type() == TupleProjectionType.NONE) {
return PROJECT_NONE;
@@ -106,7 +106,7 @@ public interface ProjectionFilter {
schemaFilter);
}
- public static ProjectionFilter definedSchemaFilter(
+ static ProjectionFilter definedSchemaFilter(
TupleMetadata definedSchema, CustomErrorContext errorContext) {
if (definedSchema.isEmpty()) {
return PROJECT_NONE;
@@ -120,7 +120,7 @@ public interface ProjectionFilter {
* projects special columns (those marked as not being expanded in
* SELECT *).
*/
- public static class ImplicitProjectionFilter implements ProjectionFilter {
+ class ImplicitProjectionFilter implements ProjectionFilter {
private final boolean projectAll;
public ImplicitProjectionFilter(boolean projectAll) {
@@ -152,7 +152,7 @@ public interface ProjectionFilter {
* the reader column is consistent with the form of projection (map,
* array, or plain) in the projection list.
*/
- public static class DirectProjectionFilter implements ProjectionFilter {
+ class DirectProjectionFilter implements ProjectionFilter {
private final RequestedTuple projectionSet;
private final CustomErrorContext errorContext;
@@ -185,7 +185,7 @@ public interface ProjectionFilter {
/**
* Schema-based projection.
*/
- public abstract static class BaseSchemaProjectionFilter implements ProjectionFilter {
+ abstract class BaseSchemaProjectionFilter implements ProjectionFilter {
protected final TupleMetadata schema;
protected final CustomErrorContext errorContext;
@@ -233,7 +233,7 @@ public interface ProjectionFilter {
* If the column is found, enforces that the reader schema has the same type and
* mode as the provided column.
*/
- public static class TypeProjectionFilter extends BaseSchemaProjectionFilter {
+ class TypeProjectionFilter extends BaseSchemaProjectionFilter {
public TypeProjectionFilter(TupleMetadata providedSchema, CustomErrorContext errorContext) {
super(providedSchema, errorContext);
@@ -265,7 +265,7 @@ public interface ProjectionFilter {
* Projection filter in which a schema exactly defines the set of allowed
* columns, and their types.
*/
- public static class SchemaProjectionFilter extends BaseSchemaProjectionFilter {
+ class SchemaProjectionFilter extends BaseSchemaProjectionFilter {
public SchemaProjectionFilter(TupleMetadata definedSchema, CustomErrorContext errorContext) {
super(definedSchema, errorContext);
@@ -296,7 +296,7 @@ public interface ProjectionFilter {
/**
* Compound filter for combining direct and provided schema projections.
*/
- public static class CompoundProjectionFilter implements ProjectionFilter {
+ class CompoundProjectionFilter implements ProjectionFilter {
private final ProjectionFilter filter1;
private final ProjectionFilter filter2;
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 3cb2077..af33675 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
@@ -284,7 +284,7 @@ public abstract class DrillRelOptUtil {
public static boolean isProjectOutputRowcountUnknown(Project project) {
for (RexNode rex : project.getProjects()) {
if (rex instanceof RexCall) {
- if ("flatten".equals(((RexCall) rex).getOperator().getName().toLowerCase())) {
+ if ("flatten".equalsIgnoreCase(((RexCall) rex).getOperator().getName())) {
return true;
}
}
@@ -304,7 +304,7 @@ public abstract class DrillRelOptUtil {
new RexVisitorImpl<Void>(true) {
@Override
public Void visitCall(RexCall call) {
- if ("convert_fromjson".equals(call.getOperator().getName().toLowerCase())) {
+ if ("convert_fromjson".equalsIgnoreCase(call.getOperator().getName())) {
throw new Util.FoundOne(call); /* throw exception to interrupt tree walk (this is similar to
other utility methods in RexUtil.java */
}
@@ -321,6 +321,17 @@ public abstract class DrillRelOptUtil {
return false;
}
+ public static TableScan findScan(RelNode... rels) {
+ for (RelNode rel : rels) {
+ if (rel instanceof TableScan) {
+ return (TableScan) rel;
+ } else {
+ return findScan(rel.getInputs().toArray(new RelNode[0]));
+ }
+ }
+ return null;
+ }
+
/**
* InputRefVisitor is a utility class used to collect all the RexInputRef nodes in a
* RexNode.
@@ -677,8 +688,7 @@ public abstract class DrillRelOptUtil {
}
public static DrillTable getDrillTable(final TableScan scan) {
- DrillTable drillTable = null;
- drillTable = scan.getTable().unwrap(DrillTable.class);
+ DrillTable drillTable = scan.getTable().unwrap(DrillTable.class);
if (drillTable == null) {
DrillTranslatableTable transTable = scan.getTable().unwrap(DrillTranslatableTable.class);
if (transTable != null) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjectIntoScanRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjectIntoScanRule.java
index 8b717c2..91875bb 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjectIntoScanRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillPushProjectIntoScanRule.java
@@ -28,10 +28,8 @@ import org.apache.calcite.rel.rules.ProjectRemoveRule;
import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexNode;
import org.apache.drill.common.exceptions.DrillRuntimeException;
-import org.apache.drill.exec.planner.common.DrillProjectRelBase;
import org.apache.drill.exec.planner.common.DrillRelOptUtil;
import org.apache.drill.exec.planner.common.DrillRelOptUtil.ProjectPushInfo;
-import org.apache.drill.exec.planner.common.DrillScanRelBase;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.planner.physical.ProjectPrel;
import org.apache.drill.exec.planner.physical.ScanPrel;
@@ -89,7 +87,7 @@ public class DrillPushProjectIntoScanRule extends RelOptRule {
}
};
- private DrillPushProjectIntoScanRule(Class<? extends Project> projectClass, Class<? extends TableScan> scanClass, String description) {
+ protected DrillPushProjectIntoScanRule(Class<? extends Project> projectClass, Class<? extends TableScan> scanClass, String description) {
super(RelOptHelper.some(projectClass, RelOptHelper.any(scanClass)), description);
}
@@ -110,14 +108,14 @@ public class DrillPushProjectIntoScanRule extends RelOptRule {
return;
}
- DrillScanRelBase newScan = createScan(scan, projectPushInfo);
+ TableScan newScan = createScan(scan, projectPushInfo);
List<RexNode> newProjects = new ArrayList<>();
for (RexNode n : project.getChildExps()) {
newProjects.add(n.accept(projectPushInfo.getInputReWriter()));
}
- DrillProjectRelBase newProject =
+ Project newProject =
createProject(project, newScan, newProjects);
if (ProjectRemoveRule.isTrivial(newProject)) {
@@ -151,7 +149,7 @@ public class DrillPushProjectIntoScanRule extends RelOptRule {
* @param newProjects new project expressions
* @return new project instance
*/
- protected DrillProjectRelBase createProject(Project project, TableScan newScan, List<RexNode> newProjects) {
+ protected Project createProject(Project project, TableScan newScan, List<RexNode> newProjects) {
return new DrillProjectRel(project.getCluster(),
project.getTraitSet().plus(DrillRel.DRILL_LOGICAL),
newScan,
@@ -168,7 +166,7 @@ public class DrillPushProjectIntoScanRule extends RelOptRule {
* @param projectPushInfo the source of row type and fields list
* @return new scan instance
*/
- protected DrillScanRelBase createScan(TableScan scan, ProjectPushInfo projectPushInfo) {
+ protected TableScan createScan(TableScan scan, ProjectPushInfo projectPushInfo) {
return new DrillScanRel(scan.getCluster(),
scan.getTraitSet().plus(DrillRel.DRILL_LOGICAL),
scan.getTable(),
@@ -184,7 +182,7 @@ public class DrillPushProjectIntoScanRule extends RelOptRule {
* @param projectPushInfo fields information
* @return true if push project into scan can be performed, false otherwise
*/
- private boolean canPushProjectIntoScan(RelOptTable table, ProjectPushInfo projectPushInfo) throws IOException {
+ protected boolean canPushProjectIntoScan(RelOptTable table, ProjectPushInfo projectPushInfo) throws IOException {
DrillTable drillTable = Utilities.getDrillTable(table);
return !Utilities.isStarQuery(projectPushInfo.getFields())
&& drillTable.getGroupScan().canPushdownProjects(projectPushInfo.getFields());
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillTable.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillTable.java
index 104fb6a..7a9dc0c 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillTable.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/logical/DrillTable.java
@@ -140,10 +140,6 @@ public abstract class DrillTable implements Table {
return metadataProviderManager;
}
- public TableMetadataProvider getMetadataProvider() throws IOException {
- return getGroupScan().getMetadataProvider();
- }
-
public StoragePluginConfig getStorageEngineConfig() {
return storageEngineConfig;
}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/LeafPrel.java
similarity index 51%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
copy to exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/LeafPrel.java
index 7107e9f..55a7955 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/LeafPrel.java
@@ -15,25 +15,23 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
+package org.apache.drill.exec.planner.physical;
-import java.util.function.Predicate;
+import java.util.Collections;
+import java.util.Iterator;
-import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.convert.ConverterRule;
-import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRelFactories;
-
-class JdbcDrelConverterRule extends ConverterRule {
+/**
+ * Prel without children.
+ */
+public interface LeafPrel extends Prel {
- JdbcDrelConverterRule(DrillJdbcConvention in) {
- super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
+ @Override
+ default boolean needsFinalColumnReordering() {
+ return true;
}
@Override
- public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
- convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
+ default Iterator<Prel> iterator() {
+ return Collections.emptyIterator();
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
index 69cd603..82fdc4b 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/ScanPrel.java
@@ -19,7 +19,6 @@ package org.apache.drill.exec.planner.physical;
import java.io.IOException;
import java.util.Collections;
-import java.util.Iterator;
import java.util.List;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelOptCost;
@@ -43,7 +42,7 @@ import org.apache.drill.exec.record.BatchSchema.SelectionVectorMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-public class ScanPrel extends DrillScanRelBase implements Prel, HasDistributionAffinity {
+public class ScanPrel extends DrillScanRelBase implements LeafPrel, HasDistributionAffinity {
private static final Logger logger = LoggerFactory.getLogger(ScanPrel.class);
private final RelDataType rowType;
@@ -137,11 +136,6 @@ public class ScanPrel extends DrillScanRelBase implements Prel, HasDistributionA
}
@Override
- public Iterator<Prel> iterator() {
- return Collections.emptyIterator();
- }
-
- @Override
public <T, X, E extends Throwable> T accept(PrelVisitor<T, X, E> logicalVisitor, X value) throws E {
return logicalVisitor.visitScan(this, value);
}
@@ -157,11 +151,6 @@ public class ScanPrel extends DrillScanRelBase implements Prel, HasDistributionA
}
@Override
- public boolean needsFinalColumnReordering() {
- return true;
- }
-
- @Override
public DistributionAffinity getDistributionAffinity() {
return this.getGroupScan().getDistributionAffinity();
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/UnnestPrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/UnnestPrel.java
index a344915..456f966 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/UnnestPrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/UnnestPrel.java
@@ -36,11 +36,9 @@ import org.apache.drill.exec.record.BatchSchema;
import java.io.IOException;
import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Iterator;
import java.util.List;
-public class UnnestPrel extends DrillUnnestRelBase implements Prel {
+public class UnnestPrel extends DrillUnnestRelBase implements LeafPrel {
protected final UnnestPOP unnestPOP;
@@ -52,11 +50,6 @@ public class UnnestPrel extends DrillUnnestRelBase implements Prel {
}
@Override
- public Iterator<Prel> iterator() {
- return Collections.emptyIterator();
- }
-
- @Override
public <T, X, E extends Throwable> T accept(PrelVisitor<T, X, E> visitor, X value) throws E {
return visitor.visitUnnest(this, value);
}
@@ -77,11 +70,6 @@ public class UnnestPrel extends DrillUnnestRelBase implements Prel {
return BatchSchema.SelectionVectorMode.NONE;
}
- @Override
- public boolean needsFinalColumnReordering() {
- return true;
- }
-
public Class<?> getParentClass() {
return LateralJoinPrel.class;
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/BasePrelVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/BasePrelVisitor.java
index b260080..dd11d03 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/BasePrelVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/BasePrelVisitor.java
@@ -20,6 +20,7 @@ package org.apache.drill.exec.planner.physical.visitor;
import org.apache.drill.exec.planner.physical.DirectScanPrel;
import org.apache.drill.exec.planner.physical.ExchangePrel;
import org.apache.drill.exec.planner.physical.JoinPrel;
+import org.apache.drill.exec.planner.physical.LeafPrel;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.planner.physical.ProjectPrel;
import org.apache.drill.exec.planner.physical.ScanPrel;
@@ -81,4 +82,8 @@ public class BasePrelVisitor<RETURN, EXTRA, EXCEP extends Throwable> implements
return visitPrel(prel, value);
}
+ @Override
+ public RETURN visitLeaf(LeafPrel prel, EXTRA value) throws EXCEP {
+ return visitPrel(prel, value);
+ }
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisitor.java
index c138c83..a97c10f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisitor.java
@@ -20,6 +20,7 @@ package org.apache.drill.exec.planner.physical.visitor;
import org.apache.drill.exec.planner.physical.DirectScanPrel;
import org.apache.drill.exec.planner.physical.ExchangePrel;
import org.apache.drill.exec.planner.physical.JoinPrel;
+import org.apache.drill.exec.planner.physical.LeafPrel;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.planner.physical.ProjectPrel;
import org.apache.drill.exec.planner.physical.ScanPrel;
@@ -42,4 +43,5 @@ public interface PrelVisitor<RETURN, EXTRA, EXCEP extends Throwable> {
RETURN visitPrel(Prel prel, EXTRA value) throws EXCEP;
RETURN visitUnnest(UnnestPrel prel, EXTRA value) throws EXCEP;
RETURN visitLateral(LateralJoinPrel prel, EXTRA value) throws EXCEP;
+ RETURN visitLeaf(LeafPrel prel, EXTRA value) throws EXCEP;
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisualizerVisitor.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisualizerVisitor.java
index f02ad6f..3dfc030 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisualizerVisitor.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/PrelVisualizerVisitor.java
@@ -20,6 +20,7 @@ package org.apache.drill.exec.planner.physical.visitor;
import org.apache.drill.exec.planner.physical.DirectScanPrel;
import org.apache.drill.exec.planner.physical.ExchangePrel;
import org.apache.drill.exec.planner.physical.JoinPrel;
+import org.apache.drill.exec.planner.physical.LeafPrel;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.planner.physical.ProjectPrel;
import org.apache.drill.exec.planner.physical.ScanPrel;
@@ -246,4 +247,10 @@ public class PrelVisualizerVisitor
visitPrel(prel, value);
return null;
}
+
+ @Override
+ public Void visitLeaf(LeafPrel prel, VisualizationState value) throws Exception {
+ visitPrel(prel, value);
+ return null;
+ }
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/StarColumnConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/StarColumnConverter.java
index e4b8800..8d91628 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/StarColumnConverter.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/physical/visitor/StarColumnConverter.java
@@ -24,6 +24,7 @@ import java.util.concurrent.atomic.AtomicLong;
import org.apache.calcite.rel.rules.ProjectRemoveRule;
import org.apache.drill.exec.planner.StarColumnHelper;
+import org.apache.drill.exec.planner.physical.LeafPrel;
import org.apache.drill.exec.planner.physical.MetadataControllerPrel;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.planner.physical.ProjectAllowDupPrel;
@@ -240,12 +241,17 @@ public class StarColumnConverter extends BasePrelVisitor<Prel, Void, RuntimeExce
@Override
public Prel visitScan(ScanPrel scanPrel, Void value) throws RuntimeException {
- return prefixTabNameToStar(scanPrel, value);
+ return visitLeaf(scanPrel, value);
+ }
+
+ @Override
+ public Prel visitLeaf(LeafPrel prel, Void value) throws RuntimeException {
+ return prefixTabNameToStar(prel, value);
}
@Override
public Prel visitUnnest(UnnestPrel unnestPrel, Void value) throws RuntimeException {
- return prefixTabNameToStar(unnestPrel, value);
+ return visitLeaf(unnestPrel, value);
}
private List<String> makeUniqueNames(List<String> names) {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SchemaUtilites.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SchemaUtilites.java
index ed0dda3..5e53fd9 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SchemaUtilites.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/SchemaUtilites.java
@@ -92,7 +92,7 @@ public class SchemaUtilites {
}
/** Utility method to search for schema path starting from the given <i>schema</i> reference */
- private static SchemaPlus searchSchemaTree(SchemaPlus schema, final List<String> schemaPath) {
+ public static SchemaPlus searchSchemaTree(SchemaPlus schema, final List<String> schemaPath) {
for (String schemaName : schemaPath) {
// schemas in Drill are case insensitive and stored in lower case
schema = schema.getSubSchema(schemaName.toLowerCase());
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/AnalyzeTableHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/AnalyzeTableHandler.java
index ee08d07..c2b7e0f 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/AnalyzeTableHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/AnalyzeTableHandler.java
@@ -37,7 +37,6 @@ import org.apache.drill.exec.physical.base.PhysicalOperator;
import org.apache.drill.exec.planner.common.DrillStatsTable;
import org.apache.drill.exec.planner.logical.DrillAnalyzeRel;
import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillScanRel;
import org.apache.drill.exec.planner.logical.DrillScreenRel;
import org.apache.drill.exec.planner.logical.DrillTable;
import org.apache.drill.exec.planner.logical.DrillWriterRel;
@@ -210,16 +209,6 @@ public class AnalyzeTableHandler extends DefaultSqlHandler {
return new DrillScreenRel(writerRel.getCluster(), writerRel.getTraitSet(), writerRel);
}
- public static DrillScanRel findScan(RelNode... rels) {
- for (RelNode rel : rels) {
- if (rel instanceof DrillScanRel) {
- return (DrillScanRel) rel;
- } else {
- return findScan(rel.getInputs().toArray(new RelNode[0]));
- }
- }
- return null;
- }
// Make sure no unsupported features in ANALYZE statement are used
private static void verifyNoUnsupportedFunctions(final SqlAnalyzeTable analyzeTable) {
// throw unsupported error for functions that are not yet implemented
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/MetastoreAnalyzeTableHandler.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/MetastoreAnalyzeTableHandler.java
index 23a6b1b..e8378e6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/MetastoreAnalyzeTableHandler.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/MetastoreAnalyzeTableHandler.java
@@ -45,6 +45,7 @@ import org.apache.drill.exec.metastore.analyze.MetadataInfoCollector;
import org.apache.drill.exec.metastore.analyze.MetastoreAnalyzeConstants;
import org.apache.drill.exec.physical.PhysicalPlan;
import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.common.DrillRelOptUtil;
import org.apache.drill.exec.planner.logical.DrillAnalyzeRel;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillScreenRel;
@@ -237,7 +238,7 @@ public class MetastoreAnalyzeTableHandler extends DefaultSqlHandler {
if (metastoreTableInfo.isExists()) {
RelNode finalRelNode = relNode;
CheckedSupplier<TableScan, SqlUnsupportedException> tableScanSupplier =
- () -> AnalyzeTableHandler.findScan(convertToDrel(finalRelNode.getInput(0)));
+ () -> DrillRelOptUtil.findScan(convertToDrel(finalRelNode.getInput(0)));
MetadataInfoCollector metadataInfoCollector = analyzeInfoProvider.getMetadataInfoCollector(basicRequests,
tableInfo, (FormatSelection) table.getSelection(), context.getPlannerSettings(),
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/ColumnConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/ColumnConverter.java
new file mode 100644
index 0000000..3688972
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/ColumnConverter.java
@@ -0,0 +1,277 @@
+/*
+ * 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.record;
+
+import java.math.BigDecimal;
+import java.sql.Date;
+import java.sql.Time;
+import java.sql.Timestamp;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.function.Consumer;
+
+import org.apache.drill.common.types.TypeProtos.DataMode;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.exec.record.metadata.ColumnMetadata;
+import org.apache.drill.exec.record.metadata.MapColumnMetadata;
+import org.apache.drill.exec.record.metadata.MetadataUtils;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.record.metadata.TupleSchema;
+import org.apache.drill.exec.vector.accessor.ArrayWriter;
+import org.apache.drill.exec.vector.accessor.DictWriter;
+import org.apache.drill.exec.vector.accessor.TupleWriter;
+
+/**
+ * Converts and sets given value into the specific column writer.
+ */
+public interface ColumnConverter {
+
+ void convert(Object value);
+
+ /**
+ * Does nothing, is used when column is not projected to avoid unnecessary
+ * column values conversions and writes.
+ */
+ class DummyColumnConverter implements ColumnConverter {
+
+ public static final DummyColumnConverter INSTANCE = new DummyColumnConverter();
+
+ @Override
+ public void convert(Object value) {
+ // do nothing
+ }
+ }
+
+ /**
+ * Converts and writes scalar values using provided {@link #valueConverter}.
+ * {@link #valueConverter} has different implementation depending
+ * on the scalar value type.
+ */
+ class ScalarColumnConverter implements ColumnConverter {
+
+ private final Consumer<Object> valueConverter;
+
+ public ScalarColumnConverter(Consumer<Object> valueConverter) {
+ this.valueConverter = valueConverter;
+ }
+
+ @Override
+ public void convert(Object value) {
+ if (value == null) {
+ return;
+ }
+
+ valueConverter.accept(value);
+ }
+ }
+
+ /**
+ * Converts and writes array values using {@link #valueConverter}
+ * into {@link #arrayWriter}.
+ */
+ class ArrayColumnConverter implements ColumnConverter {
+
+ private final ArrayWriter arrayWriter;
+ private final ColumnConverter valueConverter;
+
+ public ArrayColumnConverter(ArrayWriter arrayWriter, ColumnConverter valueConverter) {
+ this.arrayWriter = arrayWriter;
+ this.valueConverter = valueConverter;
+ }
+
+ @Override
+ public void convert(Object value) {
+ if (value == null || !arrayWriter.isProjected()) {
+ return;
+ }
+
+ Iterable<?> array = (Iterable<?>) value;
+ array.forEach(arrayValue -> {
+ valueConverter.convert(arrayValue);
+ arrayWriter.save();
+ });
+ }
+ }
+
+ /**
+ * Converts and writes all map children using provided {@link #converters}.
+ * If {@link #converters} are empty, generates their converters based on
+ * specified schema.
+ */
+ class MapColumnConverter implements ColumnConverter {
+
+ private final ColumnConverterFactory factory;
+ private final TupleMetadata providedSchema;
+ private final TupleWriter tupleWriter;
+ private final Map<String, ColumnConverter> converters;
+
+ public MapColumnConverter(ColumnConverterFactory factory,
+ TupleMetadata providedSchema,
+ TupleWriter tupleWriter, Map<String, ColumnConverter> converters) {
+ this.factory = factory;
+ this.providedSchema = providedSchema;
+ this.tupleWriter = tupleWriter;
+ this.converters = new HashMap<>(converters);
+ }
+
+ @Override
+ @SuppressWarnings("unchecked")
+ public void convert(Object value) {
+ if (value == null) {
+ return;
+ }
+
+ Map<String, Object> record = (Map<String, Object>) value;
+
+ record.forEach(this::processValue);
+ }
+
+ private void processValue(String name, Object columnValue) {
+ ColumnConverter columnConverter = converters.computeIfAbsent(name,
+ columnName -> getColumnConverter(columnValue, columnName));
+ if (columnConverter != null) {
+ columnConverter.convert(columnValue);
+ }
+ }
+
+ private ColumnConverter getColumnConverter(Object columnValue, String columnName) {
+ if (columnValue != null) {
+ ColumnMetadata providedColumn = providedSchema != null
+ ? providedSchema.metadata(columnName)
+ : null;
+ ColumnMetadata column = buildColumnMetadata(columnName, columnValue);
+ if (column != null) {
+ tupleWriter.addColumn(providedColumn != null ? providedColumn : column);
+ return factory.getConverter(providedSchema, column, tupleWriter.column(columnName));
+ }
+ }
+ return null;
+ }
+
+ @SuppressWarnings("unchecked")
+ private ColumnMetadata buildColumnMetadata(String name, Object value) {
+ Class<?> clazz = value.getClass();
+ if (Map.class.isAssignableFrom(clazz)) {
+ return buildMapColumnMetadata(name, (Map<String, Object>) value);
+ } else if (List.class.isAssignableFrom(clazz)) {
+ List<?> list = (List<?>) value;
+ if (!list.isEmpty()) {
+ Object innerValue = list.iterator().next();
+ return buildListColumnMetadata(name, innerValue);
+ } else {
+ return null;
+ }
+ } else if (clazz.isArray()) {
+ Object[] array = (Object[]) value;
+ if (array.length > 0) {
+ return buildListColumnMetadata(name, array[0]);
+ } else {
+ return null;
+ }
+ }
+ return MetadataUtils.newScalar(name, getScalarMinorType(clazz), DataMode.OPTIONAL);
+ }
+
+ private MapColumnMetadata buildMapColumnMetadata(String name, Map<String, Object> map) {
+ TupleMetadata schema = new TupleSchema();
+ map.forEach((key, value) -> {
+ if (value != null) {
+ schema.addColumn(buildColumnMetadata(key, value));
+ }
+ });
+
+ return MetadataUtils.newMap(name, schema);
+ }
+
+ private ColumnMetadata buildListColumnMetadata(String name, Object innerValue) {
+ Class<?> clazz = innerValue.getClass();
+ Class<?> componentType = clazz.getComponentType();
+ if (List.class.isAssignableFrom(clazz) || componentType!= null && componentType.isArray()) {
+ return MetadataUtils.newRepeatedList(name, buildColumnMetadata(name, innerValue));
+ } else if (Map.class.isAssignableFrom(clazz)) {
+ return MetadataUtils.newMapArray(name);
+ } else {
+ return MetadataUtils.newScalar(name, getScalarMinorType(clazz), DataMode.REPEATED);
+ }
+ }
+
+ private MinorType getScalarMinorType(Class<?> clazz) {
+ if (clazz == byte.class || clazz == Byte.class) {
+ return MinorType.TINYINT;
+ } else if (clazz == short.class || clazz == Short.class) {
+ return MinorType.SMALLINT;
+ } else if (clazz == int.class || clazz == Integer.class) {
+ return MinorType.INT;
+ } else if (clazz == long.class || clazz == Long.class) {
+ return MinorType.BIGINT;
+ } else if (clazz == Date.class) {
+ return MinorType.DATE;
+ } else if (clazz == Time.class) {
+ return MinorType.TIME;
+ } else if (clazz == Timestamp.class) {
+ return MinorType.TIMESTAMP;
+ } else if (clazz == float.class || clazz == Float.class) {
+ return MinorType.FLOAT4;
+ } else if (clazz == double.class || clazz == Double.class) {
+ return MinorType.FLOAT8;
+ } else if (clazz == boolean.class || clazz == Boolean.class) {
+ return MinorType.BIT;
+ } else if (clazz == char.class || clazz == Character.class) {
+ return MinorType.VARCHAR;
+ } else if (clazz == String.class) {
+ return MinorType.VARCHAR;
+ } else if (clazz == byte[].class) {
+ return MinorType.VARBINARY;
+ } else if (clazz == BigDecimal.class) {
+ return MinorType.VARDECIMAL;
+ }
+ throw new IllegalArgumentException("Cannot determine minor type for " + clazz.getName());
+ }
+ }
+
+ /**
+ * Converts and writes dict values using provided key / value converters.
+ */
+ class DictColumnConverter implements ColumnConverter {
+
+ private final DictWriter dictWriter;
+ private final ColumnConverter keyConverter;
+ private final ColumnConverter valueConverter;
+
+ public DictColumnConverter(DictWriter dictWriter, ColumnConverter keyConverter, ColumnConverter valueConverter) {
+ this.dictWriter = dictWriter;
+ this.keyConverter = keyConverter;
+ this.valueConverter = valueConverter;
+ }
+
+ @Override
+ public void convert(Object value) {
+ if (value == null) {
+ return;
+ }
+
+ @SuppressWarnings("unchecked") Map<Object, Object> map = (Map<Object, Object>) value;
+ map.forEach((key, val) -> {
+ keyConverter.convert(key);
+ valueConverter.convert(val);
+ dictWriter.save();
+ });
+ }
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/record/ColumnConverterFactory.java b/exec/java-exec/src/main/java/org/apache/drill/exec/record/ColumnConverterFactory.java
new file mode 100644
index 0000000..b408476
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/record/ColumnConverterFactory.java
@@ -0,0 +1,149 @@
+/*
+ * 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.record;
+
+import org.apache.drill.exec.physical.impl.scan.convert.StandardConversions;
+import org.apache.drill.exec.record.metadata.ColumnMetadata;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.record.ColumnConverter.ArrayColumnConverter;
+import org.apache.drill.exec.record.ColumnConverter.DictColumnConverter;
+import org.apache.drill.exec.record.ColumnConverter.DummyColumnConverter;
+import org.apache.drill.exec.record.ColumnConverter.MapColumnConverter;
+import org.apache.drill.exec.record.ColumnConverter.ScalarColumnConverter;
+import org.apache.drill.exec.vector.accessor.ArrayWriter;
+import org.apache.drill.exec.vector.accessor.DictWriter;
+import org.apache.drill.exec.vector.accessor.ObjectWriter;
+import org.apache.drill.exec.vector.accessor.ScalarWriter;
+import org.apache.drill.exec.vector.accessor.TupleWriter;
+import org.apache.drill.exec.vector.accessor.ValueWriter;
+import org.apache.drill.exec.vector.complex.DictVector;
+
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.StreamSupport;
+
+public class ColumnConverterFactory {
+
+ private final StandardConversions standardConversions;
+
+ public ColumnConverterFactory(TupleMetadata providedSchema) {
+ if (providedSchema == null) {
+ standardConversions = null;
+ } else {
+ standardConversions = StandardConversions.builder().withSchema(providedSchema).build();
+ }
+ }
+
+ public ColumnConverter getRootConverter(TupleMetadata providedSchema,
+ TupleMetadata readerSchema, TupleWriter tupleWriter) {
+ return getMapConverter(providedSchema, readerSchema, tupleWriter);
+ }
+
+ /**
+ * Based on column type, creates corresponding column converter
+ * which holds conversion logic and appropriate writer to set converted data into.
+ * For columns which are not projected, {@link DummyColumnConverter} is used.
+ *
+ * @param readerSchema column metadata
+ * @param writer column writer
+ * @return column converter
+ */
+ public ColumnConverter getConverter(TupleMetadata providedSchema,
+ ColumnMetadata readerSchema, ObjectWriter writer) {
+ if (!writer.isProjected()) {
+ return DummyColumnConverter.INSTANCE;
+ }
+
+ if (readerSchema.isArray()) {
+ return getArrayConverter(providedSchema,
+ readerSchema, writer.array());
+ }
+
+ if (readerSchema.isMap()) {
+ return getMapConverter(
+ providedChildSchema(providedSchema, readerSchema),
+ readerSchema.tupleSchema(), writer.tuple());
+ }
+
+ if (readerSchema.isDict()) {
+ return getDictConverter(
+ providedChildSchema(providedSchema, readerSchema),
+ readerSchema.tupleSchema(), writer.dict());
+ }
+
+ return getScalarConverter(readerSchema, writer.scalar());
+ }
+
+ private TupleMetadata providedChildSchema(TupleMetadata providedSchema,
+ ColumnMetadata readerSchema) {
+ return providedSchema == null ? null :
+ providedSchema.metadata(readerSchema.name()).tupleSchema();
+ }
+
+ private ColumnConverter getArrayConverter(TupleMetadata providedSchema,
+ ColumnMetadata readerSchema, ArrayWriter arrayWriter) {
+ ObjectWriter valueWriter = arrayWriter.entry();
+ ColumnConverter valueConverter;
+ if (readerSchema.isMap()) {
+ valueConverter = getMapConverter(providedSchema,
+ readerSchema.tupleSchema(), valueWriter.tuple());
+ } else if (readerSchema.isDict()) {
+ valueConverter = getDictConverter(providedSchema,
+ readerSchema.tupleSchema(), valueWriter.dict());
+ } else if (readerSchema.isMultiList()) {
+ valueConverter = getConverter(null, readerSchema.childSchema(), valueWriter);
+ } else {
+ valueConverter = getScalarConverter(readerSchema, valueWriter.scalar());
+ }
+ return new ArrayColumnConverter(arrayWriter, valueConverter);
+ }
+
+ protected ColumnConverter getMapConverter(TupleMetadata providedSchema,
+ TupleMetadata readerSchema, TupleWriter tupleWriter) {
+ Map<String, ColumnConverter> converters = StreamSupport.stream(readerSchema.spliterator(), false)
+ .collect(Collectors.toMap(
+ ColumnMetadata::name,
+ columnMetadata ->
+ getConverter(providedSchema, columnMetadata, tupleWriter.column(columnMetadata.name()))));
+
+ return new MapColumnConverter(this, providedSchema, tupleWriter, converters);
+ }
+
+ private ColumnConverter getDictConverter(TupleMetadata providedSchema,
+ TupleMetadata readerSchema, DictWriter dictWriter) {
+ ColumnConverter keyConverter = getScalarConverter(
+ readerSchema.metadata(DictVector.FIELD_KEY_NAME), dictWriter.keyWriter());
+ ColumnConverter valueConverter = getConverter(providedSchema,
+ readerSchema.metadata(DictVector.FIELD_VALUE_NAME), dictWriter.valueWriter());
+ return new DictColumnConverter(dictWriter, keyConverter, valueConverter);
+ }
+
+ private ColumnConverter getScalarConverter(ColumnMetadata readerSchema, ScalarWriter scalarWriter) {
+ ValueWriter valueWriter;
+ if (standardConversions == null) {
+ valueWriter = scalarWriter;
+ } else {
+ valueWriter = standardConversions.converterFor(scalarWriter, readerSchema);
+ }
+ return buildScalar(readerSchema, valueWriter);
+ }
+
+ public ScalarColumnConverter buildScalar(ColumnMetadata readerSchema, ValueWriter writer) {
+ return new ScalarColumnConverter(writer::setValue);
+ }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/SubsetRemover.java
similarity index 53%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
copy to exec/java-exec/src/main/java/org/apache/drill/exec/store/SubsetRemover.java
index 7107e9f..69206d8 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/SubsetRemover.java
@@ -15,25 +15,27 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
-
-import java.util.function.Predicate;
+package org.apache.drill.exec.store;
+import org.apache.calcite.plan.volcano.RelSubset;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.convert.ConverterRule;
-import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRelFactories;
-
-class JdbcDrelConverterRule extends ConverterRule {
+import org.apache.calcite.rel.RelShuttle;
+import org.apache.calcite.rel.RelShuttleImpl;
- JdbcDrelConverterRule(DrillJdbcConvention in) {
- super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
- }
+/**
+ * Removes {@link RelSubset} nodes from the plan.
+ * It may be useful when we want to do some operations on the plan
+ * which aren't supported by {@link RelSubset} during the planning phase.
+ */
+public class SubsetRemover extends RelShuttleImpl {
+ public static RelShuttle INSTANCE = new SubsetRemover();
@Override
- public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
- convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
+ public RelNode visit(RelNode other) {
+ if (other instanceof RelSubset) {
+ return ((RelSubset) other).getBest().accept(this);
+ } else {
+ return super.visit(other);
+ }
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroBatchReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroBatchReader.java
index 51a77d6..ec47678 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroBatchReader.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroBatchReader.java
@@ -28,6 +28,7 @@ import org.apache.drill.exec.physical.impl.scan.framework.ManagedReader;
import org.apache.drill.exec.physical.impl.scan.v3.FixedReceiver;
import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+import org.apache.drill.exec.record.ColumnConverter;
import org.apache.drill.exec.record.metadata.TupleMetadata;
import org.apache.drill.exec.util.ImpersonationUtil;
import org.apache.hadoop.fs.FileSystem;
@@ -39,8 +40,6 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.security.PrivilegedExceptionAction;
-import java.util.List;
-import java.util.stream.IntStream;
public class AvroBatchReader implements ManagedReader<FileScanFramework.FileSchemaNegotiator> {
private static final Logger logger = LoggerFactory.getLogger(AvroBatchReader.class);
@@ -49,10 +48,10 @@ public class AvroBatchReader implements ManagedReader<FileScanFramework.FileSche
private long endPosition;
private DataFileReader<GenericRecord> reader;
private ResultSetLoader loader;
- private List<ColumnConverter> converters;
+ private ColumnConverter converter;
// re-use container instance
private GenericRecord record;
- private int maxRecords;
+ private final int maxRecords;
public AvroBatchReader(int maxRecords) {
this.maxRecords = maxRecords;
@@ -81,8 +80,8 @@ public class AvroBatchReader implements ManagedReader<FileScanFramework.FileSche
logger.debug("Avro file table schema: {}", tableSchema);
negotiator.tableSchema(tableSchema, true);
loader = negotiator.build();
- ColumnConverterFactory factory = new ColumnConverterFactory(providedSchema);
- converters = factory.initConverters(providedSchema, readerSchema, loader.writer());
+ AvroColumnConverterFactory factory = new AvroColumnConverterFactory(providedSchema);
+ converter = factory.getRootConverter(providedSchema, readerSchema, loader.writer());
return true;
}
@@ -181,8 +180,7 @@ public class AvroBatchReader implements ManagedReader<FileScanFramework.FileSche
}
rowWriter.start();
- IntStream.range(0, rowWriter.size())
- .forEach(i -> converters.get(i).convert(record.get(i)));
+ converter.convert(record);
rowWriter.save();
return true;
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroColumnConverterFactory.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroColumnConverterFactory.java
new file mode 100644
index 0000000..cfd7d50
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/AvroColumnConverterFactory.java
@@ -0,0 +1,215 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.avro;
+
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.nio.ByteBuffer;
+import java.nio.ByteOrder;
+import java.nio.IntBuffer;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import org.apache.avro.generic.GenericFixed;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.util.Utf8;
+import org.apache.drill.exec.physical.impl.scan.v3.FixedReceiver;
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+import org.apache.drill.exec.record.ColumnConverter;
+import org.apache.drill.exec.record.ColumnConverterFactory;
+import org.apache.drill.exec.record.metadata.ColumnMetadata;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.vector.accessor.TupleWriter;
+import org.apache.drill.exec.vector.accessor.ValueWriter;
+import org.apache.drill.shaded.guava.com.google.common.base.Charsets;
+import org.joda.time.DateTimeConstants;
+import org.joda.time.Period;
+
+public class AvroColumnConverterFactory extends ColumnConverterFactory {
+ public AvroColumnConverterFactory(TupleMetadata providedSchema) {
+ super(providedSchema);
+ }
+
+ /**
+ * Based on given converted Avro schema and current row writer generates list of
+ * column converters based on column type.
+ *
+ * @param readerSchema converted Avro schema
+ * @param rowWriter current row writer
+ * @return list of column converters
+ */
+ public List<ColumnConverter> initConverters(TupleMetadata providedSchema,
+ TupleMetadata readerSchema, RowSetLoader rowWriter) {
+ return IntStream.range(0, readerSchema.size())
+ .mapToObj(i -> getConverter(providedSchema, readerSchema.metadata(i), rowWriter.column(i)))
+ .collect(Collectors.toList());
+ }
+
+ @Override
+ public ColumnConverter.ScalarColumnConverter buildScalar(ColumnMetadata readerSchema, ValueWriter writer) {
+ switch (readerSchema.type()) {
+ case VARCHAR:
+ return new ColumnConverter.ScalarColumnConverter(value -> {
+ byte[] binary;
+ int length;
+ if (value instanceof Utf8) {
+ Utf8 utf8 = (Utf8) value;
+ binary = utf8.getBytes();
+ length = utf8.getByteLength();
+ } else {
+ binary = value.toString().getBytes(Charsets.UTF_8);
+ length = binary.length;
+ }
+ writer.setBytes(binary, length);
+ });
+ case VARBINARY:
+ return new ColumnConverter.ScalarColumnConverter(value -> {
+ if (value instanceof ByteBuffer) {
+ ByteBuffer buf = (ByteBuffer) value;
+ writer.setBytes(buf.array(), buf.remaining());
+ } else {
+ byte[] bytes = ((GenericFixed) value).bytes();
+ writer.setBytes(bytes, bytes.length);
+ }
+ });
+ case VARDECIMAL:
+ return new ColumnConverter.ScalarColumnConverter(value -> {
+ BigInteger bigInteger;
+ if (value instanceof ByteBuffer) {
+ ByteBuffer decBuf = (ByteBuffer) value;
+ bigInteger = new BigInteger(decBuf.array());
+ } else {
+ GenericFixed genericFixed = (GenericFixed) value;
+ bigInteger = new BigInteger(genericFixed.bytes());
+ }
+ BigDecimal decimalValue = new BigDecimal(bigInteger, readerSchema.scale());
+ writer.setDecimal(decimalValue);
+ });
+ case TIMESTAMP:
+ return new ColumnConverter.ScalarColumnConverter(value -> {
+ String avroLogicalType = readerSchema.property(AvroSchemaUtil.AVRO_LOGICAL_TYPE_PROPERTY);
+ if (AvroSchemaUtil.TIMESTAMP_MILLIS_LOGICAL_TYPE.equals(avroLogicalType)) {
+ writer.setLong((long) value);
+ } else {
+ writer.setLong((long) value / 1000);
+ }
+ });
+ case DATE:
+ return new ColumnConverter.ScalarColumnConverter(value -> writer.setLong((int) value * (long) DateTimeConstants.MILLIS_PER_DAY));
+ case TIME:
+ return new ColumnConverter.ScalarColumnConverter(value -> {
+ if (value instanceof Long) {
+ writer.setInt((int) ((long) value / 1000));
+ } else {
+ writer.setInt((int) value);
+ }
+ });
+ case INTERVAL:
+ return new ColumnConverter.ScalarColumnConverter(value -> {
+ GenericFixed genericFixed = (GenericFixed) value;
+ IntBuffer intBuf = ByteBuffer.wrap(genericFixed.bytes())
+ .order(ByteOrder.LITTLE_ENDIAN)
+ .asIntBuffer();
+
+ Period period = Period.months(intBuf.get(0))
+ .withDays(intBuf.get(1)
+ ).withMillis(intBuf.get(2));
+
+ writer.setPeriod(period);
+ });
+ case FLOAT4:
+ return new ColumnConverter.ScalarColumnConverter(value -> writer.setDouble((Float) value));
+ case BIT:
+ return new ColumnConverter.ScalarColumnConverter(value -> writer.setBoolean((Boolean) value));
+ default:
+ return super.buildScalar(readerSchema, writer);
+ }
+ }
+
+ /**
+ * Based on provided schema, given converted Avro schema and current row writer
+ * generates list of column converters based on column type for {@link MapColumnConverter} and returns it.
+ *
+ * @param providedSchema provided schema
+ * @param readerSchema converted Avro schema
+ * @param tupleWriter current row writer
+ * @return {@link MapColumnConverter} with column converters
+ */
+ @Override
+ protected ColumnConverter getMapConverter(TupleMetadata providedSchema,
+ TupleMetadata readerSchema, TupleWriter tupleWriter) {
+ List<ColumnConverter> converters = IntStream.range(0, readerSchema.size())
+ .mapToObj(i -> getConverter(providedSchema, readerSchema.metadata(i), tupleWriter.column(i)))
+ .collect(Collectors.toList());
+
+ return new MapColumnConverter(this, providedSchema, tupleWriter, converters);
+ }
+
+ public void buildMapMembers(GenericRecord genericRecord, TupleMetadata providedSchema,
+ TupleWriter tupleWriter, List<ColumnConverter> converters) {
+ // fill in tuple schema for cases when it contains recursive named record types
+ TupleMetadata readerSchema = AvroSchemaUtil.convert(genericRecord.getSchema());
+ TupleMetadata tableSchema = FixedReceiver.Builder.mergeSchemas(providedSchema, readerSchema);
+ tableSchema.toMetadataList().forEach(tupleWriter::addColumn);
+
+ IntStream.range(0, tableSchema.size())
+ .mapToObj(i -> getConverter(providedSchema,
+ readerSchema.metadata(i), tupleWriter.column(i)))
+ .forEach(converters::add);
+ }
+
+ /**
+ * Converts and writes all map children using provided {@link #converters}.
+ * If {@link #converters} are empty, generates their converters based on
+ * {@link GenericRecord} schema.
+ */
+ public static class MapColumnConverter implements ColumnConverter {
+
+ private final AvroColumnConverterFactory factory;
+ private final TupleMetadata providedSchema;
+ private final TupleWriter tupleWriter;
+ private final List<ColumnConverter> converters;
+
+ public MapColumnConverter(AvroColumnConverterFactory factory,
+ TupleMetadata providedSchema,
+ TupleWriter tupleWriter, List<ColumnConverter> converters) {
+ this.factory = factory;
+ this.providedSchema = providedSchema;
+ this.tupleWriter = tupleWriter;
+ this.converters = new ArrayList<>(converters);
+ }
+
+ @Override
+ public void convert(Object value) {
+ if (value == null) {
+ return;
+ }
+
+ GenericRecord genericRecord = (GenericRecord) value;
+
+ if (converters.isEmpty()) {
+ factory.buildMapMembers(genericRecord, providedSchema, tupleWriter, converters);
+ }
+
+ IntStream.range(0, converters.size())
+ .forEach(i -> converters.get(i).convert(genericRecord.get(i)));
+ }
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverter.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverter.java
deleted file mode 100644
index ee856de..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverter.java
+++ /dev/null
@@ -1,172 +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.store.avro;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
-import java.util.function.Consumer;
-import java.util.stream.IntStream;
-
-import org.apache.avro.generic.GenericArray;
-import org.apache.avro.generic.GenericRecord;
-import org.apache.drill.exec.record.metadata.TupleMetadata;
-import org.apache.drill.exec.vector.accessor.ArrayWriter;
-import org.apache.drill.exec.vector.accessor.DictWriter;
-import org.apache.drill.exec.vector.accessor.TupleWriter;
-
-/**
- * Converts and sets given value into the specific column writer.
- */
-public interface ColumnConverter {
-
- void convert(Object value);
-
- /**
- * Does nothing, is used when column is not projected to avoid unnecessary
- * column values conversions and writes.
- */
- class DummyColumnConverter implements ColumnConverter {
-
- public static final DummyColumnConverter INSTANCE = new DummyColumnConverter();
-
- @Override
- public void convert(Object value) {
- // do nothing
- }
- }
-
- /**
- * Converts and writes scalar values using provided {@link #valueConverter}.
- * {@link #valueConverter} has different implementation depending
- * on the scalar value type.
- */
- class ScalarColumnConverter implements ColumnConverter {
-
- private final Consumer<Object> valueConverter;
-
- public ScalarColumnConverter(Consumer<Object> valueConverter) {
- this.valueConverter = valueConverter;
- }
-
- @Override
- public void convert(Object value) {
- if (value == null) {
- return;
- }
-
- valueConverter.accept(value);
- }
- }
-
- /**
- * Converts and writes array values using {@link #valueConverter}
- * into {@link #arrayWriter}.
- */
- class ArrayColumnConverter implements ColumnConverter {
-
- private final ArrayWriter arrayWriter;
- private final ColumnConverter valueConverter;
-
- public ArrayColumnConverter(ArrayWriter arrayWriter, ColumnConverter valueConverter) {
- this.arrayWriter = arrayWriter;
- this.valueConverter = valueConverter;
- }
-
- @Override
- public void convert(Object value) {
- if (value == null || !arrayWriter.isProjected()) {
- return;
- }
-
- GenericArray<?> array = (GenericArray<?>) value;
- array.forEach(arrayValue -> {
- valueConverter.convert(arrayValue);
- arrayWriter.save();
- });
- }
- }
-
- /**
- * Converts and writes all map children using provided {@link #converters}.
- * If {@link #converters} are empty, generates their converters based on
- * {@link GenericRecord} schema.
- */
- class MapColumnConverter implements ColumnConverter {
-
- private final ColumnConverterFactory factory;
- private final TupleMetadata providedSchema;
- private final TupleWriter tupleWriter;
- private final List<ColumnConverter> converters;
-
- public MapColumnConverter(ColumnConverterFactory factory,
- TupleMetadata providedSchema,
- TupleWriter tupleWriter, List<ColumnConverter> converters) {
- this.factory = factory;
- this.providedSchema = providedSchema;
- this.tupleWriter = tupleWriter;
- this.converters = new ArrayList<>(converters);
- }
-
- @Override
- public void convert(Object value) {
- if (value == null) {
- return;
- }
-
- GenericRecord genericRecord = (GenericRecord) value;
-
- if (converters.isEmpty()) {
- factory.buildMapMembers(genericRecord, providedSchema, tupleWriter, converters);
- }
-
- IntStream.range(0, converters.size())
- .forEach(i -> converters.get(i).convert(genericRecord.get(i)));
- }
- }
-
- /**
- * Converts and writes dict values using provided key / value converters.
- */
- class DictColumnConverter implements ColumnConverter {
-
- private final DictWriter dictWriter;
- private final ColumnConverter keyConverter;
- private final ColumnConverter valueConverter;
-
- public DictColumnConverter(DictWriter dictWriter, ColumnConverter keyConverter, ColumnConverter valueConverter) {
- this.dictWriter = dictWriter;
- this.keyConverter = keyConverter;
- this.valueConverter = valueConverter;
- }
-
- @Override
- public void convert(Object value) {
- if (value == null) {
- return;
- }
-
- @SuppressWarnings("unchecked") Map<Object, Object> map = (Map<Object, Object>) value;
- map.forEach((key, val) -> {
- keyConverter.convert(key);
- valueConverter.convert(val);
- dictWriter.save();
- });
- }
- }
-}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverterFactory.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverterFactory.java
deleted file mode 100644
index 527f21d..0000000
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/avro/ColumnConverterFactory.java
+++ /dev/null
@@ -1,258 +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.store.avro;
-
-import org.apache.avro.generic.GenericFixed;
-import org.apache.avro.generic.GenericRecord;
-import org.apache.avro.util.Utf8;
-import org.apache.drill.exec.physical.impl.scan.convert.StandardConversions;
-import org.apache.drill.exec.physical.impl.scan.v3.FixedReceiver;
-import org.apache.drill.exec.physical.resultSet.RowSetLoader;
-import org.apache.drill.exec.record.metadata.ColumnMetadata;
-import org.apache.drill.exec.record.metadata.TupleMetadata;
-import org.apache.drill.exec.store.avro.ColumnConverter.ArrayColumnConverter;
-import org.apache.drill.exec.store.avro.ColumnConverter.DictColumnConverter;
-import org.apache.drill.exec.store.avro.ColumnConverter.DummyColumnConverter;
-import org.apache.drill.exec.store.avro.ColumnConverter.MapColumnConverter;
-import org.apache.drill.exec.store.avro.ColumnConverter.ScalarColumnConverter;
-import org.apache.drill.exec.vector.accessor.ArrayWriter;
-import org.apache.drill.exec.vector.accessor.DictWriter;
-import org.apache.drill.exec.vector.accessor.ObjectWriter;
-import org.apache.drill.exec.vector.accessor.ScalarWriter;
-import org.apache.drill.exec.vector.accessor.TupleWriter;
-import org.apache.drill.exec.vector.accessor.ValueWriter;
-import org.apache.drill.exec.vector.complex.DictVector;
-import org.apache.drill.shaded.guava.com.google.common.base.Charsets;
-import org.joda.time.DateTimeConstants;
-import org.joda.time.Period;
-
-import java.math.BigDecimal;
-import java.math.BigInteger;
-import java.nio.ByteBuffer;
-import java.nio.ByteOrder;
-import java.nio.IntBuffer;
-import java.util.List;
-import java.util.stream.Collectors;
-import java.util.stream.IntStream;
-
-public class ColumnConverterFactory {
-
- private final StandardConversions standardConversions;
-
- public ColumnConverterFactory(TupleMetadata providedSchema) {
- if (providedSchema == null) {
- standardConversions = null;
- } else {
- standardConversions = StandardConversions.builder().withSchema(providedSchema).build();
- }
- }
-
- /**
- * Based on given converted Avro schema and current row writer generates list of
- * column converters based on column type.
- *
- * @param readerSchema converted Avro schema
- * @param rowWriter current row writer
- * @return list of column converters
- */
- public List<ColumnConverter> initConverters(TupleMetadata providedSchema,
- TupleMetadata readerSchema, RowSetLoader rowWriter) {
- return IntStream.range(0, readerSchema.size())
- .mapToObj(i -> getConverter(providedSchema, readerSchema.metadata(i), rowWriter.column(i)))
- .collect(Collectors.toList());
- }
-
- /**
- * Based on column type, creates corresponding column converter
- * which holds conversion logic and appropriate writer to set converted data into.
- * For columns which are not projected, {@link DummyColumnConverter} is used.
- *
- * @param readerSchema column metadata
- * @param writer column writer
- * @return column converter
- */
- public ColumnConverter getConverter(TupleMetadata providedSchema,
- ColumnMetadata readerSchema, ObjectWriter writer) {
- if (!writer.isProjected()) {
- return DummyColumnConverter.INSTANCE;
- }
-
- if (readerSchema.isArray()) {
- return getArrayConverter(providedSchema,
- readerSchema, writer.array());
- }
-
- if (readerSchema.isMap()) {
- return getMapConverter(
- providedChildSchema(providedSchema, readerSchema),
- readerSchema.tupleSchema(), writer.tuple());
- }
-
- if (readerSchema.isDict()) {
- return getDictConverter(
- providedChildSchema(providedSchema, readerSchema),
- readerSchema.tupleSchema(), writer.dict());
- }
-
- return getScalarConverter(readerSchema, writer.scalar());
- }
-
- private TupleMetadata providedChildSchema(TupleMetadata providedSchema,
- ColumnMetadata readerSchema) {
- return providedSchema == null ? null :
- providedSchema.metadata(readerSchema.name()).tupleSchema();
- }
-
- private ColumnConverter getArrayConverter(TupleMetadata providedSchema,
- ColumnMetadata readerSchema, ArrayWriter arrayWriter) {
- ObjectWriter valueWriter = arrayWriter.entry();
- ColumnConverter valueConverter;
- if (readerSchema.isMap()) {
- valueConverter = getMapConverter(providedSchema,
- readerSchema.tupleSchema(), valueWriter.tuple());
- } else if (readerSchema.isDict()) {
- valueConverter = getDictConverter(providedSchema,
- readerSchema.tupleSchema(), valueWriter.dict());
- } else if (readerSchema.isMultiList()) {
- valueConverter = getConverter(null, readerSchema.childSchema(), valueWriter);
- } else {
- valueConverter = getScalarConverter(readerSchema, valueWriter.scalar());
- }
- return new ArrayColumnConverter(arrayWriter, valueConverter);
- }
-
- private ColumnConverter getMapConverter(TupleMetadata providedSchema,
- TupleMetadata readerSchema, TupleWriter tupleWriter) {
- List<ColumnConverter> converters = IntStream.range(0, readerSchema.size())
- .mapToObj(i -> getConverter(providedSchema, readerSchema.metadata(i), tupleWriter.column(i)))
- .collect(Collectors.toList());
- return new MapColumnConverter(this, providedSchema, tupleWriter, converters);
- }
-
- private ColumnConverter getDictConverter(TupleMetadata providedSchema,
- TupleMetadata readerSchema, DictWriter dictWriter) {
- ColumnConverter keyConverter = getScalarConverter(
- readerSchema.metadata(DictVector.FIELD_KEY_NAME), dictWriter.keyWriter());
- ColumnConverter valueConverter = getConverter(providedSchema,
- readerSchema.metadata(DictVector.FIELD_VALUE_NAME), dictWriter.valueWriter());
- return new DictColumnConverter(dictWriter, keyConverter, valueConverter);
- }
-
- private ColumnConverter getScalarConverter(ColumnMetadata readerSchema, ScalarWriter scalarWriter) {
- ValueWriter valueWriter;
- if (standardConversions == null) {
- valueWriter = scalarWriter;
- } else {
- valueWriter = standardConversions.converterFor(scalarWriter, readerSchema);
- }
- return buildScalar(readerSchema, valueWriter);
- }
-
- public ScalarColumnConverter buildScalar(ColumnMetadata readerSchema, ValueWriter writer) {
- switch (readerSchema.type()) {
- case VARCHAR:
- return new ScalarColumnConverter(value -> {
- byte[] binary;
- int length;
- if (value instanceof Utf8) {
- Utf8 utf8 = (Utf8) value;
- binary = utf8.getBytes();
- length = utf8.getByteLength();
- } else {
- binary = value.toString().getBytes(Charsets.UTF_8);
- length = binary.length;
- }
- writer.setBytes(binary, length);
- });
- case VARBINARY:
- return new ScalarColumnConverter(value -> {
- if (value instanceof ByteBuffer) {
- ByteBuffer buf = (ByteBuffer) value;
- writer.setBytes(buf.array(), buf.remaining());
- } else {
- byte[] bytes = ((GenericFixed) value).bytes();
- writer.setBytes(bytes, bytes.length);
- }
- });
- case VARDECIMAL:
- return new ScalarColumnConverter(value -> {
- BigInteger bigInteger;
- if (value instanceof ByteBuffer) {
- ByteBuffer decBuf = (ByteBuffer) value;
- bigInteger = new BigInteger(decBuf.array());
- } else {
- GenericFixed genericFixed = (GenericFixed) value;
- bigInteger = new BigInteger(genericFixed.bytes());
- }
- BigDecimal decimalValue = new BigDecimal(bigInteger, readerSchema.scale());
- writer.setDecimal(decimalValue);
- });
- case TIMESTAMP:
- return new ScalarColumnConverter(value -> {
- String avroLogicalType = readerSchema.property(AvroSchemaUtil.AVRO_LOGICAL_TYPE_PROPERTY);
- if (AvroSchemaUtil.TIMESTAMP_MILLIS_LOGICAL_TYPE.equals(avroLogicalType)) {
- writer.setLong((long) value);
- } else {
- writer.setLong((long) value / 1000);
- }
- });
- case DATE:
- return new ScalarColumnConverter(value -> writer.setLong((int) value * (long) DateTimeConstants.MILLIS_PER_DAY));
- case TIME:
- return new ScalarColumnConverter(value -> {
- if (value instanceof Long) {
- writer.setInt((int) ((long) value / 1000));
- } else {
- writer.setInt((int) value);
- }
- });
- case INTERVAL:
- return new ScalarColumnConverter(value -> {
- GenericFixed genericFixed = (GenericFixed) value;
- IntBuffer intBuf = ByteBuffer.wrap(genericFixed.bytes())
- .order(ByteOrder.LITTLE_ENDIAN)
- .asIntBuffer();
-
- Period period = Period.months(intBuf.get(0))
- .withDays(intBuf.get(1)
- ).withMillis(intBuf.get(2));
-
- writer.setPeriod(period);
- });
- case FLOAT4:
- return new ScalarColumnConverter(value -> writer.setDouble((Float) value));
- case BIT:
- return new ScalarColumnConverter(value -> writer.setBoolean((Boolean) value));
- default:
- return new ScalarColumnConverter(writer::setValue);
- }
- }
-
- public void buildMapMembers(GenericRecord genericRecord, TupleMetadata providedSchema,
- TupleWriter tupleWriter, List<ColumnConverter> converters) {
- // fill in tuple schema for cases when it contains recursive named record types
- TupleMetadata readerSchema = AvroSchemaUtil.convert(genericRecord.getSchema());
- TupleMetadata tableSchema = FixedReceiver.Builder.mergeSchemas(providedSchema, readerSchema);
- tableSchema.toMetadataList().forEach(tupleWriter::addColumn);
-
- IntStream.range(0, tableSchema.size())
- .mapToObj(i -> getConverter(providedSchema,
- readerSchema.metadata(i), tupleWriter.column(i)))
- .forEach(converters::add);
- }
-}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/DrillDataContext.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/DrillDataContext.java
new file mode 100644
index 0000000..72237d2
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/DrillDataContext.java
@@ -0,0 +1,58 @@
+/*
+ * 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.enumerable;
+
+import org.apache.calcite.DataContext;
+import org.apache.calcite.adapter.java.JavaTypeFactory;
+import org.apache.calcite.linq4j.QueryProvider;
+import org.apache.calcite.schema.SchemaPlus;
+
+import java.util.Map;
+
+public class DrillDataContext implements DataContext {
+ private final SchemaPlus rootSchema;
+ private final JavaTypeFactory typeFactory;
+ private final Map<String, Object> properties;
+
+ public DrillDataContext(
+ SchemaPlus rootSchema, JavaTypeFactory typeFactory, Map<String, Object> properties) {
+ this.rootSchema = rootSchema;
+ this.typeFactory = typeFactory;
+ this.properties = properties;
+ }
+
+ @Override
+ public SchemaPlus getRootSchema() {
+ return rootSchema;
+ }
+
+ @Override
+ public JavaTypeFactory getTypeFactory() {
+ return typeFactory;
+ }
+
+ @Override
+ public QueryProvider getQueryProvider() {
+ return null;
+ }
+
+ @Override
+ public Object get(String name) {
+ return properties.get(name);
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableBatchCreator.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableBatchCreator.java
new file mode 100644
index 0000000..9c4ca54
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableBatchCreator.java
@@ -0,0 +1,70 @@
+/*
+ * 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.enumerable;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.types.TypeProtos;
+import org.apache.drill.common.types.Types;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
+import org.apache.drill.exec.physical.impl.BatchCreator;
+import org.apache.drill.exec.physical.impl.scan.framework.BasicScanFactory;
+import org.apache.drill.exec.physical.impl.scan.framework.ManagedReader;
+import org.apache.drill.exec.physical.impl.scan.framework.ManagedScanFramework;
+import org.apache.drill.exec.physical.impl.scan.framework.SchemaNegotiator;
+import org.apache.drill.exec.record.CloseableRecordBatch;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+
+import java.util.Collections;
+import java.util.List;
+
+@SuppressWarnings("unused")
+public class EnumerableBatchCreator implements BatchCreator<EnumerableSubScan> {
+
+ @Override
+ public CloseableRecordBatch getBatch(ExecutorFragmentContext context,
+ EnumerableSubScan subScan, List<RecordBatch> children) throws ExecutionSetupException {
+ Preconditions.checkArgument(children.isEmpty());
+
+ try {
+ ManagedScanFramework.ScanFrameworkBuilder builder = createBuilder(subScan);
+ return builder.buildScanOperator(context, subScan);
+ } catch (UserException e) {
+ // Rethrow user exceptions directly
+ throw e;
+ } catch (Throwable e) {
+ // Wrap all others
+ throw new ExecutionSetupException(e);
+ }
+ }
+
+ private ManagedScanFramework.ScanFrameworkBuilder createBuilder(EnumerableSubScan subScan) {
+ ManagedScanFramework.ScanFrameworkBuilder builder = new ManagedScanFramework.ScanFrameworkBuilder();
+ builder.projection(subScan.getColumns());
+ builder.setUserName(subScan.getUserName());
+ builder.providedSchema(subScan.getSchema());
+
+ ManagedReader<SchemaNegotiator> reader = new EnumerableRecordReader(subScan.getColumns(),
+ subScan.getFieldsMap(), subScan.getCode(), subScan.getSchemaPath());
+ ManagedScanFramework.ReaderFactory readerFactory = new BasicScanFactory(Collections.singletonList(reader).iterator());
+ builder.setReaderFactory(readerFactory);
+ builder.nullType(Types.optional(TypeProtos.MinorType.VARCHAR));
+ return builder;
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableGroupScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableGroupScan.java
new file mode 100644
index 0000000..815a5c2
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableGroupScan.java
@@ -0,0 +1,128 @@
+/*
+ * 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.enumerable;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.physical.base.AbstractGroupScan;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.physical.base.ScanStats;
+import org.apache.drill.exec.physical.base.SubScan;
+import org.apache.drill.exec.proto.CoordinationProtos;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+
+import java.util.List;
+import java.util.Map;
+import java.util.StringJoiner;
+
+public class EnumerableGroupScan extends AbstractGroupScan {
+ private final String code;
+ private final String schemaPath;
+ private final Map<String, Integer> fieldsMap;
+ private final List<SchemaPath> columns;
+ private final double rows;
+ private final TupleMetadata schema;
+
+ @JsonCreator
+ public EnumerableGroupScan(
+ @JsonProperty("sql") String code,
+ @JsonProperty("columns") List<SchemaPath> columns,
+ @JsonProperty("fieldsMap") Map<String, Integer> fieldsMap,
+ @JsonProperty("rows") double rows,
+ @JsonProperty("schema") TupleMetadata schema,
+ @JsonProperty("schemaPath") String schemaPath) {
+ super("");
+ this.code = code;
+ this.columns = columns;
+ this.fieldsMap = fieldsMap;
+ this.rows = rows;
+ this.schema = schema;
+ this.schemaPath = schemaPath;
+ }
+
+ @Override
+ public void applyAssignments(List<CoordinationProtos.DrillbitEndpoint> endpoints) {
+ }
+
+ @Override
+ public SubScan getSpecificScan(int minorFragmentId) {
+ return new EnumerableSubScan(code, columns, fieldsMap, schema, schemaPath);
+ }
+
+ @Override
+ public int getMaxParallelizationWidth() {
+ return 1;
+ }
+
+ @Override
+ public ScanStats getScanStats() {
+ return new ScanStats(
+ ScanStats.GroupScanProperty.NO_EXACT_ROW_COUNT,
+ (long) Math.max(rows, 1),
+ 1,
+ 1);
+ }
+
+ public String getCode() {
+ return code;
+ }
+
+ @Override
+ public List<SchemaPath> getColumns() {
+ return columns;
+ }
+
+ public Map<String, Integer> getFieldsMap() {
+ return fieldsMap;
+ }
+
+ public double getRows() {
+ return rows;
+ }
+
+ public TupleMetadata getSchema() {
+ return schema;
+ }
+
+ public String getSchemaPath() {
+ return schemaPath;
+ }
+
+ @Override
+ public String getDigest() {
+ return toString();
+ }
+
+ @Override
+ public PhysicalOperator getNewWithChildren(List<PhysicalOperator> children) {
+ Preconditions.checkArgument(children.isEmpty());
+ return new EnumerableGroupScan(code, columns, fieldsMap, rows, schema, schemaPath);
+ }
+
+ @Override
+ public String toString() {
+ return new StringJoiner(", ", EnumerableGroupScan.class.getSimpleName() + "[", "]")
+ .add("code='" + code + "'")
+ .add("columns=" + columns)
+ .add("fieldsMap=" + fieldsMap)
+ .add("rows=" + rows)
+ .toString();
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableRecordReader.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableRecordReader.java
new file mode 100644
index 0000000..66e8a24
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableRecordReader.java
@@ -0,0 +1,162 @@
+/*
+ * 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.enumerable;
+
+import org.apache.calcite.DataContext;
+import org.apache.calcite.jdbc.JavaTypeFactoryImpl;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.util.BuiltInMethod;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.compile.ClassBuilder;
+import org.apache.drill.exec.exception.ClassTransformationException;
+import org.apache.drill.exec.ops.OperatorContext;
+import org.apache.drill.exec.physical.impl.scan.framework.ManagedReader;
+import org.apache.drill.exec.physical.impl.scan.framework.SchemaNegotiator;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+import org.apache.drill.exec.planner.sql.SchemaUtilites;
+import org.apache.drill.exec.record.ColumnConverter;
+import org.apache.drill.exec.record.ColumnConverterFactory;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.record.metadata.TupleSchema;
+import org.codehaus.commons.compiler.CompileException;
+
+import java.io.IOException;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.StreamSupport;
+
+/**
+ * {@link ManagedReader} implementation that compiles and executes specified code,
+ * calls the method on it for obtaining the values, and reads the results using column converters.
+ */
+public class EnumerableRecordReader implements ManagedReader<SchemaNegotiator> {
+
+ private static final String CLASS_NAME = "Baz";
+
+ private final List<SchemaPath> columns;
+
+ private final Map<String, Integer> fieldsMap;
+
+ private final String code;
+
+ private final String schemaPath;
+
+ private ColumnConverter converter;
+
+ private Iterator<Map<String, Object>> records;
+
+ private ResultSetLoader loader;
+
+ public EnumerableRecordReader(List<SchemaPath> columns, Map<String, Integer> fieldsMap, String code, String schemaPath) {
+ this.columns = columns;
+ this.fieldsMap = fieldsMap;
+ this.code = code;
+ this.schemaPath = schemaPath;
+ }
+
+ @SuppressWarnings("unchecked")
+ private void setup(OperatorContext context) {
+ SchemaPlus rootSchema = context.getFragmentContext().getFullRootSchema();
+ DataContext root = new DrillDataContext(
+ schemaPath != null ? SchemaUtilites.searchSchemaTree(rootSchema, SchemaUtilites.getSchemaPathAsList(schemaPath)) : rootSchema,
+ new JavaTypeFactoryImpl(),
+ Collections.emptyMap());
+
+ try {
+ Class<?> implementationClass = ClassBuilder.getCompiledClass(code, CLASS_NAME,
+ context.getFragmentContext().getConfig(), context.getFragmentContext().getOptions());
+ Iterable<?> iterable =
+ (Iterable<Map<String, Object>>) implementationClass.getMethod(BuiltInMethod.BINDABLE_BIND.method.getName(), DataContext.class)
+ .invoke(implementationClass.newInstance(), root);
+ if (fieldsMap.keySet().size() == 1) {
+ // for the case of projecting single column, its value is returned
+ records = StreamSupport.stream(iterable.spliterator(), false)
+ .map(this::wrap)
+ .iterator();
+ } else {
+ // for the case when all columns were projected, array is returned
+ records = StreamSupport.stream(iterable.spliterator(), false)
+ .map(row -> wrap((Object[]) row))
+ .iterator();
+ }
+ } catch (CompileException | IOException | ClassTransformationException | ReflectiveOperationException e) {
+ throw new RuntimeException("Exception happened when executing generated code", e.getCause());
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<String, Object> wrap(Object[] values) {
+ Map<String, Object> row = new HashMap<>();
+ columns.stream()
+ .map(SchemaPath::getRootSegmentPath)
+ .forEach(fieldName -> {
+ if (fieldName.equals(SchemaPath.DYNAMIC_STAR)) {
+ row.putAll((Map<? extends String, ?>) values[fieldsMap.get(fieldName)]);
+ } else {
+ row.put(fieldName, values[fieldsMap.get(fieldName)]);
+ }
+ });
+ return row;
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<String, Object> wrap(Object value) {
+ SchemaPath schemaPath = columns.iterator().next();
+ if (schemaPath.equals(SchemaPath.STAR_COLUMN)) {
+ return (Map<String, Object>) value;
+ }
+ return Collections.singletonMap(schemaPath.getRootSegmentPath(), value);
+ }
+
+ @Override
+ public boolean open(SchemaNegotiator negotiator) {
+ TupleMetadata providedSchema = negotiator.providedSchema();
+ loader = negotiator.build();
+ setup(negotiator.context());
+ ColumnConverterFactory factory = new ColumnConverterFactory(providedSchema);
+ converter = factory.getRootConverter(providedSchema, new TupleSchema(), loader.writer());
+ return true;
+ }
+
+ @Override
+ public boolean next() {
+ RowSetLoader rowWriter = loader.writer();
+ while (!rowWriter.isFull()) {
+ if (records.hasNext()) {
+ processRecord(rowWriter, records.next());
+ } else {
+ return false;
+ }
+ }
+ return true;
+ }
+
+ private void processRecord(RowSetLoader writer, Map<String, Object> record) {
+ writer.start();
+ converter.convert(record);
+ writer.save();
+ }
+
+ @Override
+ public void close() {
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableSubScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableSubScan.java
new file mode 100644
index 0000000..85d282b
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/EnumerableSubScan.java
@@ -0,0 +1,78 @@
+/*
+ * 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.enumerable;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.physical.base.AbstractSubScan;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+
+import java.util.List;
+import java.util.Map;
+
+public class EnumerableSubScan extends AbstractSubScan {
+
+ public static final String OPERATOR_TYPE = "ENUMERABLE_SUB_SCAN";
+
+ private final String code;
+ private final String schemaPath;
+ private final List<SchemaPath> columns;
+ private final Map<String, Integer> fieldsMap;
+ private final TupleMetadata schema;
+
+ @JsonCreator
+ public EnumerableSubScan(
+ @JsonProperty("code") String code,
+ @JsonProperty("columns") List<SchemaPath> columns,
+ @JsonProperty("fieldsMap") Map<String, Integer> fieldsMap,
+ @JsonProperty("schema") TupleMetadata schema,
+ @JsonProperty("schemaPath") String schemaPath) {
+ super("");
+ this.code = code;
+ this.columns = columns;
+ this.fieldsMap = fieldsMap;
+ this.schema = schema;
+ this.schemaPath = schemaPath;
+ }
+
+ @Override
+ public String getOperatorType() {
+ return OPERATOR_TYPE;
+ }
+
+ public TupleMetadata getSchema() {
+ return schema;
+ }
+
+ public String getCode() {
+ return code;
+ }
+
+ public List<SchemaPath> getColumns() {
+ return columns;
+ }
+
+ public Map<String, Integer> getFieldsMap() {
+ return fieldsMap;
+ }
+
+ public String getSchemaPath() {
+ return schemaPath;
+ }
+}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrel.java
new file mode 100644
index 0000000..a7efc04
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrel.java
@@ -0,0 +1,76 @@
+/*
+ * 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.enumerable.plan;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.RelNode;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.physical.PhysicalPlanCreator;
+import org.apache.drill.exec.planner.physical.Prel;
+import org.apache.drill.exec.planner.physical.SinglePrel;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
+import org.apache.drill.exec.planner.sql.handlers.PrelFinalizable;
+import org.apache.drill.exec.record.BatchSchema;
+
+import java.util.List;
+
+public class EnumerableIntermediatePrel extends SinglePrel implements PrelFinalizable {
+
+ private final EnumerablePrelContext context;
+
+ public EnumerableIntermediatePrel(RelOptCluster cluster, RelTraitSet traits, RelNode child, EnumerablePrelContext context) {
+ super(cluster, traits, child);
+ this.context = context;
+ }
+
+ @Override
+ public PhysicalOperator getPhysicalOperator(PhysicalPlanCreator creator) {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
+ return new EnumerableIntermediatePrel(getCluster(), traitSet, inputs.iterator().next(), context);
+ }
+
+ @Override
+ protected Object clone() {
+ return copy(getTraitSet(), getInputs());
+ }
+
+ @Override
+ public BatchSchema.SelectionVectorMode getEncoding() {
+ return BatchSchema.SelectionVectorMode.NONE;
+ }
+
+ @Override
+ public Prel finalizeRel() {
+ return new EnumerablePrel(getCluster(), getTraitSet(), getInput(), context);
+ }
+
+ @Override
+ public <T, X, E extends Throwable> T accept(PrelVisitor<T, X, E> logicalVisitor, X value) {
+ throw new UnsupportedOperationException("This needs to be finalized before using a PrelVisitor.");
+ }
+
+ @Override
+ public boolean needsFinalColumnReordering() {
+ return false;
+ }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcIntermediatePrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrelConverterRule.java
similarity index 67%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcIntermediatePrelConverterRule.java
copy to exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrelConverterRule.java
index 1fc58b4..615ec61 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcIntermediatePrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrelConverterRule.java
@@ -15,9 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
-
-import java.util.function.Predicate;
+package org.apache.drill.exec.store.enumerable.plan;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.convert.ConverterRule;
@@ -25,20 +23,25 @@ import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillRelFactories;
import org.apache.drill.exec.planner.physical.Prel;
-final class JdbcIntermediatePrelConverterRule extends ConverterRule {
+import java.util.function.Predicate;
+
+public class EnumerableIntermediatePrelConverterRule extends ConverterRule {
- static final JdbcIntermediatePrelConverterRule INSTANCE = new JdbcIntermediatePrelConverterRule();
+ private final EnumerablePrelContext context;
- private JdbcIntermediatePrelConverterRule() {
- super(JdbcDrel.class, (Predicate<RelNode>) input -> true, DrillRel.DRILL_LOGICAL,
- Prel.DRILL_PHYSICAL, DrillRelFactories.LOGICAL_BUILDER, "JDBC_PREL_Converter");
+ public EnumerableIntermediatePrelConverterRule(EnumerablePrelContext context) {
+ super(VertexDrel.class, (Predicate<RelNode>) input -> true, DrillRel.DRILL_LOGICAL,
+ Prel.DRILL_PHYSICAL, DrillRelFactories.LOGICAL_BUILDER,
+ "EnumerableIntermediatePrelConverterRule:" + context.getPlanPrefix());
+ this.context = context;
}
@Override
public RelNode convert(RelNode in) {
- return new JdbcIntermediatePrel(
+ return new EnumerableIntermediatePrel(
in.getCluster(),
in.getTraitSet().replace(getOutTrait()),
- in.getInput(0));
+ in.getInput(0),
+ context);
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerablePrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerablePrel.java
new file mode 100644
index 0000000..0df66b5
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerablePrel.java
@@ -0,0 +1,116 @@
+/*
+ * 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.enumerable.plan;
+
+import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptUtil;
+import org.apache.calcite.plan.RelTraitSet;
+import org.apache.calcite.rel.AbstractRelNode;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rel.RelWriter;
+import org.apache.calcite.rel.core.TableScan;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.exec.physical.base.PhysicalOperator;
+import org.apache.drill.exec.planner.common.DrillRelOptUtil;
+import org.apache.drill.exec.planner.physical.LeafPrel;
+import org.apache.drill.exec.planner.physical.PhysicalPlanCreator;
+import org.apache.drill.exec.planner.physical.visitor.PrelVisitor;
+import org.apache.drill.exec.record.BatchSchema;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.record.metadata.schema.SchemaProvider;
+import org.apache.drill.exec.store.enumerable.EnumerableGroupScan;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.stream.Collectors;
+
+/**
+ * {@link LeafPrel} implementation that generates java code that may be executed to obtain results
+ * for the provided plan part.
+ */
+public class EnumerablePrel extends AbstractRelNode implements LeafPrel {
+ private final String code;
+ private final String schemaPath;
+ private final String plan;
+ private final double rows;
+ private final Map<String, Integer> fieldsMap;
+ private final TupleMetadata schema;
+ private final String planPrefix;
+
+ public EnumerablePrel(RelOptCluster cluster, RelTraitSet traitSet, RelNode input, EnumerablePrelContext context) {
+ super(cluster, traitSet);
+ this.rowType = input.getRowType();
+ this.rows = input.estimateRowCount(cluster.getMetadataQuery());
+
+ this.planPrefix = context.getPlanPrefix();
+ RelNode transformedNode = context.transformNode(input);
+
+ this.fieldsMap = context.getFieldsMap(transformedNode);
+
+ this.plan = RelOptUtil.toString(transformedNode);
+ this.code = context.generateCode(cluster, transformedNode);
+
+ this.schemaPath = context.getTablePath(input);
+ try {
+ TableScan scan = Objects.requireNonNull(DrillRelOptUtil.findScan(input));
+ SchemaProvider schemaProvider = DrillRelOptUtil.getDrillTable(scan)
+ .getMetadataProviderManager().getSchemaProvider();
+ this.schema = schemaProvider != null ? schemaProvider.read().getSchema() : null;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ @Override
+ public PhysicalOperator getPhysicalOperator(PhysicalPlanCreator creator) {
+ List<SchemaPath> columns = rowType.getFieldNames().stream()
+ .map(SchemaPath::getSimplePath)
+ .collect(Collectors.toList());
+ EnumerableGroupScan groupScan = new EnumerableGroupScan(code, columns, fieldsMap, rows, schema, schemaPath);
+ return creator.addMetadata(this, groupScan);
+ }
+
+ @Override
+ public RelWriter explainTerms(RelWriter pw) {
+ pw.item(planPrefix, plan);
+ return super.explainTerms(pw);
+ }
+
+ @Override
+ public double estimateRowCount(RelMetadataQuery mq) {
+ return rows;
+ }
+
+ @Override
+ public <T, X, E extends Throwable> T accept(PrelVisitor<T, X, E> logicalVisitor, X value) throws E {
+ return logicalVisitor.visitLeaf(this, value);
+ }
+
+ @Override
+ public BatchSchema.SelectionVectorMode[] getSupportedEncodings() {
+ return BatchSchema.SelectionVectorMode.DEFAULT;
+ }
+
+ @Override
+ public BatchSchema.SelectionVectorMode getEncoding() {
+ return BatchSchema.SelectionVectorMode.NONE;
+ }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerablePrelContext.java
similarity index 53%
copy from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
copy to exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerablePrelContext.java
index 7107e9f..43b9216 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerablePrelContext.java
@@ -15,25 +15,22 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
-
-import java.util.function.Predicate;
+package org.apache.drill.exec.store.enumerable.plan;
+import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.rel.RelNode;
-import org.apache.calcite.rel.convert.ConverterRule;
-import org.apache.drill.exec.planner.logical.DrillRel;
-import org.apache.drill.exec.planner.logical.DrillRelFactories;
-
-class JdbcDrelConverterRule extends ConverterRule {
-
- JdbcDrelConverterRule(DrillJdbcConvention in) {
- super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
- }
-
- @Override
- public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
- convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
- }
+
+import java.util.Map;
+
+public interface EnumerablePrelContext {
+
+ String generateCode(RelOptCluster cluster, RelNode elasticNode);
+
+ RelNode transformNode(RelNode input);
+
+ Map<String, Integer> getFieldsMap(RelNode transformedNode);
+
+ String getPlanPrefix();
+
+ String getTablePath(RelNode input);
}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrel.java
similarity index 74%
rename from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrel.java
rename to exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrel.java
index 14e29d6..edbc591 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrel.java
@@ -15,9 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
-
-import java.util.List;
+package org.apache.drill.exec.store.enumerable.plan;
import org.apache.calcite.plan.RelOptCluster;
import org.apache.calcite.plan.RelTraitSet;
@@ -27,19 +25,25 @@ import org.apache.drill.common.logical.data.LogicalOperator;
import org.apache.drill.exec.planner.logical.DrillImplementor;
import org.apache.drill.exec.planner.logical.DrillRel;
-public class JdbcDrel extends SingleRel implements DrillRel {
+import java.util.List;
+
+/**
+ * The vertex simply holds the child nodes but contains its own traits.
+ * Used for completing Drill logical planning when child nodes have some specific traits.
+ */
+public class VertexDrel extends SingleRel implements DrillRel {
- public JdbcDrel(RelOptCluster cluster, RelTraitSet traits, RelNode child) {
- super(cluster, traits, child);
+ public VertexDrel(RelOptCluster cluster, RelTraitSet traits, RelNode input) {
+ super(cluster, traits, input);
}
@Override
public RelNode copy(RelTraitSet traitSet, List<RelNode> inputs) {
- return new JdbcDrel(getCluster(), traitSet, inputs.iterator().next());
+ return new VertexDrel(getCluster(), traitSet, inputs.iterator().next());
}
@Override
- protected Object clone() throws CloneNotSupportedException {
+ protected Object clone() {
return copy(getTraitSet(), getInputs());
}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrelConverterRule.java
similarity index 76%
rename from contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
rename to exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrelConverterRule.java
index 7107e9f..f69ba7c 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcDrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrelConverterRule.java
@@ -15,25 +15,26 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.exec.store.jdbc;
-
-import java.util.function.Predicate;
+package org.apache.drill.exec.store.enumerable.plan;
+import org.apache.calcite.plan.Convention;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.convert.ConverterRule;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillRelFactories;
-class JdbcDrelConverterRule extends ConverterRule {
+import java.util.function.Predicate;
+
+public class VertexDrelConverterRule extends ConverterRule {
- JdbcDrelConverterRule(DrillJdbcConvention in) {
+ public VertexDrelConverterRule(Convention in) {
super(RelNode.class, (Predicate<RelNode>) input -> true, in, DrillRel.DRILL_LOGICAL,
- DrillRelFactories.LOGICAL_BUILDER, "JDBC_DREL_Converter" + in.getName());
+ DrillRelFactories.LOGICAL_BUILDER, "VertexDrelConverterRule" + in.getName());
}
@Override
public RelNode convert(RelNode in) {
- return new JdbcDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
+ return new VertexDrel(in.getCluster(), in.getTraitSet().replace(DrillRel.DRILL_LOGICAL),
convert(in, in.getTraitSet().replace(this.getInTrait()).simplify()));
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/table/function/WithOptionsTableMacro.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/table/function/WithOptionsTableMacro.java
index 620dad5..2a6bebc 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/table/function/WithOptionsTableMacro.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/table/function/WithOptionsTableMacro.java
@@ -54,7 +54,9 @@ public class WithOptionsTableMacro implements TableMacro {
.message("Unable to find table [%s]", sig.getName())
.build(logger);
}
- return new DrillTranslatableTable(drillTable);
+ return drillTable instanceof TranslatableTable
+ ? (TranslatableTable) drillTable :
+ new DrillTranslatableTable(drillTable);
}
@Override
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/common/HashPartitionTest.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/common/HashPartitionTest.java
index e98b4c0..1675919 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/common/HashPartitionTest.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/common/HashPartitionTest.java
@@ -88,9 +88,9 @@ public class HashPartitionTest extends BaseTest {
@Override
public CloseableRecordBatch createProbeBatch(BatchSchema schema, FragmentContext context) {
probeRowSet = new RowSetBuilder(context.getAllocator(), schema)
- .addRow(.5, "yellow")
- .addRow(1.5, "blue")
- .addRow(2.5, "black")
+ .addRow(.5f, "yellow")
+ .addRow(1.5f, "blue")
+ .addRow(2.5f, "black")
.build();
return new MockRecordBatch.Builder().
sendData(probeRowSet).
@@ -187,9 +187,9 @@ public class HashPartitionTest extends BaseTest {
@Override
public CloseableRecordBatch createProbeBatch(BatchSchema schema, FragmentContext context) {
probeRowSet = new RowSetBuilder(context.getAllocator(), schema)
- .addRow(.5, "yellow")
- .addRow(1.5, "blue")
- .addRow(2.5, "black")
+ .addRow(.5f, "yellow")
+ .addRow(1.5f, "blue")
+ .addRow(2.5f, "black")
.build();
return new MockRecordBatch.Builder().
sendData(probeRowSet).
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/convert/TestDirectConverter.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/convert/TestDirectConverter.java
index aa106c2..1615827 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/convert/TestDirectConverter.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/impl/scan/convert/TestDirectConverter.java
@@ -708,7 +708,7 @@ public class TestDirectConverter extends SubOperatorTest {
expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, bigIntCol));
expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, float4Col));
expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, float8Col));
- expect(ConversionType.IMPLICIT, conversions.analyze(tinyIntCol, decimalCol));
+ expect(ConversionType.EXPLICIT, conversions.analyze(tinyIntCol, decimalCol));
expect(ConversionType.EXPLICIT, conversions.analyze(tinyIntCol, stringCol));
// SmallInt --> x
@@ -718,7 +718,7 @@ public class TestDirectConverter extends SubOperatorTest {
expect(ConversionType.IMPLICIT, conversions.analyze(smallIntCol, bigIntCol));
expect(ConversionType.IMPLICIT, conversions.analyze(smallIntCol, float4Col));
expect(ConversionType.IMPLICIT, conversions.analyze(smallIntCol, float8Col));
- expect(ConversionType.IMPLICIT, conversions.analyze(smallIntCol, decimalCol));
+ expect(ConversionType.EXPLICIT, conversions.analyze(smallIntCol, decimalCol));
expect(ConversionType.EXPLICIT, conversions.analyze(smallIntCol, stringCol));
// Int --> x
@@ -728,7 +728,7 @@ public class TestDirectConverter extends SubOperatorTest {
expect(ConversionType.IMPLICIT, conversions.analyze(intCol, bigIntCol));
expect(ConversionType.IMPLICIT, conversions.analyze(intCol, float4Col));
expect(ConversionType.IMPLICIT, conversions.analyze(intCol, float8Col));
- expect(ConversionType.IMPLICIT, conversions.analyze(intCol, decimalCol));
+ expect(ConversionType.EXPLICIT, conversions.analyze(intCol, decimalCol));
expect(ConversionType.EXPLICIT, conversions.analyze(intCol, stringCol));
// BigInt --> x
@@ -738,7 +738,7 @@ public class TestDirectConverter extends SubOperatorTest {
expect(ConversionType.NONE, conversions.analyze(bigIntCol, bigIntCol));
expect(ConversionType.IMPLICIT, conversions.analyze(bigIntCol, float4Col));
expect(ConversionType.IMPLICIT, conversions.analyze(bigIntCol, float8Col));
- expect(ConversionType.IMPLICIT, conversions.analyze(bigIntCol, decimalCol));
+ expect(ConversionType.EXPLICIT, conversions.analyze(bigIntCol, decimalCol));
expect(ConversionType.EXPLICIT, conversions.analyze(bigIntCol, stringCol));
// Float4 --> x
@@ -748,7 +748,7 @@ public class TestDirectConverter extends SubOperatorTest {
expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float4Col, bigIntCol));
expect(ConversionType.NONE, conversions.analyze(float4Col, float4Col));
expect(ConversionType.IMPLICIT, conversions.analyze(float4Col, float8Col));
- expect(ConversionType.IMPLICIT, conversions.analyze(float4Col, decimalCol));
+ expect(ConversionType.EXPLICIT, conversions.analyze(float4Col, decimalCol));
expect(ConversionType.EXPLICIT, conversions.analyze(float4Col, stringCol));
// Float8 --> x
@@ -758,7 +758,7 @@ public class TestDirectConverter extends SubOperatorTest {
expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float8Col, bigIntCol));
expect(ConversionType.IMPLICIT_UNSAFE, conversions.analyze(float8Col, float4Col));
expect(ConversionType.NONE, conversions.analyze(float8Col, float8Col));
- expect(ConversionType.IMPLICIT, conversions.analyze(float8Col, decimalCol));
+ expect(ConversionType.EXPLICIT, conversions.analyze(float8Col, decimalCol));
expect(ConversionType.EXPLICIT, conversions.analyze(float8Col, stringCol));
// Decimal --> x
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/rowSet/TestRowSet.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/rowSet/TestRowSet.java
index da4705b..987bd02 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/rowSet/TestRowSet.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/rowSet/TestRowSet.java
@@ -901,7 +901,7 @@ public class TestRowSet extends SubOperatorTest {
final String dictName = "d";
final TupleMetadata schema = new SchemaBuilder()
.add("id", MinorType.INT)
- .addDictArray(dictName, MinorType.FLOAT4)
+ .addDictArray(dictName, MinorType.FLOAT8)
.value(MinorType.VARCHAR)
.resumeSchema()
.buildSchema();
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/rowSet/TestScalarAccessors.java b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/rowSet/TestScalarAccessors.java
index eaf2754..7f990c8 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/physical/rowSet/TestScalarAccessors.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/physical/rowSet/TestScalarAccessors.java
@@ -533,21 +533,21 @@ public class TestScalarAccessors extends SubOperatorTest {
RowSetReader reader = rs.reader();
ScalarReader colReader = reader.scalar(0);
- assertEquals(ValueType.DOUBLE, colReader.valueType());
+ assertEquals(ValueType.FLOAT, colReader.valueType());
assertTrue(reader.next());
assertFalse(colReader.isNull());
- assertEquals(0, colReader.getDouble(), 0.000001);
+ assertEquals(0, colReader.getFloat(), 0.000001);
assertTrue(reader.next());
- assertEquals(Float.MAX_VALUE, colReader.getDouble(), 0.000001);
- assertEquals(Float.MAX_VALUE, (double) colReader.getObject(), 0.000001);
+ assertEquals(Float.MAX_VALUE, colReader.getFloat(), 0.000001);
+ assertEquals(Float.MAX_VALUE, (float) colReader.getObject(), 0.000001);
assertTrue(reader.next());
- assertEquals(Float.MIN_VALUE, colReader.getDouble(), 0.000001);
+ assertEquals(Float.MIN_VALUE, colReader.getFloat(), 0.000001);
assertTrue(reader.next());
- assertEquals(100, colReader.getDouble(), 0.000001);
+ assertEquals(100, colReader.getFloat(), 0.000001);
assertEquals("100.0", colReader.getAsString());
assertFalse(reader.next());
@@ -587,7 +587,7 @@ public class TestScalarAccessors extends SubOperatorTest {
@Test
public void testNullableFloat() {
- nullableDoubleTester(MinorType.FLOAT4);
+ nullableDoubleTester(MinorType.FLOAT8);
}
private void doubleArrayTester(MinorType type) {
@@ -636,7 +636,7 @@ public class TestScalarAccessors extends SubOperatorTest {
@Test
public void testFloatArray() {
- doubleArrayTester(MinorType.FLOAT4);
+ doubleArrayTester(MinorType.FLOAT8);
}
@Test
diff --git a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithSchema.java b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithSchema.java
index ac27d2f..e56fcfe 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithSchema.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/exec/store/easy/text/compliant/TestCsvWithSchema.java
@@ -1022,7 +1022,7 @@ public class TestCsvWithSchema extends BaseCsvTest {
"col_int integer not null default '10', " +
"col_bigint bigint not null default '10', " +
"col_double double not null default '10.5', " +
- "col_float float not null default '10.5', " +
+ "col_float float not null default '10.5f', " +
"col_var varchar not null default 'foo', " +
"col_boolean boolean not null default '1', " +
"col_interval interval not null default 'P10D', " +
@@ -1051,7 +1051,7 @@ public class TestCsvWithSchema extends BaseCsvTest {
LocalDate ld = new LocalDate(2019, 3, 28);
Instant ts = ld.toDateTime(lt).toInstant();
RowSet expected = new RowSetBuilder(client.allocator(), expectedSchema)
- .addRow(10, 10L, 10.5, 10.5D, "foo", true, new Period(0).plusDays(10),
+ .addRow(10, 10L, 10.5, 10.5f, "foo", true, new Period(0).plusDays(10),
lt, ld, ts, "1")
.build();
RowSetUtilities.verify(expected, actual);
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSetComparison.java b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSetComparison.java
index f084091..86e9e63 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSetComparison.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSetComparison.java
@@ -347,6 +347,12 @@ public class RowSetComparison {
// Double must be handled specially since BigDecimal cannot handle
// INF or NAN double values.
+ case FLOAT:
+ assertEquals(label, ec.getFloat(), ac.getFloat(), delta);
+ break;
+
+ // Double must be handled specially since BigDecimal cannot handle
+ // INF or NAN double values.
case DOUBLE:
assertEquals(label, ec.getDouble(), ac.getDouble(), delta);
break;
@@ -367,6 +373,8 @@ public class RowSetComparison {
switch (scalarReader.valueType()) {
case BYTES:
return ByteBuffer.wrap(scalarReader.getBytes());
+ case FLOAT:
+ return new BigDecimal(scalarReader.getFloat(), this.scale).stripTrailingZeros();
case DOUBLE:
return new BigDecimal(scalarReader.getDouble(), this.scale).stripTrailingZeros();
default:
diff --git a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSetUtilities.java b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSetUtilities.java
index 925130f..95758d1 100644
--- a/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSetUtilities.java
+++ b/exec/java-exec/src/test/java/org/apache/drill/test/rowSet/RowSetUtilities.java
@@ -88,6 +88,8 @@ public class RowSetUtilities {
switch (valueType) {
case BYTES:
return Integer.toHexString(value).getBytes();
+ case FLOAT:
+ return (float) value;
case DOUBLE:
return (double) value;
case INTEGER:
@@ -165,6 +167,9 @@ public class RowSetUtilities {
assertTrue(msg, Arrays.equals(expected, actual));
break;
}
+ case FLOAT:
+ assertEquals(msg, (float) expectedObj, (float) actualObj, 0.0001);
+ break;
case DOUBLE:
assertEquals(msg, (double) expectedObj, (double) actualObj, 0.0001);
break;
diff --git a/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd b/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd
index 50a4258..d30cf7f 100644
--- a/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd
+++ b/exec/vector/src/main/codegen/data/ValueVectorTypes.tdd
@@ -68,7 +68,7 @@
minor: [
{ class: "Int", valueHolder: "IntHolder"},
{ class: "UInt4", valueHolder: "UInt4Holder" },
- { class: "Float4", javaType: "float" , boxedType: "Float", accessorType: "double", accessorCast: "set",
+ { class: "Float4", javaType: "float" , boxedType: "Float", accessorType: "float", accessorCast: "set",
fields: [{name: "value", type: "float"}]},
{ class: "Time", javaType: "int", friendlyType: "LocalTime", accessorType: "int" },
{ class: "IntervalYear", javaType: "int", friendlyType: "Period" }
diff --git a/exec/vector/src/main/codegen/templates/ColumnAccessors.java b/exec/vector/src/main/codegen/templates/ColumnAccessors.java
index c6dc033..06a3d4c 100644
--- a/exec/vector/src/main/codegen/templates/ColumnAccessors.java
+++ b/exec/vector/src/main/codegen/templates/ColumnAccessors.java
@@ -448,6 +448,13 @@ public class ColumnAccessors {
}
@Override
+ public final void setFloat(final float value) {
+ // Does not catch overflow from
+ // double. See Math.round for details.
+ setLong(Math.round(value));
+ }
+
+ @Override
public final void setDouble(final double value) {
// Does not catch overflow from
// double. See Math.round for details.
@@ -463,7 +470,7 @@ public class ColumnAccessors {
throw InvalidConversionError.writeError(schema(), value, e);
}
}
- <#elseif drillType == "Float4" || drillType == "Float8">
+ <#elseif drillType == "Float8">
@Override
public final void setInt(final int value) {
@@ -476,9 +483,35 @@ public class ColumnAccessors {
}
@Override
+ public final void setFloat(final float value) {
+ setDouble(value);
+ }
+
+ @Override
public final void setDecimal(final BigDecimal value) {
setDouble(value.doubleValue());
}
+ <#elseif drillType == "Float4">
+
+ @Override
+ public final void setInt(final int value) {
+ setFloat(value);
+ }
+
+ @Override
+ public final void setLong(final long value) {
+ setFloat(value);
+ }
+
+ @Override
+ public final void setDouble(final double value) {
+ setFloat((float) value);
+ }
+
+ @Override
+ public final void setDecimal(final BigDecimal value) {
+ setFloat(value.floatValue());
+ }
<#elseif decimal>
@Override
@@ -495,6 +528,11 @@ public class ColumnAccessors {
public final void setDouble(final double value) {
setDecimal(BigDecimal.valueOf(value));
}
+
+ @Override
+ public final void setFloat(final float value) {
+ setDecimal(BigDecimal.valueOf(value));
+ }
<#if drillType == "VarDecimal">
@Override
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/record/metadata/PrimitiveColumnMetadata.java b/exec/vector/src/main/java/org/apache/drill/exec/record/metadata/PrimitiveColumnMetadata.java
index ad46473..256f55c 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/record/metadata/PrimitiveColumnMetadata.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/record/metadata/PrimitiveColumnMetadata.java
@@ -249,7 +249,7 @@ public class PrimitiveColumnMetadata extends AbstractColumnMetadata {
case BIGINT:
return Long.parseLong(value);
case FLOAT4:
- return (double) Float.parseFloat(value);
+ return Float.parseFloat(value);
case FLOAT8:
return Double.parseDouble(value);
case VARDECIMAL:
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ScalarReader.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ScalarReader.java
index 30d0c3f..8758e55 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ScalarReader.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ScalarReader.java
@@ -80,6 +80,7 @@ public interface ScalarReader extends ColumnReader {
int getInt();
boolean getBoolean();
long getLong();
+ float getFloat();
double getDouble();
String getString();
byte[] getBytes();
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ValueType.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ValueType.java
index 0dde3ba..00a2ec8 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ValueType.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ValueType.java
@@ -49,6 +49,11 @@ public enum ValueType {
LONG,
/**
+ * Type is set from a float: FLOAT4.
+ */
+ FLOAT,
+
+ /**
* Type is set from a double: FLOAT4 and FLOAT8.
*/
DOUBLE,
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ValueWriter.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ValueWriter.java
index 56105f9..1ec1fd1 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ValueWriter.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/ValueWriter.java
@@ -45,6 +45,7 @@ public interface ValueWriter {
void setBoolean(boolean value);
void setInt(int value);
void setLong(long value);
+ void setFloat(float value);
void setDouble(double value);
void setString(String value);
void appendBytes(byte[] value, int len);
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/reader/AbstractScalarReader.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/reader/AbstractScalarReader.java
index 1fe9c69..e5e221e 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/reader/AbstractScalarReader.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/reader/AbstractScalarReader.java
@@ -143,6 +143,11 @@ public abstract class AbstractScalarReader implements ScalarReader, ReaderEvents
}
@Override
+ public float getFloat() {
+ throw conversionError("double");
+ }
+
+ @Override
public double getDouble() {
throw conversionError("double");
}
@@ -194,6 +199,8 @@ public abstract class AbstractScalarReader implements ScalarReader, ReaderEvents
return getBytes();
case DECIMAL:
return getDecimal();
+ case FLOAT:
+ return getFloat();
case DOUBLE:
return getDouble();
case INTEGER:
@@ -240,6 +247,8 @@ public abstract class AbstractScalarReader implements ScalarReader, ReaderEvents
switch (extendedType()) {
case BYTES:
return AccessorUtilities.bytesToString(getBytes());
+ case FLOAT:
+ return Double.toString(getFloat());
case DOUBLE:
return Double.toString(getDouble());
case INTEGER:
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractFixedWidthWriter.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractFixedWidthWriter.java
index 58fbf8e..5e5fc1e 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractFixedWidthWriter.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractFixedWidthWriter.java
@@ -138,6 +138,17 @@ public abstract class AbstractFixedWidthWriter extends BaseScalarWriter {
}
@Override
+ public final void setFloat(final float value) {
+ try {
+ // Catches int overflow. Does not catch overflow from
+ // double. See Math.round for details.
+ setInt(Math.toIntExact(Math.round(value)));
+ } catch (final ArithmeticException e) {
+ throw InvalidConversionError.writeError(schema(), value, e);
+ }
+ }
+
+ @Override
public final void setDouble(final double value) {
try {
// Catches int overflow. Does not catch overflow from
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractScalarWriter.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractScalarWriter.java
index 5a0e8ba..1d52318 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractScalarWriter.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractScalarWriter.java
@@ -50,7 +50,7 @@ public abstract class AbstractScalarWriter implements ScalarWriter {
} else if (value instanceof Double) {
setDouble((Double) value);
} else if (value instanceof Float) {
- setDouble((Float) value);
+ setFloat((Float) value);
} else if (value instanceof BigDecimal) {
setDecimal((BigDecimal) value);
} else if (value instanceof Period) {
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractTupleWriter.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractTupleWriter.java
index 9cb674e..16155f4 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractTupleWriter.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/AbstractTupleWriter.java
@@ -139,7 +139,7 @@ public abstract class AbstractTupleWriter implements TupleWriter, WriterEvents {
* tuple writer. If no listener is bound, then an attempt to add a column
* throws an exception.
*/
- public static interface TupleWriterListener {
+ public interface TupleWriterListener {
ObjectWriter addColumn(TupleWriter tuple, ColumnMetadata column);
@@ -190,8 +190,9 @@ public abstract class AbstractTupleWriter implements TupleWriter, WriterEvents {
this.writers = writers;
}
+ @SuppressWarnings("unused")
protected AbstractTupleWriter(TupleMetadata schema) {
- this(schema, new ArrayList<AbstractObjectWriter>());
+ this(schema, new ArrayList<>());
}
@Override
@@ -201,8 +202,8 @@ public abstract class AbstractTupleWriter implements TupleWriter, WriterEvents {
vectorIndex = index;
this.childIndex = childIndex;
- for (int i = 0; i < writers.size(); i++) {
- writers.get(i).events().bindIndex(childIndex);
+ for (AbstractObjectWriter writer : writers) {
+ writer.events().bindIndex(childIndex);
}
}
@@ -239,8 +240,7 @@ public abstract class AbstractTupleWriter implements TupleWriter, WriterEvents {
@Override
public boolean isProjected(String columnName) {
- return listener == null ? true
- : listener.isProjected(columnName);
+ return listener == null || listener.isProjected(columnName);
}
@Override
@@ -288,8 +288,8 @@ public abstract class AbstractTupleWriter implements TupleWriter, WriterEvents {
public void startWrite() {
assert state == State.IDLE;
state = State.IN_WRITE;
- for (int i = 0; i < writers.size(); i++) {
- writers.get(i).events().startWrite();
+ for (AbstractObjectWriter writer : writers) {
+ writer.events().startWrite();
}
}
@@ -300,16 +300,16 @@ public abstract class AbstractTupleWriter implements TupleWriter, WriterEvents {
assert state == State.IN_WRITE;
state = State.IN_ROW;
- for (int i = 0; i < writers.size(); i++) {
- writers.get(i).events().startRow();
+ for (AbstractObjectWriter writer : writers) {
+ writer.events().startRow();
}
}
@Override
public void endArrayValue() {
assert state == State.IN_ROW;
- for (int i = 0; i < writers.size(); i++) {
- writers.get(i).events().endArrayValue();
+ for (AbstractObjectWriter writer : writers) {
+ writer.events().endArrayValue();
}
}
@@ -325,16 +325,16 @@ public abstract class AbstractTupleWriter implements TupleWriter, WriterEvents {
// the current row.
assert state == State.IN_ROW;
- for (int i = 0; i < writers.size(); i++) {
- writers.get(i).events().restartRow();
+ for (AbstractObjectWriter writer : writers) {
+ writer.events().restartRow();
}
}
@Override
public void saveRow() {
assert state == State.IN_ROW;
- for (int i = 0; i < writers.size(); i++) {
- writers.get(i).events().saveRow();
+ for (AbstractObjectWriter writer : writers) {
+ writer.events().saveRow();
}
state = State.IN_WRITE;
}
@@ -345,8 +345,8 @@ public abstract class AbstractTupleWriter implements TupleWriter, WriterEvents {
// Rollover can only happen while a row is in progress.
assert state == State.IN_ROW;
- for (int i = 0; i < writers.size(); i++) {
- writers.get(i).events().preRollover();
+ for (AbstractObjectWriter writer : writers) {
+ writer.events().preRollover();
}
}
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/BaseScalarWriter.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/BaseScalarWriter.java
index 56af1dc..7938b29 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/BaseScalarWriter.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/BaseScalarWriter.java
@@ -241,6 +241,11 @@ public abstract class BaseScalarWriter extends AbstractScalarWriterImpl {
}
@Override
+ public void setFloat(float value) {
+ throw conversionError("float");
+ }
+
+ @Override
public void setDouble(double value) {
throw conversionError("double");
}
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/NullableScalarWriter.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/NullableScalarWriter.java
index 4d5353d..02343df 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/NullableScalarWriter.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/NullableScalarWriter.java
@@ -155,6 +155,13 @@ public class NullableScalarWriter extends AbstractScalarWriterImpl {
}
@Override
+ public void setFloat(float value) {
+ baseWriter.setFloat(value);
+ isSetWriter.setInt(1);
+ writerIndex.nextElement();
+ }
+
+ @Override
public void setDouble(double value) {
baseWriter.setDouble(value);
isSetWriter.setInt(1);
diff --git a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/dummy/DummyScalarWriter.java b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/dummy/DummyScalarWriter.java
index 85a7fe1..5d738c3 100644
--- a/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/dummy/DummyScalarWriter.java
+++ b/exec/vector/src/main/java/org/apache/drill/exec/vector/accessor/writer/dummy/DummyScalarWriter.java
@@ -64,6 +64,9 @@ public class DummyScalarWriter extends AbstractScalarWriterImpl {
public void setLong(long value) { }
@Override
+ public void setFloat(float value) { }
+
+ @Override
public void setDouble(double value) { }
@Override
diff --git a/pom.xml b/pom.xml
index 1e72b15..75e83c1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -58,7 +58,7 @@
avoid_bad_dependencies plugin found in the file.
-->
<calcite.groupId>com.github.vvysotskyi.drill-calcite</calcite.groupId>
- <calcite.version>1.21.0-drill-r0</calcite.version>
+ <calcite.version>1.21.0-drill-r1</calcite.version>
<avatica.version>1.15.0</avatica.version>
<janino.version>3.0.11</janino.version>
<sqlline.version>1.9.0</sqlline.version>
@@ -398,7 +398,6 @@
<exclude>**/*.csvh-test</exclude>
<exclude>**/*.tsv</exclude>
<exclude>**/*.txt</exclude>
- <exclude>**/*.yaml</exclude>
<exclude>**/*.ssv</exclude>
<exclude>**/.buildpath</exclude>
<exclude>**/target/**</exclude>
@@ -699,7 +698,6 @@
<exclude>**/*.md</exclude>
<exclude>**/*.eps</exclude>
<exclude>**/*.json</exclude>
- <exclude>**/*.yaml</exclude>
<exclude>**/*.seq</exclude>
<exclude>**/*.parquet</exclude>
<exclude>**/*.avro</exclude>