[
https://issues.apache.org/jira/browse/DRILL-8555?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18132971#comment-18132971
]
ASF GitHub Bot commented on DRILL-8555:
---------------------------------------
letian-jiang commented on code in PR #3086:
URL: https://github.com/apache/drill/pull/3086#discussion_r4390303319
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,744 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.planner.sql;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlWith;
+import org.apache.calcite.sql.SqlWithItem;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
+import org.apache.calcite.util.NlsString;
+import org.apache.drill.common.expression.AnyValueExpression;
+import org.apache.drill.common.expression.BooleanOperator;
+import org.apache.drill.common.expression.CastExpression;
+import org.apache.drill.common.expression.ConvertExpression;
+import org.apache.drill.common.expression.ExpressionStringBuilder;
+import org.apache.drill.common.expression.FunctionCall;
+import org.apache.drill.common.expression.IfExpression;
+import org.apache.drill.common.expression.LiteralExpression;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.common.expression.NullExpression;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.expression.TypedNullConstant;
+import org.apache.drill.common.expression.ValueExpressions;
+import org.apache.drill.common.logical.StoragePluginConfig;
+import org.apache.drill.common.parser.LogicalExpressionParser;
+import org.apache.drill.common.types.TypeProtos.MajorType;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.metrics.DrillMetrics;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.logical.DrillTable;
+import org.apache.drill.exec.planner.logical.DrillTableSelection;
+import org.apache.drill.exec.rpc.NamedThreadFactory;
+import org.apache.drill.exec.server.options.OptionValue;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.PlanCacheTable;
+import org.apache.drill.exec.util.Pointer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Ticker;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.hash.Hashing;
+import com.google.common.primitives.Ints;
+import com.codahale.metrics.Gauge;
+
+/** Drillbit-scoped, immutable JSON snapshots of query plans. */
+public final class PlanCache implements AutoCloseable {
+ private static final Logger logger =
LoggerFactory.getLogger(PlanCache.class);
+ private static final ObjectMapper JSON = new ObjectMapper();
+ private static final String MARKER =
ExpressionStringBuilder.BOUND_DYNAMIC_PARAM + "(";
+ private final Cache<String, Entry> entries;
+ private final AtomicLong successfulBinds = new AtomicLong();
+ private final AtomicLong misses = new AtomicLong();
+ private final AtomicLong invalidations = new AtomicLong();
+ private final Map<String, Gauge<Long>> metrics = new LinkedHashMap<>();
+ private volatile long generation;
+ private final ThreadPoolExecutor writer = new ThreadPoolExecutor(1, 1, 0L,
+ TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
+ new NamedThreadFactory("plan-cache-writer-"));
+
+ /** A zero expiration duration disables that policy; a zero capacity
disables storage. */
+ public PlanCache(long maxSizeBytes, Duration expireAfterWrite, Duration
expireAfterAccess) {
+ this(maxSizeBytes, expireAfterWrite, expireAfterAccess,
Ticker.systemTicker());
+ }
+
+ @VisibleForTesting
+ PlanCache(long maxSizeBytes, Duration expireAfterWrite, Duration
expireAfterAccess, Ticker ticker) {
+ checkArgument(maxSizeBytes >= 0, "%s must be non-negative",
ExecConstants.PLAN_CACHE_MAX_SIZE_BYTES);
+ checkArgument(!expireAfterWrite.isNegative(), "%s must be non-negative",
+ ExecConstants.PLAN_CACHE_EXPIRE_AFTER_WRITE);
+ checkArgument(!expireAfterAccess.isNegative(), "%s must be non-negative",
+ ExecConstants.PLAN_CACHE_EXPIRE_AFTER_ACCESS);
+ CacheBuilder<String, Entry> builder = CacheBuilder.newBuilder()
+ .maximumWeight(maxSizeBytes)
+ .weigher((String key, Entry value) -> Ints.saturatedCast(
+ key.getBytes(StandardCharsets.UTF_8).length +
value.contentSizeBytes))
+ .ticker(ticker)
+ .recordStats();
+ // Passing zero to Guava would expire entries immediately, rather than
disable the policy.
+ if (!expireAfterWrite.isZero()) {
+ builder.expireAfterWrite(expireAfterWrite);
+ }
+ if (!expireAfterAccess.isZero()) {
+ builder.expireAfterAccess(expireAfterAccess);
+ }
+ entries = builder.build();
+ logger.info("Plan cache settings: max_size_bytes={},
expire_after_write={}, expire_after_access={}"
+ + " (zero expiration disables that policy)", maxSizeBytes,
expireAfterWrite, expireAfterAccess);
+ registerMetric("hits", successfulBinds::get);
+ registerMetric("misses", misses::get);
+ registerMetric("invalidations", invalidations::get);
+ registerMetric("evictions", () -> entries.stats().evictionCount());
+ registerMetric("entries", entries::size);
+ }
+
+ private void registerMetric(String suffix, Gauge<Long> gauge) {
+ String name = "drill.plan_cache." + suffix;
+ metrics.put(name, gauge);
+ DrillMetrics.register(name, gauge);
+ }
+
+ public long getHitCount() {
+ return successfulBinds.get();
+ }
+
+ public void recordBind() {
+ successfulBinds.incrementAndGet();
+ }
+
+ /** Counts an eligible lookup that could not produce a compatible, bound
plan. */
+ public void recordMiss() {
+ misses.incrementAndGet();
+ }
+
+ /** Updates optional explain output using the cached template. */
+ static void bindTextPlan(Pointer<String> textPlan, String template,
List<SqlLiteral> literals) {
+ if (textPlan != null) {
+ textPlan.value = textWithBindings(template, literals);
+ }
+ }
+
+ /** Updates optional explain output and returns its original template for
cache publication. */
+ static String bindTextPlan(Pointer<String> textPlan, List<SqlLiteral>
literals) {
+ String template = textPlan == null ? null : textPlan.value;
+ bindTextPlan(textPlan, template, literals);
+ return template;
+ }
+
+ /** Appends values from this query without changing the cached explain
template. */
+ private static String textWithBindings(String template, List<SqlLiteral>
literals) {
+ if (template == null || literals.isEmpty()) {
+ return template;
+ }
+ StringBuilder text = new StringBuilder(template);
+ if (text.length() > 0 && text.charAt(text.length() - 1) != '\n') {
+ text.append('\n');
+ }
+ text.append("Parameters: ");
+ for (int i = 0; i < literals.size(); i++) {
+ if (i > 0) {
+ text.append(", ");
+ }
+ text.append('?').append(i).append(" = ")
+ .append(literals.get(i).toString().replace("\r",
"\\r").replace("\n", "\\n"));
+ }
+ return text.toString();
+ }
+
+ public Entry get(String key) {
+ return entries.getIfPresent(key);
+ }
+
+ /** Waits for writes already submitted by completed queries; intended for
tests. */
+ public void awaitWrites() throws InterruptedException, ExecutionException,
TimeoutException {
+ writer.submit(() -> { }).get(10, TimeUnit.SECONDS);
+ }
+
+ public void writeAfterSuccess(String key, String json, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context, long
expectedGeneration) {
+ try {
+ writer.execute(() -> {
+ try {
+ put(key, json, textPlan, reader, context, expectedGeneration);
+ } catch (RuntimeException | IOException e) {
+ logger.debug("Plan cache entry could not be published", e);
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ logger.debug("Plan cache writer is busy; skipping cache entry", e);
+ }
+ }
+
+ public void invalidate(String key) {
+ invalidations.incrementAndGet();
+ entries.invalidate(key);
+ }
+
+ long getGeneration() {
+ return generation;
+ }
+
+ /** Clears local entries and prevents pre-clear plans from being published
later. */
+ public synchronized void clear() {
+ generation++;
+ invalidations.incrementAndGet();
+ entries.invalidateAll();
+ entries.cleanUp();
+ logger.info("Local plan cache cleared; generation={}", generation);
+ }
+
+ public boolean put(String key, String json, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context) throws IOException {
+ return put(key, json, textPlan, reader, context, generation);
+ }
+
+ private boolean put(String key, String json, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context, long
expectedGeneration) throws IOException {
+ // The engine checks generic context and expression compatibility. Each
+ // opted-in plugin is responsible for rebuilding its scans'
value-dependent state.
+ if (context == null || expectedGeneration != generation) {
+ return false;
+ }
+ // Confirm that standard PhysicalPlan JSON can be read back before
publishing.
+ reader.readPhysicalPlan(json);
Review Comment:
Added trial binding before publication, replacing JSON-only validation.
Unbindable templates—for example, ordinary text containing the parameter
marker—are skipped without affecting the completed query.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,744 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.planner.sql;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlWith;
+import org.apache.calcite.sql.SqlWithItem;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
+import org.apache.calcite.util.NlsString;
+import org.apache.drill.common.expression.AnyValueExpression;
+import org.apache.drill.common.expression.BooleanOperator;
+import org.apache.drill.common.expression.CastExpression;
+import org.apache.drill.common.expression.ConvertExpression;
+import org.apache.drill.common.expression.ExpressionStringBuilder;
+import org.apache.drill.common.expression.FunctionCall;
+import org.apache.drill.common.expression.IfExpression;
+import org.apache.drill.common.expression.LiteralExpression;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.common.expression.NullExpression;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.expression.TypedNullConstant;
+import org.apache.drill.common.expression.ValueExpressions;
+import org.apache.drill.common.logical.StoragePluginConfig;
+import org.apache.drill.common.parser.LogicalExpressionParser;
+import org.apache.drill.common.types.TypeProtos.MajorType;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.metrics.DrillMetrics;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.logical.DrillTable;
+import org.apache.drill.exec.planner.logical.DrillTableSelection;
+import org.apache.drill.exec.rpc.NamedThreadFactory;
+import org.apache.drill.exec.server.options.OptionValue;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.PlanCacheTable;
+import org.apache.drill.exec.util.Pointer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Ticker;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.hash.Hashing;
+import com.google.common.primitives.Ints;
+import com.codahale.metrics.Gauge;
+
+/** Drillbit-scoped, immutable JSON snapshots of query plans. */
+public final class PlanCache implements AutoCloseable {
+ private static final Logger logger =
LoggerFactory.getLogger(PlanCache.class);
+ private static final ObjectMapper JSON = new ObjectMapper();
+ private static final String MARKER =
ExpressionStringBuilder.BOUND_DYNAMIC_PARAM + "(";
+ private final Cache<String, Entry> entries;
+ private final AtomicLong successfulBinds = new AtomicLong();
+ private final AtomicLong misses = new AtomicLong();
+ private final AtomicLong invalidations = new AtomicLong();
+ private final Map<String, Gauge<Long>> metrics = new LinkedHashMap<>();
+ private volatile long generation;
+ private final ThreadPoolExecutor writer = new ThreadPoolExecutor(1, 1, 0L,
+ TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
+ new NamedThreadFactory("plan-cache-writer-"));
+
+ /** A zero expiration duration disables that policy; a zero capacity
disables storage. */
+ public PlanCache(long maxSizeBytes, Duration expireAfterWrite, Duration
expireAfterAccess) {
+ this(maxSizeBytes, expireAfterWrite, expireAfterAccess,
Ticker.systemTicker());
+ }
+
+ @VisibleForTesting
+ PlanCache(long maxSizeBytes, Duration expireAfterWrite, Duration
expireAfterAccess, Ticker ticker) {
+ checkArgument(maxSizeBytes >= 0, "%s must be non-negative",
ExecConstants.PLAN_CACHE_MAX_SIZE_BYTES);
+ checkArgument(!expireAfterWrite.isNegative(), "%s must be non-negative",
+ ExecConstants.PLAN_CACHE_EXPIRE_AFTER_WRITE);
+ checkArgument(!expireAfterAccess.isNegative(), "%s must be non-negative",
+ ExecConstants.PLAN_CACHE_EXPIRE_AFTER_ACCESS);
+ CacheBuilder<String, Entry> builder = CacheBuilder.newBuilder()
+ .maximumWeight(maxSizeBytes)
+ .weigher((String key, Entry value) -> Ints.saturatedCast(
+ key.getBytes(StandardCharsets.UTF_8).length +
value.contentSizeBytes))
+ .ticker(ticker)
+ .recordStats();
+ // Passing zero to Guava would expire entries immediately, rather than
disable the policy.
+ if (!expireAfterWrite.isZero()) {
+ builder.expireAfterWrite(expireAfterWrite);
+ }
+ if (!expireAfterAccess.isZero()) {
+ builder.expireAfterAccess(expireAfterAccess);
+ }
+ entries = builder.build();
+ logger.info("Plan cache settings: max_size_bytes={},
expire_after_write={}, expire_after_access={}"
+ + " (zero expiration disables that policy)", maxSizeBytes,
expireAfterWrite, expireAfterAccess);
+ registerMetric("hits", successfulBinds::get);
+ registerMetric("misses", misses::get);
+ registerMetric("invalidations", invalidations::get);
+ registerMetric("evictions", () -> entries.stats().evictionCount());
+ registerMetric("entries", entries::size);
+ }
+
+ private void registerMetric(String suffix, Gauge<Long> gauge) {
+ String name = "drill.plan_cache." + suffix;
+ metrics.put(name, gauge);
+ DrillMetrics.register(name, gauge);
+ }
+
+ public long getHitCount() {
+ return successfulBinds.get();
+ }
+
+ public void recordBind() {
+ successfulBinds.incrementAndGet();
+ }
+
+ /** Counts an eligible lookup that could not produce a compatible, bound
plan. */
+ public void recordMiss() {
+ misses.incrementAndGet();
+ }
+
+ /** Updates optional explain output using the cached template. */
+ static void bindTextPlan(Pointer<String> textPlan, String template,
List<SqlLiteral> literals) {
+ if (textPlan != null) {
+ textPlan.value = textWithBindings(template, literals);
+ }
+ }
+
+ /** Updates optional explain output and returns its original template for
cache publication. */
+ static String bindTextPlan(Pointer<String> textPlan, List<SqlLiteral>
literals) {
+ String template = textPlan == null ? null : textPlan.value;
+ bindTextPlan(textPlan, template, literals);
+ return template;
+ }
+
+ /** Appends values from this query without changing the cached explain
template. */
+ private static String textWithBindings(String template, List<SqlLiteral>
literals) {
+ if (template == null || literals.isEmpty()) {
+ return template;
+ }
+ StringBuilder text = new StringBuilder(template);
+ if (text.length() > 0 && text.charAt(text.length() - 1) != '\n') {
+ text.append('\n');
+ }
+ text.append("Parameters: ");
+ for (int i = 0; i < literals.size(); i++) {
+ if (i > 0) {
+ text.append(", ");
+ }
+ text.append('?').append(i).append(" = ")
+ .append(literals.get(i).toString().replace("\r",
"\\r").replace("\n", "\\n"));
+ }
+ return text.toString();
+ }
+
+ public Entry get(String key) {
+ return entries.getIfPresent(key);
+ }
+
+ /** Waits for writes already submitted by completed queries; intended for
tests. */
+ public void awaitWrites() throws InterruptedException, ExecutionException,
TimeoutException {
+ writer.submit(() -> { }).get(10, TimeUnit.SECONDS);
+ }
+
+ public void writeAfterSuccess(String key, String json, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context, long
expectedGeneration) {
+ try {
+ writer.execute(() -> {
+ try {
+ put(key, json, textPlan, reader, context, expectedGeneration);
+ } catch (RuntimeException | IOException e) {
+ logger.debug("Plan cache entry could not be published", e);
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ logger.debug("Plan cache writer is busy; skipping cache entry", e);
+ }
+ }
+
+ public void invalidate(String key) {
+ invalidations.incrementAndGet();
+ entries.invalidate(key);
+ }
+
+ long getGeneration() {
+ return generation;
+ }
+
+ /** Clears local entries and prevents pre-clear plans from being published
later. */
+ public synchronized void clear() {
+ generation++;
+ invalidations.incrementAndGet();
+ entries.invalidateAll();
+ entries.cleanUp();
+ logger.info("Local plan cache cleared; generation={}", generation);
+ }
+
+ public boolean put(String key, String json, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context) throws IOException {
+ return put(key, json, textPlan, reader, context, generation);
+ }
+
+ private boolean put(String key, String json, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context, long
expectedGeneration) throws IOException {
+ // The engine checks generic context and expression compatibility. Each
+ // opted-in plugin is responsible for rebuilding its scans'
value-dependent state.
+ if (context == null || expectedGeneration != generation) {
+ return false;
+ }
+ // Confirm that standard PhysicalPlan JSON can be read back before
publishing.
+ reader.readPhysicalPlan(json);
+ // Readback may perform plugin I/O. Keep it outside the clear/publication
lock,
+ // then check again so a concurrent clear cannot be undone by that
readback.
+ synchronized (this) {
+ if (expectedGeneration != generation) {
+ logger.debug("Skipping plan cache publication after a cache clear");
+ return false;
+ }
+ entries.put(key, new Entry(json, textPlan, context));
Review Comment:
The next compatibility check rejects the stale entry. Since schema changes
are infrequent, I’m accepting the occasional extra replanning cycle for this PR.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,744 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.planner.sql;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.TreeMap;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.RejectedExecutionException;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+
+import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlIdentifier;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.SqlWith;
+import org.apache.calcite.sql.SqlWithItem;
+import org.apache.calcite.schema.SchemaPlus;
+import org.apache.calcite.schema.Table;
+import org.apache.calcite.util.NlsString;
+import org.apache.drill.common.expression.AnyValueExpression;
+import org.apache.drill.common.expression.BooleanOperator;
+import org.apache.drill.common.expression.CastExpression;
+import org.apache.drill.common.expression.ConvertExpression;
+import org.apache.drill.common.expression.ExpressionStringBuilder;
+import org.apache.drill.common.expression.FunctionCall;
+import org.apache.drill.common.expression.IfExpression;
+import org.apache.drill.common.expression.LiteralExpression;
+import org.apache.drill.common.expression.LogicalExpression;
+import org.apache.drill.common.expression.NullExpression;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.expression.TypedNullConstant;
+import org.apache.drill.common.expression.ValueExpressions;
+import org.apache.drill.common.logical.StoragePluginConfig;
+import org.apache.drill.common.parser.LogicalExpressionParser;
+import org.apache.drill.common.types.TypeProtos.MajorType;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.exec.ExecConstants;
+import org.apache.drill.exec.metrics.DrillMetrics;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.planner.PhysicalPlanReader;
+import org.apache.drill.exec.planner.logical.DrillTable;
+import org.apache.drill.exec.planner.logical.DrillTableSelection;
+import org.apache.drill.exec.rpc.NamedThreadFactory;
+import org.apache.drill.exec.server.options.OptionValue;
+import org.apache.drill.exec.store.StoragePlugin;
+import org.apache.drill.exec.store.StoragePluginRegistry;
+import org.apache.drill.exec.store.PlanCacheTable;
+import org.apache.drill.exec.util.Pointer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Ticker;
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.hash.Hashing;
+import com.google.common.primitives.Ints;
+import com.codahale.metrics.Gauge;
+
+/** Drillbit-scoped, immutable JSON snapshots of query plans. */
+public final class PlanCache implements AutoCloseable {
+ private static final Logger logger =
LoggerFactory.getLogger(PlanCache.class);
+ private static final ObjectMapper JSON = new ObjectMapper();
+ private static final String MARKER =
ExpressionStringBuilder.BOUND_DYNAMIC_PARAM + "(";
+ private final Cache<String, Entry> entries;
+ private final AtomicLong successfulBinds = new AtomicLong();
+ private final AtomicLong misses = new AtomicLong();
+ private final AtomicLong invalidations = new AtomicLong();
+ private final Map<String, Gauge<Long>> metrics = new LinkedHashMap<>();
+ private volatile long generation;
+ private final ThreadPoolExecutor writer = new ThreadPoolExecutor(1, 1, 0L,
+ TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
+ new NamedThreadFactory("plan-cache-writer-"));
+
+ /** A zero expiration duration disables that policy; a zero capacity
disables storage. */
+ public PlanCache(long maxSizeBytes, Duration expireAfterWrite, Duration
expireAfterAccess) {
+ this(maxSizeBytes, expireAfterWrite, expireAfterAccess,
Ticker.systemTicker());
+ }
+
+ @VisibleForTesting
+ PlanCache(long maxSizeBytes, Duration expireAfterWrite, Duration
expireAfterAccess, Ticker ticker) {
+ checkArgument(maxSizeBytes >= 0, "%s must be non-negative",
ExecConstants.PLAN_CACHE_MAX_SIZE_BYTES);
+ checkArgument(!expireAfterWrite.isNegative(), "%s must be non-negative",
+ ExecConstants.PLAN_CACHE_EXPIRE_AFTER_WRITE);
+ checkArgument(!expireAfterAccess.isNegative(), "%s must be non-negative",
+ ExecConstants.PLAN_CACHE_EXPIRE_AFTER_ACCESS);
+ CacheBuilder<String, Entry> builder = CacheBuilder.newBuilder()
+ .maximumWeight(maxSizeBytes)
+ .weigher((String key, Entry value) -> Ints.saturatedCast(
+ key.getBytes(StandardCharsets.UTF_8).length +
value.contentSizeBytes))
+ .ticker(ticker)
+ .recordStats();
+ // Passing zero to Guava would expire entries immediately, rather than
disable the policy.
+ if (!expireAfterWrite.isZero()) {
+ builder.expireAfterWrite(expireAfterWrite);
+ }
+ if (!expireAfterAccess.isZero()) {
+ builder.expireAfterAccess(expireAfterAccess);
+ }
+ entries = builder.build();
+ logger.info("Plan cache settings: max_size_bytes={},
expire_after_write={}, expire_after_access={}"
+ + " (zero expiration disables that policy)", maxSizeBytes,
expireAfterWrite, expireAfterAccess);
+ registerMetric("hits", successfulBinds::get);
+ registerMetric("misses", misses::get);
+ registerMetric("invalidations", invalidations::get);
+ registerMetric("evictions", () -> entries.stats().evictionCount());
+ registerMetric("entries", entries::size);
+ }
+
+ private void registerMetric(String suffix, Gauge<Long> gauge) {
+ String name = "drill.plan_cache." + suffix;
+ metrics.put(name, gauge);
+ DrillMetrics.register(name, gauge);
+ }
+
+ public long getHitCount() {
+ return successfulBinds.get();
+ }
+
+ public void recordBind() {
+ successfulBinds.incrementAndGet();
+ }
+
+ /** Counts an eligible lookup that could not produce a compatible, bound
plan. */
+ public void recordMiss() {
+ misses.incrementAndGet();
+ }
+
+ /** Updates optional explain output using the cached template. */
+ static void bindTextPlan(Pointer<String> textPlan, String template,
List<SqlLiteral> literals) {
+ if (textPlan != null) {
+ textPlan.value = textWithBindings(template, literals);
+ }
+ }
+
+ /** Updates optional explain output and returns its original template for
cache publication. */
+ static String bindTextPlan(Pointer<String> textPlan, List<SqlLiteral>
literals) {
+ String template = textPlan == null ? null : textPlan.value;
+ bindTextPlan(textPlan, template, literals);
+ return template;
+ }
+
+ /** Appends values from this query without changing the cached explain
template. */
+ private static String textWithBindings(String template, List<SqlLiteral>
literals) {
+ if (template == null || literals.isEmpty()) {
+ return template;
+ }
+ StringBuilder text = new StringBuilder(template);
+ if (text.length() > 0 && text.charAt(text.length() - 1) != '\n') {
+ text.append('\n');
+ }
+ text.append("Parameters: ");
+ for (int i = 0; i < literals.size(); i++) {
+ if (i > 0) {
+ text.append(", ");
+ }
+ text.append('?').append(i).append(" = ")
+ .append(literals.get(i).toString().replace("\r",
"\\r").replace("\n", "\\n"));
+ }
+ return text.toString();
+ }
+
+ public Entry get(String key) {
+ return entries.getIfPresent(key);
+ }
+
+ /** Waits for writes already submitted by completed queries; intended for
tests. */
+ public void awaitWrites() throws InterruptedException, ExecutionException,
TimeoutException {
+ writer.submit(() -> { }).get(10, TimeUnit.SECONDS);
+ }
+
+ public void writeAfterSuccess(String key, String json, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context, long
expectedGeneration) {
+ try {
+ writer.execute(() -> {
+ try {
+ put(key, json, textPlan, reader, context, expectedGeneration);
+ } catch (RuntimeException | IOException e) {
+ logger.debug("Plan cache entry could not be published", e);
+ }
+ });
+ } catch (RejectedExecutionException e) {
+ logger.debug("Plan cache writer is busy; skipping cache entry", e);
+ }
+ }
+
+ public void invalidate(String key) {
+ invalidations.incrementAndGet();
+ entries.invalidate(key);
+ }
+
+ long getGeneration() {
+ return generation;
+ }
+
+ /** Clears local entries and prevents pre-clear plans from being published
later. */
+ public synchronized void clear() {
+ generation++;
+ invalidations.incrementAndGet();
+ entries.invalidateAll();
+ entries.cleanUp();
+ logger.info("Local plan cache cleared; generation={}", generation);
+ }
+
+ public boolean put(String key, String json, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context) throws IOException {
+ return put(key, json, textPlan, reader, context, generation);
+ }
+
+ private boolean put(String key, String json, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context, long
expectedGeneration) throws IOException {
+ // The engine checks generic context and expression compatibility. Each
+ // opted-in plugin is responsible for rebuilding its scans'
value-dependent state.
+ if (context == null || expectedGeneration != generation) {
+ return false;
+ }
+ // Confirm that standard PhysicalPlan JSON can be read back before
publishing.
+ reader.readPhysicalPlan(json);
+ // Readback may perform plugin I/O. Keep it outside the clear/publication
lock,
+ // then check again so a concurrent clear cannot be undone by that
readback.
+ synchronized (this) {
+ if (expectedGeneration != generation) {
+ logger.debug("Skipping plan cache publication after a cache clear");
+ return false;
+ }
+ entries.put(key, new Entry(json, textPlan, context));
+ }
+ logger.debug("Plan cache entry published; entries={}", entries.size());
+ return true;
+ }
+
+ @Override
+ public void close() {
+ writer.shutdownNow();
+ // Test JVMs may contain several Drillbits. A closing cache must not remove
+ // gauges registered by a newer cache under the same names.
+ synchronized (DrillMetrics.class) {
+ metrics.forEach((name, gauge) -> {
+ if (DrillMetrics.getRegistry().getMetrics().get(name) == gauge) {
+ DrillMetrics.getRegistry().remove(name);
+ }
+ });
+ }
+ }
+
+ public static final class Entry {
+ private final String json;
+ private final long contentSizeBytes;
+ private final String textPlan;
+ private final ContextSnapshot context;
+
+ private Entry(String json, String textPlan, ContextSnapshot context) {
+ this.json = json;
+ this.contentSizeBytes = (long)
json.getBytes(StandardCharsets.UTF_8).length
+ + (textPlan == null ? 0 :
textPlan.getBytes(StandardCharsets.UTF_8).length);
+ this.textPlan = textPlan;
+ this.context = context;
+ }
+
+ public boolean matchesContext(ContextSnapshot current) {
+ return context.matches(current);
+ }
+
+ public String getTextPlan() {
+ return textPlan;
+ }
+
+ public PhysicalPlan bind(List<SqlLiteral> literals, PhysicalPlanReader
reader)
+ throws IOException {
+ JsonNode tree = JSON.readTree(json);
+ walk(tree, literals);
+ return reader.readPhysicalPlan(JSON.writeValueAsString(tree));
+ }
+ }
+
+ /** Options, plugin configurations, and table versions captured before
lookup and planning. */
+ public static final class ContextSnapshot {
+ private final String optionsFingerprint;
+ private final Map<TableIdentifier, String> tableVersions;
+ private final Map<String, String> pluginConfigs;
+
+ private ContextSnapshot(String optionsFingerprint, Map<TableIdentifier,
String> tableVersions,
+ Map<String, String> pluginConfigs) {
+ this.optionsFingerprint = Objects.requireNonNull(optionsFingerprint,
"optionsFingerprint");
+ this.tableVersions = Collections.unmodifiableMap(tableVersions);
+ this.pluginConfigs = Collections.unmodifiableMap(pluginConfigs);
+ }
+
+ private static final class TableIdentifier {
+ private final String storageName;
+ private final String tableId;
+
+ private TableIdentifier(String storageName, String tableId) {
+ this.storageName = storageName;
+ this.tableId = tableId;
+ }
+
+ @Override
+ public boolean equals(Object other) {
+ if (this == other) {
+ return true;
+ }
+ if (!(other instanceof TableIdentifier)) {
+ return false;
+ }
+ TableIdentifier that = (TableIdentifier) other;
+ return Objects.equals(storageName, that.storageName)
+ && Objects.equals(tableId, that.tableId);
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(storageName, tableId);
+ }
+ }
+
+ private static final class ResolvedTable {
+ private final List<String> tableNames;
+ private final String storageName;
+ private final StoragePlugin plugin;
+ private final DrillTableSelection selection;
+
+ private ResolvedTable(List<String> tableNames, String storageName,
+ StoragePlugin plugin, DrillTableSelection selection) {
+ this.tableNames = tableNames;
+ this.storageName = storageName;
+ this.plugin = plugin;
+ this.selection = selection;
+ }
+ }
+
+ /** Resolves every physical table in a query, including joins and
subqueries. */
+ public static ContextSnapshot resolve(SchemaPlus defaultSchema, SqlNode
query,
+ QueryContext context) {
+ StoragePluginRegistry plugins = context.getStorage();
+ List<ResolvedTable> dependencies = new ArrayList<>();
+ try {
+ // Check every participating plugin before reading any table version.
+ if (!collectQuery(defaultSchema, query, new HashSet<>(),
dependencies)) {
+ return null;
+ }
+ } catch (RuntimeException e) {
+ logger.debug("Query sources could not be validated for the plan
cache", e);
+ return null;
+ }
+ Map<TableIdentifier, String> versions = new LinkedHashMap<>();
+ Map<String, String> configs = new LinkedHashMap<>();
+ for (ResolvedTable dependency : dependencies) {
+ try {
+ PlanCacheTable source =
dependency.plugin.planCacheTable(dependency.selection);
+ if (source == null) {
+ return null;
+ }
+ TableIdentifier identifier = new
TableIdentifier(dependency.storageName,
+ source.getIdentifier());
+ String previous = versions.putIfAbsent(identifier,
source.getVersion());
+ if (previous != null && !previous.equals(source.getVersion())) {
+ return null;
+ }
+ String fingerprint = configFingerprint(plugins,
dependency.plugin.getConfig());
+ String previousConfig = configs.putIfAbsent(dependency.storageName,
fingerprint);
+ if (previousConfig != null && !previousConfig.equals(fingerprint)) {
+ return null;
+ }
+ } catch (IOException | RuntimeException e) {
+ logger.debug("Table context could not be captured for the plan
cache", e);
+ return null;
+ }
+ }
+ return new ContextSnapshot(optionFingerprint(context), versions,
configs);
+ }
+
+ private static boolean collectQuery(SchemaPlus schema, SqlNode node,
+ Set<String> ctes, List<ResolvedTable> dependencies) {
+ if (node == null) {
+ return true;
+ }
+ if (node instanceof SqlNodeList) {
+ for (SqlNode child : (SqlNodeList) node) {
+ if (!collectQuery(schema, child, ctes, dependencies)) {
+ return false;
+ }
+ }
+ return true;
+ }
+ if (node instanceof SqlWith) {
+ SqlWith with = (SqlWith) node;
+ Set<String> local = new HashSet<>(ctes);
+ for (SqlNode itemNode : with.withList) {
+ SqlWithItem item = (SqlWithItem) itemNode;
+ if (!collectQuery(schema, item.query, local, dependencies)) {
+ return false;
+ }
+ local.add(item.name.getSimple());
+ }
+ return collectQuery(schema, with.body, local, dependencies);
+ }
+ if (node instanceof SqlSelect) {
+ SqlSelect select = (SqlSelect) node;
+ if (!collectFrom(schema, select.getFrom(), ctes, dependencies)) {
+ return false;
+ }
+ for (SqlNode operand : select.getOperandList()) {
+ if (operand != select.getFrom()
+ && !collectQuery(schema, operand, ctes, dependencies)) {
+ return false;
+ }
+ }
+ return true;
+ }
+ if (node instanceof SqlCall) {
+ for (SqlNode operand : ((SqlCall) node).getOperandList()) {
+ if (!collectQuery(schema, operand, ctes, dependencies)) {
+ return false;
+ }
+ }
+ }
+ return true;
+ }
+
+ private static boolean collectFrom(SchemaPlus schema, SqlNode from,
+ Set<String> ctes, List<ResolvedTable> dependencies) {
+ if (from == null || from.getKind() == SqlKind.VALUES) {
+ return true;
+ }
+ if (from instanceof SqlIdentifier) {
+ List<String> names = ((SqlIdentifier) from).names;
+ if (names.size() == 1 && ctes.contains(names.get(0))) {
+ return true;
+ }
+ for (ResolvedTable dependency : dependencies) {
+ if (dependency.tableNames.equals(names)) {
+ return true;
+ }
+ }
+ ResolvedTable source = resolveTable(schema, names);
+ if (source == null) {
+ return false;
+ }
+ dependencies.add(source);
+ return true;
+ }
+ if (from instanceof SqlCall) {
+ SqlCall call = (SqlCall) from;
+ if (call.getKind() == SqlKind.AS || call.getKind() == SqlKind.LATERAL)
{
+ return collectFrom(schema, call.operand(0), ctes, dependencies);
+ }
+ if (call.getKind() == SqlKind.JOIN) {
+ return collectFrom(schema, call.operand(0), ctes, dependencies)
+ && collectFrom(schema, call.operand(3), ctes, dependencies)
+ && collectQuery(schema, call.operand(5), ctes, dependencies);
+ }
+ if (call.getKind().belongsTo(SqlKind.QUERY)) {
+ return collectQuery(schema, call, ctes, dependencies);
+ }
+ }
+ return false;
+ }
+
+ private static ResolvedTable resolveTable(SchemaPlus defaultSchema,
List<String> names) {
+ if (names.isEmpty() || names.get(names.size() - 1).contains("*")
+ || names.get(names.size() - 1).contains("?")) {
+ return null;
+ }
+ SchemaPlus schema = SchemaUtilities.findSchema(defaultSchema,
+ names.subList(0, names.size() - 1));
+ if (schema == null) {
+ return null;
+ }
+ Table table = schema.getTable(names.get(names.size() - 1));
+ if (!(table instanceof DrillTable)) {
+ return null;
+ }
+ DrillTable drillTable = (DrillTable) table;
+ Object selection = drillTable.getSelection();
+ if (!(selection instanceof DrillTableSelection)) {
+ return null;
+ }
+ StoragePlugin storagePlugin = drillTable.getPlugin();
+ if (!storagePlugin.supportPlanCache((DrillTableSelection) selection)) {
+ return null;
+ }
+ return new ResolvedTable(
+ Collections.unmodifiableList(new ArrayList<>(names)),
+ drillTable.getStorageEngineName(), storagePlugin,
(DrillTableSelection) selection);
+ }
+
+ private boolean matches(ContextSnapshot current) {
+ return current != null
+ && optionsFingerprint.equals(current.optionsFingerprint)
+ && tableVersions.equals(current.tableVersions)
+ && pluginConfigs.equals(current.pluginConfigs);
+ }
+
+ /** Options and plugin configurations identify separate reusable
templates. */
+ String keyFingerprint() {
+ StringBuilder configs = new StringBuilder();
+ new TreeMap<>(pluginConfigs).forEach((name, fingerprint) -> configs
+
.append(name.length()).append(':').append(name).append('=').append(fingerprint).append('\n'));
+ // Keep table versions outside the key so incompatible entries can be
invalidated.
+ return optionsFingerprint + '\n'
+ + Hashing.sha256().hashString(configs,
StandardCharsets.UTF_8).toString();
+ }
+
+ private static String optionFingerprint(QueryContext context) {
+ // OptionManager iteration walks system defaults, system overrides,
session
+ // overrides, and query overrides in precedence order. getOptionList()
only
+ // exposes local session/query values and misses system-level changes.
+ Map<String, OptionValue> options = new
TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+ for (OptionValue option : context.getOptions()) {
+ options.put(option.name, option);
+ }
+ StringBuilder fingerprint = new StringBuilder();
+ for (OptionValue option : options.values()) {
+
fingerprint.append(option.name).append(':').append(option.kind).append('=');
+ switch (option.kind) {
+ case BOOLEAN:
+ fingerprint.append(option.bool_val);
+ break;
+ case LONG:
+ fingerprint.append(option.num_val);
+ break;
+ case DOUBLE:
+ fingerprint.append(option.float_val);
+ break;
+ case STRING:
+
fingerprint.append(option.string_val.length()).append(':').append(option.string_val);
+ break;
+ default:
+ throw new IllegalArgumentException("Unsupported option kind");
+ }
+ fingerprint.append('\n');
+ }
+ // Keep the cache key bounded even when the system has many registered
options.
+ return Hashing.sha256().hashString(fingerprint,
StandardCharsets.UTF_8).toString();
+ }
+
+ private static String configFingerprint(StoragePluginRegistry plugins,
+ StoragePluginConfig config) {
+ if (config == null) {
+ throw new IllegalArgumentException("Missing plugin configuration");
+ }
+ String encoded = plugins.encode(config);
+ if (encoded == null) {
+ throw new IllegalArgumentException("Plugin configuration could not be
encoded");
+ }
+ return Hashing.sha256().hashString(encoded,
StandardCharsets.UTF_8).toString();
+ }
+ }
+
+ private static LiteralExpression bindLiteral(SqlLiteral literal, MajorType
type, int index) {
+ Object value = literal.getValue();
+ LogicalExpression expression;
+ if (value instanceof BigDecimal) {
+ BigDecimal number = (BigDecimal) value;
+ if (type.getMinorType() == MinorType.VARDECIMAL) {
+ BigDecimal decimal = number.setScale(type.getScale(),
RoundingMode.UNNECESSARY);
+ if (decimal.precision() > type.getPrecision()) {
+ throw new IllegalArgumentException("Decimal parameter exceeds cached
precision");
+ }
+ expression = ValueExpressions.getVarDecimal(decimal,
+ type.getPrecision(), type.getScale());
+ } else if (type.getMinorType() == MinorType.FLOAT8) {
+ expression = ValueExpressions.getFloat8(number.doubleValue());
+ } else if (type.getMinorType() == MinorType.FLOAT4) {
+ expression = ValueExpressions.getFloat4(number.floatValue());
+ } else if (type.getMinorType() == MinorType.BIGINT) {
+ expression = ValueExpressions.getBigInt(number.longValueExact());
+ } else if (type.getMinorType() == MinorType.INT) {
+ expression = ValueExpressions.getInt(number.intValueExact());
+ } else {
+ throw new IllegalArgumentException("Unsupported numeric parameter
type");
+ }
+ } else if (value instanceof Boolean) {
+ expression = ValueExpressions.getBit((Boolean) value);
+ } else if (value instanceof NlsString) {
+ String string = ((NlsString) value).getValue();
+ expression = ValueExpressions.getChar(string, type.getPrecision());
+ } else {
+ throw new IllegalArgumentException("Unsupported parameter type");
+ }
+ LiteralExpression bound = (LiteralExpression) expression;
+ bound.setDynamicParamIndex(index);
Review Comment:
Documented that caching must remain disabled during rolling upgrades and can
only be enabled after all Drillbits are upgraded.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/handlers/ClearPlanCacheHandler.java:
##########
@@ -0,0 +1,45 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.planner.sql.handlers;
+
+import org.apache.calcite.sql.SqlNode;
+import org.apache.drill.exec.ops.QueryContext;
+import org.apache.drill.exec.physical.PhysicalPlan;
+import org.apache.drill.exec.planner.sql.DirectPlan;
+import org.apache.drill.exec.planner.sql.parser.SqlClearPlanCache;
+import org.apache.drill.exec.proto.CoordinationProtos.DrillbitEndpoint;
+import org.apache.drill.exec.work.foreman.ForemanSetupException;
+
+/** Clears the cache on the Drillbit that plans this command. */
+public final class ClearPlanCacheHandler extends AbstractSqlHandler {
+ private final QueryContext context;
+
+ public ClearPlanCacheHandler(QueryContext context) {
+ this.context = context;
+ }
+
+ @Override
+ public PhysicalPlan getPlan(SqlNode sqlNode) throws ForemanSetupException {
+ unwrap(sqlNode, SqlClearPlanCache.class);
+ checkAdminPrivileges(context, context.getOptions(), "Not authorized to
clear the SYSTEM plan cache.");
+ context.getDrillbitContext().getPlanCache().clear();
Review Comment:
Documented that CLEAR affects only the receiving Drillbit. For example,
clearing node A does not clear node B behind the same load balancer. The
response identifies the cleared node.
##########
contrib/format-iceberg/src/main/java/org/apache/drill/exec/store/iceberg/format/IcebergFormatPlugin.java:
##########
@@ -104,6 +109,27 @@ public boolean supportsRead() {
return true;
}
+ @Override
+ public boolean supportPlanCache(FileSelection selection) {
+ return selection != null && config.getSnapshot() == null
+ && !(selection instanceof IcebergMetadataFileSelection) &&
selection.getSelectionRoot() != null;
+ }
+
+ @Override
+ public PlanCacheTable planCacheTable(FileSelection selection) throws
IOException {
+ if (!supportPlanCache(selection)) {
+ return null;
+ }
+ String identifier = selection.getSelectionRoot().toString();
+ Table table = new HadoopTables(fsConf).load(identifier);
+ if (!(table instanceof HasTableOperations)) {
+ return null;
+ }
+ TableMetadata metadata = ((HasTableOperations)
table).operations().current();
+ return metadata == null || metadata.uuid() == null ? null
+ : new PlanCacheTable(identifier, metadata.uuid() + ":" +
metadata.currentSchemaId());
Review Comment:
Provided schema contents now contribute to table compatibility. For example,
changing `id VARCHAR` to `id BIGINT` invalidates the old entry. Added tests
with and without provided schemas, including native schema changes. Tables
without an Iceberg UUID still bypass caching; this is now explicit in the
plugin guide.
> Physical plan cache for parameterized SQL queries
> -------------------------------------------------
>
> Key: DRILL-8555
> URL: https://issues.apache.org/jira/browse/DRILL-8555
> Project: Apache Drill
> Issue Type: New Feature
> Components: Query Planning & Optimization
> Reporter: Letian Jiang
> Priority: Major
>
> h2. Motivation
> Repeated queries that differ only in literals are a common workload, yet each
> query currently requires full planning. This adds unnecessary latency and
> wastes CPU time.
> h2. Brief Design
> Add a per-Drillbit, in-memory plan cache. Drill automatically parameterizes
> eligible SQL queries and reuses a cached physical plan with the current
> literal values, without changes to client queries.
> h2. Plan reuse safety
> * *Engine:* matches the SQL template and checks relevant option changes,
> plugin configuration changes, and whether base table versions still match.
> * *Storage plugin:* explicitly opts in, provides a plan-compatible table
> version, and is responsible for rebuilding its predicate-dependent scan work
> from the rebound expressions.
> h2. Expected benefit
> Short, frequently executed queries such as HBase reads can benefit
> substantially. Analytical queries with significant planning cost can benefit
> as well.
> h2. Implementation status
> A working implementation is available, including HBase, Paimon and Iceberg
> integrations. A pull request will follow.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
