DRILL-8555: Physical plan cache for parameterized SQL queries

contrib/format-iceberg/src/main/java/org/apache/drill/exec/store/iceberg/format/IcebergFormatPlugin.java

@@ -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;
@@ -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;
@@ -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());
+  }
+
   @Override
   public boolean supportsWrite() {
     return false;

contrib/format-iceberg/src/main/java/org/apache/drill/exec/store/iceberg/plan/DrillExprToIcebergTranslator.java

@@ -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;
@@ -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()) {
@@ -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);
         if (nameRef instanceof SchemaPath && expression instanceof ConstantExpression) {
           String name = IcebergGroupScan.getPath((SchemaPath) nameRef);
           return Expressions.greaterThanOrEqual(name, ((ConstantExpression<?>) expression).getValue());
@@ -174,17 +216,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

contrib/format-iceberg/src/main/java/org/apache/drill/exec/store/iceberg/plan/IcebergPluginImplementor.java

@@ -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;
@@ -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;
 
@@ -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) {
+      expression = FunctionCallFactory.createBooleanOperator(FunctionNames.AND,
+          Arrays.asList(groupScan.getCondition(), expression));
+    }
     groupScan = groupScan.toBuilder().condition(expression).build();
   }