Enable PPL fillnull on the analytics-engine route via DataFusion COALESCE

sandbox/plugins/analytics-backend-datafusion/src/main/java/org/opensearch/be/datafusion/DataFusionAnalyticsBackendPlugin.java

@@ -17,6 +17,7 @@
 import org.opensearch.analytics.spi.FieldType;
 import org.opensearch.analytics.spi.FilterCapability;
 import org.opensearch.analytics.spi.FragmentConvertor;
+import org.opensearch.analytics.spi.ProjectCapability;
 import org.opensearch.analytics.spi.ScalarFunction;
 import org.opensearch.analytics.spi.ScanCapability;
 import org.opensearch.analytics.spi.SearchExecEngineProvider;
@@ -61,6 +62,13 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP
         ScalarFunction.LIKE
     );
 
+    // Project-side scalar functions DataFusion can evaluate natively. Each entry corresponds to a
+    // PPL command/function we want the analytics-engine planner to route through DataFusion. Add
+    // here only after verifying the function deserializes through Substrait isthmus into a plan
+    // DataFusion's native runtime can execute (see DataFusionFragmentConvertor for the conversion
+    // path). COALESCE is the lowering target of PPL `fillnull`.
+    private static final Set<ScalarFunction> STANDARD_PROJECT_OPS = Set.of(ScalarFunction.COALESCE, ScalarFunction.CEIL);
+
     private static final Set<AggregateFunction> AGG_FUNCTIONS = Set.of(
         AggregateFunction.SUM,
         AggregateFunction.SUM0,
@@ -107,6 +115,16 @@ public Set<FilterCapability> filterCapabilities() {
                 return Set.copyOf(caps);
             }
 
+            @Override
+            public Set<ProjectCapability> projectCapabilities() {
+                Set<String> formats = Set.copyOf(plugin.getSupportedFormats());
+                Set<ProjectCapability> caps = new HashSet<>();
+                for (ScalarFunction op : STANDARD_PROJECT_OPS) {
+                    caps.add(new ProjectCapability.Scalar(op, Set.copyOf(SUPPORTED_FIELD_TYPES), formats, true));
+                }
+                return Set.copyOf(caps);
+            }
+
             @Override
             public Set<AggregateCapability> aggregateCapabilities() {
                 Set<String> formats = Set.copyOf(plugin.getSupportedFormats());

sandbox/plugins/analytics-engine/src/main/java/org/opensearch/analytics/planner/rel/OpenSearchProject.java

@@ -18,8 +18,10 @@
 import org.apache.calcite.rel.logical.LogicalProject;
 import org.apache.calcite.rel.metadata.RelMetadataQuery;
 import org.apache.calcite.rel.type.RelDataType;
+import org.apache.calcite.rex.RexCall;
 import org.apache.calcite.rex.RexInputRef;
 import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.rex.RexShuttle;
 import org.opensearch.analytics.planner.RelNodeUtils;
 import org.opensearch.analytics.spi.FieldStorageInfo;
 
@@ -122,10 +124,28 @@ public RelNode stripAnnotations(List<RelNode> strippedChildren) {
 
     @Override
     public RelNode stripAnnotations(List<RelNode> strippedChildren, Function<OperatorAnnotation, RexNode> annotationResolver) {
+        // OpenSearchProjectRule.annotateExpr recurses into operands when validating viable
+        // backends, so a top-level call like COALESCE(num0, CEIL(num1)) ends up with the inner
+        // CEIL also wrapped. The supplied annotationResolver controls how each top-level
+        // wrapper is unwrapped (defaults to OperatorAnnotation::unwrap, returning the original
+        // RexNode); a RexShuttle then sweeps the resolver's result to strip any remaining
+        // nested wrappers. Substrait conversion only recognizes the underlying RexCall shape,
+        // so every wrapper at every depth must be removed before the plan is handed to a
+        // backend's FragmentConvertor.
+        RexShuttle nestedAnnotationStripper = new RexShuttle() {
+            @Override
+            public RexNode visitCall(RexCall call) {
+                if (call instanceof AnnotatedProjectExpression nested) {
+                    return nested.getOriginal().accept(this);
+                }
+                return super.visitCall(call);
+            }
+        };
         List<RexNode> strippedExprs = new ArrayList<>();
         for (RexNode expr : getProjects()) {
             if (expr instanceof AnnotatedProjectExpression annotated) {
-                strippedExprs.add(annotationResolver.apply(annotated));
+                RexNode resolved = annotationResolver.apply(annotated);
+                strippedExprs.add(resolved.accept(nestedAnnotationStripper));
             } else {
                 // Plain expressions have no annotation to strip — pass through.
                 strippedExprs.add(expr);

sandbox/plugins/analytics-engine/src/test/java/org/opensearch/analytics/planner/ProjectRuleTests.java

@@ -236,6 +236,50 @@ public void testNestedScalarFunctions() {
         assertAnnotation(result.getProjects().get(0), MockDataFusionBackend.NAME);
     }
 
+    public void testStripAnnotationsRecursivelyUnwrapsNestedExpressions() {
+        // PLUS(CEIL(value), value) — a scalar call with another scalar call as an operand.
+        // The project rule recurses into operands (annotateExpr lines 127-139), so both PLUS
+        // and the inner CEIL get wrapped in AnnotatedProjectExpression. stripAnnotations must
+        // remove every wrapper at every depth before the plan reaches the backend
+        // FragmentConvertor — Substrait isthmus has no converter for ANNOTATED_PROJECT_EXPR and
+        // would throw "Unable to convert call". (COALESCE would be the natural shape here since
+        // PPL fillnull lowers to it, but Calcite's makeCall simplifies COALESCE on non-nullable
+        // operands away into the first arg, defeating the test. PLUS+CEIL preserves the
+        // nested-call structure we want to exercise.)
+        RexNode value = rexBuilder.makeInputRef(typeFactory.createSqlType(SqlTypeName.INTEGER), 1);
+        RexNode ceilCall = rexBuilder.makeCall(SqlStdOperatorTable.CEIL, value);
+        RexNode plusCall = rexBuilder.makeCall(SqlStdOperatorTable.PLUS, ceilCall, value);
+        OpenSearchProject annotated = runProject(plusCall);
+
+        // Sanity: confirm the rule produced the nested-wrapper shape this test exercises.
+        RexNode topLevel = annotated.getProjects().get(0);
+        assertTrue("Outer PLUS must be annotated", topLevel instanceof AnnotatedProjectExpression);
+        RexCall outerOriginal = (RexCall) ((AnnotatedProjectExpression) topLevel).getOriginal();
+        assertTrue(
+            "Inner CEIL must also be annotated (recursive annotateExpr behavior)",
+            outerOriginal.getOperands().get(0) instanceof AnnotatedProjectExpression
+        );
+
+        // Strip and assert no AnnotatedProjectExpression survives anywhere in the RexNode tree.
+        RelNode stripped = annotated.stripAnnotations(annotated.getInputs());
+        assertTrue("Stripped plan should be a plain LogicalProject", stripped instanceof LogicalProject);
+        for (RexNode expr : ((LogicalProject) stripped).getProjects()) {
+            assertNoAnnotationInTree(expr);
+        }
+    }
+
+    private static void assertNoAnnotationInTree(RexNode node) {
+        assertFalse(
+            "Expression tree must not contain AnnotatedProjectExpression after strip: " + node,
+            node instanceof AnnotatedProjectExpression
+        );
+        if (node instanceof RexCall call) {
+            for (RexNode operand : call.getOperands()) {
+                assertNoAnnotationInTree(operand);
+            }
+        }
+    }
+
     // ---- Mixed backends in one projection ----
 
     public void testMixedBackendsInProjection() {