[FEATURE] PPL UDF and UDAF unification in Spark

[dai-chen]

Is your feature request related to a problem?

Unifying function behavior removes a major source of drift between engines and lets the PPL frontend stay portable. Today, many PPL UDFs/UDAFs have their definitions and implementations tightly coupled to Calcite APIs. This leads to duplicated engine-specific function code paths, and friction when porting these PPL functions to other engine such as Spark.

What solution would you like?

Proposed approach:

1. unified-query-api module provides engine-independent UnifiedFunction and UnifiedFunctionRepository abstraction.
2. Spark integrates unified function into Catalyst expression system by a thin UnifiedFunctionSparkWrapper

What alternatives have you considered?

[Update after PoC]

Do you have any additional context?

N/A

[dai-chen]

Proof of Concept

• PoC branch: https://github.com/dai-chen/opensearch-spark/tree/refactor-unified-function-calcite-to-spark

Issues Identified

1. Dependency issues

   • 1.1 Assembly failures with Calcite + ASM (JarJar plugin): see [BUG] Assembly fails with Calcite classes from unified-query-api #1228
   • 1.2 Runtime version mismatches with Spark: transitive dependency conflicts (e.g., Guava) cause runtime linkage errors.

2. Function-related

   • 2.1 PPL scalar overrides Spark aggregates (e.g., min/max): ambiguous resolution between scalar and aggregate forms alters query behavior.
   • 2.2 OpenSearch-only functions registered (e.g., match): engine-specific functions are registered without required OpenSearch data source and fail at runtime.
   • 2.3 Shared SparkSession contamination: global function registration leaks into other SparkSQL workloads and affects results.

spark-sql (default)> search source=test_events | where match(host, 'server1');

25/11/07 09:21:38 ERROR SparkSQLDriver: Failed in [search source=test_events | where match(host, 'server1')]
java.lang.UnsupportedOperationException: Relevance search query functions are only supported when
they are pushed down

spark-sql (default)> SELECT MIN(packets) FROM test_events;
min(packets)
60
120
60
180

3. Spark SQL dialect translation
   • 3.1 SAFE_CAST not mapped: missing conversion to Spark-compatible CAST/TRY_CAST semantics.
   • 3.2 Interval units emitted as strings: interval literals are produced as strings rather than proper interval expressions/types.

search source=test_events | spath input=message category | eval cat = abs(category);

== SQL ==
SELECT `@timestamp`, `host`, `packets`, `message`, `JSON_EXTRACT`(`message`, 'category') `category`,
ABS(SAFE_CAST(`JSON_EXTRACT`(`message`, 'category') AS DOUBLE)) `cat`
---------------------------------------------------------------------------------------------------------------------------------------------------------^^^
FROM `spark_catalog`.`default`.`test_events`

spark-sql (default)> search source=test_events | timechart per_second(packets);

[INVALID_PARAMETER_VALUE.DATETIME_UNIT] The value of parameter(s) `unit` in `TIMESTAMPDIFF` is invalid: 
expects one of the units without quotes YEAR, QUARTER, MONTH, WEEK, DAY, DAYOFYEAR, HOUR, MINUTE, SECOND,
MILLISECOND, MICROSECOND, but got the string literal 'MILLISECOND'.(line 1, pos 64)

== SQL ==
SELECT `@timestamp`, `DIVIDE`(`per_second(packets)` * 1.0000E3, TIMESTAMPDIFF('MILLISECOND',
`@timestamp`, TIMESTAMPADD('MINUTE', 1, `@timestamp`))) `per_second(packets)`
----------------------------------------------------------------^^^
...

---

Demo

• Test with Spark 3.5.1:

SPARK_LOCAL_IP=127.0.0.1 /.../spark-3.5.1-bin-hadoop3/bin/spark-sql --silent \
  --master 'local[1]' \
  --jars /.../opensearch-spark/flint-spark-integration/target/scala-2.12/flint-spark-integration-assembly-1.0.0-SNAPSHOT.jar \
  --conf spark.sql.extensions=org.opensearch.flint.spark.FlintNewSparkPPLExtensions \
  --conf spark.driver.bindAddress=127.0.0.1 \
  --conf spark.driver.host=127.0.0.1 \
  --conf spark.sql.cli.print.header=true \
  --conf "spark.hadoop.hive.cli.print.header=true" 2> >(grep -Ev '^(ANTLR (Tool|Runtime) version|line [0-9]+:.*extraneous input)' >&2)

UDFs

Run the following queries to verify:

# Create test table
spark-sql> CREATE TABLE test_events (
  `@timestamp` TIMESTAMP, host STRING, packets INT, message STRING) USING JSON;
spark-sql> INSERT INTO test_events VALUES
        (TIMESTAMP '2025-09-08 10:00:00', 'server1', 60, '{"category":1, "resource":"A"}'),
        (TIMESTAMP '2025-09-08 10:01:00', 'server1', 120, '{"category":2, "resource":"B"}'),
        (TIMESTAMP '2025-09-08 10:02:00', 'server1', 60, '{"category":3, "resource":"C"}'),
        (TIMESTAMP '2025-09-08 10:02:30', 'server2', 180, '{"category":4, "resource":"D"}');

spark-sql> SHOW USER FUNCTIONS;
spark-sql> DESC FUNCTION json_delete;
function_desc
Function: json_delete
Class: org.opensearch.flint.spark.query.wrapper.UnifiedFunctionSparkWrapper
Usage: N/A.

##### Show test data #####
spark-sql (default)> search source=test_events;

UnifiedQueryParser: PPL translated to Spark SQL: SELECT *
FROM `spark_catalog`.`default`.`test_events`
----------------------------------------------------------------------
@timestamp	host	packets	message
2025-09-08 10:00:00	server1	60	{"category":1, "resource":"A"}
2025-09-08 10:01:00	server1	120	{"category":2, "resource":"B"}
2025-09-08 10:02:00	server1	60	{"category":3, "resource":"C"}
2025-09-08 10:02:30	server2	180	{"category":4, "resource":"D"}

##### Simple UDF test #####
spark-sql> search source=test_events 
| eval cleanMsg = json_delete(message, 'category')
| fields cleanMsg;

UnifiedQueryParser: PPL translated to Spark SQL:
 SELECT `JSON_DELETE`(`message`, 'category') `cleanMsg`
 FROM `spark_catalog`.`default`.`test_events`
----------------
cleanMsg
{"resource":"A"}
{"resource":"B"}
{"resource":"C"}
{"resource":"D"}

spark-sql> EXPLAIN EXTENDED
 SELECT `JSON_DELETE`(`message`, 'category') `cleanMsg`
 FROM `spark_catalog`.`default`.`test_events`;
---------------------------------------------------------------------
== Parsed Logical Plan ==
'Project ['JSON_DELETE('message, category) AS cleanMsg#140]
+- 'UnresolvedRelation [spark_catalog, default, test_events], [], false

== Analyzed Logical Plan ==
cleanMsg: string
Project [json_delete(message#10,category) AS cleanMsg#140]
+- SubqueryAlias spark_catalog.default.test_events
   +- Relation spark_catalog.default.test_events[@timestamp#7,host#8,packets#9,message#10] json

== Optimized Logical Plan ==
Project [json_delete(message#10,category) AS cleanMsg#140]
+- Relation spark_catalog.default.test_events[@timestamp#7,host#8,packets#9,message#10] json

== Physical Plan ==
Project [json_delete(message#10,category) AS cleanMsg#140]
+- FileScan json spark_catalog.default.test_events[message#10] Batched: false, DataFilters: [], Format: JSON,
 Location: InMemoryFileIndex(1 paths)[file:/.../spark-3.5.1-bin-hadoop3/spark-warehouse/t...,
 PartitionFilters: [], PushedFilters: [], ReadSchema: struct

##### Mix use with Spark builtin functions #####
spark-sql> search source=test_events
 | eval col1 = COALESCE('hello', substring(host, 3)),
        col2 = substring(COALESCE(host, 'world'), 3)
 | fields host, col1, col2;

UnifiedQueryParser: PPL translated to Spark SQL:
SELECT `host`, COALESCE('hello', SUBSTRING(`host`, 3)) `col1`, SUBSTRING(COALESCE(`host`, 'world'), 3) `col2`
FROM `spark_catalog`.`default`.`test_events`
-------------------------
host	col1	col2
server1	hello	rver1
server1	hello	rver1
server1	hello	rver1
server2	hello	rver2

##### Complex command test: timechart #####
spark-sql (default)> search source=test_events | timechart span=2m limit=1 sum(packets) by host;

UnifiedQueryParser: PPL translated to Spark SQL: SELECT `t1`.`@timestamp`, 
  CASE WHEN `t7`.`host` IS NOT NULL THEN `t1`.`host`
  ELSE
    CASE WHEN `t1`.`host` IS NULL THEN NULL ELSE 'OTHER'
  END END `host`, SUM(`t1`.`$f2`) `sum(packets)`
FROM (SELECT `SPAN`(`@timestamp`, 2, 'm') `@timestamp`, `host`, SUM(`packets`) `$f2`
FROM `spark_catalog`.`default`.`test_events`
GROUP BY `host`, `SPAN`(`@timestamp`, 2, 'm')) `t1`
LEFT JOIN (SELECT `host`, SUM(`$f2`) `grand_total`
FROM (SELECT `SPAN`(`@timestamp`, 2, 'm') `@timestamp`, `host`, SUM(`packets`) `$f2`
FROM `spark_catalog`.`default`.`test_events`
GROUP BY `host`, `SPAN`(`@timestamp`, 2, 'm')) `t4`
WHERE `host` IS NOT NULL
GROUP BY `host`
ORDER BY 2 DESC NULLS FIRST
LIMIT 10) `t7` ON `t1`.`host` = `t7`.`host`
GROUP BY `t1`.`@timestamp`, CASE WHEN `t7`.`host` IS NOT NULL THEN `t1`.`host`
  ELSE
    CASE WHEN `t1`.`host` IS NULL THEN NULL ELSE 'OTHER' END
  END
ORDER BY `t1`.`@timestamp` NULLS LAST, 2 NULLS LAST
---------------------------------------
@timestamp	host	sum(packets)
2025-09-08 10:00:00	server1	180
2025-09-08 10:02:00	OTHER	180
2025-09-08 10:02:00	server1	60

UDAFs

spark-sql (default)> search source=test_events | stats values(host);
25/11/06 19:49:36 WARN UnifiedQueryParser: PPL translated to Spark SQL:
 SELECT `VALUES`(`host`) `values(host)`
FROM `spark_catalog`.`default`.`test_events`

values(host)
[server1, server2]
spark-sql (default)> EXPLAIN SELECT `VALUES`(`host`) `values(host)`
                   FROM `spark_catalog`.`default`.`test_events`;
line 1:23 no viable alternative at input 'EXPLAIN SELECT `VALUES`('
plan
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- ObjectHashAggregate(keys=[], functions=[ValuesAggFunction(class org.opensearch.sql.calcite.udf.udaf.ValuesAggFunction,
org.opensearch.flint.spark.query.api.UnifiedFunctionRepository$$$Lambda$1182/0x0000000800922040@3f27cfbf,
host#8, 0, 0)])
   +- Exchange SinglePartition, ENSURE_REQUIREMENTS, [plan_id=540]
      +- ObjectHashAggregate(keys=[], functions=[partial_ValuesAggFunction(class org.opensearch.sql.calcite.udf.udaf.ValuesAggFunction,
org.opensearch.flint.spark.query.api.UnifiedFunctionRepository$$$Lambda$1182/0x0000000800922040@3f27cfbf,
host#8, 0, 0)])
         +- FileScan json spark_catalog.default.test_events[host#8] Batched: false, DataFilters: [], Format: JSON,
Location: InMemoryFileIndex(1 paths)[file:/.../spark-3.5.1-bin-hadoop3/spark-warehouse/t...,
PartitionFilters: [], PushedFilters: [], ReadSchema: struct