cgivre commented on code in PR #3086:
URL: https://github.com/apache/drill/pull/3086#discussion_r4220447783
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java:
##########
@@ -295,7 +298,79 @@ private static PhysicalPlan getQueryPlan(QueryContext
context, String sql, Point
context.getOptions().setLocalOption(ExecConstants.RETURN_RESULT_SET_FOR_DDL,
true);
}
- return handler.getPlan(sqlNode);
+ // Keep the original SQL if no safe snapshot is available or cache
preparation fails.
+ SqlNode planningSql = sqlNode;
+ Consumer<PhysicalPlan> prepareCacheInsert = null;
+ if (context.getOptions().getOption(PlannerSettings.PLAN_CACHE)
+ && !context.getSession().hasTemporaryTables()
+ && !hasMutableAliases(context)
+ && PlanCacheEligibility.isSafeToCache(sqlNode,
context.getFunctionRegistry())) {
+ try {
+ // Capture the context once, before lookup and planning. Changes during
+ // planning invalidate this entry on its first hit.
+ PlanCache.ContextSnapshot snapshot = PlanCache.ContextSnapshot.resolve(
+ parser.getDefaultSchema(), sqlNode, context);
+ if (snapshot != null) {
+ PlanCacheParameterizer.Candidate candidate =
+ PlanCacheParameterizer.parameterize(sqlNode,
parser.getTypeFactory());
+ PlanCache cache = context.getDrillbitContext().getPlanCache();
+ String key = context.getQueryUserName() + '\n'
+ + context.getSession().getDefaultSchemaPath() + '\n'
+ + candidate.template;
+ PlanCache.Entry entry = cache.get(key);
+ if (entry != null) {
+ try {
+ if (entry.matchesContext(snapshot)) {
+ PhysicalPlan bound = entry.bind(candidate.literals,
+ context.getDrillbitContext().getPlanReader());
+ PlanCache.bindTextPlan(textPlan, entry.getTextPlan(),
candidate.literals);
+ cache.recordBind();
+ context.setPlanCacheHit();
+ return bound;
+ }
+ } catch (RuntimeException | IOException e) {
+ logger.debug("Cached plan could not be rebound", e);
+ }
+ cache.invalidate(key);
+ }
+ final PhysicalPlanReader cacheReader =
+ context.getDrillbitContext().getPlanReader();
+ // Prepare insertion after planning; Foreman publishes it only after
the query succeeds.
+ prepareCacheInsert = planned -> {
+ String templateTextPlan = PlanCache.bindTextPlan(textPlan,
candidate.literals);
+ context.setPendingPlanCacheInsert(() ->
cache.writeAfterSuccess(key, planned,
+ templateTextPlan, cacheReader, snapshot));
+ };
+ // On a cache miss, plan the parameterized SQL to produce a reusable
template.
+ planningSql = candidate.sql;
+ }
+ } catch (RuntimeException e) {
+ logger.debug("Plan cache attempt fell back to ordinary planning", e);
+ }
+ }
+
+ // Plan exactly once, outside cache fallback. Validation can mutate the
SQL tree,
+ // so retrying a failed planning attempt could reuse partially rewritten
nodes.
+ PhysicalPlan planned = handler.getPlan(planningSql);
+ if (prepareCacheInsert != null) {
+ prepareCacheInsert.accept(planned);
+ }
+ return planned;
Review Comment:
**[Correctness] Fall back to ordinary planning when the parameterized SQL
fails to plan.**
On a miss, the query is planned once from `candidate.sql`, and the
`try/catch` above only covers lookup and preparation. Any query that only plans
with real literals fails, but only when the cache is enabled. Examples: `JOIN
... ON TRUE` (becomes `ON ?`), `PERCENTILE_CONT(0.5) WITHIN GROUP`, and
functions that need a literal operand but aren't in `configurationOperand`.
The comment above is right that the mutated tree can't be retried. This
suggestion avoids that by re-entering `getQueryPlan` from the SQL string with
`planner.enable_plan_cache` set to false at query level, the same pattern the
Metastore retry in `getPlan` uses. The fresh parse, converter and handler start
clean. A query that fails for a real reason (bad column, etc.) fails again on
the second attempt and reports the error from the original SQL.
```suggestion
if (prepareCacheInsert == null) {
return handler.getPlan(planningSql);
}
PhysicalPlan planned;
try {
planned = handler.getPlan(planningSql);
} catch (Exception e) {
// Some queries only plan with real literal values (e.g. ON TRUE, or
functions that
// need a literal operand). Validation mutates the SQL tree, so
re-parse and plan the
// original SQL with the cache off for this query, like the Metastore
retry above.
logger.debug("Parameterized planning failed; replanning without the
plan cache", e);
context.getOptions().setLocalOption(PlannerSettings.ENABLE_PLAN_CACHE_OPTION,
false);
return getQueryPlan(context, sql, textPlan);
}
prepareCacheInsert.accept(planned);
return planned;
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCacheParameterizer.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * 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 java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Objects;
+
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.util.SqlShuttle;
+import org.apache.calcite.util.NlsString;
+import org.apache.drill.exec.planner.sql.conversion.SqlBoundDynamicParam;
+
+/** Extracts SQL literals into typed slots independently of cache eligibility.
*/
+final class PlanCacheParameterizer extends SqlShuttle {
+ private final List<SqlLiteral> literals = new ArrayList<>();
+ private final List<String> types = new ArrayList<>();
+ private final RelDataTypeFactory typeFactory;
+ private boolean preserveProjection;
+
+ private PlanCacheParameterizer(RelDataTypeFactory typeFactory) {
+ this.typeFactory = Objects.requireNonNull(typeFactory, "typeFactory");
+ }
+
+ static Candidate parameterize(SqlNode parsed, RelDataTypeFactory
typeFactory) {
+ Objects.requireNonNull(parsed, "parsed");
+ PlanCacheParameterizer visitor = new PlanCacheParameterizer(typeFactory);
+ SqlNode parameterized = parsed.accept(visitor);
+ if (visitor.literals.isEmpty()) {
+ parameterized = parsed;
+ }
+ return new Candidate(parameterized,
+ Collections.unmodifiableList(visitor.literals),
+ parameterized.toString() + visitor.types);
+ }
+
+ @Override
+ public SqlNode visit(SqlCall call) {
+ int configurationOperand = configurationOperand(call);
+ if (configurationOperand >= 0 && configurationOperand <
call.operandCount()) {
+ // These operands become function names, encodings or resolved plugin
+ // references during conversion. They are not bindable expressions in
+ // the physical plan, so keep their entire subtree in the template key.
+ SqlCall parameterized = (SqlCall) call.clone(call.getParserPosition());
+ for (int i = 0; i < call.operandCount(); i++) {
+ if (i != configurationOperand) {
+ parameterized.setOperand(i, visitNullable(call.operand(i)));
+ }
+ }
+ return parameterized;
+ }
+ if ("COALESCE".equalsIgnoreCase(call.getOperator().getName())
+ && call.getOperandList().stream()
+ .anyMatch(operand -> operand instanceof SqlLiteral
+ && ((SqlLiteral) operand).getTypeName() == SqlTypeName.NULL)) {
+ // Calcite cannot infer a dynamic parameter's type alongside an untyped
+ // NULL. Keep this call in the key; explicit casts still allow slots.
+ return call;
+ }
+ if (call instanceof SqlJoin) {
+ SqlJoin join = (SqlJoin) call;
+ return new SqlJoin(join.getParserPosition(),
visitNullable(join.getLeft()),
+ join.isNaturalNode(), join.getJoinTypeNode(),
visitNullable(join.getRight()),
+ join.getConditionTypeNode(), visitNullable(join.getCondition()));
+ }
+ if (call.getKind() == SqlKind.ITEM && call.operand(1) instanceof
SqlLiteral) {
+ // A map/array key identifies a field in the input schema. Changing it
can
+ // change the scan projection and the HBase column used by pushdown.
+ return call.getOperator().createCall(call.getParserPosition(),
+ visitNullable(call.operand(0)), call.operand(1));
+ }
+ // These literals control rows, ordinals, window frames or operator
+ // configuration. Keep them verbatim in the template key.
+ if (call.getKind() == SqlKind.VALUES || call.getKind() == SqlKind.ORDER_BY
+ || call.getKind() == SqlKind.OVER) {
+ if (call instanceof SqlOrderBy) {
+ SqlOrderBy orderBy = (SqlOrderBy) call;
+ boolean previous = preserveProjection;
+ preserveProjection = true;
+ SqlNode query;
+ try {
+ query = orderBy.query.accept(this);
+ } finally {
+ preserveProjection = previous;
+ }
+ return new SqlOrderBy(call.getParserPosition(), query,
orderBy.orderList,
+ orderBy.offset, orderBy.fetch);
+ }
+ return call;
+ }
+ if (call instanceof SqlSelect) {
+ SqlSelect select = (SqlSelect) call;
+ // Keep structural SQL operands verbatim while visiting expressions.
+ boolean structuralProjection = preserveProjection
+ || (select.getGroup() != null && !select.getGroup().isEmpty())
+ || (select.getOrderList() != null &&
!select.getOrderList().isEmpty())
+ || (select.getWindowList() != null &&
!select.getWindowList().isEmpty());
+ return new SqlSelect(select.getParserPosition(),
+ (SqlNodeList) select.getOperandList().get(0),
+ structuralProjection ? select.getSelectList()
+ : (SqlNodeList) visitNullable(select.getSelectList()),
+ visitNullable(select.getFrom()), visitNullable(select.getWhere()),
+ select.getGroup(), visitNullable(select.getHaving()),
+ select.getWindowList(), visitNullable(select.getQualify()),
Review Comment:
**[Correctness] Keep HAVING and QUALIFY verbatim when GROUP BY / the select
list are.**
When `structuralProjection` is true, GROUP BY and the select list keep their
literals, but HAVING and QUALIFY still get slots. So they no longer match the
grouped or windowed expressions:
```sql
SELECT x || 'a', COUNT(*) FROM hbase.t GROUP BY x || 'a' HAVING x || 'a' =
'ba'
```
HAVING becomes `x || ? = ?`, and Calcite raises *"Expression 'x' is not
being grouped"*. Without a fallback, the query fails only when the cache is on.
Using the same condition for HAVING and QUALIFY keeps them consistent with what
they have to match.
```suggestion
select.getGroup(),
structuralProjection ? select.getHaving() :
visitNullable(select.getHaving()),
select.getWindowList(),
structuralProjection ? select.getQualify() :
visitNullable(select.getQualify()),
```
##########
logical/src/main/java/org/apache/drill/common/expression/ExpressionStringBuilder.java:
##########
@@ -147,7 +185,14 @@ public Void visitSchemaPath(SchemaPath path, StringBuilder
sb) throws RuntimeExc
@Override
public Void visitLongConstant(LongExpression lExpr, StringBuilder sb) throws
RuntimeException {
- sb.append(lExpr.getLong());
+ startBoundLiteral(lExpr, sb);
+ if (!lExpr.isDynamicParam()) {
+ // Preserve integer width in JSON; the parser restores this as a BIGINT
literal.
+ sb.append("cast(").append(lExpr.getLong()).append(" as BIGINT)");
+ } else {
+ sb.append(lExpr.getLong());
+ }
Review Comment:
**[Correctness / compatibility] Narrow the BIGINT cast to values that need
it.**
This now writes `cast(N as BIGINT)` for every non-parameter BIGINT literal
in every plan, even with the cache off. That changes EXPLAIN output, scan
digests and filter strings (e.g.
`AbstractGroupScanWithMetadata.getFilterString`) for all users.
Only values in int range are ambiguous on read-back, because
`ValueExpressions.getNumericExpression` tries `Integer.parseInt` first and then
`Long.parseLong`. Values outside int range already round-trip as
`LongExpression`, so they don't need the cast. This suggestion limits the cast
to the ambiguous case. The `createCast` special case in `ExprParser.g4` still
handles it.
Is the cast needed at all outside the plan cache? Physical plans were
already serialized to JSON for remote fragments before this PR. If only cached
plans need it, could it be applied only when serializing for the cache?
```suggestion
long value = lExpr.getLong();
if (!lExpr.isDynamicParam() && value >= Integer.MIN_VALUE && value <=
Integer.MAX_VALUE) {
// Only int-range values are ambiguous: the parser reads them back as
INT.
sb.append("cast(").append(value).append(" as BIGINT)");
} else {
sb.append(value);
}
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java:
##########
@@ -295,7 +298,79 @@ private static PhysicalPlan getQueryPlan(QueryContext
context, String sql, Point
context.getOptions().setLocalOption(ExecConstants.RETURN_RESULT_SET_FOR_DDL,
true);
}
- return handler.getPlan(sqlNode);
+ // Keep the original SQL if no safe snapshot is available or cache
preparation fails.
+ SqlNode planningSql = sqlNode;
+ Consumer<PhysicalPlan> prepareCacheInsert = null;
+ if (context.getOptions().getOption(PlannerSettings.PLAN_CACHE)
+ && !context.getSession().hasTemporaryTables()
+ && !hasMutableAliases(context)
+ && PlanCacheEligibility.isSafeToCache(sqlNode,
context.getFunctionRegistry())) {
+ try {
+ // Capture the context once, before lookup and planning. Changes during
+ // planning invalidate this entry on its first hit.
+ PlanCache.ContextSnapshot snapshot = PlanCache.ContextSnapshot.resolve(
+ parser.getDefaultSchema(), sqlNode, context);
+ if (snapshot != null) {
+ PlanCacheParameterizer.Candidate candidate =
+ PlanCacheParameterizer.parameterize(sqlNode,
parser.getTypeFactory());
+ PlanCache cache = context.getDrillbitContext().getPlanCache();
+ String key = context.getQueryUserName() + '\n'
+ + context.getSession().getDefaultSchemaPath() + '\n'
+ + candidate.template;
Review Comment:
**[Performance] Include the session/plugin fingerprint in the key** (2 of 2;
see `ContextSnapshot.keyFingerprint()` in `PlanCache`). This keeps sessions
with different option values from evicting each other's entries.
```suggestion
String key = context.getQueryUserName() + '\n'
+ context.getSession().getDefaultSchemaPath() + '\n'
+ snapshot.keyFingerprint() + '\n'
+ candidate.template;
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,640 @@
+/*
+ * 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 java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+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.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.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.hash.Hashing;
+
+/** 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 = CacheBuilder.newBuilder()
+ .maximumWeight(32 * 1024 * 1024)
+ .weigher((String key, Entry value) -> value.json.length())
+ .expireAfterWrite(10, TimeUnit.MINUTES)
+ .build();
+ private final AtomicLong successfulBinds = new AtomicLong();
+ private final ThreadPoolExecutor writer = new ThreadPoolExecutor(1, 1, 0L,
+ TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
+ new NamedThreadFactory("plan-cache-writer-"));
+
+ public long getHitCount() {
+ return successfulBinds.get();
+ }
+
+ public void recordBind() {
+ successfulBinds.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, PhysicalPlan plan, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context) {
+ try {
+ writer.execute(() -> {
+ try {
+ put(key, plan, textPlan, reader, context);
+ } 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) {
+ entries.invalidate(key);
+ }
+
+ public boolean put(String key, PhysicalPlan plan, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context) 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) {
+ return false;
+ }
+ String json = reader.writeJson(plan);
+ // Confirm that standard PhysicalPlan JSON can be read back before
publishing.
+ reader.readPhysicalPlan(json);
+ entries.put(key, new Entry(json, textPlan, context));
+ return true;
+ }
+
+ @Override
+ public void close() {
+ writer.shutdownNow();
+ }
+
+ public static final class Entry {
+ private final String json;
+ private final String textPlan;
+ private final ContextSnapshot context;
+
+ private Entry(String json, String textPlan, ContextSnapshot context) {
+ this.json = json;
+ 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);
+ }
Review Comment:
**[Performance] Put the options and plugin-config fingerprints in the key**
(1 of 2; the other suggestion is in `DrillSqlWorker`).
Today, a mismatch in `matchesContext` invalidates the shared entry. Two
sessions running the same template with different options (e.g. `ALTER SESSION
SET planner.slice_target = ...`) keep evicting each other's entry, so the hit
rate is 0 and every query also pays to republish. With the fingerprints in the
key, each option set gets its own entry. Only a table-version change, which
should invalidate, falls through to `invalidate`.
```suggestion
}
/**
* Session- and plugin-dependent part of the context. It belongs in the
cache key, so
* sessions with different options keep separate entries instead of
evicting each other.
* Table versions stay in {@link #matches}, where a change should
invalidate the entry.
*/
String keyFingerprint() {
return optionsFingerprint + '\n' + pluginConfigs;
}
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java:
##########
@@ -295,7 +298,79 @@ private static PhysicalPlan getQueryPlan(QueryContext
context, String sql, Point
context.getOptions().setLocalOption(ExecConstants.RETURN_RESULT_SET_FOR_DDL,
true);
}
- return handler.getPlan(sqlNode);
+ // Keep the original SQL if no safe snapshot is available or cache
preparation fails.
+ SqlNode planningSql = sqlNode;
+ Consumer<PhysicalPlan> prepareCacheInsert = null;
+ if (context.getOptions().getOption(PlannerSettings.PLAN_CACHE)
+ && !context.getSession().hasTemporaryTables()
+ && !hasMutableAliases(context)
+ && PlanCacheEligibility.isSafeToCache(sqlNode,
context.getFunctionRegistry())) {
+ try {
+ // Capture the context once, before lookup and planning. Changes during
+ // planning invalidate this entry on its first hit.
+ PlanCache.ContextSnapshot snapshot = PlanCache.ContextSnapshot.resolve(
+ parser.getDefaultSchema(), sqlNode, context);
+ if (snapshot != null) {
+ PlanCacheParameterizer.Candidate candidate =
+ PlanCacheParameterizer.parameterize(sqlNode,
parser.getTypeFactory());
+ PlanCache cache = context.getDrillbitContext().getPlanCache();
+ String key = context.getQueryUserName() + '\n'
+ + context.getSession().getDefaultSchemaPath() + '\n'
+ + candidate.template;
+ PlanCache.Entry entry = cache.get(key);
+ if (entry != null) {
+ try {
+ if (entry.matchesContext(snapshot)) {
+ PhysicalPlan bound = entry.bind(candidate.literals,
+ context.getDrillbitContext().getPlanReader());
+ PlanCache.bindTextPlan(textPlan, entry.getTextPlan(),
candidate.literals);
+ cache.recordBind();
+ context.setPlanCacheHit();
+ return bound;
+ }
+ } catch (RuntimeException | IOException e) {
+ logger.debug("Cached plan could not be rebound", e);
+ }
+ cache.invalidate(key);
+ }
+ final PhysicalPlanReader cacheReader =
+ context.getDrillbitContext().getPlanReader();
+ // Prepare insertion after planning; Foreman publishes it only after
the query succeeds.
+ prepareCacheInsert = planned -> {
+ String templateTextPlan = PlanCache.bindTextPlan(textPlan,
candidate.literals);
+ context.setPendingPlanCacheInsert(() ->
cache.writeAfterSuccess(key, planned,
+ templateTextPlan, cacheReader, snapshot));
+ };
+ // On a cache miss, plan the parameterized SQL to produce a reusable
template.
+ planningSql = candidate.sql;
Review Comment:
**[Plan quality] On a miss, the query that actually runs loses its literal
values.**
The results returned to the user come from planning `candidate.sql`. `WHERE
1 = 1` and `flag = TRUE` become `? = ?` / `flag = ?`, so Calcite can't reduce
constants or estimate selectivity from the values. Join order and distribution
for this run can be worse than with the cache off, even if the plan is never
published (queue full, query failed).
No line-level fix here, but some options:
- Plan the literal SQL for execution, and build the parameterized template
in the background on the first miss (or the second, to skip one-off queries).
- At minimum, document the trade-off in the design doc.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,640 @@
+/*
+ * 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 java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+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.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.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.hash.Hashing;
+
+/** 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 = CacheBuilder.newBuilder()
+ .maximumWeight(32 * 1024 * 1024)
+ .weigher((String key, Entry value) -> value.json.length())
+ .expireAfterWrite(10, TimeUnit.MINUTES)
+ .build();
+ private final AtomicLong successfulBinds = new AtomicLong();
+ private final ThreadPoolExecutor writer = new ThreadPoolExecutor(1, 1, 0L,
+ TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
+ new NamedThreadFactory("plan-cache-writer-"));
+
+ public long getHitCount() {
+ return successfulBinds.get();
+ }
+
+ public void recordBind() {
+ successfulBinds.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, PhysicalPlan plan, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context) {
+ try {
+ writer.execute(() -> {
+ try {
+ put(key, plan, textPlan, reader, context);
+ } 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) {
+ entries.invalidate(key);
+ }
Review Comment:
**[Observability / operations] Count invalidations and misses, and allow a
full flush.**
- `invalidations` and `misses` feed the metrics registered in the
constructor (see the suggestion on the cache fields).
- There's currently no way to clear the cache short of waiting for expiry or
restarting the Drillbit. Turning off `planner.enable_plan_cache` doesn't clear
it either. `invalidateAll()` gives an admin REST endpoint (e.g. `POST
/plancache/clear`, admin-only) something to call. It would also be reasonable
to call it when a storage plugin config is updated.
```suggestion
public void invalidate(String key) {
invalidations.incrementAndGet();
entries.invalidate(key);
}
public void recordMiss() {
misses.incrementAndGet();
}
/** Removes every entry; for an admin endpoint or after bulk metadata
changes. */
public void invalidateAll() {
entries.invalidateAll();
}
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,640 @@
+/*
+ * 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 java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+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.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.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.hash.Hashing;
+
+/** 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 = CacheBuilder.newBuilder()
+ .maximumWeight(32 * 1024 * 1024)
+ .weigher((String key, Entry value) -> value.json.length())
+ .expireAfterWrite(10, TimeUnit.MINUTES)
+ .build();
+ private final AtomicLong successfulBinds = new AtomicLong();
+ private final ThreadPoolExecutor writer = new ThreadPoolExecutor(1, 1, 0L,
+ TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
+ new NamedThreadFactory("plan-cache-writer-"));
+
+ public long getHitCount() {
+ return successfulBinds.get();
+ }
+
+ public void recordBind() {
+ successfulBinds.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, PhysicalPlan plan, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context) {
+ try {
+ writer.execute(() -> {
+ try {
+ put(key, plan, textPlan, reader, context);
+ } 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);
+ }
+ }
Review Comment:
**[Memory]** This goes with the `DrillSqlWorker` suggestion that serializes
right after planning: take the JSON string (and the literals for the trial
bind) instead of the live `PhysicalPlan`.
```suggestion
public void writeAfterSuccess(String key, String json, String textPlan,
PhysicalPlanReader reader, ContextSnapshot context, List<SqlLiteral>
literals) {
try {
writer.execute(() -> {
try {
put(key, json, textPlan, reader, context, literals);
} 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);
}
}
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,640 @@
+/*
+ * 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 java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+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.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.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.hash.Hashing;
+
+/** 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 = CacheBuilder.newBuilder()
+ .maximumWeight(32 * 1024 * 1024)
+ .weigher((String key, Entry value) -> value.json.length())
+ .expireAfterWrite(10, TimeUnit.MINUTES)
+ .build();
+ private final AtomicLong successfulBinds = new AtomicLong();
+ private final ThreadPoolExecutor writer = new ThreadPoolExecutor(1, 1, 0L,
+ TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
+ new NamedThreadFactory("plan-cache-writer-"));
+
+ public long getHitCount() {
+ return successfulBinds.get();
+ }
+
+ public void recordBind() {
+ successfulBinds.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, PhysicalPlan plan, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context) {
+ try {
+ writer.execute(() -> {
+ try {
+ put(key, plan, textPlan, reader, context);
+ } 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) {
+ entries.invalidate(key);
+ }
+
+ public boolean put(String key, PhysicalPlan plan, String textPlan,
+ PhysicalPlanReader reader, ContextSnapshot context) 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) {
+ return false;
+ }
+ String json = reader.writeJson(plan);
+ // Confirm that standard PhysicalPlan JSON can be read back before
publishing.
+ reader.readPhysicalPlan(json);
+ entries.put(key, new Entry(json, textPlan, context));
+ return true;
+ }
Review Comment:
**[Correctness / performance] Trial-bind before publishing, not just read
back.**
`readPhysicalPlan(json)` proves the JSON parses, but not that `Entry.bind`
works. Some entries fail on every hit, e.g. a textual field containing
`bound_dynamic_param(` that isn't a parseable expression, or an expression
class `bindExpression` doesn't handle. Those get published, fail on each hit,
are invalidated, replanned and republished, forever. Binding once with the
query's own literals runs on the writer thread, so it adds no latency to the
query. It also covers the read-back, because `bind` ends in `readPhysicalPlan`.
```suggestion
public boolean put(String key, String json, String textPlan,
PhysicalPlanReader reader, ContextSnapshot context, List<SqlLiteral>
literals)
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) {
return false;
}
Entry entry = new Entry(json, textPlan, context);
// A read-back alone does not prove the entry can be reused. Bind it
once with this
// query's own values, so entries whose bind always fails are never
published.
entry.bind(literals, reader);
entries.put(key, entry);
return true;
}
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java:
##########
@@ -295,7 +298,79 @@ private static PhysicalPlan getQueryPlan(QueryContext
context, String sql, Point
context.getOptions().setLocalOption(ExecConstants.RETURN_RESULT_SET_FOR_DDL,
true);
}
- return handler.getPlan(sqlNode);
+ // Keep the original SQL if no safe snapshot is available or cache
preparation fails.
+ SqlNode planningSql = sqlNode;
+ Consumer<PhysicalPlan> prepareCacheInsert = null;
+ if (context.getOptions().getOption(PlannerSettings.PLAN_CACHE)
+ && !context.getSession().hasTemporaryTables()
+ && !hasMutableAliases(context)
+ && PlanCacheEligibility.isSafeToCache(sqlNode,
context.getFunctionRegistry())) {
+ try {
+ // Capture the context once, before lookup and planning. Changes during
+ // planning invalidate this entry on its first hit.
+ PlanCache.ContextSnapshot snapshot = PlanCache.ContextSnapshot.resolve(
+ parser.getDefaultSchema(), sqlNode, context);
+ if (snapshot != null) {
+ PlanCacheParameterizer.Candidate candidate =
+ PlanCacheParameterizer.parameterize(sqlNode,
parser.getTypeFactory());
+ PlanCache cache = context.getDrillbitContext().getPlanCache();
+ String key = context.getQueryUserName() + '\n'
+ + context.getSession().getDefaultSchemaPath() + '\n'
+ + candidate.template;
+ PlanCache.Entry entry = cache.get(key);
+ if (entry != null) {
+ try {
+ if (entry.matchesContext(snapshot)) {
+ PhysicalPlan bound = entry.bind(candidate.literals,
+ context.getDrillbitContext().getPlanReader());
+ PlanCache.bindTextPlan(textPlan, entry.getTextPlan(),
candidate.literals);
+ cache.recordBind();
+ context.setPlanCacheHit();
+ return bound;
+ }
+ } catch (RuntimeException | IOException e) {
+ logger.debug("Cached plan could not be rebound", e);
+ }
+ cache.invalidate(key);
+ }
+ final PhysicalPlanReader cacheReader =
+ context.getDrillbitContext().getPlanReader();
+ // Prepare insertion after planning; Foreman publishes it only after
the query succeeds.
+ prepareCacheInsert = planned -> {
+ String templateTextPlan = PlanCache.bindTextPlan(textPlan,
candidate.literals);
+ context.setPendingPlanCacheInsert(() ->
cache.writeAfterSuccess(key, planned,
+ templateTextPlan, cacheReader, snapshot));
+ };
Review Comment:
**[Memory] Serialize the plan right after planning, and queue only the
JSON.**
`writeAfterSuccess` currently captures the live `PhysicalPlan`. Up to 64
full operator graphs (HBase region maps, Iceberg task lists, plugin references)
can stay alive in the writer queue. They're also serialized only after
parallelization and execution have touched the operators. Serializing here
costs little next to planning and avoids both problems. This also passes
`candidate.literals` through so the writer can trial-bind before publishing
(see the suggestion on `put`).
```suggestion
final PhysicalPlanReader cacheReader =
context.getDrillbitContext().getPlanReader();
// Prepare insertion after planning; Foreman publishes it only
after the query succeeds.
prepareCacheInsert = planned -> {
String templateTextPlan = PlanCache.bindTextPlan(textPlan,
candidate.literals);
String json;
try {
// Serialize now, so the writer queue holds a string rather
than the live plan
// graph, and execution cannot change what gets cached.
json = cacheReader.writeJson(planned);
} catch (IOException | RuntimeException e) {
logger.debug("Plan could not be serialized for the plan
cache", e);
return;
}
final String planJson = json;
context.setPendingPlanCacheInsert(() ->
cache.writeAfterSuccess(key, planJson,
templateTextPlan, cacheReader, snapshot,
candidate.literals));
};
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java:
##########
@@ -295,7 +298,79 @@ private static PhysicalPlan getQueryPlan(QueryContext
context, String sql, Point
context.getOptions().setLocalOption(ExecConstants.RETURN_RESULT_SET_FOR_DDL,
true);
}
- return handler.getPlan(sqlNode);
+ // Keep the original SQL if no safe snapshot is available or cache
preparation fails.
+ SqlNode planningSql = sqlNode;
+ Consumer<PhysicalPlan> prepareCacheInsert = null;
+ if (context.getOptions().getOption(PlannerSettings.PLAN_CACHE)
+ && !context.getSession().hasTemporaryTables()
+ && !hasMutableAliases(context)
+ && PlanCacheEligibility.isSafeToCache(sqlNode,
context.getFunctionRegistry())) {
+ try {
+ // Capture the context once, before lookup and planning. Changes during
+ // planning invalidate this entry on its first hit.
+ PlanCache.ContextSnapshot snapshot = PlanCache.ContextSnapshot.resolve(
+ parser.getDefaultSchema(), sqlNode, context);
+ if (snapshot != null) {
+ PlanCacheParameterizer.Candidate candidate =
+ PlanCacheParameterizer.parameterize(sqlNode,
parser.getTypeFactory());
+ PlanCache cache = context.getDrillbitContext().getPlanCache();
+ String key = context.getQueryUserName() + '\n'
+ + context.getSession().getDefaultSchemaPath() + '\n'
+ + candidate.template;
+ PlanCache.Entry entry = cache.get(key);
+ if (entry != null) {
+ try {
+ if (entry.matchesContext(snapshot)) {
+ PhysicalPlan bound = entry.bind(candidate.literals,
+ context.getDrillbitContext().getPlanReader());
+ PlanCache.bindTextPlan(textPlan, entry.getTextPlan(),
candidate.literals);
+ cache.recordBind();
+ context.setPlanCacheHit();
+ return bound;
+ }
+ } catch (RuntimeException | IOException e) {
+ logger.debug("Cached plan could not be rebound", e);
+ }
+ cache.invalidate(key);
+ }
Review Comment:
**[Observability] Log hits, misses and why an entry was invalidated, and
count misses.**
Right now a hit is only visible in the query profile, and stale-entry
invalidations aren't logged at all. That makes it hard to tell whether the
cache is working in production. Foreman threads are named after the query ID,
so these debug lines can be matched to a query. `recordMiss()` is added in
`PlanCache` (see the suggestion on `invalidate`) and feeds the new metrics.
```suggestion
PlanCache.Entry entry = cache.get(key);
if (entry != null) {
try {
if (entry.matchesContext(snapshot)) {
PhysicalPlan bound = entry.bind(candidate.literals,
context.getDrillbitContext().getPlanReader());
PlanCache.bindTextPlan(textPlan, entry.getTextPlan(),
candidate.literals);
cache.recordBind();
context.setPlanCacheHit();
logger.debug("Plan cache hit");
return bound;
}
logger.debug("Plan cache entry is stale (table version
changed); replanning");
} catch (RuntimeException | IOException e) {
logger.debug("Cached plan could not be rebound; replanning",
e);
}
cache.invalidate(key);
} else {
logger.debug("Plan cache miss");
}
cache.recordMiss();
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,640 @@
+/*
+ * 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 java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+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.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.cache.Cache;
+import com.google.common.cache.CacheBuilder;
+import com.google.common.hash.Hashing;
+
+/** 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 = CacheBuilder.newBuilder()
+ .maximumWeight(32 * 1024 * 1024)
+ .weigher((String key, Entry value) -> value.json.length())
+ .expireAfterWrite(10, TimeUnit.MINUTES)
+ .build();
+ private final AtomicLong successfulBinds = new AtomicLong();
+ private final ThreadPoolExecutor writer = new ThreadPoolExecutor(1, 1, 0L,
+ TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
+ new NamedThreadFactory("plan-cache-writer-"));
Review Comment:
**[Configuration / observability] Read size and expiry from boot options,
and expose metrics.**
This replaces the hard-coded 32 MB / 10-minute limits with boot options (as
discussed in my earlier comment) and registers
hit/miss/invalidation/eviction/size gauges. The gauges show up in the Web UI
metrics page and over JMX. `getHitCount()` currently has no caller outside
tests.
Three other changes are needed for this to compile and run:
1. Imports (suggestions on the import lines in this file):
`org.apache.drill.common.config.DrillConfig`,
`org.apache.drill.exec.metrics.DrillMetrics`, `com.codahale.metrics.Gauge`.
2. `DrillbitContext`: `new PlanCache(config)` (suggestion there).
3. Defaults in `exec/java-exec/src/main/resources/drill-module.conf`, inside
`drill.exec`, e.g. after the `compile` block. I can't attach a suggestion to
that spot because it's outside the diff:
```
plan_cache: {
# Upper bound on the total size of cached plan JSON on each Drillbit.
max_size: 32MB,
# How long an entry lives after it is written. 0 disables time-based
expiry.
expire_after_write: 10m,
# Evict entries that have not been used for this long. 0 disables.
expire_after_access: 0
},
```
`DrillMetrics.register` replaces an existing gauge of the same name, so
several Drillbits in one test JVM are fine (same as `WorkManager`'s
`drill.fragments.running`).
```suggestion
static final String CONFIG_PREFIX = "drill.exec.plan_cache.";
static final String MAX_SIZE_CONFIG = CONFIG_PREFIX + "max_size";
static final String EXPIRE_AFTER_WRITE_CONFIG = CONFIG_PREFIX +
"expire_after_write";
static final String EXPIRE_AFTER_ACCESS_CONFIG = CONFIG_PREFIX +
"expire_after_access";
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 ThreadPoolExecutor writer = new ThreadPoolExecutor(1, 1, 0L,
TimeUnit.MILLISECONDS, new ArrayBlockingQueue<>(64),
new NamedThreadFactory("plan-cache-writer-"));
public PlanCache(DrillConfig config) {
long maxSize = config.getBytes(MAX_SIZE_CONFIG);
long expireAfterWrite = config.getDuration(EXPIRE_AFTER_WRITE_CONFIG,
TimeUnit.SECONDS);
long expireAfterAccess = config.getDuration(EXPIRE_AFTER_ACCESS_CONFIG,
TimeUnit.SECONDS);
CacheBuilder<String, Entry> builder = CacheBuilder.newBuilder()
.maximumWeight(maxSize)
// Approximate: counts JSON characters, which equals bytes for ASCII
plans.
.weigher((String key, Entry value) -> value.json.length())
.recordStats();
if (expireAfterWrite > 0) {
builder.expireAfterWrite(expireAfterWrite, TimeUnit.SECONDS);
}
if (expireAfterAccess > 0) {
builder.expireAfterAccess(expireAfterAccess, TimeUnit.SECONDS);
}
entries = builder.build();
logger.info("Plan cache: max size {} bytes, expire after write {}s,
expire after access {}s (0 = never)",
maxSize, expireAfterWrite, expireAfterAccess);
DrillMetrics.register("drill.plan_cache.hits", (Gauge<Long>)
successfulBinds::get);
DrillMetrics.register("drill.plan_cache.misses", (Gauge<Long>)
misses::get);
DrillMetrics.register("drill.plan_cache.invalidations", (Gauge<Long>)
invalidations::get);
DrillMetrics.register("drill.plan_cache.evictions", (Gauge<Long>) () ->
entries.stats().evictionCount());
DrillMetrics.register("drill.plan_cache.entries", (Gauge<Long>)
entries::size);
}
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,640 @@
+/*
+ * 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 java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+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.ops.QueryContext;
Review Comment:
Import for the plan-cache metrics.
```suggestion
import org.apache.drill.exec.metrics.DrillMetrics;
import org.apache.drill.exec.ops.QueryContext;
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,640 @@
+/*
+ * 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 java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+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;
Review Comment:
Import for the configurable cache constructor.
```suggestion
import org.apache.drill.common.config.DrillConfig;
import org.apache.drill.common.expression.AnyValueExpression;
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/server/DrillbitContext.java:
##########
@@ -122,6 +124,7 @@ public DrillbitContext(
ExecConstants.STORAGE_PLUGIN_REGISTRY_IMPL,
StoragePluginRegistry.class, this);
reader = new PhysicalPlanReader(config, classpathScan, lpPersistence,
endpoint, storagePlugins);
+ planCache = new PlanCache();
Review Comment:
Pass the boot config so the cache size and expiry can be configured (see the
suggestion on `PlanCache`'s fields).
```suggestion
planCache = new PlanCache(config);
```
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCache.java:
##########
@@ -0,0 +1,640 @@
+/*
+ * 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 java.io.IOException;
+import java.math.BigDecimal;
+import java.math.RoundingMode;
+import java.nio.charset.StandardCharsets;
+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.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;
Review Comment:
Import for the plan-cache metrics.
```suggestion
import com.codahale.metrics.Gauge;
import com.fasterxml.jackson.databind.JsonNode;
```
##########
contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseStoragePlugin.java:
##########
@@ -58,6 +63,27 @@ public boolean supportsRead() {
return true;
}
+ @Override
+ public boolean supportPlanCache(DrillTableSelection selection) {
+ return selection instanceof HBaseScanSpec;
+ }
+
+ @Override
+ public PlanCacheTable planCacheTable(DrillTableSelection selection) throws
IOException {
+ if (!supportPlanCache(selection)) {
+ return null;
+ }
+ String identifier = ((HBaseScanSpec) selection).getTableName();
+ try (Admin admin = getConnection().getAdmin()) {
Review Comment:
**[Performance] Remote metadata I/O on every lookup, including hits.**
`ContextSnapshot.resolve` calls this before every lookup. Each call opens an
`Admin` and makes two master RPCs (`tableExists` + `getTableDescriptor`).
Iceberg is similar: `IcebergFormatPlugin.planCacheTable` runs
`HadoopTables.load`, and `IcebergGroupScan` loads the table again during bind.
A hit can end up costing more than the planning it skips.
Possible fixes:
- Drop `tableExists` and call `getDescriptor` directly (it throws
`TableNotFoundException`, which can map to `null`). That's one RPC instead of
two.
- Use `Connection.getTable(name).getDescriptor()` rather than opening an
`Admin` per query.
- Cache the descriptor hash briefly (a few seconds) per Drillbit, accepting
a short window where a schema change is caught at bind time instead.
Could you include hit latency against full planning time for an HBase point
lookup in the PR benchmarks?
##########
PLAN_CACHE_DESIGN.md:
##########
@@ -0,0 +1,103 @@
+<!--
Review Comment:
**[Docs]** Please move `PLAN_CACHE_DESIGN.md` and
`PLAN_CACHE_PLUGIN_GUIDE.md` from the repo root to `docs/dev/`, and link them
from `docs/dev/DevDocs.md`. It would also help to document the new boot options
and metrics, and the plan for supporting other plugins (see my earlier comment).
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCacheParameterizer.java:
##########
@@ -0,0 +1,205 @@
+/*
+ * 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 java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.Locale;
+import java.util.Objects;
+
+import org.apache.calcite.rel.type.RelDataTypeFactory;
+import org.apache.calcite.sql.SqlCall;
+import org.apache.calcite.sql.SqlJoin;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlLiteral;
+import org.apache.calcite.sql.SqlNode;
+import org.apache.calcite.sql.SqlOrderBy;
+import org.apache.calcite.sql.SqlNodeList;
+import org.apache.calcite.sql.SqlSelect;
+import org.apache.calcite.sql.type.SqlTypeName;
+import org.apache.calcite.sql.util.SqlShuttle;
+import org.apache.calcite.util.NlsString;
+import org.apache.drill.exec.planner.sql.conversion.SqlBoundDynamicParam;
+
+/** Extracts SQL literals into typed slots independently of cache eligibility.
*/
+final class PlanCacheParameterizer extends SqlShuttle {
+ private final List<SqlLiteral> literals = new ArrayList<>();
+ private final List<String> types = new ArrayList<>();
+ private final RelDataTypeFactory typeFactory;
+ private boolean preserveProjection;
+
+ private PlanCacheParameterizer(RelDataTypeFactory typeFactory) {
+ this.typeFactory = Objects.requireNonNull(typeFactory, "typeFactory");
+ }
+
+ static Candidate parameterize(SqlNode parsed, RelDataTypeFactory
typeFactory) {
+ Objects.requireNonNull(parsed, "parsed");
+ PlanCacheParameterizer visitor = new PlanCacheParameterizer(typeFactory);
+ SqlNode parameterized = parsed.accept(visitor);
+ if (visitor.literals.isEmpty()) {
+ parameterized = parsed;
+ }
+ return new Candidate(parameterized,
+ Collections.unmodifiableList(visitor.literals),
+ parameterized.toString() + visitor.types);
+ }
+
+ @Override
+ public SqlNode visit(SqlCall call) {
+ int configurationOperand = configurationOperand(call);
+ if (configurationOperand >= 0 && configurationOperand <
call.operandCount()) {
+ // These operands become function names, encodings or resolved plugin
+ // references during conversion. They are not bindable expressions in
+ // the physical plan, so keep their entire subtree in the template key.
+ SqlCall parameterized = (SqlCall) call.clone(call.getParserPosition());
+ for (int i = 0; i < call.operandCount(); i++) {
+ if (i != configurationOperand) {
+ parameterized.setOperand(i, visitNullable(call.operand(i)));
+ }
+ }
+ return parameterized;
+ }
+ if ("COALESCE".equalsIgnoreCase(call.getOperator().getName())
+ && call.getOperandList().stream()
+ .anyMatch(operand -> operand instanceof SqlLiteral
+ && ((SqlLiteral) operand).getTypeName() == SqlTypeName.NULL)) {
+ // Calcite cannot infer a dynamic parameter's type alongside an untyped
+ // NULL. Keep this call in the key; explicit casts still allow slots.
+ return call;
+ }
+ if (call instanceof SqlJoin) {
+ SqlJoin join = (SqlJoin) call;
+ return new SqlJoin(join.getParserPosition(),
visitNullable(join.getLeft()),
+ join.isNaturalNode(), join.getJoinTypeNode(),
visitNullable(join.getRight()),
+ join.getConditionTypeNode(), visitNullable(join.getCondition()));
+ }
+ if (call.getKind() == SqlKind.ITEM && call.operand(1) instanceof
SqlLiteral) {
+ // A map/array key identifies a field in the input schema. Changing it
can
+ // change the scan projection and the HBase column used by pushdown.
+ return call.getOperator().createCall(call.getParserPosition(),
+ visitNullable(call.operand(0)), call.operand(1));
+ }
+ // These literals control rows, ordinals, window frames or operator
+ // configuration. Keep them verbatim in the template key.
+ if (call.getKind() == SqlKind.VALUES || call.getKind() == SqlKind.ORDER_BY
+ || call.getKind() == SqlKind.OVER) {
+ if (call instanceof SqlOrderBy) {
+ SqlOrderBy orderBy = (SqlOrderBy) call;
+ boolean previous = preserveProjection;
+ preserveProjection = true;
+ SqlNode query;
+ try {
+ query = orderBy.query.accept(this);
+ } finally {
+ preserveProjection = previous;
+ }
+ return new SqlOrderBy(call.getParserPosition(), query,
orderBy.orderList,
+ orderBy.offset, orderBy.fetch);
+ }
+ return call;
+ }
+ if (call instanceof SqlSelect) {
+ SqlSelect select = (SqlSelect) call;
+ // Keep structural SQL operands verbatim while visiting expressions.
+ boolean structuralProjection = preserveProjection
+ || (select.getGroup() != null && !select.getGroup().isEmpty())
+ || (select.getOrderList() != null &&
!select.getOrderList().isEmpty())
+ || (select.getWindowList() != null &&
!select.getWindowList().isEmpty());
+ return new SqlSelect(select.getParserPosition(),
+ (SqlNodeList) select.getOperandList().get(0),
+ structuralProjection ? select.getSelectList()
+ : (SqlNodeList) visitNullable(select.getSelectList()),
+ visitNullable(select.getFrom()), visitNullable(select.getWhere()),
+ select.getGroup(), visitNullable(select.getHaving()),
+ select.getWindowList(), visitNullable(select.getQualify()),
+ select.getOrderList(), select.getOffset(), select.getFetch(),
+ select.getHints());
+ }
+ return super.visit(call);
+ }
+
+ private static int configurationOperand(SqlCall call) {
+ // Keep this list aligned with the value-dependent rewrites in DrillOptiq
+ // and PreProcessLogicalRel; ordinary data operands still get slots.
+ switch (call.getOperator().getName().toUpperCase(Locale.ROOT)) {
+ case "DATE_PART":
Review Comment:
**[Robustness] Hard-coded list of functions whose operands must stay
literal.**
Any function not on this list that needs a literal operand during
validation, return-type inference or `DrillOptiq` conversion silently gets a
slot and fails planning. That includes format strings, scale arguments,
percentile fractions and new UDFs. The fallback suggested in `DrillSqlWorker`
stops these from failing queries. It would still be better not to rely on the
list: could literal-only operands come from the operator's operand type checker
(e.g. `SqlOperandTypeChecker` / `OperandTypes.LITERAL`), or from an annotation
on Drill UDFs, rather than a name switch?
--
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]
