[RFC] Support Schema-on-Read in PPL

[ykmr1224]

1. Summary

This RFC proposes a solution for OpenSearch PPL to support schema-on-read data sources where fields and types are unknown until data reading begins. The proposal focuses on a long-term solution to support both schema-on-read and commands that generate fields dynamically, such as spath. The core approach introduces a field resolution phase to determine the input schema based on the AST before logical planning.

Related issues

• [FEATURE] Support dynamic columns in PPL #4112 (motivation of this RFC)
• [RFC] Dynamic Fields in PPL Calcite Implementation #4433 (This RFC will be deprecate)
• [RFC] Support permissive mode in PPL #4349

2. Motivation

The current PPL engine utilizes Apache Calcite for planning and execution, but Calcite requires a fixed schema during the planning stage. To handle the case where input schema is not known (or partially known), and also dynamic field generation commands, we need mechanism to handle query without schema.

Existing mechanisms in Calcite, such as DynamicRecordType, are insufficient for PPL due to several gaps:

• Validation Stage Missing: PPL lacks a query validation stage where DynamicRecordType typically collects fields.
• Field Generation: DynamicRecordType does not support commands that dynamically produce fields (e.g., spath, multikv).
• Merge Conflicts: PPL merges fields from multiple inputs by overwriting non-null values, whereas Calcite retains dynamic stars as separate fields.
• Complexity: Previous attempts to mix static and dynamic fields resulted in complicated planning logic.

3. Proposed Solution: Field Resolution Phase

To resolve the schema issues, we propose introducing a Field Resolution phase. This phase analyzes the Abstract Syntax Tree (AST) to decide the input schema before converting it to a Calcite RelNode tree.

3.1 The Process

The new execution flow is as follows:

1. Parse PPL: Build AST from input PPL query.
2. Field Resolution (New): Decide input schema based on AST.
   • Traverse AST from root to leaf.
   • Calculate required input fields for each data source based on output requirements.
3. Logical Plan: Convert AST to Calcite RelNode tree using the resolved static schema.
4. Physical Plan & Execution: Apply rules and execute.

3.2 Key Logic

• Static Mapping: By calculating requirements through the AST, fields used by commands are mapped as static fields in the data schema.
• Dynamic Fields (_MAP): To support wildcards (e.g., prefix*) and dynamic field generation (e.g., spath), a dynamic map field is still required to store unmapped or pattern-matched fields.
• Field Ordering: Since Calcite's Map structure does not preserve order, we will calculate the final field order based on the query AST separately and reorder the results during execution to match Splunk-like behavior (lexicographical order for unmatched fields).

4. Implementation Roadmap

The implementation is broken down into five steps:

• Step 1: spath Limited Version: Implement spath without dynamic fields. It will extract only fields identified by the field resolution phase and treat them as STRING types.
• Step 2: Dynamic spath: Extend spath to produce dynamic fields and update commands like join and append to handle dynamic field expansion.
• Step 3: Schema-on-Read (Static Type): Implement LogicalIndexScan that applies resolved fields to the schema.
• Step 4: ANY Type Support: Extend schema-on-read to handle ANY types and adjust type coercion.
• Step 5: Performance: Improve performance by automatically identifying and indexing JSON internal fields.

5. Alternative Solutions Considered

5.1 Store All Fields into _MAP

• Description: Store all available fields into a single Map<String, Any> column, similar to the Elasticsearch/MongoDB adapter implementation.
• Pros: Simplifies implementation as commands simply manipulate the map.
• Cons: High overhead for field access and copying; hard to optimize using standard column pruning.

5.2 Convert PPL to SQL

• Description: Transpile PPL queries directly to SQL and execute them via the SQL engine.
• Pros: Flexible generation logic. Utilize DynamicRecordType usable during SQL validation phase.
• Cons: Non-trivial work to convert PPL logic to SQL; requires handling dynamic field logic within the generated SQL.

5.3 Unresolved Logical Plan (ULP)

• Description: Introduce a new abstraction (ULP) between the AST and Calcite's RelNode tree to separate command logic from schema resolution.
• Pros: Decouples logic from Calcite, facilitating future engine migrations.
• Cons: Requires a full rewrite of the current Calcite RelNodeVisitor and a new abstraction layer.

6. Appendix: Field Resolution Examples

This section illustrates how the field resolution phase traverses the AST from root to leaf to identify the necessary static fields.

6.1 Basic Backward Propagation

In this example, requirements propagate backward from the final command to the source.

Query: source idx | filter a > 1 | eval b = c * 2 | fields a, b, d

• fields a, b, d: Requires inputs (a, b, d).
• eval b = c * 2:
  • Needs c to produce b.
  • Needs a and d to satisfy the subsequent fields command.
  • Result: Requires (a, c, d).
• filter a > 1: Needs a for filtering, plus (a, c, d) from the previous step.
  • Result: Requires (a, c, d).
• source idx: The final resolved schema for the source is (a, c, d).

6.2 Field Pruning

The fields command can block the propagation of unneeded fields, allowing the source to fetch fewer columns.

Query: source idx | fields a, b | eval d = c * 2

• eval d = c * 2: Needs c.
• fields a, b: explicitly requests only a and b. It does not produce or pass c.
• source idx: Resolved schema is (a, b). Field c is pruned because the explicit fields command prevents it from reaching the eval step where it is needed.

6.3 Wildcards and Dynamic Fields

When wildcards are used, the resolution phase identifies specific static fields where possible and delegates the rest to the dynamic map.

Query: source=idx1 | eval a=b+2 | fields a, c, suffix*

• fields a, c, suffix*: Requires a, c, and the wildcard pattern suffix*.
• eval a=b+2:
  • Needs b to produce a.
  • a is generated here, so it is removed from the input requirement.
  • Passes through c and suffix*.
  • Result: Requires (b, c, suffix*).
• source idx1: Resolved schema includes static fields b, c and a dynamic map for matching suffix*.

6.4 Multi-Input (Join)

For multi-input commands like join, requirements are distributed to the appropriate branches.

Query: source=idx1 | join a [source=idx2 | where b > 1] | fields prefix*

• **fields prefix* **: Requires prefix*.
• join a:
  • Requires the join key a from both sides.
  • Distributes prefix* requirement to both side.
  • Result (Left): (a, prefix*)
  • Result (Right): (a, prefix*).
• source idx2 (Right Branch):
  • Needs a and prefix* from the join.
  • Needs b for the where b > 1 clause.
  • Result: (a, b, prefix*).
• source idx1 (Left Branch):
  • Result: (a, prefix*).

[ykmr1224]

POC for field resolution: main...ykmr1224:sql:field-resolution

[yuancu]

Validation Stage Missing: PPL lacks a query validation stage where DynamicRecordType typically collects fields.

#4892 adds back Calcite's validation phase. It also fully supports the alternative approach 5.2 (convert PPL to SQL). I haven't read the PoC yet, but I'm thinking if the type can be inferred from PPL's AST, the validation phase should work for this scenario because it also performs intricate type derivation from the logical plan. Given this, should we reconsider the feasibility of the implementation with DynamicRecordType?

5.2 Convert PPL to SQL

Btw, can you explain further on why 5.2 was considered? I.e. why converting PPL to SQL solve the problem of dynamic fields generation?

Proposed Solution: Field Resolution Phase

The PoC seems to traverse a PPL query's AST from root to leaf to determine which fields each data source needs to fetch. I haven't figured how dose this help schema-on-read. Isn't the data source schema known already with existing approach with Calcite? How does knowing which fields are used (in PoC) help, considering that we have known all fields and their types with the existing approach.

[qianheng-aws]

Field Resolution (New) seems to have similar process(detecting used fields from top to down) like RelFieldTrimmer in Calcite. And trimUnusedFields is called right before optimization in CalcitePreparingStmt::prepare_. I'm thinking if we can reuse or refer to it. Based on that, another proposal is:

1. Parse PPL: No change
2. Convert AST to Logical Plan: using _MAP or _OTHER for resolving unmapping columns.
3. Override RelFieldTrimmer to handle and replace _OTHER with static fields. When in scan level, we have chance to update its schema with all used fields passed from above.
4. Physical Plan & Execution: No change

[ykmr1224]

@yuancu

Validation Stage Missing: PPL lacks a query validation stage where DynamicRecordType typically collects fields.

#4892 adds back Calcite's validation phase. It also fully supports the alternative approach 5.2 (convert PPL to SQL). I haven't read the PoC yet, but I'm thinking if the type can be inferred from PPL's AST, the validation phase should work for this scenario because it also performs intricate type derivation from the logical plan. Given this, should we reconsider the feasibility of the implementation with DynamicRecordType?

In my understanding, #4892 will convert RelNode tree to SQLNode tree, but field resolution is needed before building RelNode tree.

5.2 Convert PPL to SQL

Btw, can you explain further on why 5.2 was considered? I.e. why converting PPL to SQL solve the problem of dynamic fields generation?

By directly convert to SQL, we might be able to utilize DynamicRecordType usable in SQL validation phase. (Updated description)

Proposed Solution: Field Resolution Phase

The PoC seems to traverse a PPL query's AST from root to leaf to determine which fields each data source needs to fetch. I haven't figured how dose this help schema-on-read. Isn't the data source schema known already with existing approach with Calcite? How does knowing which fields are used (in PoC) help, considering that we have known all fields and their types with the existing approach.

The major motivation is to handle the case where not all fields are indexed (some fields are not listed in index mapping) due to 10k mapping key limit, and actual available fields depends on the data.

[ykmr1224]

@qianheng-aws

Field Resolution (New) seems to have similar process(detecting used fields from top to down) like RelFieldTrimmer in Calcite. And trimUnusedFields is called right before optimization in CalcitePreparingStmt::prepare_. I'm thinking if we can reuse or refer to it. Based on that, another proposal is:

1. Parse PPL: No change
2. Convert AST to Logical Plan: using _MAP or _OTHER for resolving unmapping columns.
3. Override RelFieldTrimmer to handle and replace _OTHER with static fields. When in scan level, we have chance to update its schema with all used fields passed from above.
4. Physical Plan & Execution: No change

I think this idea is quite similar to #4433
This approach looked simple solution, but it had major downside where all the command implementation need to consider the existence of dynamic fields (_MAP) which makes the implementation complicated and error-prone.
And several command need to merge/deduplicate between static/dynamic fields, then converting from dynamic field to static field would be very complicated.
You can refer this diff for the actual implementation of this approach. (optimization is not implemented)

The approach proposed in this RFC separate the referred fields and others, and it simplify the query planning.