[FEATURE] PPL Unification in Spark

[dai-chen]

Is your feature request related to a problem?

OpenSearch 3.0 introduces a new PPL (Piped Processing Language) engine built on Apache Calcite, which unifies logical planning and enables advanced query optimization across multiple execution backends. However, this engine currently exists only within the OpenSearch plugin. As a result, users running PPL queries in Spark (e.g., via Flint) are limited to a legacy PPL implementation that diverges in behavior and lacks recent improvements.

This inconsistency creates several challenges:

• Semantic divergence: Query results and behaviors may differ between OpenSearch and Spark for the same PPL query due to differences in parsing, type coercion, null handling, and exception propagation.
• Engineering duplication: Maintaining two independent implementations of PPL logic (in OpenSearch and Spark) increases development effort and risk of regression.
• Feature lag: New PPL features added to OpenSearch's Calcite engine are not automatically available in Spark, limiting functionality and user adoption.
• Optimization gaps: Advanced query rewrites and pushdowns enabled by Calcite's rule-based planning are not leveraged in Spark, reducing performance potential.

What solution would you like?

High-Level Design Considerations

• Approach 1 – Reuse Spark SQL Catalyst Optimizer: Leverage Spark SQL’s Catalyst optimizer and execution engine to minimize integration overhead. More details in design discussion.

Detailed Design Breakdown

• Deployment Model: Option A – Embedded Mode that embeds Calcite directly within Spark’s runtime to simplify integration and enable reuse of Calcite classes like OpenSearchSchema.
• Integration Interface: Option A – Spark SQL Interface as the lowest-effort path by routing PPL queries through Spark SQL and leveraging Calcite’s RelToSqlConverter.
• Integration Layer: Option A – Logical Plan Integration to enable faster iteration on language unification, while deferring OpenSearch-specific execution unification to Phase 2.

See further design context in this comment.

Milestones

• Phase 1: PPL language unification
  • M1: PPL-Calcite library ready for reuse
    • [FEATURE] Make PPL-Calcite engine library reusable in Spark sql#3598
    • [FEATURE] Export PPL-Calcite engine as reusable library sql#3734
  • M2: PPL-Calcite integration with Spark
    • [FEATURE] Add PPL-Calcite library as Flint Spark dependency #1202
    • [FEATURE] Enable PPL-Calcite logical plan integration with Spark #1203
  • M3: Spark PPL compatibility
    • [FEATURE] Evaluate semantic compatibility between PPL in Spark and OpenSearch #1208
    • Unify UDT/UDF/UDAFs support
• Phase 2: OpenSearch-specific unification [TBD]
  • OpenSearch schema, DSL pushdown and index scan logic unification.

Dependencies

1. PPL-Calcite backport to 2.19-dev branch: Required to resolve the versioning conflicts highlighted in [FEATURE] Make PPL-Calcite engine library reusable in Spark sql#3598.
2. PPL Compatibility Test Framework: Needed to verify behavioral consistency between PPL OpenSearch and Spark. Ideally, there is supposed to be a formal framework [FEATURE] Create a PPL compatibility verification framework piped-processing-language#32.
3. Calcite Function Contract: Ensure every PPL-Calcite function in library:
   • Can be translated into a valid SQL expression
   • Has a Java static method definition that supports registration in SparkSQL’s function catalog

---

What alternatives have you considered?

• Continue extending the legacy PPL implementation in Spark: This approach involves adding new commands or functions directly to the existing PPL Spark codebase. However, it results in duplicated logic and a growing risk of divergence from the Calcite-based PPL engine in OpenSearch. As it continues to evolve, maintaining consistency and semantic alignment would become increasingly difficult and error-prone.

Do you have any additional context?

• Proof of concepts
  • [FEATURE] PPL Unification in Spark #1136 (comment)
  • [Do Not Merge][POC] Calcite integration #993

[dai-chen]

Proof of Concept

To achieve consistent PPL semantics across different execution engines, we propose a proof of concept (PoC) to integrate the Calcite-based query engine into Spark through the Flint Spark extension. This PoC will explore the feasibility of reusing Calcite’s logical plan directly in Spark and evaluate the integration points, such as custom data type (UDT), function interoperability (UDF/UDAF), and plan translation.

---

1. Introduce PPL-Calcite library to Flint Spark extension

To enable reuse of the Calcite-based PPL engine in Spark, the first step is to package the existing PPL-Calcite components as a standalone library that can be integrated into the Flint Spark extension.

Open Questions

• Can the PPL-Calcite engine, originally built with JDK 21, be fully downgraded to JDK 17 without losing functionality or introducing subtle behavior changes?
• Are there any transitive dependencies in the PPL-Calcite library that conflict with those used in the Flint Spark extension (e.g., Jackson, Guava, Calcite core)?
• What is the minimal set of modules and dependencies required to isolate the Calcite planner logic from OpenSearch-specific components?

Key Steps

The work involves:

• Rebuilding the PPL-Calcite engine with an older JDK version (e.g., JDK 11 or JDK 17) compatible with Spark 3.x.
• Identifying and fixing code that uses JDK 21-specific features (such as pattern matching for instanceof, record classes, or sealed interfaces) that prevent compilation on the target JDK.
• Ensuring the modified Calcite-based PPL library remains functionally consistent with the original implementation.

Deliverables

• A JDK-compatible PPL-Calcite library artifact (e.g., JAR): https://github.com/dai-chen/sql-1/tree/export-ppl-calcite-library
• Documentation of any code changes or feature downgrades made: [FEATURE] Make PPL-Calcite engine library reusable in Spark sql#3598
• Verified successful compilation and basic usage of the library within the Flint Spark extension.

---

2. Integrate Spark catalog with Calcite schema

To enable PPL query execution on Spark through the Calcite-based planner, proper catalog integration is required between Spark’s table/catalog system and Calcite’s schema representation. Specifically, the Calcite planner relies on its own Schema to resolve table references, while Spark uses its Catalog and TableProvider mechanisms. Without bridging these systems, PPL queries that reference table names, index aliases, or wildcard patterns will fail during logical plan generation.

Open Questions

• How can we best implement a Calcite Schema that delegates table resolution to Spark’s catalog and reconcile differences in column type metadata between Calcite’s and Spark’s data type?
• Can this approach support unstructured or semi-structured tables, such as CloudWatch LogGroup catalogs, where schema inference happens at runtime?
• How should we handle index aliases and wildcard patterns (e.g., logs-*) that are common in OpenSearch but not natively supported by Spark's catalog? [Low priority]

Key Steps

• Implement a Calcite Schema or CatalogReader adapter that reads table metadata from Spark’s catalog (including table names, column types, and partitions if needed).
• Ensure that:
  • Table existence checks in Calcite are delegated to Spark’s catalog.
  • Column type information remains consistent between Spark and Calcite.
• Support basic catalog operations for the PoC scope:
  • Table resolution by name.
  • Alias and wildcard handling (if required by the test query set).
• Validate catalog integration by running PPL queries that involve table lookups and verifying correct schema resolution.

Deliverables

• A working Calcite catalog integration layer that allows the PPL planner to resolve Spark tables
  • The PoC implementation below integrates Spark’s catalog with Calcite by leveraging Spark’s Table abstraction. The solution extracts schema information from any data source that conforms to Spark’s interface, making the approach agnostic to the underlying data source (e.g., Parquet, JSON, CloudWatch).
  • Code: https://github.com/dai-chen/opensearch-spark/blob/unified-ppl-test/flint-spark-integration/src/main/scala/org/opensearch/flint/spark/FlintSparkPPLCalciteParser.scala#L153
• Verification that table resolution works for the initial PPL query set on Spark.
• Spark SQL to Calcite SQL type mappings

PPL Type	Calcite SQL Type	Spark SQL Type(s)	Notes / Differences
boolean	BOOLEAN	BooleanType	Fully aligned
tinyint	TINYINT	ByteType	Fully aligned
smallint	SMALLINT	ShortType	Fully aligned
int	INTEGER	IntegerType	Fully aligned
bigint	BIGINT	LongType	Fully aligned
float	FLOAT	FloatType	Fully aligned
double	DOUBLE	DoubleType	Fully aligned
string	VARCHAR	StringType	Calcite may enforce length constraints
ip	IP (UDT)	IP (UDT)
date	DATE (UDT)	?
time	TIME (UDT)	?
timestamp	TIMESTAMP (UDT)	?
binary	VARBINARY	BinaryType	Calcite may enforce length constraints
struct	STRUCT	StructType
array	STRUCT	ArrayType(StructType)
interval	X	X
geo_point	X	X

Reference:

• PPL types: TODO
• PPL type mapping to Calcite types
• PPL type mapping to SparkSQL types

---

3. Integrate Calcite logical plan with Spark via SparkSQL

The goal of this subtask is to enable execution of PPL queries on Spark by leveraging the Calcite logical plan (RelNode) generated by the Calcite-based PPL engine introduced in Subtask 1. The initial approach focuses on generating SQL query text from the Calcite logical plan (RelNode) using Calcite’s SparkSQL dialect converter. This allows Spark to execute the translated SQL using its native SQL parser and Catalyst engine, minimizing initial integration complexity.

Open Questions

• Does the SQL generated from RelNode fully preserve the semantics of the original PPL query (e.g., filters, projections, null handling)?
• Is the performance overhead of re-parsing SQL in Spark acceptable, especially for complex queries? [Low priority]

Key Steps

• Use the PPL planner in the PPL-Calcite library to generate the Calcite logical plan (RelNode) from PPL queries.
• Convert the generated RelNode to SparkSQL query text using Calcite’s built-in SparkSQL dialect converter.
• Execute the generated SparkSQL query via Spark’s sql() API.
• Validate query correctness and output consistency with existing PPL test cases.

Deliverables

• Working integration code that PPL parsing and execution through Spark’s SQL engine.
• Identify queries that cannot be translated to SparkSQL correctly.

---

4. Support Calcite custom data type (UDT) in Spark

To ensure seamless integration between the Calcite-based PPL planner and SparkSQL execution, it is critical to handle data type compatibility between Calcite’s logical plans and Spark’s Catalyst engine. The Calcite-based PPL engine defines custom user-defined types (UDTs) that are not natively recognized by Spark. Without proper type mapping and registration, these UDTs could block plan translation, query execution, or result parsing.

Open Questions

• Which Calcite UDTs are actually required by PPL queries?
• Can Spark’s UserDefinedType mechanism adequately represent all Calcite UDTs without semantic differences?
• Will these UDTs need to be exposed to end-users or can they remain internal to the execution engine?

Key Steps

• Identify UDTs Used in Calcite Logical Plans:
  Review the Calcite-based PPL planner to collect all custom data types (e.g., interval types, complex types, or other PPL-specific UDTs).
• Define Spark-Compatible UDT Implementations:
  Implement corresponding Spark UDT classes (extending UserDefinedType in Spark) for each identified Calcite UDT.
• Register UDTs with Spark Session:
  Ensure that the Spark session recognizes these UDTs through appropriate registration mechanisms, allowing them to be used transparently in SQL queries and Catalyst plans.
• Validate Compatibility:
  Test type handling by running PPL queries that involve these UDTs and verifying successful plan generation, query execution, and result decoding.

Deliverables

• Spark UDT implementations for Calcite-defined types.
• Integration of UDT registration in the Flint Spark extension initialization.
• Documentation on supported UDTs and any limitations or workarounds.
• Validation tests to confirm that UDTs work correctly with Spark SQL queries generated from PPL logical plans. [Low priority in this PoC]

---

5. Support Calcite custom function (UDF/UDAF) in Spark

The Calcite-based PPL engine defines various user-defined functions (UDFs) that are required for query parsing, logical plan generation, and execution. Many of these functions may not have direct equivalents in Spark, or their behavior might differ from Spark’s built-in implementations.

Open Questions

• Which Calcite UDF/UDAFs are required to support the initial set of PPL queries on Spark?
• Can we reuse any of Spark’s built-in functions as drop-in replacements?
• How do we ensure consistent behavior, such as function overloading, type promotion rules, division by zero, null input, etc?
• Should we register these functions using Spark’s UDFRegistration API, or define them as Catalyst Expression classes for better performance and optimizer compatibility? [Low priority]

Key Steps

• Identify UDFs Required by the Calcite PPL Planner:
  Review the PPL-Calcite function registry and collect all custom functions that are used in PPL query generation.
• Implement or Reuse UDFs in Spark:
  • For functions that already exist in Spark with matching behavior, map to the built-in Spark equivalents.
  • For Calcite-specific or behavior-different functions, implement them as Spark UDFs or use Spark’s Expression-based function registration.
• Register UDFs with Spark Session:
  Ensure that all identified functions are registered to the Spark session during initialization, making them available in both SQL execution and (potentially) Catalyst plans.
• Verify Function Behavior Consistency:
  Validate that the registered functions produce results consistent with the Calcite engine by running representative test cases from the PPL test suite.

Deliverables

• Registered UDFs in the Spark session for Calcite-defined functions.
• Function mapping documentation (including notes on reused Spark built-ins vs. custom implementations).
• Validation tests ensuring behavior consistency between PPL-on-OpenSearch and PPL-on-Spark for the registered functions. [Low priority in this PoC]

[Swiddis]

Some thoughts:

a) We've already been down the road a few times of accepting pre-made solutions for standard DB operations. Calcite marks the third time that we're throwing out our querying engine and making a new one. How can we ensure that it's extensible and we won't need a V4?

While solution B is more work, it seems like it prioritizes the long-term vision and gives us a lot more flexibility to work with more data sources down the line. I'm concerned about solution A being a cheaper win.

b) I've mentioned this a few times in side-channels but it seems relevant here too: for a lot of queries, the bottleneck isn't execution, but the list-files operation on lower data sources. I wonder if we're prematurely optimizing here? We should try and prioritize a cohesive data flow from bootstrap -> analysis -> file ops -> results, and that might mean using the more flexible framework¹.

c) In either approach, I think we should look at how we want to benchmark this. The current E2E-timing approach is cumbersome and makes it hard to figure out if optimizations are actually working². There's a lot of discussion here about what's faster/slower but nothing about how much faster or what sort of performance characteristics are acceptable.

Overall, I'm in favor of B based on what's presented. I don't think the downsides outweigh the long-term benefits.

Footnotes

1. In addition, converting PPL to SQL restricts the scope of what PPL would be capable of doing down the road. I don't know if its feature should be bound to just being SQL with different syntax. I also don't like that we need to parse 2 queries to run 1 PPL query: parse/analyze PPL, then convert to SQL, then parse/analyze SQL. ↩

2. In recent memory, consider all the adaptive rate limiter testing. If we had a microbenchmarking framework, we wouldn't need as much effort there. ↩

[dai-chen]

2. Requirements

The goal of this integration is to ensure consistent PPL language behavior across OpenSearch and Spark, while reusing the same underlying library to eliminate duplication and reduce maintenance overhead.

Note: Other aspects such as query response shaping (e.g., pagination), result formatting, and access control are out of scope for now.

3. High Level Design

3.1 Unifying PPL Language Frontend

The core of this integration is to reuse the shared PPL-Calcite library, allowing both OpenSearch PPL and Spark PPL to rely on the same language frontend. This reuse directly addresses the PPL Language Compatibility requirements outlined earlier:

• Grammar consistency is guaranteed by using the same PPL parser and syntax definitions.
• Semantic consistency is achievable assuming parity across common types, functions, and operators between Calcite and Spark SQL.

While complete semantic alignment will require extensive compatibility testing, this validation is deferred to a future phase. For now, we assume that any minor differences can be identified and addressed incrementally.

3.2 Unifying PPL Implementation with OpenSearch

With the language front-end unified, the next focus is integrating OpenSearch-specific logic into the Spark runtime, including index metadata mapping, custom types and functions, and index scan optimization and execution. This involves three key aspects, each of which will be discussed in detail in Detailed Design section below.

Before finalizing decisions on each of these aspects, we must first address a fundamental architectural question that influences all three: Should we leverage Spark SQL engine for optimization and planning, or bypass it entirely and use Spark purely as a distributed execution runtime?

The diagram below illustrates two architectural approaches for executing PPL queries in Spark: one that leverages Spark SQL's Catalyst optimizer, and another that bypasses it entirely to use Spark as a distributed execution runtime.

Examples

# PPL query
source=http_logs | where bytes > 1000 | stats count() by status

# Approach 1: Leverage Spark SQL Optimizer
## Spark Logical Plan
Aggregate [status#12], [status#12, count(1) AS count#20L]
+- Filter (bytes#13 > 1000)
   +- Relation http_logs [status#12, bytes#13]

## Spark Physical Plan
AdaptiveSparkPlan isFinalPlan=false
+- WholeStageCodegen
   +- HashAggregate(keys=[status#12], functions=[count(1)])
      +- Exchange hashpartitioning(status#12, 200)
         +- HashAggregate(keys=[status#12], functions=[partial_count(1)])
            +- IndexScan [status, bytes]
               PushedFilters: [GreaterThan(bytes,1000)]
               ...

# Approach 2: Spark as Distributed Execution Runtime
val rdd = indexScanRDD("http_logs")
  .filter(_.bytes > 1000)
  .map(_.status)
  .countByValue()

Design Decisions

Approach 1 is preferred to gain the best of both engines while minimizing engineering effort. Approach 2 may be revisited in the future if Calcite is promoted as a centralized query planner and optimizer — enabling consistent cost-based optimization (CBO) across OpenSearch, AWS Glue tables, CloudWatch LogGroups, and Flint indexes (skipping index, covering index, materialized views). However, the significant challenges mentioned above needs to be addressed.

[dai-chen]

4. Detailed Design

This section breaks down the three core aspects that determine how Calcite and Spark interact:

1. Deployment Model: Whether the Calcite engine runs within the same JVM as Spark (tightly coupled) or as a standalone service (loosely coupled).
2. Integration Interface: How query plans from Calcite are passed to Spark — via SparkSQL logical plans, DataFrame APIs, or emerging interfaces like Substrait.
3. Integration Layer: Whether to integrate at the logical plan level or at the physical plan level, where Calcite provides execution-aware plans that Spark consumes more directly.

4.1 Deployment Model

The deployment model defines where and how the Calcite-based PPL engine runs in relation to the Spark runtime.

The following diagram illustrates the structural difference between embedding Calcite in the Spark JVM versus running it as a standalone process.

Design Decision

Option A is preferred as the initial deployment strategy. Option B may be revisited in the future if JVM compatibility issues arise or if there is a need to deploy the Calcite planner as a standalone service.

4.2 Integration Interface

This section compares two approaches for translating Calcite-generated query plans into Spark execution logic.

Examples

# PPL query
source=http_logs | where status = '200' | stats sum(bytes) by ip

# Option A: Spark SQL Query
SELECT ip, SUM(bytes) FROM http_logs WHERE status = '200' GROUP BY ip

# Option B: DataFrame API
df.filter($"status" === "200")
  .groupBy("ip")
  .agg(sum("bytes"))
  
# Option C: Substrait Plan (JSON format)
{
  "relations": [
    {
      "root": {
        "input": {
          "aggregate": {
            "groupings": [
              { "grouping_expressions": [{ "selection": { "direct_reference": { "struct_field": { "field": 0 } } } }] }
            ],
            ...
}

Design Decision

Option A is preferred as the initial approach due to its simplicity and low integration cost. It is not a one-way door—if limitations or correctness issues are discovered, the system can evolve to Option B or C which offers more fine-grained control and fidelity over query translation.

4.3 Integration Layer

This section compares two approaches for integrating Calcite into the Spark query execution pipeline—at the logical level or at the physical level. Both approaches translate Calcite plans into Spark SQL, but differ in how much control is retained over execution semantics.

Option A: Logical Plan Integration

1. Calcite builds a logical plan, using schema information retrieved from SparkSchema.
2. The logical plan is converted to SQL using Calcite’s RelToSqlConverter.
3. The generated SQL query is submitted to Spark, with any required UDTs and UDFs registered beforehand.
4. Spark re-parses the SQL and delegates pushdown optimizations to the Flint data source and S3 data source.
5. Spark executes the final physical plan, including index scan operations implemented within Flint data source.

Option B: Physical Plan Integration

1. Calcite builds a physical plan, with:
   1. OpenSearch schema information retrieved from OpenSearchSchema and pushdown logic defined in CalciteIndexScanRules.
   2. An EnumerableTableScan with minimal capability to skip non-OpenSearch tables, which will be handled later by Spark SQL.
2. The physical plan is converted to SQL using Calcite’s RelToSqlConverter, preserving execution semantics:
   1. Calcite’s Linq4j expressions are dynamically bridged to Spark UDFs at runtime.
   2. Index pushdown filters are encoded in SQL via table-valued functions (UDTFs).
3. The generated SQL query is submitted to Spark, with any required UDTs registered beforehand.
4. Spark re-parses the SQL and delegates pushdown optimizations to the S3 data source.
5. Spark executes the final physical plan, including index scan operations implemented within Calcite using IndexScanEnumerator.

Design Decision

Option A is selected for the initial implementation due to its simplicity. It avoids introducing direct dependencies on opensearch modules and enable faster iteration for PPL support for non-OpenSearch data sources. Option B offers a fully unified PPL stack — including schema mapping, type/function handling, physical optimizations, and index scan operations. It will be considered in future phases when deeper execution control and end-to-end semantic unification become necessary.