Enable PPL eval string concat on the analytics-engine route via DataFusion CONCAT/CAST

sandbox/libs/analytics-framework/src/main/java/org/opensearch/analytics/spi/ScalarFunction.java

@@ -10,8 +10,12 @@
 
 import org.apache.calcite.sql.SqlFunction;
 import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.SqlOperator;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 
+import java.util.HashMap;
 import java.util.Locale;
+import java.util.Map;
 
 /**
  * All scalar functions a backend may support — comparisons, full-text search,
@@ -52,7 +56,15 @@ public enum ScalarFunction {
     LOWER(Category.STRING, SqlKind.OTHER_FUNCTION),
     TRIM(Category.STRING, SqlKind.TRIM),
     SUBSTRING(Category.STRING, SqlKind.OTHER_FUNCTION),
-    CONCAT(Category.STRING, SqlKind.OTHER_FUNCTION),
+    /**
+     * String concatenation. Calcite's {@code SqlStdOperatorTable.CONCAT} is a
+     * {@link org.apache.calcite.sql.SqlBinaryOperator} named {@code "||"} (not {@code "CONCAT"})
+     * with {@link SqlKind#OTHER}, so neither {@link #fromSqlKind(SqlKind)} nor identifier-name
+     * {@link #valueOf(String)} resolves it. The {@code referenceOperator} hook below pins the
+     * concrete Calcite operator constant so resolution is a singleton-identity match — a Calcite
+     * rename surfaces as a compile error rather than as a silent string mismatch at runtime.
+     */
+    CONCAT(Category.STRING, SqlKind.OTHER_FUNCTION, SqlStdOperatorTable.CONCAT),
     CHAR_LENGTH(Category.STRING, SqlKind.OTHER_FUNCTION),
 
     // ── Math ─────────────────────────────────────────────────────────
@@ -68,6 +80,14 @@ public enum ScalarFunction {
 
     // ── Cast / type ──────────────────────────────────────────────────
     CAST(Category.SCALAR, SqlKind.CAST),
+    /**
+     * Calcite's {@code SAFE_CAST} — emitted by PPL's explicit {@code CAST(... AS ...)} when the
+     * source value may be NULL or the conversion may fail; returns NULL on failure rather than
+     * throwing. Resolves through {@link SqlKind#SAFE_CAST}, distinct from {@link #CAST} which
+     * uses {@link SqlKind#CAST}. DataFusion's native cast already returns NULL on conversion
+     * failure, so SAFE_CAST and CAST share the same backend semantics.
+     */
+    SAFE_CAST(Category.SCALAR, SqlKind.SAFE_CAST),
 
     // ── Conditional ──────────────────────────────────────────────────
     CASE(Category.SCALAR, SqlKind.CASE),
@@ -98,10 +118,24 @@ public enum Category {
 
     private final Category category;
     private final SqlKind sqlKind;
+    /**
+     * Optional Calcite operator that this constant maps to when the operator cannot be resolved
+     * via {@link SqlKind} or via identifier-name {@link #valueOf(String)} — typically operators
+     * whose {@code getName()} returns a non-identifier token (e.g. {@code SqlStdOperatorTable.CONCAT}
+     * is named {@code "||"}). Null for the common case where SqlKind or name resolution suffices.
+     * Stored as a reference (not a string) so a Calcite-side rename of the operator surfaces as a
+     * compile error here.
+     */
+    private final SqlOperator referenceOperator;
 
     ScalarFunction(Category category, SqlKind sqlKind) {
+        this(category, sqlKind, null);
+    }
+
+    ScalarFunction(Category category, SqlKind sqlKind, SqlOperator referenceOperator) {
         this.category = category;
         this.sqlKind = sqlKind;
+        this.referenceOperator = referenceOperator;
     }
 
     public Category getCategory() {
@@ -134,4 +168,52 @@ public static ScalarFunction fromSqlFunction(SqlFunction function) {
         // valueOf(toUpperCase). This couples enum constant naming to SQL function naming convention.
         return ScalarFunction.valueOf(function.getName().toUpperCase(Locale.ROOT));
     }
+
+    /**
+     * Reverse index from {@link #referenceOperator} to enum constant. Built from the enum itself
+     * at class init — adding a new symbolic operator is a single-site change on the enum constant,
+     * no separate map to maintain. Lookup is identity-keyed because Calcite's standard operators
+     * are singletons (e.g. {@code SqlStdOperatorTable.CONCAT}). Empty in the common case (most
+     * constants resolve by SqlKind or identifier-name valueOf).
+     */
+    private static final Map<SqlOperator, ScalarFunction> BY_REFERENCE_OPERATOR;
+
+    static {
+        Map<SqlOperator, ScalarFunction> byOperator = new HashMap<>();
+        for (ScalarFunction func : values()) {
+            if (func.referenceOperator != null) {
+                byOperator.put(func.referenceOperator, func);
+            }
+        }
+        // The HashMap is private static final and never exposed beyond the get() in the resolver
+        // below — wrapping it in Map.copyOf adds an allocation without any external safety guarantee.
+        BY_REFERENCE_OPERATOR = byOperator;
+    }
+
+    /**
+     * Maps any Calcite {@link SqlOperator} to a {@link ScalarFunction}, or returns null if
+     * unrecognized. Resolution order: {@link SqlKind} match, then {@link #referenceOperator}
+     * identity match (handles {@code SqlStdOperatorTable.CONCAT} a.k.a. {@code ||}), then
+     * identifier-name {@link #valueOf(String)} match.
+     *
+     * <p>Prefer this entry point over {@link #fromSqlKind(SqlKind)} /
+     * {@link #fromSqlFunction(SqlFunction)} when resolving an arbitrary {@code RexCall}'s
+     * operator: a {@code RexCall} may be backed by a {@code SqlBinaryOperator} (e.g. {@code ||})
+     * which is neither covered by {@code OTHER} {@code SqlKind} nor by {@code SqlFunction}.
+     */
+    public static ScalarFunction fromSqlOperatorWithFallback(SqlOperator operator) {
+        ScalarFunction byKind = fromSqlKind(operator.getKind());
+        if (byKind != null) {
+            return byKind;
+        }
+        ScalarFunction byReference = BY_REFERENCE_OPERATOR.get(operator);
+        if (byReference != null) {
+            return byReference;
+        }
+        try {
+            return ScalarFunction.valueOf(operator.getName().toUpperCase(Locale.ROOT));
+        } catch (IllegalArgumentException ignored) {
+            return null;
+        }
+    }
 }

sandbox/libs/analytics-framework/src/test/java/org/opensearch/analytics/spi/ScalarFunctionTests.java

@@ -9,13 +9,42 @@
 package org.opensearch.analytics.spi;
 
 import org.apache.calcite.sql.SqlKind;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
 import org.opensearch.test.OpenSearchTestCase;
 
 import java.util.EnumMap;
 import java.util.Map;
 
+/**
+ * Unit coverage for {@link ScalarFunction}'s three resolution paths used by the analytics-engine
+ * planner ({@code OpenSearchProjectRule}, {@code OpenSearchFilterRule}, {@code BackendPlanAdapter}).
+ *
+ * <p>Each test pins one of the resolver's branches so a regression that drops a branch surfaces
+ * here rather than in IT-level "No backend supports scalar function [null]" errors.
+ */
 public class ScalarFunctionTests extends OpenSearchTestCase {
 
+    // ── fromSqlKind ─────────────────────────────────────────────────────────────
+
+    public void testFromSqlKindResolvesDedicatedKind() {
+        assertEquals(ScalarFunction.EQUALS, ScalarFunction.fromSqlKind(SqlKind.EQUALS));
+        assertEquals(ScalarFunction.PLUS, ScalarFunction.fromSqlKind(SqlKind.PLUS));
+        assertEquals(ScalarFunction.CAST, ScalarFunction.fromSqlKind(SqlKind.CAST));
+        assertEquals(ScalarFunction.SAFE_CAST, ScalarFunction.fromSqlKind(SqlKind.SAFE_CAST));
+        assertEquals(ScalarFunction.COALESCE, ScalarFunction.fromSqlKind(SqlKind.COALESCE));
+    }
+
+    public void testFromSqlKindReturnsNullForOtherKind() {
+        // SqlKind.OTHER is shared by many SqlBinaryOperators — must NOT resolve via SqlKind.
+        assertNull(ScalarFunction.fromSqlKind(SqlKind.OTHER));
+    }
+
+    public void testFromSqlKindReturnsNullForOtherFunctionKind() {
+        // SqlKind.OTHER_FUNCTION is shared by many name-distinguished SqlFunctions — must NOT
+        // resolve via SqlKind even though several enum entries declare it.
+        assertNull(ScalarFunction.fromSqlKind(SqlKind.OTHER_FUNCTION));
+    }
+
     /** Non-OTHER_FUNCTION SqlKinds must be unique: fromSqlKind picks the first match and would shadow later entries. */
     public void testNoDuplicateSqlKindBindings() {
         Map<SqlKind, ScalarFunction> claimedBy = new EnumMap<>(SqlKind.class);
@@ -34,4 +63,42 @@ public void testNoDuplicateSqlKindBindings() {
     public void testSargPredicateIsBoundToSqlKindSearch() {
         assertSame(ScalarFunction.SARG_PREDICATE, ScalarFunction.fromSqlKind(SqlKind.SEARCH));
     }
+
+    // ── fromSqlOperatorWithFallback: SqlKind branch ────────────────────────────────────────
+
+    public void testFromSqlOperatorResolvesViaSqlKind() {
+        // Calcite's CAST has a dedicated SqlKind.CAST — short-circuit before name lookup.
+        assertEquals(ScalarFunction.CAST, ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.CAST));
+        assertEquals(ScalarFunction.PLUS, ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.PLUS));
+        assertEquals(ScalarFunction.GREATER_THAN, ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.GREATER_THAN));
+        assertEquals(ScalarFunction.COALESCE, ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.COALESCE));
+    }
+
+    // ── fromSqlOperatorWithFallback: reference-operator branch ─────────────────────────────
+
+    public void testFromSqlOperatorResolvesPipeConcatViaReferenceOperator() {
+        // The original "no backend supports scalar function [null]" symptom for PPL string `+`.
+        // SqlStdOperatorTable.CONCAT is a SqlBinaryOperator named "||" with SqlKind.OTHER —
+        // neither fromSqlKind nor fromSqlFunction(SqlFunction) resolves it. CONCAT's
+        // referenceOperator field points at the singleton, so the resolver matches by identity.
+        assertEquals("||", SqlStdOperatorTable.CONCAT.getName());
+        assertEquals(SqlKind.OTHER, SqlStdOperatorTable.CONCAT.getKind());
+        assertEquals(ScalarFunction.CONCAT, ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.CONCAT));
+    }
+
+    // ── fromSqlOperatorWithFallback: identifier-name branch ────────────────────────────────
+
+    public void testFromSqlOperatorResolvesViaIdentifierName() {
+        // SqlStdOperatorTable.UPPER is a SqlFunction named "UPPER" with SqlKind.OTHER_FUNCTION;
+        // resolves through the valueOf(name.toUpperCase()) fallback after SqlKind misses.
+        assertEquals(ScalarFunction.UPPER, ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.UPPER));
+        assertEquals(ScalarFunction.LOWER, ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.LOWER));
+        assertEquals(ScalarFunction.ABS, ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.ABS));
+    }
+
+    public void testFromSqlOperatorReturnsNullForUnknownFunction() {
+        // UNARY_MINUS has SqlKind.MINUS_PREFIX (no enum) and name "-" (not a valid valueOf input);
+        // both resolution paths miss and the resolver returns null instead of throwing.
+        assertNull(ScalarFunction.fromSqlOperatorWithFallback(SqlStdOperatorTable.UNARY_MINUS));
+    }
 }

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

@@ -0,0 +1,69 @@
+/*
+ * 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.plan.RelOptCluster;
+import org.apache.calcite.rex.RexBuilder;
+import org.apache.calcite.rex.RexCall;
+import org.apache.calcite.rex.RexNode;
+import org.apache.calcite.sql.fun.SqlStdOperatorTable;
+import org.opensearch.analytics.spi.FieldStorageInfo;
+import org.opensearch.analytics.spi.ScalarFunctionAdapter;
+
+import java.util.List;
+
+/**
+ * Adapts {@code ||(a, b, ...)} (Calcite {@code SqlStdOperatorTable.CONCAT}) into a
+ * null-propagating form for the DataFusion backend.
+ *
+ * <p>Calcite's {@code ||} operator follows the SQL standard: if any operand is NULL, the result
+ * is NULL. Substrait's default {@code concat} extension is documented with the same semantics,
+ * but DataFusion's substrait reader maps it to the DataFusion {@code concat()} function — which
+ * deviates from the standard and treats NULL operands as empty strings. To preserve Calcite's
+ * semantics on the analytics-engine path, this adapter rewrites
+ *
+ * <pre>{@code
+ *   ||(a, b)
+ *     →
+ *   CASE WHEN a IS NULL OR b IS NULL THEN NULL ELSE ||(a, b) END
+ * }</pre>
+ *
+ * The inner {@code ||} is left intact and serializes through the same Substrait conversion path,
+ * but with the surrounding CASE/IS_NULL the DataFusion {@code concat()} call is short-circuited
+ * for any input that contains a NULL — restoring SQL-standard null-propagation without requiring
+ * a custom DataFusion UDF.
+ *
+ * <p>Single-operand calls fall through unchanged (the result equals the operand, so no
+ * null-handling rewrite is needed).
+ */
+class ConcatFunctionAdapter implements ScalarFunctionAdapter {
+
+    @Override
+    public RexNode adapt(RexCall original, List<FieldStorageInfo> fieldStorage, RelOptCluster cluster) {
+        List<RexNode> operands = original.getOperands();
+        if (operands.size() < 2) {
+            return original;
+        }
+        RexBuilder rexBuilder = cluster.getRexBuilder();
+        // Fold operands into a single OR(IS_NULL(o0), IS_NULL(o1), ...) predicate. IS_NULL on a
+        // non-null literal reduces to constant-false, so the OR collapses cleanly through the
+        // optimizer for cases where some operands are statically non-null.
+        RexNode anyNull = rexBuilder.makeCall(SqlStdOperatorTable.IS_NULL, operands.get(0));
+        for (int i = 1; i < operands.size(); i++) {
+            anyNull = rexBuilder.makeCall(
+                SqlStdOperatorTable.OR,
+                anyNull,
+                rexBuilder.makeCall(SqlStdOperatorTable.IS_NULL, operands.get(i))
+            );
+        }
+        // Result type stays the same as the original CONCAT — nullable VARCHAR.
+        RexNode nullLiteral = rexBuilder.makeNullLiteral(original.getType());
+        return rexBuilder.makeCall(original.getType(), SqlStdOperatorTable.CASE, List.of(anyNull, nullLiteral, original));
+    }
+}

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

@@ -85,12 +85,17 @@ public class DataFusionAnalyticsBackendPlugin implements AnalyticsSearchBackendP
     // path). COALESCE is the lowering target of PPL `fillnull`. CAST is required because
     // ReduceExpressionsRule.ProjectReduceExpressionsRule (in PlannerImpl) constant-folds field
     // references through equality filters into typed literals — e.g. after `where str0 = 'FURNITURE'`,
-    // the projection `fields str0` is rewritten to `CAST('FURNITURE' AS VARCHAR)`. The remaining
-    // comparison / arithmetic / logical operators are project-capable for eval-style projections.
+    // the projection `fields str0` is rewritten to `CAST('FURNITURE' AS VARCHAR)`. CONCAT is the
+    // lowering target of PPL `eval`'s `+` for strings (Calcite emits `||`, resolved to CONCAT in
+    // ScalarFunction); SAFE_CAST covers PPL `eval`'s explicit nullable `CAST(... AS ...)`
+    // expressions. The remaining comparison / arithmetic / logical operators are project-capable
+    // for eval-style projections.
     private static final Set<ScalarFunction> STANDARD_PROJECT_OPS = Set.of(
         ScalarFunction.COALESCE,
         ScalarFunction.CEIL,
         ScalarFunction.CAST,
+        ScalarFunction.CONCAT,
+        ScalarFunction.SAFE_CAST,
         ScalarFunction.SARG_PREDICATE,
         ScalarFunction.EQUALS,
         ScalarFunction.NOT_EQUALS,
@@ -180,15 +185,19 @@ public Set<AggregateCapability> aggregateCapabilities() {
 
             @Override
             public Map<ScalarFunction, ScalarFunctionAdapter> scalarFunctionAdapters() {
+                // Add new (ScalarFunction, ScalarFunctionAdapter) pairs in alphabetical order for
+                // readability — the Map.ofEntries form keeps spotless happy past the 5-pair point
+                // where Map.of becomes single-line and unreadable.
                 return Map.ofEntries(
-                    Map.entry(ScalarFunction.TIMESTAMP, new TimestampFunctionAdapter()),
-                    Map.entry(ScalarFunction.SARG_PREDICATE, new SargAdapter()),
+                    Map.entry(ScalarFunction.CONCAT, new ConcatFunctionAdapter()),
+                    Map.entry(ScalarFunction.CONVERT_TZ, new ConvertTzAdapter()),
                     Map.entry(ScalarFunction.DIVIDE, new StdOperatorRewriteAdapter("DIVIDE", SqlStdOperatorTable.DIVIDE)),
-                    Map.entry(ScalarFunction.MOD, new StdOperatorRewriteAdapter("MOD", SqlStdOperatorTable.MOD)),
                     Map.entry(ScalarFunction.LIKE, new LikeAdapter()),
-                    Map.entry(ScalarFunction.YEAR, new YearAdapter()),
-                    Map.entry(ScalarFunction.CONVERT_TZ, new ConvertTzAdapter()),
-                    Map.entry(ScalarFunction.UNIX_TIMESTAMP, new UnixTimestampAdapter())
+                    Map.entry(ScalarFunction.MOD, new StdOperatorRewriteAdapter("MOD", SqlStdOperatorTable.MOD)),
+                    Map.entry(ScalarFunction.SARG_PREDICATE, new SargAdapter()),
+                    Map.entry(ScalarFunction.TIMESTAMP, new TimestampFunctionAdapter()),
+                    Map.entry(ScalarFunction.UNIX_TIMESTAMP, new UnixTimestampAdapter()),
+                    Map.entry(ScalarFunction.YEAR, new YearAdapter())
                 );
             }
         };