pradeep-L
diff --git a/‎sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java‎
Lines changed: 8 additions & 0 deletions b/‎sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java‎
Lines changed: 8 additions & 0 deletions
diff --git a/‎sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java‎
Lines changed: 36 additions & 10 deletions b/‎sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionFragmentConvertor.java‎
Lines changed: 36 additions & 10 deletions
diff --git a/‎sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/UntypedNullPreprocessor.java‎
Lines changed: 120 additions & 0 deletions b/‎sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/UntypedNullPreprocessor.java‎
Lines changed: 120 additions & 0 deletions
@@ -101,6 +101,14 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP
         ScalarFunction.CAST,
         ScalarFunction.CONCAT,
         ScalarFunction.SAFE_CAST,
+        // CASE — Calcite emits CASE WHEN ... THEN ... END for conditional expressions, including
+        // PPL `count(eval(predicate))` (lowered to COUNT(CASE WHEN predicate THEN ... ELSE NULL END))
+        // and explicit `eval x = case(cond, val, ...)`. Isthmus translates SqlKind.CASE structurally
+        // to a Substrait IfThen rel — no extension lookup needed, no adapter required. DataFusion's
+        // substrait consumer handles IfThen natively. Without this entry, the analytics planner
+        // rejects the operator with "No backend supports scalar function [CASE] among [datafusion]"
+        // before substrait emission.
+        ScalarFunction.CASE,
         // ABS / SUBSTRING — `eval x = abs(...)` and `eval s = substring(...)` projections that PPL
         // sort-pushdown moves into the project tree (see CalciteSortCommandIT
         // testPushdownSortExpressionContainsNull and CalcitePPLSortIT
 
@@ -70,7 +70,7 @@
  *   <li>{@link #attachPartialAggOnTop(RelNode, byte[])} and
  *       {@link #attachFragmentOnTop(RelNode, byte[])} — convert the wrapping
  *       operator standalone, then rewire its input to the decoded inner plan's
- *       root via {@link #rewire(Plan, Rel)}.</li>
+ *       root via {@link #rewire(Plan, Rel, List)}.</li>
  * </ul>
  *
  * @opensearch.internal
@@ -126,7 +126,11 @@ public byte[] attachPartialAggOnTop(RelNode partialAggFragment, byte[] innerByte
         LOGGER.debug("Attaching partial aggregate on top of {} inner bytes", innerBytes.length);
         Plan inner = decodePlan(innerBytes);
         Rel wrapper = convertStandalone(partialAggFragment);
-        Plan rewired = rewire(inner, withAggregationPhase(wrapper, Expression.AggregationPhase.INITIAL_TO_INTERMEDIATE));
+        Plan rewired = rewire(
+            inner,
+            withAggregationPhase(wrapper, Expression.AggregationPhase.INITIAL_TO_INTERMEDIATE),
+            fieldNames(partialAggFragment)
+        );
         return serializePlan(rewired);
     }
 
@@ -150,14 +154,19 @@ public byte[] attachFragmentOnTop(RelNode fragment, byte[] innerBytes) {
         // the visitor still walks them top-down to build the wrapper rel.
         RelNode rewritten = rewriteStageInputScans(fragment);
         Rel wrapper = convertStandalone(rewritten);
-        return serializePlan(rewire(inner, wrapper));
+        return serializePlan(rewire(inner, wrapper, fieldNames(fragment)));
     }
 
     // ── Core conversion helpers ─────────────────────────────────────────────────
 
     private byte[] convertToSubstrait(RelNode fragment) {
-        RelRoot root = RelRoot.of(fragment, SqlKind.SELECT);
-        SubstraitRelVisitor visitor = createVisitor(fragment);
+        // Rewrite SqlTypeName.NULL literals (Calcite's untyped null, emitted for the
+        // implicit ELSE arm of CASE) to typed nulls — isthmus' TypeConverter rejects NULL
+        // with "Unable to convert the type NULL". The widening only changes literal type
+        // tags; semantics and field names (used by Plan.Root.names) are unchanged.
+        RelNode preprocessed = UntypedNullPreprocessor.rewrite(fragment);
+        RelRoot root = RelRoot.of(preprocessed, SqlKind.SELECT);
+        SubstraitRelVisitor visitor = createVisitor(preprocessed);
         Rel substraitRel = visitor.apply(root.rel);
 
         List<String> fieldNames = root.fields.stream().map(field -> field.getValue()).toList();
@@ -178,11 +187,15 @@ private byte[] convertToSubstrait(RelNode fragment) {
      * children (e.g. the {@code attachPartialAggOnTop} caller passes a
      * {@code LogicalAggregate} whose input is the already-stripped inner tree); we
      * deliberately discard those children by taking only the outermost rel of the
-     * conversion and rewiring its input during {@link #rewire(Plan, Rel)}.
+     * conversion and rewiring its input during {@link #rewire(Plan, Rel, List)}.
      */
     private Rel convertStandalone(RelNode operator) {
-        SubstraitRelVisitor visitor = createVisitor(operator);
-        return visitor.apply(operator);
+        // Same untyped-NULL preprocessing rationale as convertToSubstrait — the standalone
+        // wrapper conversion is just as susceptible to a SqlTypeName.NULL literal lurking in
+        // a CASE call attached on top of an inner plan.
+        RelNode preprocessed = UntypedNullPreprocessor.rewrite(operator);
+        SubstraitRelVisitor visitor = createVisitor(preprocessed);
+        return visitor.apply(preprocessed);
     }
 
     /**
@@ -191,15 +204,28 @@ private Rel convertStandalone(RelNode operator) {
      * {@code wrapper(inner.root)}. Supports the known single-input wrappers emitted
      * by our four SPI methods ({@link Aggregate}, {@link Sort}, {@link Filter},
      * {@link Project}).
+     *
+     * <p>{@code wrapperNames} must be the wrapper's output column names — typically
+     * derived from the wrapper {@link RelNode}'s row type. For schema-preserving
+     * wrappers (Sort, Filter, Fetch) these match the inner plan's names; for
+     * schema-reshaping wrappers (Aggregate, Project) they don't, and using the
+     * inner's names there causes DataFusion's substrait consumer to reject the
+     * Plan with a "Names list must match exactly to nested schema" error in
+     * {@code make_renamed_schema}.
      */
-    static Plan rewire(Plan inner, Rel wrapper) {
+    static Plan rewire(Plan inner, Rel wrapper, List<String> wrapperNames) {
         if (inner.getRoots().isEmpty()) {
             throw new IllegalArgumentException("Inner Substrait plan has no root relation to rewire under wrapper");
         }
         Plan.Root innerRoot = inner.getRoots().get(0);
         Rel innerRel = innerRoot.getInput();
         Rel rewired = replaceInput(wrapper, innerRel);
-        return Plan.builder().addRoots(Plan.Root.builder().input(rewired).names(innerRoot.getNames()).build()).build();
+        return Plan.builder().addRoots(Plan.Root.builder().input(rewired).names(wrapperNames).build()).build();
+    }
+
+    /** Extracts a wrapper's output column names from its Calcite row type. */
+    private static List<String> fieldNames(RelNode fragment) {
+        return fragment.getRowType().getFieldList().stream().map(RelDataTypeField::getName).toList();
     }
 
     private static Rel replaceInput(Rel wrapper, Rel newInput) {
 
@@ -0,0 +1,120 @@
+/*
+ * SPDX-License-Identifier: Apache-2.0
+ *
+ * The OpenSearch Contributors require contributions made to
+ * this file be licensed under the Apache-2.0 license or a
+ * compatible open source license.
+ */
+
+package org.opensearch.be.datafusion;
+
+import org.apache.calcite.rel.RelHomogeneousShuttle;
+import org.apache.calcite.rel.RelNode;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexLiteral;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
+import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.type.SqlTypeName;
+
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * Pre-isthmus pass that rewrites untyped {@code NULL} literals
+ * ({@code RexLiteral} with {@link SqlTypeName#NULL}) to typed null literals
+ * inferred from their enclosing operator.
+ *
+ * <p>Calcite emits an untyped NULL for the implicit {@code ELSE} arm of
+ * {@code CASE WHEN cond THEN val END}, which is exactly the shape PPL
+ * {@code count(eval(predicate))} lowers to:
+ *
+ * <pre>{@code
+ *   COUNT(CASE WHEN predicate THEN <projected> END)   // ELSE is implicit NULL, type=NULL
+ * }</pre>
+ *
+ * <p>Isthmus' {@code TypeConverter.toSubstrait} rejects {@link SqlTypeName#NULL}
+ * with {@code "Unable to convert the type NULL"}. The CASE call's resolved
+ * return type already carries the right widened type ({@code NULLABLE BIGINT}
+ * for the count-eval shape, etc), so we substitute that.
+ *
+ * <p>Scope: only CASE call operands are rewritten today. Other untyped-NULL
+ * sites (function arguments, comparison RHS, etc) are rare in PPL-generated
+ * plans and would need per-operator type-inference to do safely; defer until
+ * a concrete test surfaces one.
+ *
+ * @opensearch.internal
+ */
+final class UntypedNullPreprocessor {
+
+    private UntypedNullPreprocessor() {}
+
+    /**
+     * Walk the RelNode tree, applying the rewrite to every node's expressions.
+     * Returns a new tree if any rewrite occurred, otherwise the input unchanged.
+     */
+    static RelNode rewrite(RelNode root) {
+        return root.accept(new RelHomogeneousShuttle() {
+            @Override
+            public RelNode visit(RelNode other) {
+                RelNode visited = super.visit(other);
+                return visited.accept(new CaseUntypedNullShuttle(visited.getCluster().getRexBuilder()));
+            }
+        });
+    }
+
+    /**
+     * Per-node rex shuttle: for every {@code CASE} call encountered, rewrite any
+     * {@link SqlTypeName#NULL}-typed literal operand into a typed null literal
+     * matching the CASE's resolved return type.
+     */
+    private static final class CaseUntypedNullShuttle extends RexShuttle {
+
+        private final RexBuilder rexBuilder;
+
+        CaseUntypedNullShuttle(RexBuilder rexBuilder) {
+            this.rexBuilder = rexBuilder;
+        }
+
+        @Override
+        public RexNode visitCall(RexCall call) {
+            // Recurse first so nested CASE calls are rewritten bottom-up — each inner CASE is
+            // resolved against its own return type, so by the time we look at the outer one,
+            // every operand is already typed.
+            RexCall recursed = (RexCall) super.visitCall(call);
+            if (recursed.getKind() != SqlKind.CASE) {
+                return recursed;
+            }
+            List<RexNode> operands = recursed.getOperands();
+            List<RexNode> rewritten = new ArrayList<>(operands.size());
+            boolean changed = false;
+            for (int i = 0; i < operands.size(); i++) {
+                RexNode op = operands.get(i);
+                if (isCaseValueOperand(i, operands.size()) && isUntypedNullLiteral(op)) {
+                    rewritten.add(rexBuilder.makeNullLiteral(recursed.getType()));
+                    changed = true;
+                } else {
+                    rewritten.add(op);
+                }
+            }
+            return changed ? recursed.clone(recursed.getType(), rewritten) : recursed;
+        }
+
+        /**
+         * Calcite's CASE operand layout is {@code [cond1, val1, cond2, val2, …, condN, valN, else]}.
+         * Conditions sit at even indices except the last operand (the ELSE), which is always
+         * a value. Returns true for value operands (the THEN/ELSE arms).
+         */
+        private static boolean isCaseValueOperand(int index, int total) {
+            return (index % 2 == 1) || (index == total - 1);
+        }
+
+        private static boolean isUntypedNullLiteral(RexNode node) {
+            if (!(node instanceof RexLiteral lit)) {
+                return false;
+            }
+            return lit.isNull() && lit.getType().getSqlTypeName() == SqlTypeName.NULL;
+        }
+    }
+}