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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
