Skip to content

DRILL-8555: Physical plan cache for parameterized SQL queries #3086

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
letian-jiang wants to merge 17 commits into
base: master
Choose a base branch
from
+5,266 −163
Open

DRILL-8555: Physical plan cache for parameterized SQL queries #3086

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
17 commits
Select commit Hold shift + click to select a range
a4c43b3
Add opt-in physical plan cache with HBase and Iceberg support
letian-jiang Oct 2, 2026
4136460
Remove unused temporal plan-cache parameter markers
letian-jiang Oct 2, 2026
b003715
Clarify metadata table exclusion from Iceberg plan caching
letian-jiang Oct 2, 2026
f5e64d5
Regenerate plan cache profile protobufs with CI compiler
letian-jiang Oct 2, 2026
8f1249a
Fix plan cache eligibility and BIGINT plan round trips
letian-jiang Oct 6, 2026
3abcce8
Keep planning failures outside plan cache fallback
letian-jiang Oct 6, 2026
d7c9ba6
Simplify plan cache plugin APIs and expression serialization
letian-jiang Oct 6, 2026
8ff6bb6
Use imports for plan cache type references
letian-jiang Oct 6, 2026
53306c1
Preserve literal types and bind cached expression nodes safely
letian-jiang Oct 6, 2026
39d86b1
Make physical plan cache size and expiration configurable
letian-jiang Oct 8, 2026
e16de1e
Include cache keys and text plans in plan cache weight
letian-jiang Oct 8, 2026
9ce7361
Document plan cache storage plugin roadmap
letian-jiang Oct 8, 2026
419403d
Preserve matching HAVING and QUALIFY expressions in plan cache templates
letian-jiang Oct 8, 2026
6aa9618
Address plan cache review feedback and add SQL cache management
letian-jiang Oct 8, 2026
6ee1c59
Reduce HBase plan cache metadata lookup overhead
letian-jiang Oct 8, 2026
b40a70d
Add plan cache integration coverage for HBase and Iceberg
letian-jiang Oct 8, 2026
1ae6773
Address plan cache review feedback and schema compatibility
letian-jiang Oct 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 27 additions & 1 deletion ...iceberg/src/main/java/org/apache/drill/exec/store/iceberg/format/IcebergFormatPlugin.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -29,6 +29,7 @@
import org.apache.drill.exec.record.metadata.TupleMetadata;
import org.apache.drill.exec.record.metadata.schema.SchemaProvider;
import org.apache.drill.exec.server.DrillbitContext;
import org.apache.drill.exec.store.PlanCacheTable;
import org.apache.drill.exec.store.PluginRulesProviderImpl;
import org.apache.drill.exec.store.StoragePluginRulesSupplier;
import org.apache.drill.exec.store.dfs.FileSelection;
Expand All @@ -41,6 +42,10 @@
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.iceberg.HasTableOperations;
import org.apache.iceberg.Table;
import org.apache.iceberg.TableMetadata;
import org.apache.iceberg.hadoop.HadoopTables;

import java.io.IOException;
import java.util.Collections;
Expand Down Expand Up @@ -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());

@cgivre cgivre Oct 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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.

@letian-jiang letian-jiang Oct 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Provided schema contents now contribute to table compatibility. For example, changing id VARCHAR to id BIGINT invalidates the old entry. Added tests with and without provided schemas, including native schema changes. Tables without an Iceberg UUID still bypass caching; this is now explicit in the plugin guide.

}

@Override
public boolean supportsWrite() {
return false;
Expand Down Expand Up @@ -154,7 +180,7 @@ public AbstractGroupScan getGroupScan(String userName, FileSelection selection,
public AbstractGroupScan getGroupScan(String userName, FileSelection selection,
List<SchemaPath> columns, MetadataProviderManager metadataProviderManager) throws IOException {
SchemaProvider schemaProvider = metadataProviderManager.getSchemaProvider();
TupleMetadata schema = schemaProvider != null
TupleMetadata schema = schemaProvider != null && schemaProvider.exists()
? schemaProvider.read().getSchema()
: null;
return IcebergGroupScan.builder()
Expand Down
68 changes: 57 additions & 11 deletions .../src/main/java/org/apache/drill/exec/store/iceberg/plan/DrillExprToIcebergTranslator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,8 @@
package org.apache.drill.exec.store.iceberg.plan;

import org.apache.drill.common.FunctionNames;
import org.apache.drill.common.expression.BooleanOperator;
import org.apache.drill.common.expression.CastExpression;
import org.apache.drill.common.expression.FunctionCall;
import org.apache.drill.common.expression.LogicalExpression;
import org.apache.drill.common.expression.SchemaPath;
Expand All @@ -32,6 +34,46 @@ public class DrillExprToIcebergTranslator extends AbstractExprVisitor<Expression

public static final ExprVisitor<Expression, Void, RuntimeException> INSTANCE = new DrillExprToIcebergTranslator();

@Override
public Expression visitBooleanOperator(BooleanOperator op, Void value) {
Expression result = null;
for (LogicalExpression argument : op.args()) {
Expression next = argument.accept(this, value);
if (next == null) {
return null;
}
result = result == null ? next : FunctionNames.AND.equals(op.getName())
? Expressions.and(result, next) : Expressions.or(result, next);
}
return result;
}

@Override
public Expression visitCastExpression(CastExpression cast, Void value) {
Expression input = cast.getInput().accept(this, value);
if (!(input instanceof ConstantExpression)) {
return null;
}
Object literal = ((ConstantExpression<?>) input).getValue();
if (!(literal instanceof Integer) && !(literal instanceof Long)) {
return null;
}
long millis = ((Number) literal).longValue();
// Drill serializes temporal constants as casts of epoch milliseconds.
// Iceberg's DATE uses days; TIME and TIMESTAMP use microseconds.
switch (cast.getMajorType().getMinorType()) {
case BIGINT:
return new ConstantExpression<>(millis);
case DATE:
return new ConstantExpression<>(Math.toIntExact(Math.floorDiv(millis, 86_400_000L)));
case TIME:
case TIMESTAMP:
return new ConstantExpression<>(Math.multiplyExact(millis, 1_000L));
default:
return null;
}
}

@Override
public Expression visitFunctionCall(FunctionCall call, Void value) throws RuntimeException {
switch (call.getName()) {
Expand All @@ -52,10 +94,8 @@ public Expression visitFunctionCall(FunctionCall call, Void value) throws Runtim
return null;
}
case FunctionNames.NOT: {
Expression expression = call.arg(0).accept(this, null);
if (expression != null) {
return Expressions.not(expression);
}
// Complementing a two-valued Iceberg predicate does not preserve SQL
// UNKNOWN for null operands. Leave unsimplified NOT predicates in Drill.
return null;
}
case FunctionNames.IS_NULL: {
Expand All @@ -79,7 +119,7 @@ public Expression visitFunctionCall(FunctionCall call, Void value) throws Runtim
Expression expression = call.arg(1).accept(this, null);
if (nameRef instanceof SchemaPath && expression instanceof ConstantExpression) {
String name = IcebergGroupScan.getPath((SchemaPath) nameRef);
return Expressions.lessThan(name, ((ConstantExpression<?>) expression).getValue());
return rejectNull(name, Expressions.lessThan(name, ((ConstantExpression<?>) expression).getValue()));
}
return null;
}
Expand All @@ -88,7 +128,7 @@ public Expression visitFunctionCall(FunctionCall call, Void value) throws Runtim
Expression expression = call.arg(1).accept(this, null);
if (nameRef instanceof SchemaPath && expression instanceof ConstantExpression) {
String name = IcebergGroupScan.getPath((SchemaPath) nameRef);
return Expressions.lessThanOrEqual(name, ((ConstantExpression<?>) expression).getValue());
return rejectNull(name, Expressions.lessThanOrEqual(name, ((ConstantExpression<?>) expression).getValue()));
}
return null;
}
Expand All @@ -103,7 +143,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);

@cgivre cgivre Oct 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

[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.

@letian-jiang letian-jiang Oct 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added translator and cache-off query tests for >=, AND/OR and temporal values. Also fixed NULL handling and retained Drill filtering when ignoreResiduals=true.

if (nameRef instanceof SchemaPath && expression instanceof ConstantExpression) {
String name = IcebergGroupScan.getPath((SchemaPath) nameRef);
return Expressions.greaterThanOrEqual(name, ((ConstantExpression<?>) expression).getValue());
Expand All @@ -124,14 +164,20 @@ public Expression visitFunctionCall(FunctionCall call, Void value) throws Runtim
Expression expression = call.args().get(1).accept(this, null);
if (nameRef instanceof SchemaPath && expression instanceof ConstantExpression) {
String name = IcebergGroupScan.getPath((SchemaPath) nameRef);
return Expressions.notEqual(name, ((ConstantExpression<?>) expression).getValue());
return rejectNull(name, Expressions.notEqual(name, ((ConstantExpression<?>) expression).getValue()));
}
return null;
}
}
return null;
}

private static Expression rejectNull(String name, Expression predicate) {
// Iceberg 0.12.1 orders null before non-null values when evaluating residuals.
// SQL comparisons must reject null, including <, <= and !=.
return Expressions.and(Expressions.notNull(name), predicate);
}

@Override
public Expression visitFloatConstant(ValueExpressions.FloatExpression fExpr, Void value) throws RuntimeException {
return new ConstantExpression<>(fExpr.getFloat());
Expand Down Expand Up @@ -174,17 +220,17 @@ public Expression visitVarDecimalConstant(ValueExpressions.VarDecimalExpression

@Override
public Expression visitDateConstant(ValueExpressions.DateExpression dateExpr, Void value) throws RuntimeException {
return new ConstantExpression<>(dateExpr.getDate());
return new ConstantExpression<>(Math.toIntExact(Math.floorDiv(dateExpr.getDate(), 86_400_000L)));
}

@Override
public Expression visitTimeConstant(ValueExpressions.TimeExpression timeExpr, Void value) throws RuntimeException {
return new ConstantExpression<>(timeExpr.getTime());
return new ConstantExpression<>((long) timeExpr.getTime() * 1_000L);
}

@Override
public Expression visitTimeStampConstant(ValueExpressions.TimeStampExpression timestampExpr, Void value) throws RuntimeException {
return new ConstantExpression<>(timestampExpr.getTimeStamp());
return new ConstantExpression<>(Math.multiplyExact(timestampExpr.getTimeStamp(), 1_000L));
}

@Override
Expand Down
12 changes: 12 additions & 0 deletions ...berg/src/main/java/org/apache/drill/exec/store/iceberg/plan/IcebergPluginImplementor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,8 @@
import org.apache.calcite.rex.RexLiteral;
import org.apache.calcite.rex.RexNode;
import org.apache.calcite.util.Util;
import org.apache.drill.common.FunctionNames;
import org.apache.drill.common.expression.FunctionCallFactory;
import org.apache.drill.common.expression.LogicalExpression;
import org.apache.drill.common.expression.SchemaPath;
import org.apache.drill.exec.physical.base.GroupScan;
Expand All @@ -46,6 +48,7 @@

import java.io.IOException;
import java.math.BigDecimal;
import java.util.Arrays;
import java.util.List;
import java.util.stream.Collectors;

Expand All @@ -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) {

@cgivre cgivre Oct 21, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

@letian-jiang letian-jiang Oct 27, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Added an implementor regression test for two stacked filters, such as id >= 2 AND id < 4. It checks values inside and outside the range to verify that both bounds are retained.

expression = FunctionCallFactory.createBooleanOperator(FunctionNames.AND,
Arrays.asList(groupScan.getCondition(), expression));
}
groupScan = groupScan.toBuilder().condition(expression).build();
}

Expand Down Expand Up @@ -98,6 +105,11 @@ public boolean canImplement(Filter filter) {
GroupScan scan = findGroupScan(filter);
if (scan instanceof IcebergGroupScan) {
IcebergGroupScan groupScan = (IcebergGroupScan) scan;
// This rule replaces Drill's Filter. Without Iceberg residuals, file
// pruning alone cannot guarantee that every returned row matches.
if (Boolean.TRUE.equals(groupScan.getFormatPlugin().getConfig().getIgnoreResiduals())) {
return false;
}
// ensures that expression compatible with table schema
expression = Binder.bind(groupScan.getTableScan().schema().asStruct(), expression, true);
} else {
Expand Down
Loading