cgivre commented on code in PR #3086:
URL: https://github.com/apache/drill/pull/3086#discussion_r4343148969
##########
contrib/format-iceberg/src/main/java/org/apache/drill/exec/store/iceberg/plan/DrillExprToIcebergTranslator.java:
##########
@@ -103,7 +145,7 @@ public Expression visitFunctionCall(FunctionCall call, Void
value) throws Runtim
}
case FunctionNames.GE: {
LogicalExpression nameRef = call.args().get(0);
- Expression expression = call.args().get(0).accept(this, null);
+ Expression expression = call.args().get(1).accept(this, null);
Review Comment:
**[Behaviour change for all queries] This changes Iceberg pushdown with the
cache off too.**
Several changes in this file and in `IcebergPluginImplementor` change
pushdown for every Iceberg query:
- `>=` previously read `args().get(0)` and was never pushed; now it is.
- AND/OR are now pushed.
- Successive filters are combined with AND.
- DATE constants become days, and TIME/TIMESTAMP constants become
microseconds. Previously they were milliseconds, which could prune the wrong
files.
These all look like genuine fixes, but there are no tests for any of them.
When a format config sets `ignoreResiduals: true`, the newly pushed predicates
mean rows that should have been filtered can come back.
Could you add translator unit tests and a few Iceberg query tests with the
cache off? Cover `>=`, AND/OR, and date/timestamp filters. Alternatively, move
these into their own PR so they can be reviewed separately. Either is fine with
me.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/DrillSqlWorker.java:
##########
@@ -295,7 +305,95 @@ 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 {
+ PlanCache cache = context.getDrillbitContext().getPlanCache();
+ long cacheGeneration = cache.getGeneration();
+ // 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());
+ String key = context.getQueryUserName() + '\n'
+ + context.getSession().getDefaultSchemaPath() + '\n'
+ + snapshot.keyFingerprint() + '\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();
+ logger.debug("Plan cache hit");
+ return bound;
+ }
+ logger.debug("Plan cache entry invalidated because its context
changed; replanning");
+ } catch (RuntimeException | IOException e) {
+ logger.debug("Plan cache entry invalidated because rebinding
failed; replanning", e);
+ }
+ cache.invalidate(key);
+ } else {
+ logger.debug("Plan cache miss");
+ }
+ cache.recordMiss();
+ 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);
+ final String json;
+ try {
+ // Capture before parallelization/execution mutates the
operators.
+ // Pending publication and the writer queue retain only
immutable JSON.
+ json = cacheReader.writeJson(planned);
+ } catch (IOException | RuntimeException e) {
+ logger.debug("Plan could not be serialized for the plan cache;
skipping publication", e);
+ return;
+ }
+ context.setPendingPlanCacheInsert(() ->
cache.writeAfterSuccess(key, json,
+ templateTextPlan, cacheReader, snapshot, cacheGeneration));
+ };
+ // 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);
Review Comment:
**[Correctness] No fallback when the parameterized SQL fails to plan.**
I understand why you don't want to retry on the same `SqlNode`, since
validation mutates it. The fallback doesn't need to reuse it, though. On
failure, set `planner.enable_plan_cache=false` for this query and call
`getQueryPlan` again from the original SQL string. That parses a fresh tree.
Without this, queries that only plan with real literals fail only when the
cache is on, with no workaround short of turning the cache off. Examples are
`ON TRUE`, a function that needs a literal operand but isn't on the hard-coded
list, and a new UDF.
##########
exec/java-exec/src/main/java/org/apache/drill/exec/planner/sql/PlanCacheParameterizer.java:
##########
@@ -0,0 +1,208 @@
+/*
+ * 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;
+ // Preserve matching expressions across SELECT, GROUP BY, HAVING and
+ // QUALIFY. Parameterizing only some occurrences breaks validation or
rewrites.
+ boolean structuralProjection = preserveProjection
Review Comment:
**[Correctness] QUALIFY with an inline OVER and no ORDER BY is still
parameterized.**
The SELECT list, HAVING and QUALIFY are preserved only when there is a GROUP
BY, ORDER BY or WINDOW clause. An inline `OVER(...)` is preserved on its own,
but the expressions around it are not. Your zero-rows example from the last
round was fixed because it ended in `ORDER BY`. The same query without it still
gets `x + ? = ?` in both the select list and QUALIFY, and the root cause of the
zero rows was never identified.
Could `structuralProjection` also be true when the SELECT contains any
`OVER` call or a QUALIFY clause? Please also add the no-ORDER-BY variant to
`TestPlanCachePlanning`.
##########
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:
**[Robustness] A bind that always fails cycles forever.**
The read-back here proves the JSON parses, but not that `Entry.bind`
succeeds. If a plan's bind always fails, each hit fails, invalidates and
replans, and the entry is published again. An example is a non-expression text
field that happens to contain `bound_dynamic_param(`.
A trial bind with the original literals before publishing would catch this.
A small per-key failure counter that stops republishing would also work.
##########
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:
Minor: entries are published after the query finishes, so a long query that
started before a schema change can overwrite a newer entry with its older
snapshot. This causes one extra invalidate-and-replan cycle.
`entries.asMap().compute(...)` that skips the write when the existing entry's
context differs would avoid it.
##########
logical/src/main/antlr4/org/apache/drill/common/expression/parser/ExprLexer.g4:
##########
@@ -94,6 +94,8 @@ Bool
| 'false'
;
+BoundDynamicParam : 'bound_dynamic_param';
Review Comment:
Nit: this makes `bound_dynamic_param` a reserved word for all expression
parsing, so an unquoted column or function with that name no longer parses.
It's unlikely to matter, and `date`/`as` already behave this way. If you want
to avoid it, match `Identifier` with a semantic predicate instead of a lexer
token.
##########
contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseGroupScan.java:
##########
@@ -108,18 +116,44 @@ public HBaseGroupScan(@JsonProperty("userName") String
userName,
@JsonProperty("storage") HBaseStoragePluginConfig
storagePluginConfig,
@JsonProperty("columns") List<SchemaPath> columns,
@JsonProperty("maxRecords") int maxRecords,
+ @JsonProperty("baseScanSpec") HBaseScanSpec
baseScanSpec,
+ @JsonProperty("pushedFilter") String pushedFilter,
+ @JsonProperty("pushedFilterFullyConverted") boolean
pushedFilterFullyConverted,
@JacksonInject StoragePluginRegistry pluginRegistry)
throws IOException, ExecutionSetupException {
- this (userName, pluginRegistry.resolve(storagePluginConfig,
HBaseStoragePlugin.class), hbaseScanSpec, columns, maxRecords);
+ this(userName, pluginRegistry.resolve(storagePluginConfig,
HBaseStoragePlugin.class),
+ hbaseScanSpec, columns, maxRecords, baseScanSpec,
+ pushedFilter == null ? null :
LogicalExpressionParser.parse(pushedFilter),
+ pushedFilterFullyConverted, true);
}
public HBaseGroupScan(String userName, HBaseStoragePlugin storagePlugin,
HBaseScanSpec scanSpec,
List<SchemaPath> columns, int maxRecords) {
+ this(userName, storagePlugin, scanSpec, columns, maxRecords, null, null,
false, false);
+ }
+
+ HBaseGroupScan(String userName, HBaseStoragePlugin storagePlugin,
HBaseScanSpec scanSpec,
+ List<SchemaPath> columns, int maxRecords, HBaseScanSpec baseScanSpec,
+ LogicalExpression pushedFilter, boolean pushedFilterFullyConverted,
boolean rebuild) {
super(userName);
this.storagePlugin = storagePlugin;
this.storagePluginConfig = storagePlugin.getConfig();
- this.hbaseScanSpec = scanSpec;
+ this.baseScanSpec = baseScanSpec;
+ this.pushedFilter = pushedFilter;
+ this.pushedFilterFullyConverted = pushedFilterFullyConverted;
+ this.hbaseScanSpec = rebuild && pushedFilter != null ? baseScanSpec :
scanSpec;
this.columns = columns == null ? ALL_COLUMNS : columns;
this.maxRecords = maxRecords;
+ if (rebuild && pushedFilter != null) {
+ if (baseScanSpec == null) {
+ throw new IllegalArgumentException("Cached HBase filter has no base
scan specification");
+ }
+ HBaseFilterBuilder builder = new HBaseFilterBuilder(this, pushedFilter);
+ HBaseScanSpec rebuilt = builder.parseTree();
+ if (rebuilt == null || (pushedFilterFullyConverted &&
!builder.isAllExpressionsConverted())) {
Review Comment:
Minor: this rebuild runs whenever an HBase scan is read from JSON, not only
on a cache bind, and a failure is now a hard `IllegalArgumentException`. Plans
submitted as physical-plan JSON, or replayed from a profile, take this path too.
Could non-cache reads fall back to the serialized `hbaseScanSpec`? Also,
`TestHBasePlanCache` uses a single region and fully converted predicates only.
A test for a region split between warm-up and hit would be valuable, as would a
partially converted filter and a value that flips full conversion to partial.
##########
contrib/format-iceberg/src/main/java/org/apache/drill/exec/store/iceberg/plan/IcebergPluginImplementor.java:
##########
@@ -67,6 +70,10 @@ public void implement(PluginFilterRel filter) throws
IOException {
new
DrillParseContext(PrelUtil.getPlannerSettings(filter.getCluster().getPlanner())),
filter.getInput(),
condition);
+ if (groupScan.getCondition() != null) {
Review Comment:
Related to the translator comment: combining with the existing condition is
the right behaviour, but it also affects queries with the cache off. A test
with two stacked filters would pin it down.
##########
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:
**[Operations]** This clears only the Drillbit that plans the command.
Behind a load balancer, `ALTER SYSTEM CLEAR PLAN CACHE` reads as cluster-wide,
but the other Drillbits keep their entries. Could it be broadcast, through the
control RPC or a ZK generation counter? If that's too much for this PR, please
document it as per-Drillbit (and consider the wording of the command).
##########
contrib/storage-hbase/src/main/java/org/apache/drill/exec/store/hbase/HBaseStoragePlugin.java:
##########
@@ -58,6 +65,26 @@ 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 (Table table =
getConnection().getTable(TableName.valueOf(identifier))) {
+ String version = Hashing.sha256().hashBytes(
+
TableDescriptorBuilder.toByteArray(table.getDescriptor())).toString();
Review Comment:
**[Performance]** Thanks for removing `tableExists` here. The descriptor is
still fetched about three times per query:
- once by the existing `AbstractHBaseDrillTable` lookup, which also opens an
`Admin` it never closes;
- once here;
- once in `HBaseGroupScan.init()` on a hit.
Could this hash the descriptor `DrillHBaseTable` already holds, or memoize
it briefly? Iceberg has the same pattern: `HadoopTables.load` runs here and
again in `IcebergGroupScan.initTableScan`.
##########
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:
**[Correctness] A change to the provided schema doesn't invalidate the
entry.**
The version is `uuid:currentSchemaId`. However, the cached
`IcebergGroupScan` also carries Drill's provided schema from `.drill.schema`
when `store.table.use_schema_file=true`. After `CREATE OR REPLACE SCHEMA ...
FOR TABLE`, hits keep using the old types and casts. The options fingerprint
only covers the flag, not the file contents.
Could you fold a hash of the provided schema into the version (or the schema
file's modification time and length), or return null when a schema provider is
present?
Separately, tables without a UUID silently never cache. That's safe, but
worth a line in the plugin guide.
##########
logical/src/main/java/org/apache/drill/common/expression/LiteralExpression.java:
##########
@@ -0,0 +1,47 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.common.expression;
+
+import com.fasterxml.jackson.annotation.JsonIgnore;
+
+/** A literal that may carry a bound SQL parameter slot in a cached plan. */
+public abstract class LiteralExpression extends LogicalExpressionBase {
+ private int dynamicParamIndex = -1;
+
+ protected LiteralExpression(ExpressionPosition position) {
+ super(position);
+ }
+
+ @JsonIgnore
+ public int getDynamicParamIndex() {
+ return dynamicParamIndex;
+ }
+
+ /** Whether this literal is bound to a zero-based SQL parameter slot. */
+ @JsonIgnore
+ public boolean isDynamicParam() {
+ return dynamicParamIndex >= 0;
+ }
+
+ public void setDynamicParamIndex(int dynamicParamIndex) {
Review Comment:
Minor hardening: this public mutator is inherited by shared singletons
(`BooleanExpression.TRUE/FALSE`, `QuotedString.EMPTY_STRING`). It's unreachable
today because `DrillOptiq` allocates fresh literals. But if anything ever
returns a singleton and sets a slot on it, every query in the JVM would
serialize that constant as a slot.
A `withDynamicParamIndex(i)` that returns a copy would make this impossible.
Relatedly, the `Decimal9`–`38` classes now extend `LiteralExpression`, but
their serializers would silently drop a slot. Throwing for slotted types
without a wrapper would be safer.
##########
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:
**[Compatibility] Markers reach executed fragments.**
The bound expressions keep their slot index, so executed fragments, profiles
and plan text contain `bound_dynamic_param(i, TYPE, v)`. On the first miss and
on hits, remote fragments therefore carry syntax a Drillbit without this PR
can't parse. A rolling upgrade with the cache enabled would fail queries at
fragment setup.
Could the markers stay in the cached template only, and be cleared (or
rewritten to plain literals) before the plan is parallelized? That would also
make profiles show the actual values.
--
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]
