You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@drill.apache.org by dz...@apache.org on 2022/12/19 08:43:37 UTC
[drill] branch 1.20 updated (3a23665775 -> 1a20a931a3)
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a change to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
from 3a23665775 DRILL-8282: Update hadoop.dll and winutils.exe to 3.2.4. (#2635)
new fb1be8d204 DRILL-8291: Allow Case Sensitive Filters in HTTP Plugin (#2639)
new bb8524116a DRILL-8293: Add a docker-compose file to run Drill in cluster mode (#2640)
new 1cfedd677a DRILL-8295: Probable resource leak in the HTTP storage plugin (#2641)
new 67f7b74e57 DRILL-8300: Upgrade to snakeyaml 1.31 due to cve (#2643)
new 3b949bc31a DRILL-8307: Ensure thread safety in the Druid plugin HTTP client (#2650)
new e82d308fad DRILL-8308: POI 5.2.3 (#2651)
new 12e47c8dcb DRILL-8190: Fix mongo project pushdown for queries with joins (#2652)
new 7400628512 DRILL-8321: Change kafka_2.13 dependency scope to test (#2662)
new ee8e30bf8a DRILL-8323: upgrade commons-text to 1.10.0 (#2663)
new 1a20a931a3 DRILL-8324: Sizeof refactor (#2665)
The 10 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
contrib/format-excel/pom.xml | 4 +-
.../exec/store/druid/rest/DruidAdminClient.java | 30 +-
.../exec/store/druid/rest/DruidQueryClient.java | 32 +-
.../drill/exec/store/druid/rest/RestClient.java | 20 +-
.../exec/store/druid/rest/RestClientWrapper.java | 48 +--
.../drill/exec/store/druid/DruidTestBase.java | 4 +-
.../{DruidTestSuit.java => DruidTestSuite.java} | 4 +-
.../drill/exec/store/druid/TestDataGenerator.java | 85 ++--
.../store/druid/rest/DruidQueryClientTest.java | 37 +-
contrib/storage-http/README.md | 7 +-
.../drill/exec/store/http/HttpApiConfig.java | 19 +-
.../drill/exec/store/http/HttpBatchReader.java | 1 +
.../exec/store/http/HttpPushDownListener.java | 9 +-
.../drill/exec/store/http/HttpXMLBatchReader.java | 2 +-
.../drill/exec/store/http/util/SimpleHttp.java | 20 +-
.../jdbc/JdbcIntermediatePrelConverterRule.java | 3 +-
contrib/storage-kafka/pom.xml | 1 +
.../drill/exec/store/kafka/MessageIterator.java | 2 +-
.../store/mongo/plan/MongoPluginImplementor.java | 5 +-
.../exec/store/mongo/TestMongoProjectPushDown.java | 22 +
contrib/storage-opentsdb/pom.xml | 5 -
.../exec/store/openTSDB/OpenTSDBGroupScan.java | 1 -
.../drill/exec/store/openTSDB/SizeEstimator.java | 457 +++++++++++++++++++++
.../{ => exec}/store/openTSDB/TestDataHolder.java | 2 +-
.../store/openTSDB/TestOpenTSDBPlugin.java | 31 +-
.../exec/store/openTSDB/TestSizeEstimator.java | 87 ++++
.../PhoenixIntermediatePrelConverterRule.java | 3 +-
.../exec/udfs/TestStringDistanceFunctions.java | 6 +-
.../docker-cluster-mode/Dockerfile | 14 +-
distribution/docker-cluster-mode/README | 25 ++
.../docker-cluster-mode/docker-compose.yml | 49 ++-
.../drill-override.conf | 2 +-
.../docker-cluster-mode/logback.xml | 28 +-
.../exec/planner/cost/DrillRelMdRowCount.java | 12 +-
.../enumerable/ColumnConverterFactoryProvider.java | 2 +
.../enumerable/DynamicTypeResolverBuilder.java | 47 +++
.../store/enumerable/EnumerableBatchCreator.java | 2 +-
.../exec/store/enumerable/EnumerableSubScan.java | 2 +-
.../EnumerableIntermediatePrelConverterRule.java | 3 +-
.../exec/store/enumerable/plan/VertexDrel.java | 18 +
.../drill/exec/store/http/oauth/OAuthUtils.java | 18 +-
.../exec/store/plan/rel/PluginAggregateRel.java | 2 +-
.../store/plan/rel/StoragePluginTableScan.java | 20 +
.../rule/PluginIntermediatePrelConverterRule.java | 3 +-
.../java/org/apache/drill/exec/util/Utilities.java | 30 +-
pom.xml | 4 +-
46 files changed, 983 insertions(+), 245 deletions(-)
rename contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/{DruidTestSuit.java => DruidTestSuite.java} (98%)
create mode 100644 contrib/storage-opentsdb/src/main/java/org/apache/drill/exec/store/openTSDB/SizeEstimator.java
rename contrib/storage-opentsdb/src/test/java/org/apache/drill/{ => exec}/store/openTSDB/TestDataHolder.java (99%)
rename contrib/storage-opentsdb/src/test/java/org/apache/drill/{ => exec}/store/openTSDB/TestOpenTSDBPlugin.java (84%)
create mode 100644 contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestSizeEstimator.java
copy exec/vector/src/main/codegen/config.fmpp => distribution/docker-cluster-mode/Dockerfile (79%)
create mode 100644 distribution/docker-cluster-mode/README
copy contrib/format-maprdb/src/main/resources/drill-module.conf => distribution/docker-cluster-mode/docker-compose.yml (53%)
copy distribution/{src/main/resources => docker-cluster-mode}/drill-override.conf (97%)
copy contrib/format-maprdb/src/test/resources/core-site.xml => distribution/docker-cluster-mode/logback.xml (58%)
create mode 100644 exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/DynamicTypeResolverBuilder.java
[drill] 07/10: DRILL-8190: Fix mongo project pushdown for queries with joins (#2652)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 12e47c8dcba8414b3b8e995e9757f877a390c6f0
Author: Volodymyr Vysotskyi <vv...@gmail.com>
AuthorDate: Mon Sep 26 14:27:27 2022 +0800
DRILL-8190: Fix mongo project pushdown for queries with joins (#2652)
---
.../jdbc/JdbcIntermediatePrelConverterRule.java | 3 +-
.../store/mongo/plan/MongoPluginImplementor.java | 5 +--
.../exec/store/mongo/TestMongoProjectPushDown.java | 22 ++++++++++
.../PhoenixIntermediatePrelConverterRule.java | 3 +-
.../exec/planner/cost/DrillRelMdRowCount.java | 12 +++++-
.../enumerable/ColumnConverterFactoryProvider.java | 2 +
.../enumerable/DynamicTypeResolverBuilder.java | 47 ++++++++++++++++++++++
.../store/enumerable/EnumerableBatchCreator.java | 2 +-
.../exec/store/enumerable/EnumerableSubScan.java | 2 +-
.../EnumerableIntermediatePrelConverterRule.java | 3 +-
.../exec/store/enumerable/plan/VertexDrel.java | 18 +++++++++
.../exec/store/plan/rel/PluginAggregateRel.java | 2 +-
.../store/plan/rel/StoragePluginTableScan.java | 20 +++++++++
.../rule/PluginIntermediatePrelConverterRule.java | 3 +-
.../java/org/apache/drill/exec/util/Utilities.java | 30 +++++++-------
15 files changed, 147 insertions(+), 27 deletions(-)
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 a66888f2df..81cf232342 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
@@ -25,6 +25,7 @@ import org.apache.calcite.rel.RelNode;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillRelFactories;
import org.apache.drill.exec.planner.logical.RelOptHelper;
+import org.apache.drill.exec.planner.physical.DrillDistributionTrait;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.store.enumerable.plan.VertexDrel;
@@ -48,7 +49,7 @@ class JdbcIntermediatePrelConverterRule extends RelOptRule {
VertexDrel in = call.rel(0);
RelNode jdbcIntermediatePrel = new JdbcIntermediatePrel(
in.getCluster(),
- in.getTraitSet().replace(outTrait),
+ in.getTraitSet().replace(outTrait).plus(DrillDistributionTrait.SINGLETON),
in.getInput(0));
call.transformTo(jdbcIntermediatePrel);
}
diff --git a/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/plan/MongoPluginImplementor.java b/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/plan/MongoPluginImplementor.java
index b55fa6b173..774a23265b 100644
--- a/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/plan/MongoPluginImplementor.java
+++ b/contrib/storage-mongo/src/main/java/org/apache/drill/exec/store/mongo/plan/MongoPluginImplementor.java
@@ -55,7 +55,6 @@ import org.apache.drill.exec.store.plan.rel.PluginProjectRel;
import org.apache.drill.exec.store.plan.rel.PluginSortRel;
import org.apache.drill.exec.store.plan.rel.PluginUnionRel;
import org.apache.drill.exec.store.plan.rel.StoragePluginTableScan;
-import org.apache.drill.exec.util.Utilities;
import org.bson.BsonDocument;
import org.bson.BsonElement;
import org.bson.BsonInt32;
@@ -220,8 +219,8 @@ public class MongoPluginImplementor extends AbstractPluginImplementor {
}
@Override
- public void implement(StoragePluginTableScan scan) throws IOException {
- groupScan = (MongoGroupScan) Utilities.getDrillTable(scan.getTable()).getGroupScan();
+ public void implement(StoragePluginTableScan scan) {
+ groupScan = (MongoGroupScan) scan.getGroupScan();
operations = this.groupScan.getScanSpec().getOperations().stream()
.map(BsonDocument::parse)
.collect(Collectors.toList());
diff --git a/contrib/storage-mongo/src/test/java/org/apache/drill/exec/store/mongo/TestMongoProjectPushDown.java b/contrib/storage-mongo/src/test/java/org/apache/drill/exec/store/mongo/TestMongoProjectPushDown.java
index a691443311..372ec6d5e4 100644
--- a/contrib/storage-mongo/src/test/java/org/apache/drill/exec/store/mongo/TestMongoProjectPushDown.java
+++ b/contrib/storage-mongo/src/test/java/org/apache/drill/exec/store/mongo/TestMongoProjectPushDown.java
@@ -101,4 +101,26 @@ public class TestMongoProjectPushDown extends MongoTestBase {
.go();
}
+ @Test // DRILL-8190
+ public void testProjectWithJoin() throws Exception {
+ String query = "SELECT sum(s1.sales) s1_sales,\n" +
+ "sum(s2.sales) s2_sales\n" +
+ "FROM mongo.%s.`%s` s1\n" +
+ "JOIN mongo.%s.`%s` s2 ON s1._id = s2._id";
+
+ queryBuilder()
+ .sql(query, DONUTS_DB, DONUTS_COLLECTION, DONUTS_DB, DONUTS_COLLECTION)
+ .planMatcher()
+ .include("columns=\\[`_id`, `sales`]")
+ .exclude("columns=\\[`\\*\\*`")
+ .match();
+
+ testBuilder()
+ .sqlQuery(query, DONUTS_DB, DONUTS_COLLECTION, DONUTS_DB, DONUTS_COLLECTION)
+ .unOrdered()
+ .baselineColumns("s1_sales", "s2_sales")
+ .baselineValues(1194L, 1194L)
+ .go();
+ }
+
}
diff --git a/contrib/storage-phoenix/src/main/java/org/apache/drill/exec/store/phoenix/rules/PhoenixIntermediatePrelConverterRule.java b/contrib/storage-phoenix/src/main/java/org/apache/drill/exec/store/phoenix/rules/PhoenixIntermediatePrelConverterRule.java
index c5eaaf1dda..7d6a88d0cd 100644
--- a/contrib/storage-phoenix/src/main/java/org/apache/drill/exec/store/phoenix/rules/PhoenixIntermediatePrelConverterRule.java
+++ b/contrib/storage-phoenix/src/main/java/org/apache/drill/exec/store/phoenix/rules/PhoenixIntermediatePrelConverterRule.java
@@ -25,6 +25,7 @@ import org.apache.calcite.rel.RelNode;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillRelFactories;
import org.apache.drill.exec.planner.logical.RelOptHelper;
+import org.apache.drill.exec.planner.physical.DrillDistributionTrait;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.store.enumerable.plan.VertexDrel;
@@ -48,7 +49,7 @@ final class PhoenixIntermediatePrelConverterRule extends RelOptRule {
VertexDrel in = call.rel(0);
RelNode intermediatePrel = new PhoenixIntermediatePrel(
in.getCluster(),
- in.getTraitSet().replace(outTrait),
+ in.getTraitSet().replace(outTrait).plus(DrillDistributionTrait.SINGLETON),
in.getInput(0));
call.transformTo(intermediatePrel);
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdRowCount.java b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdRowCount.java
index 0bfb70ae90..abcc83672a 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdRowCount.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/planner/cost/DrillRelMdRowCount.java
@@ -27,6 +27,7 @@ import org.apache.calcite.rel.metadata.ReflectiveRelMetadataProvider;
import org.apache.calcite.rel.metadata.RelMdRowCount;
import org.apache.calcite.rel.metadata.RelMetadataProvider;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.schema.Table;
import org.apache.calcite.util.BuiltInMethod;
import org.apache.calcite.util.ImmutableBitSet;
import org.apache.drill.exec.planner.common.DrillLimitRelBase;
@@ -45,6 +46,8 @@ public class DrillRelMdRowCount extends RelMdRowCount{
public static final RelMetadataProvider SOURCE = ReflectiveRelMetadataProvider.reflectiveSource(BuiltInMethod.ROW_COUNT.method, INSTANCE);
+ private static final Double DEFAULT_SCAN_ROW_COUNT = 1e9;
+
@Override
public Double getRowCount(Aggregate rel, RelMetadataQuery mq) {
ImmutableBitSet groupKey = ImmutableBitSet.range(rel.getGroupCount());
@@ -96,7 +99,14 @@ public class DrillRelMdRowCount extends RelMdRowCount{
PlannerSettings settings = PrelUtil.getSettings(rel.getCluster());
// If guessing, return selectivity from RelMDRowCount
if (DrillRelOptUtil.guessRows(rel)) {
- return super.getRowCount(rel, mq);
+ if (rel instanceof DrillScanRelBase
+ || rel.getTable().unwrap(Table.class).getStatistic().getRowCount() != null) {
+ return super.getRowCount(rel, mq);
+ } else {
+ // if table doesn't have row count statistics, return large row count
+ // to make sure that limit will be pushed down
+ return DEFAULT_SCAN_ROW_COUNT;
+ }
}
// Return rowcount from statistics, if available. Otherwise, delegate to parent.
try {
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/ColumnConverterFactoryProvider.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/ColumnConverterFactoryProvider.java
index 28aec71d0f..b91e4cbb13 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/ColumnConverterFactoryProvider.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/ColumnConverterFactoryProvider.java
@@ -18,10 +18,12 @@
package org.apache.drill.exec.store.enumerable;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
+import com.fasterxml.jackson.databind.annotation.JsonTypeResolver;
import org.apache.drill.exec.record.ColumnConverterFactory;
import org.apache.drill.exec.record.metadata.TupleMetadata;
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME)
+@JsonTypeResolver(DynamicTypeResolverBuilder.class)
public interface ColumnConverterFactoryProvider {
ColumnConverterFactory getFactory(TupleMetadata schema);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/DynamicTypeResolverBuilder.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/DynamicTypeResolverBuilder.java
new file mode 100644
index 0000000000..64fa9864cf
--- /dev/null
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/DynamicTypeResolverBuilder.java
@@ -0,0 +1,47 @@
+/*
+ * 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.databind.DeserializationConfig;
+import com.fasterxml.jackson.databind.JavaType;
+import com.fasterxml.jackson.databind.jsontype.NamedType;
+import com.fasterxml.jackson.databind.jsontype.TypeDeserializer;
+import com.fasterxml.jackson.databind.jsontype.impl.StdTypeResolverBuilder;
+import org.reflections.Reflections;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.stream.Collectors;
+
+public class DynamicTypeResolverBuilder extends StdTypeResolverBuilder {
+
+ @Override
+ public TypeDeserializer buildTypeDeserializer(DeserializationConfig config,
+ JavaType baseType, Collection<NamedType> subtypes) {
+
+ Reflections reflections = new Reflections("org.apache.drill.exec.store");
+ @SuppressWarnings("unchecked")
+ Class<Object> rawClass = (Class<Object>) baseType.getRawClass();
+ List<NamedType> dynamicSubtypes = reflections.getSubTypesOf(rawClass).stream()
+ .map(NamedType::new)
+ .collect(Collectors.toList());
+ dynamicSubtypes.addAll(subtypes);
+
+ return super.buildTypeDeserializer(config, baseType, dynamicSubtypes);
+ }
+}
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
index 2dec45a61a..9930484a2b 100644
--- 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
@@ -61,7 +61,7 @@ public class EnumerableBatchCreator implements BatchCreator<EnumerableSubScan> {
builder.providedSchema(subScan.getSchema());
ManagedReader<SchemaNegotiator> reader = new EnumerableRecordReader(subScan.getColumns(),
- subScan.getFieldsMap(), subScan.getCode(), subScan.getSchemaPath(), subScan.factoryProvider());
+ subScan.getFieldsMap(), subScan.getCode(), subScan.getSchemaPath(), subScan.getConverterFactoryProvider());
ManagedScanFramework.ReaderFactory readerFactory = new BasicScanFactory(Collections.singletonList(reader).iterator());
builder.setReaderFactory(readerFactory);
builder.nullType(Types.optional(TypeProtos.MinorType.VARCHAR));
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
index 4476be8c53..0c7245b6cc 100644
--- 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
@@ -79,7 +79,7 @@ public class EnumerableSubScan extends AbstractSubScan {
return schemaPath;
}
- public ColumnConverterFactoryProvider factoryProvider() {
+ public ColumnConverterFactoryProvider getConverterFactoryProvider() {
return converterFactoryProvider;
}
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrelConverterRule.java
index 7272a36bf6..c5ede3a22e 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/EnumerableIntermediatePrelConverterRule.java
@@ -25,6 +25,7 @@ import org.apache.calcite.rel.RelNode;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillRelFactories;
import org.apache.drill.exec.planner.logical.RelOptHelper;
+import org.apache.drill.exec.planner.physical.DrillDistributionTrait;
import org.apache.drill.exec.planner.physical.Prel;
public class EnumerableIntermediatePrelConverterRule extends RelOptRule {
@@ -48,7 +49,7 @@ public class EnumerableIntermediatePrelConverterRule extends RelOptRule {
VertexDrel in = call.rel(0);
RelNode intermediatePrel = new EnumerableIntermediatePrel(
in.getCluster(),
- in.getTraitSet().replace(outTrait),
+ in.getTraitSet().replace(outTrait).plus(DrillDistributionTrait.SINGLETON),
in.getInput(0),
context);
call.transformTo(intermediatePrel);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrel.java
index edbc5912fb..d202ebdb89 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/enumerable/plan/VertexDrel.java
@@ -18,15 +18,22 @@
package org.apache.drill.exec.store.enumerable.plan;
import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptCost;
+import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.RelNode;
import org.apache.calcite.rel.SingleRel;
+import org.apache.calcite.rel.metadata.RelMetadataQuery;
import org.apache.drill.common.logical.data.LogicalOperator;
import org.apache.drill.exec.planner.logical.DrillImplementor;
import org.apache.drill.exec.planner.logical.DrillRel;
+import org.apache.drill.exec.util.Utilities;
+import org.checkerframework.checker.nullness.qual.Nullable;
import java.util.List;
+import static org.apache.drill.exec.planner.logical.DrillScanRel.STAR_COLUMN_COST;
+
/**
* 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.
@@ -51,4 +58,15 @@ public class VertexDrel extends SingleRel implements DrillRel {
public LogicalOperator implement(DrillImplementor implementor) {
throw new UnsupportedOperationException();
}
+
+ @Override
+ public @Nullable RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+ double rowCount = estimateRowCount(mq);
+ double columnCount = Utilities.isStarQuery(getRowType()) ? STAR_COLUMN_COST : getRowType().getFieldCount();
+ double valueCount = rowCount * columnCount;
+ // columns count is considered during cost calculation to make preferable plans
+ // with pushed plugin project operators since in the opposite case planner wouldn't consider
+ // a plan with additional plugin projection that reduces columns as better than a plan without it
+ return planner.getCostFactory().makeCost(rowCount, valueCount, 0).multiplyBy(0.1);
+ }
}
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rel/PluginAggregateRel.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rel/PluginAggregateRel.java
index 02885e90ab..3c7f115843 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rel/PluginAggregateRel.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rel/PluginAggregateRel.java
@@ -51,7 +51,7 @@ public class PluginAggregateRel extends DrillAggregateRelBase implements PluginR
@Override
public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
- return super.computeSelfCost(planner, mq).multiplyBy(0.1);
+ return super.computeLogicalAggCost(planner, mq).multiplyBy(0.1);
}
@Override
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rel/StoragePluginTableScan.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rel/StoragePluginTableScan.java
index ce8ce9ec98..3e7d07c5ce 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rel/StoragePluginTableScan.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rel/StoragePluginTableScan.java
@@ -18,6 +18,8 @@
package org.apache.drill.exec.store.plan.rel;
import org.apache.calcite.plan.RelOptCluster;
+import org.apache.calcite.plan.RelOptCost;
+import org.apache.calcite.plan.RelOptPlanner;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.plan.RelTraitSet;
import org.apache.calcite.rel.RelWriter;
@@ -27,11 +29,14 @@ import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.exec.physical.base.GroupScan;
import org.apache.drill.exec.planner.common.DrillScanRelBase;
import org.apache.drill.exec.store.plan.PluginImplementor;
+import org.apache.drill.exec.util.Utilities;
import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;
+import static org.apache.drill.exec.planner.logical.DrillScanRel.STAR_COLUMN_COST;
+
/**
* Storage plugin table scan rel implementation.
*/
@@ -80,6 +85,21 @@ public class StoragePluginTableScan extends DrillScanRelBase implements PluginRe
return implementor.canImplement(this);
}
+ @Override
+ public RelOptCost computeSelfCost(RelOptPlanner planner, RelMetadataQuery mq) {
+ List<SchemaPath> columns = groupScan.getColumns();
+ // column count should be adjusted to consider the case of projecting nested columns,
+ // such a scan should be preferable compared to the scan where root columns are projected only
+ double columnCount = Utilities.isStarQuery(columns)
+ ? STAR_COLUMN_COST
+ : Math.pow(getRowType().getFieldCount(), 2) / Math.max(columns.size(), 1);
+
+ double rowCount = estimateRowCount(mq);
+ double valueCount = rowCount * columnCount;
+
+ return planner.getCostFactory().makeCost(rowCount, valueCount, 0).multiplyBy(0.1);
+ }
+
private static List<SchemaPath> getColumns(RelDataType rowType) {
return rowType.getFieldList().stream()
.map(filed -> filed.isDynamicStar()
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginIntermediatePrelConverterRule.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginIntermediatePrelConverterRule.java
index 279241efea..a13dc25e32 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginIntermediatePrelConverterRule.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/plan/rule/PluginIntermediatePrelConverterRule.java
@@ -25,6 +25,7 @@ import org.apache.calcite.rel.RelNode;
import org.apache.drill.exec.planner.logical.DrillRel;
import org.apache.drill.exec.planner.logical.DrillRelFactories;
import org.apache.drill.exec.planner.logical.RelOptHelper;
+import org.apache.drill.exec.planner.physical.DrillDistributionTrait;
import org.apache.drill.exec.planner.physical.Prel;
import org.apache.drill.exec.store.enumerable.plan.VertexDrel;
import org.apache.drill.exec.store.plan.PluginImplementor;
@@ -53,7 +54,7 @@ public class PluginIntermediatePrelConverterRule extends RelOptRule {
VertexDrel in = call.rel(0);
RelNode intermediatePrel = new PluginIntermediatePrel(
in.getCluster(),
- in.getTraitSet().replace(outTrait),
+ in.getTraitSet().replace(outTrait).plus(DrillDistributionTrait.SINGLETON),
in.getInput(0),
implementorFactory);
call.transformTo(intermediatePrel);
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java b/exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java
index 87f2201804..1d160d07ed 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/util/Utilities.java
@@ -21,6 +21,7 @@ import java.util.Collection;
import org.apache.calcite.plan.RelOptTable;
import org.apache.calcite.rel.metadata.JaninoRelMetadataProvider;
import org.apache.calcite.rel.metadata.RelMetadataQuery;
+import org.apache.calcite.rel.type.RelDataType;
import org.apache.calcite.rex.RexLiteral;
import org.apache.drill.common.expression.PathSegment;
import org.apache.drill.common.expression.SchemaPath;
@@ -33,8 +34,6 @@ import org.apache.drill.exec.proto.BitControl.QueryContextInformation;
import org.apache.drill.exec.proto.ExecProtos;
import org.apache.drill.exec.proto.helper.QueryIdHelper;
import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
-import org.apache.drill.shaded.guava.com.google.common.base.Predicate;
-import org.apache.drill.shaded.guava.com.google.common.collect.Iterables;
public class Utilities {
@@ -52,17 +51,13 @@ public class Utilities {
int majorFragmentId = handle.getMajorFragmentId();
int minorFragmentId = handle.getMinorFragmentId();
- String fileName = String.format("%s//%s_%s_%s_%s", location, qid, majorFragmentId, minorFragmentId, tag);
-
- return fileName;
+ return String.format("%s//%s_%s_%s_%s", location, qid, majorFragmentId, minorFragmentId, tag);
}
/**
* Create {@link org.apache.drill.exec.proto.BitControl.QueryContextInformation} with given <i>defaultSchemaName</i>. Rest of the members of the
* QueryContextInformation is derived from the current state of the process.
*
- * @param defaultSchemaName
- * @param sessionId
* @return A {@link org.apache.drill.exec.proto.BitControl.QueryContextInformation} with given <i>defaultSchemaName</i>.
*/
public static QueryContextInformation createQueryContextInfo(final String defaultSchemaName,
@@ -82,22 +77,25 @@ public class Utilities {
* @return The Drill version.
*/
public static String getDrillVersion() {
- String v = Utilities.class.getPackage().getImplementationVersion();
- return v;
+ return Utilities.class.getPackage().getImplementationVersion();
}
/**
* Return true if list of schema path has star column.
- * @param projected
+ *
* @return True if the list of {@link org.apache.drill.common.expression.SchemaPath}s has star column.
*/
public static boolean isStarQuery(Collection<SchemaPath> projected) {
- return Iterables.tryFind(Preconditions.checkNotNull(projected, COL_NULL_ERROR), new Predicate<SchemaPath>() {
- @Override
- public boolean apply(SchemaPath path) {
- return Preconditions.checkNotNull(path).equals(SchemaPath.STAR_COLUMN);
- }
- }).isPresent();
+ return Preconditions.checkNotNull(projected, COL_NULL_ERROR).stream()
+ .anyMatch(SchemaPath::isDynamicStar);
+ }
+
+ /**
+ * Return true if the row type has star column.
+ */
+ public static boolean isStarQuery(RelDataType projected) {
+ return projected.getFieldNames().stream()
+ .anyMatch(SchemaPath.DYNAMIC_STAR::equals);
}
/**
[drill] 08/10: DRILL-8321: Change kafka_2.13 dependency scope to test (#2662)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 7400628512099df77e1382db5da2121ab61d386a
Author: Rymar Maksym <ri...@gmail.com>
AuthorDate: Sun Oct 2 05:58:43 2022 +0300
DRILL-8321: Change kafka_2.13 dependency scope to test (#2662)
(cherry picked from commit 88f3c684bb64385c9136ba2cf43efce9578109c1)
---
contrib/storage-kafka/pom.xml | 1 +
.../main/java/org/apache/drill/exec/store/kafka/MessageIterator.java | 2 +-
2 files changed, 2 insertions(+), 1 deletion(-)
diff --git a/contrib/storage-kafka/pom.xml b/contrib/storage-kafka/pom.xml
index c4e021cf5f..4e12a11dd1 100644
--- a/contrib/storage-kafka/pom.xml
+++ b/contrib/storage-kafka/pom.xml
@@ -81,6 +81,7 @@
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_2.13</artifactId>
<version>${kafka.version}</version>
+ <scope>test</scope>
<exclusions>
<exclusion>
<groupId>org.slf4j</groupId>
diff --git a/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/MessageIterator.java b/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/MessageIterator.java
index 9827298257..7282147ee8 100644
--- a/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/MessageIterator.java
+++ b/contrib/storage-kafka/src/main/java/org/apache/drill/exec/store/kafka/MessageIterator.java
@@ -28,13 +28,13 @@ import org.apache.drill.exec.ExecConstants;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.drill.shaded.guava.com.google.common.base.Stopwatch;
-import kafka.common.KafkaException;
public class MessageIterator implements Iterator<ConsumerRecord<byte[], byte[]>>, AutoCloseable {
[drill] 04/10: DRILL-8300: Upgrade to snakeyaml 1.31 due to cve (#2643)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 67f7b74e576b9227369233b74f70f5655d8a615a
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Thu Sep 8 09:19:31 2022 +0100
DRILL-8300: Upgrade to snakeyaml 1.31 due to cve (#2643)
---
pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/pom.xml b/pom.xml
index cd134b1623..7c43b54ea1 100644
--- a/pom.xml
+++ b/pom.xml
@@ -124,7 +124,7 @@
<libthrift.version>0.14.0</libthrift.version>
<derby.version>10.14.2.0</derby.version>
<commons.cli.version>1.4</commons.cli.version>
- <snakeyaml.version>1.26</snakeyaml.version>
+ <snakeyaml.version>1.31</snakeyaml.version>
<commons.lang3.version>3.10</commons.lang3.version>
<testcontainers.version>1.16.3</testcontainers.version>
<typesafe.config.version>1.0.0</typesafe.config.version>
[drill] 06/10: DRILL-8308: POI 5.2.3 (#2651)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit e82d308fadd08baee4dbc63be3949849d6e955e4
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Mon Sep 19 20:31:34 2022 +0800
DRILL-8308: POI 5.2.3 (#2651)
---
contrib/format-excel/pom.xml | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
diff --git a/contrib/format-excel/pom.xml b/contrib/format-excel/pom.xml
index 9fb10fb520..6fb0821f2e 100644
--- a/contrib/format-excel/pom.xml
+++ b/contrib/format-excel/pom.xml
@@ -31,7 +31,7 @@
<name>Drill : Contrib : Format : Excel</name>
<properties>
- <poi.version>5.2.2</poi.version>
+ <poi.version>5.2.3</poi.version>
</properties>
<dependencies>
<dependency>
[drill] 03/10: DRILL-8295: Probable resource leak in the HTTP storage plugin (#2641)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 1cfedd677a4aeeea8e02e9787f4d994739cf62dc
Author: James Turton <91...@users.noreply.github.com>
AuthorDate: Fri Sep 9 05:34:13 2022 +0800
DRILL-8295: Probable resource leak in the HTTP storage plugin (#2641)
---
.../apache/drill/exec/store/http/HttpBatchReader.java | 1 +
.../drill/exec/store/http/HttpXMLBatchReader.java | 2 +-
.../apache/drill/exec/store/http/util/SimpleHttp.java | 16 ++++++++++------
.../apache/drill/exec/store/http/oauth/OAuthUtils.java | 18 ++++++++++--------
4 files changed, 22 insertions(+), 15 deletions(-)
diff --git a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpBatchReader.java b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpBatchReader.java
index ad056d7fd0..8bbe855626 100644
--- a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpBatchReader.java
+++ b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpBatchReader.java
@@ -116,6 +116,7 @@ public class HttpBatchReader implements ManagedReader<SchemaNegotiator> {
buildImplicitColumns();
}
+ // inStream is expected to be closed by the JsonLoader.
InputStream inStream = http.getInputStream();
populateImplicitFieldMap(http);
diff --git a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpXMLBatchReader.java b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpXMLBatchReader.java
index 8e0aa684d4..c7a2857b74 100644
--- a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpXMLBatchReader.java
+++ b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpXMLBatchReader.java
@@ -143,7 +143,7 @@ public class HttpXMLBatchReader extends HttpBatchReader {
@Override
public void close() {
- AutoCloseables.closeSilently(inStream);
AutoCloseables.closeSilently(xmlReader);
+ AutoCloseables.closeSilently(inStream);
}
}
diff --git a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/util/SimpleHttp.java b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/util/SimpleHttp.java
index 03e4d5e5cc..e5940de184 100644
--- a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/util/SimpleHttp.java
+++ b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/util/SimpleHttp.java
@@ -28,6 +28,7 @@ import okhttp3.Request;
import okhttp3.Response;
import org.apache.commons.lang3.StringUtils;
+import org.apache.drill.common.AutoCloseables;
import org.apache.drill.common.exceptions.CustomErrorContext;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.oauth.PersistentTokenTable;
@@ -239,7 +240,8 @@ public class SimpleHttp {
/**
* Returns an InputStream based on the URL and config in the scanSpec. If anything goes wrong
* the method throws a UserException.
- * @return An Inputstream of the data from the URL call.
+ * @return An Inputstream of the data from the URL call. The caller is responsible for calling
+ * close() on the InputStream.
*/
public InputStream getInputStream() {
@@ -268,15 +270,14 @@ public class SimpleHttp {
// Build the request object
Request request = requestBuilder.build();
+ Response response = null;
try {
logger.debug("Executing request: {}", request);
logger.debug("Headers: {}", request.headers());
// Execute the request
- Response response = client
- .newCall(request)
- .execute();
+ response = client.newCall(request).execute();
// Preserve the response
responseMessage = response.message();
@@ -291,8 +292,9 @@ public class SimpleHttp {
paginator.notifyPartialPage();
}
- // If the request is unsuccessful, throw a UserException
+ // If the request is unsuccessful, clean up and throw a UserException
if (!isSuccessful(responseCode)) {
+ AutoCloseables.closeSilently(response);
throw UserException
.dataReadError()
.message("HTTP request failed")
@@ -304,9 +306,11 @@ public class SimpleHttp {
logger.debug("HTTP Request for {} successful.", url());
logger.debug("Response Headers: {} ", response.headers());
- // Return the InputStream of the response
+ // Return the InputStream of the response. Note that it is necessary and
+ // and sufficient that the caller invokes close() on the returned stream.
return Objects.requireNonNull(response.body()).byteStream();
} catch (IOException e) {
+ // response can only be null at this location so we do not attempt to close it.
throw UserException
.dataReadError(e)
.message("Failed to read the HTTP response body")
diff --git a/exec/java-exec/src/main/java/org/apache/drill/exec/store/http/oauth/OAuthUtils.java b/exec/java-exec/src/main/java/org/apache/drill/exec/store/http/oauth/OAuthUtils.java
index 573df78cba..8b80499bb6 100644
--- a/exec/java-exec/src/main/java/org/apache/drill/exec/store/http/oauth/OAuthUtils.java
+++ b/exec/java-exec/src/main/java/org/apache/drill/exec/store/http/oauth/OAuthUtils.java
@@ -38,13 +38,13 @@ public class OAuthUtils {
private static final Logger logger = LoggerFactory.getLogger(OAuthUtils.class);
/**
- * Craft a GET request to obtain an access token.
+ * Crafts a POST request to obtain an access token.
* @param credentialsProvider A credential provider containing the clientID, clientSecret and authorizationCode
* @param authorizationCode The authorization code from the OAuth2.0 enabled API
* @param callbackURL The callback URL. For our purposes this is obtained from the incoming Drill request as it all goes to the same place.
* @return A Request Body to obtain an access token
*/
- public static RequestBody getPostResponse(CredentialsProvider credentialsProvider, String authorizationCode, String callbackURL) {
+ public static RequestBody getPostRequest(CredentialsProvider credentialsProvider, String authorizationCode, String callbackURL) {
return new FormBody.Builder()
.add("grant_type", "authorization_code")
.add("client_id", credentialsProvider.getCredentials().get(OAuthTokenCredentials.CLIENT_ID))
@@ -55,12 +55,12 @@ public class OAuthUtils {
}
/**
- * Crafts a POST response for refreshing an access token when a refresh token is present.
+ * Crafts a POST request for refreshing an access token when a refresh token is present.
* @param credentialsProvider A credential provider containing the clientID, clientSecret and refreshToken
* @param refreshToken The refresh token
* @return A Request Body with the correct parameters for obtaining an access token
*/
- public static RequestBody getPostResponseForTokenRefresh(CredentialsProvider credentialsProvider, String refreshToken) {
+ public static RequestBody getPostRequestForTokenRefresh(CredentialsProvider credentialsProvider, String refreshToken) {
return new FormBody.Builder()
.add("grant_type", "refresh_token")
.add("client_id", credentialsProvider.getCredentials().get(OAuthTokenCredentials.CLIENT_ID))
@@ -90,7 +90,7 @@ public class OAuthUtils {
.url(buildAccessTokenURL(credentialsProvider))
.header("Content-Type", "application/json")
.addHeader("Accept", "application/json")
- .post(getPostResponse(credentialsProvider, authenticationCode, callbackURL))
+ .post(getPostRequest(credentialsProvider, authenticationCode, callbackURL))
.build();
}
@@ -110,7 +110,7 @@ public class OAuthUtils {
.url(tokenURI)
.header("Content-Type", "application/json")
.addHeader("Accept", "application/json")
- .post(getPostResponseForTokenRefresh(credentialsProvider, refreshToken))
+ .post(getPostRequestForTokenRefresh(credentialsProvider, refreshToken))
.build();
}
@@ -127,9 +127,10 @@ public class OAuthUtils {
String accessToken;
String refreshToken;
Map<String, String> tokens = new HashMap<>();
+ Response response = null;
try {
- Response response = client.newCall(request).execute();
+ response = client.newCall(request).execute();
String responseBody = response.body().string();
if (!response.isSuccessful()) {
@@ -164,13 +165,14 @@ public class OAuthUtils {
refreshToken = (String) parsedJson.get("refresh_token");
tokens.put(OAuthTokenCredentials.REFRESH_TOKEN, refreshToken);
}
- response.close();
return tokens;
} catch (NullPointerException | IOException e) {
throw UserException.connectionError()
.message("Error refreshing access OAuth2 access token. " + e.getMessage())
.build(logger);
+ } finally {
+ response.close();
}
}
}
[drill] 09/10: DRILL-8323: upgrade commons-text to 1.10.0 (#2663)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit ee8e30bf8a8cf341fc142e70307bb1b7f7cebdda
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Fri Sep 30 03:28:35 2022 +0800
DRILL-8323: upgrade commons-text to 1.10.0 (#2663)
---
contrib/format-excel/pom.xml | 2 +-
.../org/apache/drill/exec/udfs/TestStringDistanceFunctions.java | 6 +++---
pom.xml | 2 +-
3 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/contrib/format-excel/pom.xml b/contrib/format-excel/pom.xml
index 6fb0821f2e..918896bdfc 100644
--- a/contrib/format-excel/pom.xml
+++ b/contrib/format-excel/pom.xml
@@ -52,7 +52,7 @@
<dependency>
<groupId>com.github.pjfanning</groupId>
<artifactId>excel-streaming-reader</artifactId>
- <version>4.0.1</version>
+ <version>4.0.3</version>
</dependency>
<dependency>
<groupId>org.apache.logging.log4j</groupId>
diff --git a/contrib/udfs/src/test/java/org/apache/drill/exec/udfs/TestStringDistanceFunctions.java b/contrib/udfs/src/test/java/org/apache/drill/exec/udfs/TestStringDistanceFunctions.java
index e90c0b60c1..17b43fb4d4 100644
--- a/contrib/udfs/src/test/java/org/apache/drill/exec/udfs/TestStringDistanceFunctions.java
+++ b/contrib/udfs/src/test/java/org/apache/drill/exec/udfs/TestStringDistanceFunctions.java
@@ -59,7 +59,7 @@ public class TestStringDistanceFunctions extends ClusterTest {
double result = queryBuilder()
.sql("select jaccard_distance( 'Big car', 'red car' ) as distance FROM (VALUES(1))")
.singletonDouble();
- assertEquals(0.56, result, 0.0);
+ assertEquals(0.5555555555555556, result, 0.0);
}
@Test
@@ -67,7 +67,7 @@ public class TestStringDistanceFunctions extends ClusterTest {
double result = queryBuilder()
.sql("select jaro_distance( 'Big car', 'red car' ) as distance FROM (VALUES(1))")
.singletonDouble();
- assertEquals(0.7142857142857143, result, 0.0);
+ assertEquals(0.2857142857142857, result, 0.0);
}
@Test
@@ -77,4 +77,4 @@ public class TestStringDistanceFunctions extends ClusterTest {
.singletonDouble();
assertEquals(3.0, result, 0.0);
}
-}
\ No newline at end of file
+}
diff --git a/pom.xml b/pom.xml
index 7c43b54ea1..9b317b0ea5 100644
--- a/pom.xml
+++ b/pom.xml
@@ -109,7 +109,7 @@
<maven.min.version>3.6.3</maven.min.version>
<commons.net.version>3.6</commons.net.version>
<commons.validator.version>1.6</commons.validator.version>
- <commons.text.version>1.6</commons.text.version>
+ <commons.text.version>1.10.0</commons.text.version>
<protobuf.version>3.16.1</protobuf.version>
<protostuff.version>1.7.1</protostuff.version>
<codemodel.version>2.6</codemodel.version>
[drill] 05/10: DRILL-8307: Ensure thread safety in the Druid plugin HTTP client (#2650)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 3b949bc31ae0b0ece2cf9b8e00e7862a95efb88e
Author: James Turton <91...@users.noreply.github.com>
AuthorDate: Sun Sep 18 14:54:14 2022 +0800
DRILL-8307: Ensure thread safety in the Druid plugin HTTP client (#2650)
---
.../exec/store/druid/rest/DruidAdminClient.java | 30 ++++----
.../exec/store/druid/rest/DruidQueryClient.java | 32 ++++----
.../drill/exec/store/druid/rest/RestClient.java | 20 ++++-
.../exec/store/druid/rest/RestClientWrapper.java | 48 ++++++------
.../drill/exec/store/druid/DruidTestBase.java | 4 +-
.../{DruidTestSuit.java => DruidTestSuite.java} | 4 +-
.../drill/exec/store/druid/TestDataGenerator.java | 85 +++++++++++-----------
.../store/druid/rest/DruidQueryClientTest.java | 37 ++++------
8 files changed, 135 insertions(+), 125 deletions(-)
diff --git a/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/DruidAdminClient.java b/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/DruidAdminClient.java
index bbfd336a42..09c99fc834 100755
--- a/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/DruidAdminClient.java
+++ b/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/DruidAdminClient.java
@@ -21,12 +21,12 @@ import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.store.druid.druid.SimpleDatasourceInfo;
-import org.apache.http.HttpResponse;
-import org.apache.http.HttpStatus;
-import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import okhttp3.Response;
+
+import java.io.InputStream;
import java.io.IOException;
import java.util.List;
@@ -34,7 +34,6 @@ public class DruidAdminClient {
private static final Logger logger = LoggerFactory.getLogger(DruidAdminClient.class);
private static final String DATASOURCES_BASE_URI = "/druid/coordinator/v1/datasources?simple";
- private static final String DEFAULT_ENCODING = "UTF-8";
private static final ObjectMapper mapper = new ObjectMapper();
private final String coordinatorAddress;
@@ -47,18 +46,19 @@ public class DruidAdminClient {
public List<SimpleDatasourceInfo> getDataSources() throws IOException {
String url = this.coordinatorAddress + DATASOURCES_BASE_URI;
- HttpResponse response = restClient.get(url);
+ try (Response response = restClient.get(url)) {
+ if (!response.isSuccessful()) {
+ // TODO: Add a CustomErrorContext when this plugin is converted to EVF.
+ throw UserException
+ .dataReadError()
+ .message("Error getting druid datasources. HTTP request failed")
+ .addContext("Response code", response.code())
+ .addContext("Response message", response.message())
+ .build(logger);
+ }
- if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
- throw UserException
- .dataReadError()
- .message("Error getting druid datasources. HTTP request failed")
- .addContext("Response code", response.getStatusLine().getStatusCode())
- .addContext("Response message", response.getStatusLine().getReasonPhrase())
- .build(logger);
+ InputStream responseStream = response.body().byteStream();
+ return mapper.readValue(responseStream, new TypeReference<List<SimpleDatasourceInfo>>(){});
}
-
- String responseJson = EntityUtils.toString(response.getEntity(), DEFAULT_ENCODING);
- return mapper.readValue(responseJson, new TypeReference<List<SimpleDatasourceInfo>>(){});
}
}
diff --git a/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/DruidQueryClient.java b/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/DruidQueryClient.java
index bd05c3ee5d..fe82650199 100755
--- a/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/DruidQueryClient.java
+++ b/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/DruidQueryClient.java
@@ -22,12 +22,12 @@ import com.fasterxml.jackson.databind.node.ArrayNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.store.druid.druid.DruidScanResponse;
-import org.apache.http.HttpResponse;
-import org.apache.http.HttpStatus;
-import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import okhttp3.Response;
+
+import java.io.InputStream;
import java.util.ArrayList;
public class DruidQueryClient {
@@ -48,20 +48,22 @@ public class DruidQueryClient {
public DruidScanResponse executeQuery(String query) throws Exception {
logger.debug("Executing Query - {}", query);
- HttpResponse response = restClient.post(queryUrl, query);
- if (response.getStatusLine().getStatusCode() != HttpStatus.SC_OK) {
- throw UserException
- .dataReadError()
- .message("Error executing druid query. HTTP request failed")
- .addContext("Response code", response.getStatusLine().getStatusCode())
- .addContext("Response message", response.getStatusLine().getReasonPhrase())
- .build(logger);
- }
+ try (Response response = restClient.post(queryUrl, query)) {
+ if (!response.isSuccessful()) {
+ // TODO: Add a CustomErrorContext when this plugin is converted to EVF.
+ throw UserException
+ .dataReadError()
+ .message("Error executing druid query. HTTP request failed")
+ .addContext("Response code", response.code())
+ .addContext("Response message", response.message())
+ .build(logger);
+ }
- String data = EntityUtils.toString(response.getEntity());
- ArrayNode responses = mapper.readValue(data, ArrayNode.class);
- return parseResponse(responses);
+ InputStream responseStream = response.body().byteStream();
+ ArrayNode responses = mapper.readValue(responseStream, ArrayNode.class);
+ return parseResponse(responses);
+ }
}
private DruidScanResponse parseResponse(ArrayNode responses) {
diff --git a/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/RestClient.java b/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/RestClient.java
index d88a41f5b4..fa8a4cf5e1 100644
--- a/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/RestClient.java
+++ b/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/RestClient.java
@@ -17,11 +17,25 @@
*/
package org.apache.drill.exec.store.druid.rest;
-import org.apache.http.HttpResponse;
+import okhttp3.Response;
import java.io.IOException;
public interface RestClient {
- HttpResponse get(String url) throws IOException;
- HttpResponse post(String url, String body) throws IOException;
+ /**
+ * Executes an HTTP GET.
+ * @param url request URL
+ * @return a Response object that the caller is responsible for closing.
+ * @throws IOException
+ */
+ Response get(String url) throws IOException;
+
+ /**
+ * Executes an HTTP POST.
+ * @param url request URL.
+ * @param body request body.
+ * @return a Response object that the caller is responsible for closing.
+ * @throws IOException
+ */
+ Response post(String url, String body) throws IOException;
}
diff --git a/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/RestClientWrapper.java b/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/RestClientWrapper.java
index 3dd47ee261..a5cb6e73f1 100644
--- a/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/RestClientWrapper.java
+++ b/contrib/storage-druid/src/main/java/org/apache/drill/exec/store/druid/rest/RestClientWrapper.java
@@ -17,36 +17,36 @@
*/
package org.apache.drill.exec.store.druid.rest;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.entity.ByteArrayEntity;
-import org.apache.http.impl.client.DefaultHttpClient;
-
-import javax.ws.rs.core.HttpHeaders;
-import java.io.IOException;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
-import static javax.ws.rs.core.MediaType.APPLICATION_JSON;
-import static org.apache.http.protocol.HTTP.CONTENT_TYPE;
+import java.nio.charset.StandardCharsets;
+import java.io.IOException;
public class RestClientWrapper implements RestClient {
- private static final HttpClient httpClient = new DefaultHttpClient();
- private static final String DEFAULT_ENCODING = "UTF-8";
+ // OkHttp client is designed to be shared across threads.
+ private final OkHttpClient httpClient = new OkHttpClient();
- public HttpResponse get(String url) throws IOException {
- HttpGet httpget = new HttpGet(url);
- httpget.addHeader(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON);
- return httpClient.execute(httpget);
+ public Response get(String url) throws IOException {
+ Request get = new Request.Builder()
+ .url(url)
+ .addHeader("Content-Type", "application/json")
+ .build();
+
+ return httpClient.newCall(get).execute();
}
- public HttpResponse post(String url, String body) throws IOException {
- HttpPost httppost = new HttpPost(url);
- httppost.addHeader(CONTENT_TYPE, APPLICATION_JSON);
- HttpEntity entity = new ByteArrayEntity(body.getBytes(DEFAULT_ENCODING));
- httppost.setEntity(entity);
+ public Response post(String url, String body) throws IOException {
+ RequestBody postBody = RequestBody.create(body.getBytes(StandardCharsets.UTF_8));
+
+ Request post = new Request.Builder()
+ .url(url)
+ .addHeader("Content-Type", "application/json")
+ .post(postBody)
+ .build();
- return httpClient.execute(httppost);
+ return httpClient.newCall(post).execute();
}
}
diff --git a/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestBase.java b/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestBase.java
index ef47e26c0e..d52a8fc28e 100644
--- a/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestBase.java
+++ b/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestBase.java
@@ -35,7 +35,7 @@ public class DruidTestBase extends ClusterTest implements DruidTestConstants {
startCluster(ClusterFixture.builder(dirTestWatcher));
pluginRegistry = cluster.drillbit().getContext().getStorage();
- DruidTestSuit.initDruid();
+ DruidTestSuite.initDruid();
initDruidStoragePlugin();
}
@@ -43,7 +43,7 @@ public class DruidTestBase extends ClusterTest implements DruidTestConstants {
pluginRegistry
.put(
DruidStoragePluginConfig.NAME,
- DruidTestSuit.getDruidStoragePluginConfig());
+ DruidTestSuite.getDruidStoragePluginConfig());
}
@AfterClass
diff --git a/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestSuit.java b/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestSuite.java
similarity index 98%
rename from contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestSuit.java
rename to contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestSuite.java
index 5e1c149556..92d3c3486a 100644
--- a/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestSuit.java
+++ b/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/DruidTestSuite.java
@@ -43,8 +43,8 @@ import java.io.File;
DruidScanSpecBuilderTest.class
})
@Category({SlowTest.class, DruidStorageTest.class})
-public class DruidTestSuit {
- private static final Logger logger = LoggerFactory.getLogger(DruidTestSuit.class);
+public class DruidTestSuite {
+ private static final Logger logger = LoggerFactory.getLogger(DruidTestSuite.class);
private static final ObjectMapper mapper = new ObjectMapper();
diff --git a/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/TestDataGenerator.java b/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/TestDataGenerator.java
index f4abcd3ca5..3b96db7cdb 100644
--- a/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/TestDataGenerator.java
+++ b/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/TestDataGenerator.java
@@ -23,38 +23,28 @@ import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.drill.shaded.guava.com.google.common.io.Resources;
-import org.apache.http.HttpEntity;
-import org.apache.http.HttpResponse;
-import org.apache.http.HttpStatus;
-import org.apache.http.StatusLine;
-import org.apache.http.client.HttpClient;
-import org.apache.http.client.methods.HttpGet;
-import org.apache.http.client.methods.HttpPost;
-import org.apache.http.entity.ByteArrayEntity;
-import org.apache.http.impl.client.DefaultHttpClient;
-import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import javax.ws.rs.core.HttpHeaders;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+
+import java.io.InputStream;
import java.io.IOException;
import java.net.URISyntaxException;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.concurrent.TimeUnit;
-import static javax.ws.rs.core.MediaType.APPLICATION_JSON;
-import static org.apache.http.protocol.HTTP.CONTENT_TYPE;
-
public class TestDataGenerator {
private static final Logger logger = LoggerFactory.getLogger(TestDataGenerator.class);
- private static final HttpClient httpClient = new DefaultHttpClient();
+ private static final OkHttpClient httpClient = new OkHttpClient();
private static final ObjectMapper mapper = new ObjectMapper();
- private static final String DEFAULT_ENCODING = "UTF-8";
-
private static final String RESPONSE_SUCCESS = "SUCCESS";
public static void startImport(DruidStoragePluginConfig druidStoragePluginConfig) throws Exception {
@@ -72,11 +62,13 @@ public class TestDataGenerator {
private static boolean isDruidRunning(DruidStoragePluginConfig druidStoragePluginConfig) {
try {
String healthCheckUrl = druidStoragePluginConfig.getCoordinatorAddress() + "/status/health";
- HttpGet httpGet = new HttpGet(healthCheckUrl);
- HttpResponse response = httpClient.execute(httpGet);
- StatusLine statusLine = response.getStatusLine();
- String status = EntityUtils.toString(response.getEntity(), DEFAULT_ENCODING);
- return statusLine.getStatusCode() == HttpStatus.SC_OK && status.equalsIgnoreCase("true");
+ Request get = new Request.Builder()
+ .url(healthCheckUrl)
+ .build();
+
+ try (Response resp = httpClient.newCall(get).execute()) {
+ return resp.isSuccessful() && resp.body().string().equalsIgnoreCase("true");
+ }
} catch (Exception ex) {
logger.error("Error getting druid status", ex);
return false;
@@ -90,18 +82,24 @@ public class TestDataGenerator {
private static String startImportTask(DruidStoragePluginConfig druidStoragePluginConfig) throws URISyntaxException, IOException {
try {
String url = taskUrl(druidStoragePluginConfig);
- byte[] taskConfig = Files.readAllBytes(Paths.get(Resources.getResource("wikipedia-index.json").toURI()));
-
- HttpPost httpPost = new HttpPost(url);
- httpPost.addHeader(CONTENT_TYPE, APPLICATION_JSON);
- HttpEntity entity = new ByteArrayEntity(taskConfig);
- httpPost.setEntity(entity);
-
- HttpResponse response = httpClient.execute(httpPost);
- String data = EntityUtils.toString(response.getEntity());
- TaskStartResponse taskStartResponse = mapper.readValue(data, TaskStartResponse.class);
- logger.debug("Started Indexing Task - " + taskStartResponse.getTaskId());
- return taskStartResponse.getTaskId();
+ RequestBody postBody = RequestBody.create(
+ Files.readAllBytes(Paths.get(Resources.getResource("wikipedia-index.json").toURI()))
+ );
+ Request post = new Request.Builder()
+ .url(url)
+ .addHeader("Content-Type", "application/json")
+ .post(postBody)
+ .build();
+
+ try (Response resp = httpClient.newCall(post).execute()) {
+ String respBodyStr = resp.body().string();
+ TaskStartResponse taskStartResponse = mapper.readValue(
+ respBodyStr,
+ TaskStartResponse.class
+ );
+ logger.debug("Started Indexing Task - {}", taskStartResponse.getTaskId());
+ return taskStartResponse.getTaskId();
+ }
} catch (Exception ex) {
logger.error("Error starting Indexing Task");
throw ex;
@@ -114,14 +112,17 @@ public class TestDataGenerator {
Thread.sleep(TimeUnit.MINUTES.toMillis(sleepMinutes));
String url = taskUrl(druidStoragePluginConfig) + "/" + taskId + "/status";
- HttpGet httpget = new HttpGet(url);
- httpget.addHeader(HttpHeaders.CONTENT_TYPE, APPLICATION_JSON);
-
- HttpResponse response = httpClient.execute(httpget);
- String responseJson = EntityUtils.toString(response.getEntity(), DEFAULT_ENCODING);
- TaskStatusResponse taskStatusResponse = mapper.readValue(responseJson, TaskStatusResponse.class);
- if (!taskStatusResponse.taskStatus.status.equalsIgnoreCase(RESPONSE_SUCCESS)) {
- throw new Exception(String.format("Task %s finished with status %s", taskId, taskStatusResponse.taskStatus.status));
+ Request get = new Request.Builder()
+ .url(url)
+ .addHeader("Content-Type", "application/json")
+ .build();
+
+ try (Response resp = httpClient.newCall(get).execute()) {
+ InputStream jsonStream = resp.body().byteStream();
+ TaskStatusResponse taskStatusResponse = mapper.readValue(jsonStream, TaskStatusResponse.class);
+ if (!taskStatusResponse.taskStatus.status.equalsIgnoreCase(RESPONSE_SUCCESS)) {
+ throw new Exception(String.format("Task %s finished with status %s", taskId, taskStatusResponse.taskStatus.status));
+ }
}
logger.debug("Task {} finished successfully", taskId);
diff --git a/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/rest/DruidQueryClientTest.java b/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/rest/DruidQueryClientTest.java
index de560d9557..0fe579c7ef 100644
--- a/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/rest/DruidQueryClientTest.java
+++ b/contrib/storage-druid/src/test/java/org/apache/drill/exec/store/druid/rest/DruidQueryClientTest.java
@@ -18,10 +18,6 @@
package org.apache.drill.exec.store.druid.rest;
import org.apache.drill.exec.store.druid.druid.DruidScanResponse;
-import org.apache.http.HttpStatus;
-import org.apache.http.HttpResponse;
-import org.apache.http.StatusLine;
-import org.apache.http.HttpEntity;
import org.apache.http.Header;
import org.apache.http.HttpHeaders;
import org.apache.http.message.BasicHeader;
@@ -29,6 +25,9 @@ import org.junit.Before;
import org.junit.Test;
import org.mockito.Mock;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+
import java.io.ByteArrayInputStream;
import java.io.IOException;
import java.io.InputStream;
@@ -45,13 +44,10 @@ public class DruidQueryClientTest {
private RestClient restClient;
@Mock
- private HttpResponse httpResponse;
-
- @Mock
- private StatusLine statusLine;
+ private Response httpResponse;
@Mock
- private HttpEntity httpEntity;
+ private ResponseBody httpResponseBody;
private DruidQueryClient druidQueryClient;
private static final String BROKER_URI = "some broker uri";
@@ -62,14 +58,11 @@ public class DruidQueryClientTest {
@Before
public void setup() throws IOException {
restClient = mock(RestClient.class);
- httpResponse = mock(HttpResponse.class);
- statusLine = mock(StatusLine.class);
- httpEntity = mock(HttpEntity.class);
-
- when(statusLine.getStatusCode()).thenReturn(HttpStatus.SC_OK);
- when(httpEntity.getContentEncoding()).thenReturn(ENCODING_HEADER);
- when(httpResponse.getStatusLine()).thenReturn(statusLine);
- when(httpResponse.getEntity()).thenReturn(httpEntity);
+ httpResponse = mock(Response.class);
+ httpResponseBody = mock(ResponseBody.class);
+
+ when(httpResponse.isSuccessful()).thenReturn(true);
+ when(httpResponse.body()).thenReturn(httpResponseBody);
when(restClient.post(BROKER_URI + "/druid/v2", QUERY))
.thenReturn(httpResponse);
@@ -79,7 +72,7 @@ public class DruidQueryClientTest {
@Test(expected=Exception.class)
public void executeQueryCalledDruidReturnsNon200ShouldThrowError()
throws Exception {
- when(statusLine.getStatusCode()).thenReturn(HttpStatus.SC_INTERNAL_SERVER_ERROR);
+ when(httpResponse.isSuccessful()).thenReturn(false);
druidQueryClient.executeQuery(QUERY);
}
@@ -87,8 +80,8 @@ public class DruidQueryClientTest {
public void executeQueryCalledNoResponsesFoundReturnsEmptyEventList()
throws Exception {
InputStream inputStream =
- new ByteArrayInputStream("[]".getBytes(StandardCharsets.UTF_8.name()));
- when(httpEntity.getContent()).thenReturn(inputStream);
+ new ByteArrayInputStream("[]".getBytes(StandardCharsets.UTF_8));
+ when(httpResponseBody.byteStream()).thenReturn(inputStream);
DruidScanResponse response = druidQueryClient.executeQuery(QUERY);
assertThat(response.getEvents()).isEmpty();
@@ -99,8 +92,8 @@ public class DruidQueryClientTest {
throws Exception {
String result = "[{\"segmentId\":\"wikipedia_2016-06-27T14:00:00.000Z_2016-06-27T15:00:00.000Z_2021-12-11T11:12:16.106Z\",\"columns\":[\"__time\",\"channel\",\"cityName\",\"comment\",\"countryIsoCode\",\"countryName\",\"diffUrl\",\"flags\",\"isAnonymous\",\"isMinor\",\"isNew\",\"isRobot\",\"isUnpatrolled\",\"metroCode\",\"namespace\",\"page\",\"regionIsoCode\",\"regionName\",\"user\",\"sum_deleted\",\"sum_deltaBucket\",\"sum_added\",\"sum_commentLength\",\"count\",\"sum_delta\"],\"ev [...]
InputStream inputStream =
- new ByteArrayInputStream(result.getBytes(StandardCharsets.UTF_8.name()));
- when(httpEntity.getContent()).thenReturn(inputStream);
+ new ByteArrayInputStream(result.getBytes(StandardCharsets.UTF_8));
+ when(httpResponseBody.byteStream()).thenReturn(inputStream);
DruidScanResponse response = druidQueryClient.executeQuery(QUERY);
assertThat(response.getEvents()).isNotEmpty();
[drill] 02/10: DRILL-8293: Add a docker-compose file to run Drill in cluster mode (#2640)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit bb8524116a788be01301956b3eac28e2d0b15f60
Author: James Turton <91...@users.noreply.github.com>
AuthorDate: Fri Sep 9 07:12:52 2022 +0200
DRILL-8293: Add a docker-compose file to run Drill in cluster mode (#2640)
---
distribution/docker-cluster-mode/Dockerfile | 26 ++++++++++++
distribution/docker-cluster-mode/README | 25 +++++++++++
.../docker-cluster-mode/docker-compose.yml | 48 ++++++++++++++++++++++
.../docker-cluster-mode/drill-override.conf | 28 +++++++++++++
distribution/docker-cluster-mode/logback.xml | 39 ++++++++++++++++++
5 files changed, 166 insertions(+)
diff --git a/distribution/docker-cluster-mode/Dockerfile b/distribution/docker-cluster-mode/Dockerfile
new file mode 100644
index 0000000000..d4e360865d
--- /dev/null
+++ b/distribution/docker-cluster-mode/Dockerfile
@@ -0,0 +1,26 @@
+#
+# 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.
+#
+
+FROM apache/drill:latest
+
+ENV DRILL_LOG_LEVEL=info
+
+COPY drill-override.conf $DRILL_HOME/conf/drill-override.conf
+COPY logback.xml $DRILL_HOME/conf/logback.xml
+
+ENTRYPOINT $DRILL_HOME/bin/drillbit.sh run
diff --git a/distribution/docker-cluster-mode/README b/distribution/docker-cluster-mode/README
new file mode 100644
index 0000000000..bd8bb6d040
--- /dev/null
+++ b/distribution/docker-cluster-mode/README
@@ -0,0 +1,25 @@
+This directory contains source code artifacts to launch Drill in cluster mode
+along with a ZooKeeper. The Drill image is based on a minor customisation of
+the official Drill image that switches it from an embedded to a cluster mode
+launch. Logging is redirected to stdout.
+
+In the docker-cluster-mode directory:
+
+1. docker build -t apache/drill-cluster-mode
+2. docker-compose up
+
+Then access the web UI at http://localhost:8047 or connect a JDBC client to
+jdbc:drill:drillbit=localhost or jdbc:drill:zk=localhost but note that you
+will need to make the drillbit container hostnames resolvable from the host
+to use a ZooKeeper JDBC URL.
+
+To launch a cluster of 3 Drillbits
+
+3. docker-compose up --scale drillbit=3
+
+but first note that to use docker-compose's "scale" feature to run multiple
+Drillbit containers on a single host you will need to remove the host port
+mappings from the compose file to prevent collisions (see the comments
+on the relevant lines in that file). Once the Drillbits are launched run
+`docker-compose ps` to list the ephemeral ports that have been allocated on
+the host.
diff --git a/distribution/docker-cluster-mode/docker-compose.yml b/distribution/docker-cluster-mode/docker-compose.yml
new file mode 100644
index 0000000000..d43bfa9e6f
--- /dev/null
+++ b/distribution/docker-cluster-mode/docker-compose.yml
@@ -0,0 +1,48 @@
+#
+# 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.
+#
+
+version: "3.9"
+services:
+ zookeeper:
+ container_name: zookeeper
+ hostname: zookeeper
+ image: zookeeper:latest
+ volumes:
+ - zk_data:/data
+ - zk_datalog:/datalog
+ - zk_logs:/logs
+ ports:
+ - "2181:2181"
+
+ drillbit:
+ # container_name not set so that --scale can be used
+ image: apache/drill-cluster-mode:latest
+ volumes:
+ - drill_data:/data
+ ports:
+ - "8047:8047" # user HTTP, replace with just "8047" for --scale
+ - "31010:31010" # user RPC and JDBC, replace with just "31010" for --scale
+ depends_on: ["zookeeper"]
+ environment:
+ - DRILL_MAX_PROC_MEM=8G
+
+volumes:
+ zk_data:
+ zk_datalog:
+ zk_logs:
+ drill_data:
diff --git a/distribution/docker-cluster-mode/drill-override.conf b/distribution/docker-cluster-mode/drill-override.conf
new file mode 100644
index 0000000000..c65652a72b
--- /dev/null
+++ b/distribution/docker-cluster-mode/drill-override.conf
@@ -0,0 +1,28 @@
+#
+# 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.
+
+# See 'drill-override-example.conf' for example configurations
+
+drill.exec: {
+ cluster-id: "drillbits1",
+ zk.connect: "zookeeper:2181"
+}
diff --git a/distribution/docker-cluster-mode/logback.xml b/distribution/docker-cluster-mode/logback.xml
new file mode 100644
index 0000000000..c1cdbfdd27
--- /dev/null
+++ b/distribution/docker-cluster-mode/logback.xml
@@ -0,0 +1,39 @@
+<?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.
+
+-->
+<configuration>
+ <appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
+ <encoder>
+ <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>
+ </encoder>
+ </appender>
+ <logger name="org.apache.drill" additivity="false">
+ <level value="${DRILL_LOG_LEVEL:-info}" />
+ <appender-ref ref="STDOUT" />
+ </logger>
+ <logger name="query.logger" additivity="false">
+ <level value="${DRILL_LOG_LEVEL:-info}" />
+ <appender-ref ref="STDOUT" />
+ </logger>
+ <root>
+ <level value="${DRILL_LOG_LEVEL:-info}" />
+ <appender-ref ref="STDOUT" />
+ </root>
+</configuration>
[drill] 10/10: DRILL-8324: Sizeof refactor (#2665)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit 1a20a931a31f8f9bc33b9d24eda3793b5662023a
Author: PJ Fanning <pj...@users.noreply.github.com>
AuthorDate: Sun Oct 16 05:10:25 2022 +0100
DRILL-8324: Sizeof refactor (#2665)
---
contrib/storage-opentsdb/pom.xml | 5 -
.../exec/store/openTSDB/OpenTSDBGroupScan.java | 1 -
.../drill/exec/store/openTSDB/SizeEstimator.java | 457 +++++++++++++++++++++
.../{ => exec}/store/openTSDB/TestDataHolder.java | 2 +-
.../store/openTSDB/TestOpenTSDBPlugin.java | 31 +-
.../exec/store/openTSDB/TestSizeEstimator.java | 87 ++++
6 files changed, 560 insertions(+), 23 deletions(-)
diff --git a/contrib/storage-opentsdb/pom.xml b/contrib/storage-opentsdb/pom.xml
index 3c771a3121..53a40f203d 100644
--- a/contrib/storage-opentsdb/pom.xml
+++ b/contrib/storage-opentsdb/pom.xml
@@ -75,11 +75,6 @@
<artifactId>converter-jackson</artifactId>
<version>${retrofit.version}</version>
</dependency>
- <dependency>
- <groupId>com.madhukaraphatak</groupId>
- <artifactId>java-sizeof_2.11</artifactId>
- <version>0.1</version>
- </dependency>
</dependencies>
</project>
diff --git a/contrib/storage-opentsdb/src/main/java/org/apache/drill/exec/store/openTSDB/OpenTSDBGroupScan.java b/contrib/storage-opentsdb/src/main/java/org/apache/drill/exec/store/openTSDB/OpenTSDBGroupScan.java
index 680090368b..8969a2d1b1 100644
--- a/contrib/storage-opentsdb/src/main/java/org/apache/drill/exec/store/openTSDB/OpenTSDBGroupScan.java
+++ b/contrib/storage-opentsdb/src/main/java/org/apache/drill/exec/store/openTSDB/OpenTSDBGroupScan.java
@@ -24,7 +24,6 @@ import com.fasterxml.jackson.annotation.JsonProperty;
import com.fasterxml.jackson.annotation.JsonTypeName;
import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
import org.apache.drill.shaded.guava.com.google.common.collect.Lists;
-import com.madhukaraphatak.sizeof.SizeEstimator;
import org.apache.drill.common.exceptions.ExecutionSetupException;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.exec.physical.base.AbstractGroupScan;
diff --git a/contrib/storage-opentsdb/src/main/java/org/apache/drill/exec/store/openTSDB/SizeEstimator.java b/contrib/storage-opentsdb/src/main/java/org/apache/drill/exec/store/openTSDB/SizeEstimator.java
new file mode 100644
index 0000000000..89fd96f889
--- /dev/null
+++ b/contrib/storage-opentsdb/src/main/java/org/apache/drill/exec/store/openTSDB/SizeEstimator.java
@@ -0,0 +1,457 @@
+/*
+ * 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.openTSDB;
+
+import org.apache.drill.shaded.guava.com.google.common.collect.MapMaker;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.management.MBeanServer;
+import java.lang.management.ManagementFactory;
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.lang.reflect.Modifier;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.IdentityHashMap;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.Random;
+
+// based on https://github.com/apache/spark/blob/master/core/src/main/scala/org/apache/spark/util/SizeEstimator.scala
+// which itself is based on https://www.infoworld.com/article/2077408/sizeof-for-java.html
+class SizeEstimator {
+
+ private static final Logger logger = LoggerFactory.getLogger(SizeEstimator.class);
+ // Estimate the size of arrays larger than ARRAY_SIZE_FOR_SAMPLING by sampling.
+ private static final int ARRAY_SIZE_FOR_SAMPLING = 400;
+ private static final int ARRAY_SAMPLE_SIZE = 100; // should be lower than ARRAY_SIZE_FOR_SAMPLING
+
+ /**
+ * The state of an ongoing size estimation. Contains a stack of objects to visit as well as an
+ * IdentityHashMap of visited objects, and provides utility methods for enqueueing new objects
+ * to visit.
+ */
+ private static class SearchState {
+
+ private final IdentityHashMap<Object, Object> visited;
+ private final LinkedList<Object> stack = new LinkedList<>();
+ long size = 0L;
+
+ SearchState(IdentityHashMap<Object, Object> visited) {
+ this.visited = visited;
+ }
+
+ void enqueue(Object obj) {
+ if (obj != null && !visited.containsKey(obj)) {
+ visited.put(obj, null);
+ stack.add(obj);
+ }
+ }
+
+ boolean isFinished() {
+ return stack.isEmpty();
+ }
+
+ Object dequeue() {
+ return stack.removeLast();
+ }
+ }
+
+ /**
+ * Cached information about each class. We remember two things: the "shell size" of the class
+ * (size of all non-static fields plus the java.lang.Object size), and any fields that are
+ * pointers to objects.
+ */
+ private static class ClassInfo {
+ private final long shellSize;
+ private final LinkedList<Field> pointerFields;
+
+ ClassInfo(final long shellSize, final LinkedList<Field> pointerFields) {
+ this.shellSize = shellSize;
+ this.pointerFields = pointerFields;
+ }
+
+ long getShellSize() {
+ return shellSize;
+ }
+
+ LinkedList<Field> getPointerFields() {
+ return pointerFields;
+ }
+ }
+
+ // Sizes of primitive types
+ private static final int BYTE_SIZE = 1;
+ private static final int BOOLEAN_SIZE = 1;
+ private static final int CHAR_SIZE = 2;
+ private static final int SHORT_SIZE = 2;
+ private static final int INT_SIZE = 4;
+ private static final int LONG_SIZE = 8;
+ private static final int FLOAT_SIZE = 4;
+ private static final int DOUBLE_SIZE = 8;
+
+ // Fields can be primitive types, sizes are: 1, 2, 4, 8. Or fields can be pointers. The size of
+ // a pointer is 4 or 8 depending on the JVM (32-bit or 64-bit) and UseCompressedOops flag.
+ // The sizes should be in descending order, as we will use that information for fields placement.
+ private static final int MAX_FIELD_SIZE = 8;
+ private static final List<Integer> fieldSizes = new ArrayList<>();
+
+ // Alignment boundary for objects
+ // TODO: Is this arch dependent ?
+ private static final int ALIGN_SIZE = 8;
+
+ // A cache of ClassInfo objects for each class
+ private static final Map<Class<?>, ClassInfo> classInfos = new MapMaker().weakKeys().makeMap();
+
+ // Object and pointer sizes are arch dependent
+ private static final boolean is64bit;
+
+ private static int pointerSize = 4;
+
+ // Minimum size of a java.lang.Object
+ private static int objectSize = 8;
+
+ static {
+ fieldSizes.add(8);
+ fieldSizes.add(4);
+ fieldSizes.add(2);
+ fieldSizes.add(1);
+
+ // Sets object size, pointer size based on architecture and CompressedOops settings
+ // from the JVM.
+ final String arch = System.getProperty("os.arch");
+ is64bit = arch.contains("64") || arch.contains("s390x");
+ // Size of an object reference
+ // Based on https://wikis.oracle.com/display/HotSpotInternals/CompressedOops
+ final boolean isCompressedOops = getIsCompressedOops();
+
+ objectSize = !is64bit ? 8 : (!isCompressedOops ? 16 : 12);
+ pointerSize = (is64bit && !isCompressedOops) ? 8 : 4;
+ classInfos.clear();
+ classInfos.put(Object.class, new ClassInfo(objectSize, new LinkedList<>()));
+ }
+
+ private static boolean getIsCompressedOops() {
+ // This is only used by tests to override the detection of compressed oops. The test
+ // actually uses a system property instead of a SparkConf, so we'll stick with that.
+ if (System.getProperty("spark.test.useCompressedOops") != null) {
+ return Boolean.getBoolean("spark.test.useCompressedOops");
+ }
+
+ final String javaVendor = System.getProperty("java.vendor");
+ if (javaVendor.contains("IBM") || javaVendor.contains("OpenJ9")) {
+ return System.getProperty("java.vm.info").contains("Compressed Ref");
+ }
+
+ try {
+ final String hotSpotMBeanName = "com.sun.management:type=HotSpotDiagnostic";
+ final MBeanServer server = ManagementFactory.getPlatformMBeanServer();
+
+ // NOTE: This should throw an exception in non-Sun JVMs
+ final Class<?> hotSpotMBeanClass = Class.forName("com.sun.management.HotSpotDiagnosticMXBean");
+ final Method getVMMethod = hotSpotMBeanClass.getDeclaredMethod("getVMOption",
+ Class.forName("java.lang.String"));
+
+ final Object bean = ManagementFactory.newPlatformMXBeanProxy(server,
+ hotSpotMBeanName, hotSpotMBeanClass);
+ // TODO: We could use reflection on the VMOption returned ?
+ return getVMMethod.invoke(bean, "UseCompressedOops").toString().contains("true");
+ } catch(Exception e) {
+ // Guess whether they've enabled UseCompressedOops based on whether maxMemory < 32 GB
+ final boolean guess = Runtime.getRuntime().maxMemory() < (32L*1024*1024*1024);
+ logger.warn("Failed to check whether UseCompressedOops is set; assuming {}", guess);
+ return guess;
+ }
+ }
+
+ /**
+ * Estimate the number of bytes that the given object takes up on the JVM heap. The estimate
+ * includes space taken up by objects referenced by the given object, their references, and so on
+ * and so forth.
+ *
+ * This is useful for determining the amount of heap space a broadcast variable will occupy on
+ * each executor or the amount of space each object will take when caching objects in
+ * deserialized form. This is not the same as the serialized size of the object, which will
+ * typically be much smaller.
+ */
+ public static long estimate(final Object obj) {
+ return estimate(obj, new IdentityHashMap<>());
+ }
+
+ private static long estimate(final Object obj, final IdentityHashMap<Object, Object> visited) {
+ final SearchState state = new SearchState(visited);
+ state.enqueue(obj);
+ while (!state.isFinished()) {
+ visitSingleObject(state.dequeue(), state);
+ }
+ return state.size;
+ }
+
+ private static void visitSingleObject(final Object obj, final SearchState state) {
+ final Class<?> cls = obj.getClass();
+ if (cls.isArray()) {
+ visitArray(obj, cls, state);
+ } else if (cls.getName().startsWith("scala.reflect")) {
+ // Many objects in the scala.reflect package reference global reflection objects which, in
+ // turn, reference many other large global objects. Do nothing in this case.
+ } else if (obj instanceof ClassLoader || obj instanceof Class<?>) {
+ // Hadoop JobConfs created in the interpreter have a ClassLoader, which greatly confuses
+ // the size estimator since it references the whole REPL. Do nothing in this case. In
+ // general all ClassLoaders and Classes will be shared between objects anyway.
+ } else {
+ final Long calculatedSize = knownSize(obj);
+ if (calculatedSize != null) {
+ state.size += calculatedSize;
+ } else {
+ final ClassInfo classInfo = getClassInfo(cls);
+ state.size += alignSize(classInfo.shellSize);
+ for (Field field : classInfo.pointerFields) {
+ try {
+ state.enqueue(field.get(obj));
+ } catch (Exception e) {
+ //skip this field
+ }
+ }
+ }
+ }
+ }
+
+ private static void visitArray(final Object array, final Class<?> arrayClass, final SearchState state) {
+ final long length = arrayLength(array);
+ final Class<?> elementClass = arrayClass.getComponentType();
+
+ // Arrays have object header and length field which is an integer
+ long arrSize = alignSize(objectSize + INT_SIZE);
+
+ if (elementClass.isPrimitive()) {
+ arrSize += alignSize(length * primitiveSize(elementClass));
+ state.size += arrSize;
+ } else {
+ arrSize += alignSize(length * pointerSize);
+ state.size += arrSize;
+
+ if (length <= ARRAY_SIZE_FOR_SAMPLING) {
+ int arrayIndex = 0;
+ while (arrayIndex < length) {
+ state.enqueue(arrayApply(array, arrayIndex));
+ arrayIndex += 1;
+ }
+ } else {
+ // Estimate the size of a large array by sampling elements without replacement.
+ // To exclude the shared objects that the array elements may link, sample twice
+ // and use the min one to calculate array size.
+ final Random rand = new Random(42);
+ final HashSet<Integer> drawn = new HashSet<>(2 * ARRAY_SAMPLE_SIZE);
+ final long s1 = sampleArray(array, state, rand, drawn, (int) length);
+ final long s2 = sampleArray(array, state, rand, drawn, (int) length);
+ final long size = Math.min(s1, s2);
+ state.size += Math.max(s1, s2) +
+ (size * ((length - ARRAY_SAMPLE_SIZE) / ARRAY_SAMPLE_SIZE));
+ }
+ }
+ }
+
+ private static long sampleArray(final Object array, final SearchState state, final Random rand,
+ final HashSet<Integer> drawn, final int length) {
+ long size = 0L;
+ for (int i = 0; i < ARRAY_SAMPLE_SIZE; i++) {
+ int index;
+ do {
+ index = rand.nextInt(length);
+ } while (drawn.contains(index));
+ drawn.add(index);
+ final Object obj = arrayApply(array, index);
+ if (obj != null) {
+ size += SizeEstimator.estimate(obj, state.visited);
+ }
+ }
+ return size;
+ }
+
+ private static Object arrayApply(final Object obj, final int index) {
+ if (obj instanceof Object[]) {
+ return ((Object[])obj)[index];
+ } else if (obj instanceof byte[]) {
+ return ((byte[]) obj)[index];
+ } else if (obj instanceof int[]) {
+ return ((int[]) obj)[index];
+ } else if (obj instanceof short[]) {
+ return ((short[])obj)[index];
+ } else if (obj instanceof long[]) {
+ return ((long[])obj)[index];
+ } else if (obj instanceof boolean[]) {
+ return ((boolean[]) obj)[index];
+ } else if (obj instanceof float[]) {
+ return ((float[]) obj)[index];
+ } else if (obj instanceof double[]) {
+ return ((double[]) obj)[index];
+ } else if (obj instanceof char[]) {
+ return ((char[]) obj)[index];
+ }
+ throw new IllegalArgumentException("illegal input for arrayApply " + obj);
+ }
+
+ private static int arrayLength(final Object obj) {
+ if (obj instanceof Object[]) {
+ return ((Object[])obj).length;
+ } else if (obj instanceof byte[]) {
+ return ((byte[]) obj).length;
+ } else if (obj instanceof int[]) {
+ return ((int[]) obj).length;
+ } else if (obj instanceof short[]) {
+ return ((short[])obj).length;
+ } else if (obj instanceof long[]) {
+ return ((long[])obj).length;
+ } else if (obj instanceof boolean[]) {
+ return ((boolean[]) obj).length;
+ } else if (obj instanceof float[]) {
+ return ((float[]) obj).length;
+ } else if (obj instanceof double[]) {
+ return ((double[])obj).length;
+ } else if (obj instanceof char[]) {
+ return ((char[]) obj).length;
+ }
+ throw new IllegalArgumentException("illegal input for arrayLength " + obj);
+ }
+
+ private static int primitiveSize(Class<?> cls) {
+ if (cls == byte.class) {
+ return BYTE_SIZE;
+ } else if (cls == boolean.class) {
+ return BOOLEAN_SIZE;
+ } else if (cls == char.class) {
+ return CHAR_SIZE;
+ } else if (cls == short.class) {
+ return SHORT_SIZE;
+ } else if (cls == int.class) {
+ return INT_SIZE;
+ } else if (cls == long.class) {
+ return LONG_SIZE;
+ } else if (cls == float.class) {
+ return FLOAT_SIZE;
+ } else if (cls == double.class) {
+ return DOUBLE_SIZE;
+ } else {
+ throw new IllegalArgumentException(
+ "Non-primitive class " + cls + " passed to primitiveSize()");
+ }
+ }
+
+ /**
+ * Get or compute the ClassInfo for a given class.
+ */
+ private static ClassInfo getClassInfo(Class<?> cls) {
+ // Check whether we've already cached a ClassInfo for this class
+ final ClassInfo info = classInfos.get(cls);
+ if (info != null) {
+ return info;
+ }
+
+ final ClassInfo parent = getClassInfo(cls.getSuperclass());
+ long shellSize = parent.getShellSize();
+ LinkedList<Field> pointerFields = parent.getPointerFields();
+ final int[] sizeCount = new int[MAX_FIELD_SIZE + 1];
+
+ for (Field field : cls.getDeclaredFields()) {
+ if (!Modifier.isStatic(field.getModifiers())) {
+ final Class<?> fieldClass = field.getType();
+ if (fieldClass.isPrimitive()) {
+ sizeCount[primitiveSize(fieldClass)] += 1;
+ } else {
+ // Note: in Java 9+ this would be better with trySetAccessible and canAccess
+ try {
+ field.setAccessible(true); // Enable future get()'s on this field
+ pointerFields.add(0, field);
+ } catch(SecurityException se) {
+ // do nothing
+ // Java 9+ can throw InaccessibleObjectException but the class is Java 9+-only
+ } catch (RuntimeException re) {
+ if (re.getClass().getSimpleName().equals("InaccessibleObjectException")) {
+ // do nothing
+ } else {
+ throw re;
+ }
+ }
+ sizeCount[pointerSize] += 1;
+ }
+ }
+ }
+
+ // Based on the simulated field layout code in Aleksey Shipilev's report:
+ // http://cr.openjdk.java.net/~shade/papers/2013-shipilev-fieldlayout-latest.pdf
+ // The code is in Figure 9.
+ // The simplified idea of field layout consists of 4 parts (see more details in the report):
+ //
+ // 1. field alignment: HotSpot lays out the fields aligned by their size.
+ // 2. object alignment: HotSpot rounds instance size up to 8 bytes
+ // 3. consistent fields layouts throughout the hierarchy: This means we should layout
+ // superclass first. And we can use superclass's shellSize as a starting point to layout the
+ // other fields in this class.
+ // 4. class alignment: HotSpot rounds field blocks up to HeapOopSize not 4 bytes, confirmed
+ // with Aleksey. see https://bugs.openjdk.java.net/browse/CODETOOLS-7901322
+ //
+ // The real world field layout is much more complicated. There are three kinds of fields
+ // order in Java 8. And we don't consider the @contended annotation introduced by Java 8.
+ // see the HotSpot classloader code, layout_fields method for more details.
+ // hg.openjdk.java.net/jdk8/jdk8/hotspot/file/tip/src/share/vm/classfile/classFileParser.cpp
+ long alignedSize = shellSize;
+ for (Integer size : fieldSizes) {
+ if (sizeCount[size] > 0) {
+ final long count = sizeCount[size];
+ // If there are internal gaps, smaller field can fit in.
+ alignedSize = Math.max(alignedSize, alignSizeUp(shellSize, size) + size * count);
+ shellSize += size * count;
+ }
+ }
+
+ // Should choose a larger size to be new shellSize and clearly alignedSize >= shellSize, and
+ // round up the instance filed blocks
+ shellSize = alignSizeUp(alignedSize, pointerSize);
+
+ // Create and cache a new ClassInfo
+ final ClassInfo newInfo = new ClassInfo(shellSize, pointerFields);
+ classInfos.put(cls, newInfo);
+ return newInfo;
+ }
+
+ private static long alignSize(final long size) {
+ return alignSizeUp(size, ALIGN_SIZE);
+ }
+
+ /**
+ * Compute aligned size. The alignSize must be 2^n, otherwise the result will be wrong.
+ * When alignSize = 2^n, alignSize - 1 = 2^n - 1. The binary representation of (alignSize - 1)
+ * will only have n trailing 1s(0b00...001..1). ~(alignSize - 1) will be 0b11..110..0. Hence,
+ * (size + alignSize - 1) & ~(alignSize - 1) will set the last n bits to zeros, which leads to
+ * multiple of alignSize.
+ */
+ private static long alignSizeUp(final long size, final int alignSize) {
+ return (size + alignSize - 1) & ~(alignSize - 1);
+ }
+
+ private static Long knownSize(final Object obj) {
+ try {
+ final Method method = obj.getClass().getMethod("estimatedSize");
+ return (Long) method.invoke(obj);
+ } catch (Exception e) {
+ return null;
+ }
+ }
+}
diff --git a/contrib/storage-opentsdb/src/test/java/org/apache/drill/store/openTSDB/TestDataHolder.java b/contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestDataHolder.java
similarity index 99%
rename from contrib/storage-opentsdb/src/test/java/org/apache/drill/store/openTSDB/TestDataHolder.java
rename to contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestDataHolder.java
index c6e7228450..7e5fef04f9 100644
--- a/contrib/storage-opentsdb/src/test/java/org/apache/drill/store/openTSDB/TestDataHolder.java
+++ b/contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestDataHolder.java
@@ -15,7 +15,7 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.store.openTSDB;
+package org.apache.drill.exec.store.openTSDB;
public class TestDataHolder {
diff --git a/contrib/storage-opentsdb/src/test/java/org/apache/drill/store/openTSDB/TestOpenTSDBPlugin.java b/contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestOpenTSDBPlugin.java
similarity index 84%
rename from contrib/storage-opentsdb/src/test/java/org/apache/drill/store/openTSDB/TestOpenTSDBPlugin.java
rename to contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestOpenTSDBPlugin.java
index 1c8207ae0c..e021c15ee8 100644
--- a/contrib/storage-opentsdb/src/test/java/org/apache/drill/store/openTSDB/TestOpenTSDBPlugin.java
+++ b/contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestOpenTSDBPlugin.java
@@ -15,13 +15,12 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-package org.apache.drill.store.openTSDB;
+package org.apache.drill.exec.store.openTSDB;
import com.github.tomakehurst.wiremock.junit.WireMockRule;
import org.apache.drill.common.exceptions.UserRemoteException;
import org.apache.drill.exec.proto.UserBitShared.QueryType;
import org.apache.drill.exec.store.StoragePluginRegistry;
-import org.apache.drill.exec.store.openTSDB.OpenTSDBStoragePluginConfig;
import org.apache.drill.test.ClientFixture;
import org.apache.drill.test.ClusterFixture;
import org.apache.drill.test.ClusterTest;
@@ -36,20 +35,20 @@ import static com.github.tomakehurst.wiremock.client.WireMock.equalToJson;
import static com.github.tomakehurst.wiremock.client.WireMock.get;
import static com.github.tomakehurst.wiremock.client.WireMock.post;
import static com.github.tomakehurst.wiremock.client.WireMock.urlEqualTo;
-import static org.apache.drill.store.openTSDB.TestDataHolder.DOWNSAMPLE_REQUEST_WITH_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.DOWNSAMPLE_REQUEST_WTIHOUT_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.END_PARAM_REQUEST_WITH_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.END_PARAM_REQUEST_WTIHOUT_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.POST_REQUEST_WITHOUT_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.POST_REQUEST_WITH_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.REQUEST_TO_NONEXISTENT_METRIC;
-import static org.apache.drill.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_GET_TABLE_NAME_REQUEST;
-import static org.apache.drill.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_GET_TABLE_REQUEST;
-import static org.apache.drill.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_DOWNSAMPLE_REQUEST_WITHOUT_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_DOWNSAMPLE_REQUEST_WITH_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_END_REQUEST_WITHOUT_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_END_REQUEST_WITH_TAGS;
-import static org.apache.drill.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_REQUEST_WITH_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.DOWNSAMPLE_REQUEST_WITH_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.DOWNSAMPLE_REQUEST_WTIHOUT_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.END_PARAM_REQUEST_WITH_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.END_PARAM_REQUEST_WTIHOUT_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.POST_REQUEST_WITHOUT_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.POST_REQUEST_WITH_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.REQUEST_TO_NONEXISTENT_METRIC;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_GET_TABLE_NAME_REQUEST;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_GET_TABLE_REQUEST;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_DOWNSAMPLE_REQUEST_WITHOUT_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_DOWNSAMPLE_REQUEST_WITH_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_END_REQUEST_WITHOUT_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_END_REQUEST_WITH_TAGS;
+import static org.apache.drill.exec.store.openTSDB.TestDataHolder.SAMPLE_DATA_FOR_POST_REQUEST_WITH_TAGS;
import static org.junit.Assert.assertEquals;
public class TestOpenTSDBPlugin extends ClusterTest {
diff --git a/contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestSizeEstimator.java b/contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestSizeEstimator.java
new file mode 100644
index 0000000000..95e5a5b470
--- /dev/null
+++ b/contrib/storage-opentsdb/src/test/java/org/apache/drill/exec/store/openTSDB/TestSizeEstimator.java
@@ -0,0 +1,87 @@
+/*
+ * 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.openTSDB;
+
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+public class TestSizeEstimator {
+
+ private static class TestMetricDTO {
+ private final String metric;
+ private final Map<String, String> tags;
+ private final List<String> aggregateTags;
+ private final Map<String, String> dps;
+
+ TestMetricDTO(String metric, Map<String, String> tags, List<String> aggregateTags, Map<String, String> dps) {
+ this.metric = metric;
+ this.tags = tags;
+ this.aggregateTags = aggregateTags;
+ this.dps = dps;
+ }
+ }
+
+ @Test
+ public void testArrays() {
+ assertEquals(32, SizeEstimator.estimate(new byte[10]));
+ assertEquals(40, SizeEstimator.estimate(new char[10]));
+ assertEquals(40, SizeEstimator.estimate(new short[10]));
+ assertEquals(56, SizeEstimator.estimate(new int[10]));
+ assertEquals(96, SizeEstimator.estimate(new long[10]));
+ assertEquals(56, SizeEstimator.estimate(new float[10]));
+ assertEquals(96, SizeEstimator.estimate(new double[10]));
+ assertEquals(4016, SizeEstimator.estimate(new int[1000]));
+ assertEquals(8016, SizeEstimator.estimate(new long[1000]));
+ assertEquals(56, SizeEstimator.estimate(new String[10]));
+ assertEquals(56, SizeEstimator.estimate(new Object[10]));
+ }
+
+ @Test
+ public void testList() {
+ ArrayList<Long> list = new ArrayList<>();
+ for(int i = 0; i < 10; i++) {
+ list.add(Long.valueOf(i));
+ }
+ assertEquals(320, SizeEstimator.estimate(list));
+ for(int i = 0; i < 10; i++) {
+ list.add(Long.valueOf(i));
+ }
+ assertEquals(368, SizeEstimator.estimate(list));
+ }
+
+ @Test
+ public void testMetricDTO() {
+ Map<String, String> tags = new HashMap<>();
+ Map<String, String> dps = new HashMap<>();
+ tags.put("t1", "v1");
+ dps.put("dp1", "dpv1");
+ List<String> aggregateTags = new ArrayList<>();
+ TestMetricDTO dto = new TestMetricDTO("metric1", tags, aggregateTags, dps);
+ long size = SizeEstimator.estimate(dto);
+ assertTrue("size less then expected: " + size, size > 550);
+ assertTrue("size greater then expected: " + size, size < 800);
+ }
+
+}
[drill] 01/10: DRILL-8291: Allow Case Sensitive Filters in HTTP Plugin (#2639)
Posted by dz...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
dzamo pushed a commit to branch 1.20
in repository https://gitbox.apache.org/repos/asf/drill.git
commit fb1be8d2044c9981611e03a1835cb8cab4c664cb
Author: Charles S. Givre <cg...@apache.org>
AuthorDate: Mon Sep 5 19:44:08 2022 +0800
DRILL-8291: Allow Case Sensitive Filters in HTTP Plugin (#2639)
* DRILL-8291: Allow Case Sensitive Filters in HTTP Plugin
---
contrib/storage-http/README.md | 7 +++++--
.../apache/drill/exec/store/http/HttpApiConfig.java | 19 ++++++++++++++++++-
.../drill/exec/store/http/HttpPushDownListener.java | 9 ++++++++-
.../apache/drill/exec/store/http/util/SimpleHttp.java | 4 +---
4 files changed, 32 insertions(+), 7 deletions(-)
diff --git a/contrib/storage-http/README.md b/contrib/storage-http/README.md
index 79c76f6491..35105554fd 100644
--- a/contrib/storage-http/README.md
+++ b/contrib/storage-http/README.md
@@ -305,6 +305,9 @@ on 400 series errors, `false` will return an empty result set (with implicit fie
Default is `true`, but when set to false, Drill will trust all SSL certificates. Useful for debugging or on internal corporate networks using self-signed certificates or
private certificate authorities.
+#### caseSensitiveFilters
+Some APIs are case sensitive with the fields which are pushed down. If the endpoint that you are working with is in fact case sensitive, simply set this to `true`. Defaults to `false`.
+
## Usage
This plugin is different from other plugins in that it the table component of the `FROM` clause
@@ -576,9 +579,9 @@ ORDER BY issue_count DESC
you may encounter errors or empty responses if you are counting on the endpoint for
redirection.
-2. At this time, the plugin does not support any authentication other than basic authentication.
+~~2. At this time, the plugin does not support any authentication other than basic authentication.~~
-3. This plugin does not implement join filter pushdowns (only constant plushdowns are
+3. This plugin does not implement join filter pushdowns (only constant pushdowns are
supported). Join pushdown has the potential to improve performance if you use the HTTP service
joined to another table.
diff --git a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpApiConfig.java b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpApiConfig.java
index 09d6b2ff63..83dd631f09 100644
--- a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpApiConfig.java
+++ b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpApiConfig.java
@@ -94,6 +94,8 @@ public class HttpApiConfig {
private final String limitQueryParam;
@JsonProperty
private final boolean errorOn400;
+ @JsonProperty
+ private final boolean caseSensitiveFilters;
// Enables the user to configure JSON options at the connection level rather than globally.
@JsonProperty
@@ -186,6 +188,7 @@ public class HttpApiConfig {
&& errorOn400 == that.errorOn400
&& verifySSLCert == that.verifySSLCert
&& directCredentials == that.directCredentials
+ && caseSensitiveFilters == that.caseSensitiveFilters
&& Objects.equals(url, that.url)
&& Objects.equals(method, that.method)
&& Objects.equals(postBody, that.postBody)
@@ -204,7 +207,7 @@ public class HttpApiConfig {
public int hashCode() {
return Objects.hash(url, requireTail, method, postBody, headers, params, dataPath,
authType, inputType, xmlDataLevel, limitQueryParam, errorOn400, jsonOptions, verifySSLCert,
- credentialsProvider, paginator, directCredentials);
+ credentialsProvider, paginator, directCredentials, caseSensitiveFilters);
}
@Override
@@ -217,6 +220,7 @@ public class HttpApiConfig {
.field("headers", headers)
.field("params", params)
.field("dataPath", dataPath)
+ .field("caseSensitiveFilters", caseSensitiveFilters)
.field("authType", authType)
.field("inputType", inputType)
.field("xmlDataLevel", xmlDataLevel)
@@ -285,6 +289,7 @@ public class HttpApiConfig {
this.xmlDataLevel = Math.max(1, builder.xmlDataLevel);
this.errorOn400 = builder.errorOn400;
+ this.caseSensitiveFilters = builder.caseSensitiveFilters;
this.credentialsProvider = CredentialProviderUtils.getCredentialsProvider(builder.userName, builder.password, builder.credentialsProvider);
this.directCredentials = builder.credentialsProvider == null;
@@ -331,6 +336,11 @@ public class HttpApiConfig {
return credentialsProvider;
}
+ @JsonProperty
+ public boolean caseSensitiveFilters() {
+ return this.caseSensitiveFilters;
+ }
+
@JsonPOJOBuilder(withPrefix = "")
public static class HttpApiConfigBuilder {
private String userName;
@@ -359,6 +369,8 @@ public class HttpApiConfig {
private int xmlDataLevel;
+ private boolean caseSensitiveFilters;
+
private String limitQueryParam;
private boolean errorOn400;
@@ -425,6 +437,11 @@ public class HttpApiConfig {
return this;
}
+ public HttpApiConfigBuilder caseSensitiveFilters(boolean caseSensitiveFilters) {
+ this.caseSensitiveFilters = caseSensitiveFilters;
+ return this;
+ }
+
public HttpApiConfigBuilder method(String method) {
this.method = method;
return this;
diff --git a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpPushDownListener.java b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpPushDownListener.java
index c4f55967e6..30f4be57fa 100644
--- a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpPushDownListener.java
+++ b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/HttpPushDownListener.java
@@ -35,6 +35,7 @@ import org.apache.drill.exec.store.http.HttpPaginatorConfig.PaginatorMethod;
import org.apache.drill.exec.store.http.util.SimpleHttp;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -175,7 +176,13 @@ public class HttpPushDownListener implements FilterPushDownListener {
*/
@Override
public Pair<GroupScan, List<RexNode>> transform(AndNode andNode) {
- Map<String, String> filters = CaseInsensitiveMap.newHashMap();
+ Map<String, String> filters;
+ if (groupScan.getHttpConfig().caseSensitiveFilters()) {
+ filters = new HashMap<>();
+ } else {
+ filters = CaseInsensitiveMap.newHashMap();
+ }
+
double selectivity = 1;
for (ExprNode expr : andNode.children) {
ColRelOpConstNode relOp = (ColRelOpConstNode) expr;
diff --git a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/util/SimpleHttp.java b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/util/SimpleHttp.java
index 777f49d207..03e4d5e5cc 100644
--- a/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/util/SimpleHttp.java
+++ b/contrib/storage-http/src/main/java/org/apache/drill/exec/store/http/util/SimpleHttp.java
@@ -28,7 +28,6 @@ import okhttp3.Request;
import okhttp3.Response;
import org.apache.commons.lang3.StringUtils;
-import org.apache.drill.common.map.CaseInsensitiveMap;
import org.apache.drill.common.exceptions.CustomErrorContext;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.exec.oauth.PersistentTokenTable;
@@ -529,7 +528,6 @@ public class SimpleHttp {
.message("API Query with URL Parameters must be populated.")
.build(logger);
}
- CaseInsensitiveMap<String>caseInsensitiveFilterMap = (CaseInsensitiveMap<String>)filters;
List<String> params = SimpleHttp.getURLParameters(url);
String tempUrl = SimpleHttp.decodedURL(url);
@@ -543,7 +541,7 @@ public class SimpleHttp {
// of providing helpful errors in strange cases, it is there.
- String value = caseInsensitiveFilterMap.get(param);
+ String value = filters.get(param);
// Check and see if there is a default for this parameter. If not throw an error.
if (StringUtils.isEmpty(value)) {