feat(core,isthmus): add DynamicParameter expression support

core/src/main/java/io/substrait/expression/AbstractExpressionVisitor.java

@@ -603,4 +603,17 @@ public O visit(Expression.ScalarSubquery expr, C context) throws E {
   public O visit(Expression.InPredicate expr, C context) throws E {
     return visitFallback(expr, context);
   }
+
+  /**
+   * Visits a dynamic parameter expression.
+   *
+   * @param expr the dynamic parameter
+   * @param context the visitation context
+   * @return the visit result
+   * @throws E if visitation fails
+   */
+  @Override
+  public O visit(Expression.DynamicParameter expr, C context) throws E {
+    return visitFallback(expr, context);
+  }
 }

core/src/main/java/io/substrait/expression/Expression.java

@@ -2075,6 +2075,44 @@ public <R, C extends VisitationContext, E extends Throwable> R accept(
     }
   }
 
+  /** Represents a runtime parameter placeholder with a declared type and parameter reference. */
+  @Value.Immutable
+  abstract class DynamicParameter implements Expression {
+    /**
+     * Returns the declared type of the parameter value.
+     *
+     * @return the parameter type
+     */
+    public abstract Type type();
+
+    /**
+     * Returns the zero-based parameter reference.
+     *
+     * @return the parameter reference
+     */
+    public abstract int parameterReference();
+
+    @Override
+    public Type getType() {
+      return type();
+    }
+
+    /**
+     * Creates a new builder for constructing a DynamicParameter.
+     *
+     * @return a new builder instance
+     */
+    public static ImmutableExpression.DynamicParameter.Builder builder() {
+      return ImmutableExpression.DynamicParameter.builder();
+    }
+
+    @Override
+    public <R, C extends VisitationContext, E extends Throwable> R accept(
+        ExpressionVisitor<R, C, E> visitor, C context) throws E {
+      return visitor.visit(this, context);
+    }
+  }
+
   /** Defines the operation type for set predicates (EXISTS, UNIQUE). */
   enum PredicateOp {
     /** Unspecified predicate operation. */

core/src/main/java/io/substrait/expression/ExpressionVisitor.java

@@ -470,4 +470,14 @@ public interface ExpressionVisitor<R, C extends VisitationContext, E extends Thr
    * @throws E on visit failure
    */
   R visit(Expression.InPredicate expr, C context) throws E;
+
+  /**
+   * Visit a dynamic parameter expression.
+   *
+   * @param expr the dynamic parameter expression
+   * @param context visitation context
+   * @return visit result
+   * @throws E on visit failure
+   */
+  R visit(Expression.DynamicParameter expr, C context) throws E;
 }

core/src/main/java/io/substrait/expression/proto/ExpressionProtoConverter.java

@@ -733,6 +733,18 @@ public Expression visit(
         .build();
   }
 
+  @Override
+  public Expression visit(
+      io.substrait.expression.Expression.DynamicParameter expr, EmptyVisitationContext context)
+      throws RuntimeException {
+    return Expression.newBuilder()
+        .setDynamicParameter(
+            io.substrait.proto.DynamicParameter.newBuilder()
+                .setType(toProto(expr.type()))
+                .setParameterReference(expr.parameterReference()))
+        .build();
+  }
+
   public static class BoundConverter
       implements WindowBound.WindowBoundVisitor<Expression.WindowFunction.Bound, RuntimeException> {
     private static final BoundConverter TO_BOUND_VISITOR = new BoundConverter();

core/src/main/java/io/substrait/expression/proto/ProtoExpressionConverter.java

@@ -289,6 +289,14 @@ public Type visit(Type.Struct type) throws RuntimeException {
 
           return lambdaBuilder.lambdaFromStruct(parameters, () -> from(protoLambda.getBody()));
         }
+      case DYNAMIC_PARAMETER:
+        {
+          io.substrait.proto.DynamicParameter dp = expr.getDynamicParameter();
+          return Expression.DynamicParameter.builder()
+              .type(protoTypeConverter.from(dp.getType()))
+              .parameterReference(dp.getParameterReference())
+              .build();
+        }
       // TODO enum.
       case ENUM:
         throw new UnsupportedOperationException("Unsupported type: " + expr.getRexTypeCase());

core/src/main/java/io/substrait/relation/ExpressionCopyOnWriteVisitor.java

@@ -455,6 +455,12 @@ public Optional<Expression> visit(Expression.Lambda lambda, EmptyVisitationConte
             .build());
   }
 
+  @Override
+  public Optional<Expression> visit(
+      Expression.DynamicParameter expr, EmptyVisitationContext context) throws E {
+    return Optional.empty();
+  }
+
   // utilities
 
   protected Optional<List<Expression>> visitExprList(

core/src/test/java/io/substrait/type/proto/DynamicParameterRoundtripTest.java

@@ -0,0 +1,114 @@
+package io.substrait.type.proto;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import io.substrait.TestBase;
+import io.substrait.expression.Expression;
+import io.substrait.type.Type;
+import io.substrait.type.TypeCreator;
+import org.junit.jupiter.api.Test;
+
+class DynamicParameterRoundtripTest extends TestBase {
+
+  @Test
+  void dynamicParameterI64() {
+    Expression.DynamicParameter dp =
+        Expression.DynamicParameter.builder()
+            .type(TypeCreator.REQUIRED.I64)
+            .parameterReference(0)
+            .build();
+
+    assertDynamicParameter(dp, TypeCreator.REQUIRED.I64, 0);
+    verifyRoundTrip(dp);
+  }
+
+  @Test
+  void dynamicParameterNullableString() {
+    Expression.DynamicParameter dp =
+        Expression.DynamicParameter.builder()
+            .type(TypeCreator.NULLABLE.STRING)
+            .parameterReference(1)
+            .build();
+
+    assertDynamicParameter(dp, TypeCreator.NULLABLE.STRING, 1);
+    verifyRoundTrip(dp);
+  }
+
+  @Test
+  void dynamicParameterFP64() {
+    Expression.DynamicParameter dp =
+        Expression.DynamicParameter.builder()
+            .type(TypeCreator.REQUIRED.FP64)
+            .parameterReference(2)
+            .build();
+
+    assertDynamicParameter(dp, TypeCreator.REQUIRED.FP64, 2);
+    verifyRoundTrip(dp);
+  }
+
+  @Test
+  void dynamicParameterI32Nullable() {
+    Expression.DynamicParameter dp =
+        Expression.DynamicParameter.builder()
+            .type(TypeCreator.NULLABLE.I32)
+            .parameterReference(42)
+            .build();
+
+    assertDynamicParameter(dp, TypeCreator.NULLABLE.I32, 42);
+    verifyRoundTrip(dp);
+  }
+
+  @Test
+  void dynamicParameterDate() {
+    Expression.DynamicParameter dp =
+        Expression.DynamicParameter.builder()
+            .type(TypeCreator.REQUIRED.DATE)
+            .parameterReference(3)
+            .build();
+
+    assertDynamicParameter(dp, TypeCreator.REQUIRED.DATE, 3);
+    verifyRoundTrip(dp);
+  }
+
+  @Test
+  void dynamicParameterBoolean() {
+    Expression.DynamicParameter dp =
+        Expression.DynamicParameter.builder()
+            .type(TypeCreator.REQUIRED.BOOLEAN)
+            .parameterReference(0)
+            .build();
+
+    assertDynamicParameter(dp, TypeCreator.REQUIRED.BOOLEAN, 0);
+    verifyRoundTrip(dp);
+  }
+
+  @Test
+  void dynamicParameterDecimal() {
+    Expression.DynamicParameter dp =
+        Expression.DynamicParameter.builder()
+            .type(TypeCreator.REQUIRED.decimal(10, 2))
+            .parameterReference(5)
+            .build();
+
+    assertDynamicParameter(dp, TypeCreator.REQUIRED.decimal(10, 2), 5);
+    verifyRoundTrip(dp);
+  }
+
+  @Test
+  void dynamicParameterTimestamp() {
+    Expression.DynamicParameter dp =
+        Expression.DynamicParameter.builder()
+            .type(TypeCreator.NULLABLE.TIMESTAMP)
+            .parameterReference(7)
+            .build();
+
+    assertDynamicParameter(dp, TypeCreator.NULLABLE.TIMESTAMP, 7);
+    verifyRoundTrip(dp);
+  }
+
+  private void assertDynamicParameter(
+      Expression.DynamicParameter dp, Type expectedType, int expectedReference) {
+    assertEquals(expectedType, dp.getType());
+    assertEquals(expectedReference, dp.parameterReference());
+  }
+}

examples/substrait-spark/src/main/java/io/substrait/examples/util/ExpressionStringify.java

@@ -363,4 +363,10 @@ public String visit(EmptyMapLiteral expr, EmptyVisitationContext context)
       throws RuntimeException {
     return "<EmptyMapLiteral>";
   }
+
+  @Override
+  public String visit(Expression.DynamicParameter expr, EmptyVisitationContext context)
+      throws RuntimeException {
+    return "<DynamicParameter " + expr.parameterReference() + " " + expr.type() + ">";
+  }
 }

isthmus/src/main/java/io/substrait/isthmus/expression/ExpressionRexConverter.java

@@ -867,6 +867,12 @@ public RexNode visit(SetPredicate expr, Context context) throws RuntimeException
     }
   }
 
+  @Override
+  public RexNode visit(Expression.DynamicParameter expr, Context context) throws RuntimeException {
+    RelDataType calciteType = typeConverter.toCalcite(typeFactory, expr.type());
+    return rexBuilder.makeDynamicParam(calciteType, expr.parameterReference());
+  }
+
   /**
    * Helper method to create a Calcite ROW expression for encoding UDT struct literals.
    *

isthmus/src/main/java/io/substrait/isthmus/expression/RexExpressionConverter.java

@@ -193,7 +193,10 @@ public Expression visitCorrelVariable(RexCorrelVariable correlVariable) {
    */
   @Override
   public Expression visitDynamicParam(RexDynamicParam dynamicParam) {
-    throw new UnsupportedOperationException("RexDynamicParam not supported");
+    return Expression.DynamicParameter.builder()
+        .type(typeConverter.toSubstrait(dynamicParam.getType()))
+        .parameterReference(dynamicParam.getIndex())
+        .build();
   }
 
   /**