hudi-agent commented on code in PR #19997: URL: https://github.com/apache/hudi/pull/19997#discussion_r4046463461
########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,405 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +# RFC-112: Support Spark Declarative Pipelines + +## Proposers + +- @cshuo +- @danny0405 + +## Approvers + +- TBD + +## Status + +Issue: TBD + +## Abstract + +[Spark Declarative Pipelines (SDP)](https://spark.apache.org/docs/4.1.1/declarative-pipelines-programming-guide.html) lets users define tables and transformations while Spark builds the dependency graph and runs the batch and streaming flows. Hudi provides atomic commits, keyed updates, event ordering, deletes, incremental queries, and table services. + +This RFC makes Hudi a supported SDP source and target. It covers catalog-based table creation, materialized views, streaming tables, basic Hudi-to-Hudi incremental reads, and keyed upserts through Hudi's existing writer. + +SDP and Hudi retain their existing responsibilities: + +- SDP owns dataset declarations, dependency analysis, orchestration, flow checkpoints, retries, and execution order. +- Hudi owns table storage, commit atomicity, record reconciliation, incremental read semantics, and table services. + +Users can build the same pipelines today with Spark jobs, Hudi writers or Hudi Streamer, and an external orchestrator. This integration removes much of that glue code while keeping the outputs as ordinary Hudi tables. + +## Background + +### Spark Declarative Pipelines + +Apache Spark 4.1 introduced SDP for declarative batch and streaming ETL. SDP persists two main dataset types: + +- A **materialized view**, which has exactly one batch flow that computes a table. +- A **streaming table**, which has one or more streaming flows that append their output to a table. + +Users define pipelines in Python or SQL. SDP validates dataset references, builds the dataflow graph, runs independent flows in parallel, and stores Structured Streaming checkpoints in the pipeline storage location. + +In the Spark 4.1 implementation, a streaming table flow uses a Structured Streaming append-mode write to `DataStreamWriter.toTable`, and a batch table flow uses append-mode `DataFrameWriter.saveAsTable`. Before running a materialized-view batch flow, SDP truncates an existing target table. See Spark's [`FlowExecution`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala) and [`DatasetManager`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala). + +These table operations are the storage-format integration boundary; this RFC does not add a Hudi-specific execution engine. + +### Relevant Hudi capabilities + +Hudi already provides the required data-plane primitives: + +- `HoodieCatalog` can create and load Hudi tables through Spark's catalog APIs. +- The Hudi batch writer supports bulk insert, insert, upsert, insert overwrite, and table insert overwrite operations. An upsert can carry delete-marker records. +- `HoodieStreamingSink` writes Structured Streaming micro-batches and records checkpoint metadata used to avoid recommitting an already committed batch after retry. +- Hudi's streaming source and incremental query modes can expose records changed since a prior instant. + +The integration exposes these capabilities through SDP's existing table and flow abstractions. Hudi Streamer, the DataFrame writer, Spark SQL DML, and direct Structured Streaming APIs remain unchanged. + +### Current gaps + +`format="hudi"` is not yet an end-to-end supported SDP contract. Some operations may already work, but table creation, refresh, restart, upsert configuration, and Hudi-to-Hudi chaining are neither documented nor covered by an integration suite. + +Materialized-view refresh is the main correctness gap. SDP 4.1 truncates the target before appending the recomputed result, while Hudi's full-table `TRUNCATE TABLE` deletes the table path and reinitializes its properties. Using this path would discard the timeline and could expose an empty or partially rebuilt table if the refresh fails. + +The integration must also address: + +- SDP-created table properties must be persisted and reliably supplied to Hudi batch and streaming writers. +- Streaming restart must coordinate the SDP/Spark checkpoint with Hudi's sink checkpoint metadata. +- A Hudi source used through `readStream.table` must preserve incremental offset semantics. +- SDP must carry Hudi's persisted record-key, precombine, and record-merger settings into the existing upsert path. Deletes remain upsert input rows marked by `_hoodie_is_deleted`; Spark append output mode does not replace this contract. + +### Related work + +- Hudi's Spark 4.1 support ([issue #17663](https://github.com/apache/hudi/issues/17663), [PR #17674](https://github.com/apache/hudi/pull/17674)) provides the runtime baseline, but not an SDP contract. +- DataSource V2 read work ([RFC-98](../rfc-98/rfc-98.md), [issue #15023](https://github.com/apache/hudi/issues/15023)) may improve the catalog read path but is not required for the initial integration. +- The [SQL insert-overwrite discussion](https://github.com/apache/hudi/discussions/13815) shows why refresh semantics must be explicit rather than inferred from an append call. +- [Issue #13973](https://github.com/apache/hudi/issues/13973) shows that a latest-state incremental read cannot fully propagate hard deletes. Flows that require complete delete propagation can use Hudi's existing CDC incremental format. +- [Issue #5537](https://github.com/apache/hudi/issues/5537) tracks named-catalog support. The first version uses `HoodieCatalog` as `spark_catalog`. + +## Motivation + +### Representative use case + +A typical order pipeline looks like this: + +```text +MySQL/PostgreSQL CDC or Kafka + | + v + bronze_order_events (Hudi) + | + v + current_orders (Hudi) + | + v + daily_order_metrics (Hudi) +``` + +The bronze table stores incoming events. The current-state table uses a record key and precombine field to reconcile duplicates and out-of-order events, including delete-marker records. Downstream streaming tables consume changed records incrementally, while batch aggregates can be materialized as views. + +Today this usually means several Spark applications plus custom handling for table creation, write options, checkpoints, dependencies, and restarts. With SDP, the graph lives in one project and the persisted datasets remain regular Hudi tables that other engines can query. + +Related user request issues include: + +- #19281: [multi-dataset incremental processing](https://github.com/apache/hudi/issues/19281) +- #5189: [chained incremental Hudi tables](https://github.com/apache/hudi/issues/5189) +- #13973: [downstream delete propagation](https://github.com/apache/hudi/issues/13973). + +### User value + +1. **Declarative pipelines.** Users define datasets as SQL or DataFrame transformations; SDP manages dependencies, checkpoints, and Hudi writes. +2. **Managed incremental composition.** Hudi already provides incremental reads; SDP composes them into a dependency-aware pipeline, managing execution order, checkpoints, and recovery without separate Spark jobs or hand-written orchestration. +3. **Open outputs.** Each dataset remains a standard Hudi table that can be queried independently of SDP by any Hudi-compatible engine. + +SDP does not make arbitrary joins or aggregates incremental; performance benefits apply only to flows that can process changes incrementally. + +## Goals + +1. Create Hudi tables from SDP datasets through `HoodieCatalog`. +2. Write materialized views with Hudi `bulk_insert` and `insert_overwrite_table` semantics. +3. Write SDP streaming tables through Hudi's Structured Streaming sink. +4. Support a basic incremental Hudi-to-Hudi streaming-table chain. +5. Expose Hudi's existing keyed upsert path through SDP, including record-key, precombine, record-merger, and delete-marker handling. +6. Keep non-SDP Hudi behavior unchanged. + +## Non-Goals + +The first version excludes: +- A complete Hudi DataSource V2 read or streaming-write implementation. +- Cross-table transactions across multiple SDP targets. +- End-to-end exactly-once delivery across an arbitrary external source and Hudi. Guarantees remain scoped to Spark/Hudi micro-batch retries at one target. +- Multiple concurrent append flows into one Hudi target in the initial phase. +- Arbitrary named catalogs for `HoodieCatalog`; the first version uses Hudi as `spark_catalog`. + +## Proposed User Experience + +The examples below implement the representative order pipeline in dependency order. Property names will use the canonical Spark 4.1 SDP and Hudi configuration names. + +### Pipeline configuration + +An SDP project configures the Hudi extension and catalog like any other Hudi Spark application: + +```yaml +name: hudi_orders_pipeline +libraries: + - glob: + include: transformations/** +storage: s3://pipelines/checkpoints/hudi_orders_pipeline +catalog: spark_catalog +database: lakehouse +configuration: + spark.sql.extensions: org.apache.spark.sql.hudi.HoodieSparkSessionExtension + spark.sql.catalog.spark_catalog: org.apache.spark.sql.hudi.catalog.HoodieCatalog +``` + +The SQL below assumes that `order_events_source` is an external streaming source containing parsed Kafka or database CDC events. Each `CREATE ... AS SELECT` statement defines both the target dataset and its single flow. + +### Streaming table with an append flow + +```sql +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'insert' +) +AS +SELECT * +FROM STREAM order_events_source; +``` + +SDP owns the checkpoint location and trigger; Hudi commits each micro-batch. + +### Streaming table with keyed upserts + +```sql +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'upsert', + 'hoodie.datasource.write.recordkey.field' = 'order_id', + 'hoodie.datasource.write.precombine.field' = 'event_sequence', + 'hoodie.datasource.write.partitionpath.field' = 'event_date' +) +AS +SELECT + *, + operation = 'DELETE' AS _hoodie_is_deleted +FROM STREAM bronze_order_events; +``` + +Spark append output mode only controls how each micro-batch reaches the sink; it does not make the Hudi target append-only. `HoodieStreamingSink` still runs the configured Hudi operation, which defaults to `upsert`. + +For an existing Hudi table, SDP loads the record key, precombine field, partition fields, and record-merger settings from the table configuration. For a new SDP table, these settings are declared as table properties and persisted by Hudi. SDP does not implement its own deduplication, ordering, merge, or delete path. It validates the schema and configuration, then passes each micro-batch to the existing Hudi writer. + +### Batch materialized view + +```sql +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'COPY_ON_WRITE' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and may use `bulk_insert`; later runs replace the result with `insert_overwrite_table`. Refresh must not delete and recreate the table. + +## Semantics + +### Dataset-to-Hudi operation mapping + +| SDP dataset/flow | Spark 4.1 execution shape | Proposed Hudi semantics | +| ----------------------------------- | --------------------------------------------- | --------------------------------------------------------- | +| New materialized view | Batch append to newly created table | Initial bulk insert or insert | +| Existing materialized view | Full batch recomputation | Hudi `insert_overwrite_table`; no physical table deletion | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured at table level | +| Hudi upsert target | Append-mode micro-batches | Existing Hudi `upsert` | +| Hudi streaming source | Structured Streaming read with SDP checkpoint | Basic Hudi latest-state incremental source | + +### Materialized-view write mapping + +Initial materialization maps to Hudi `bulk_insert` or `insert`; refresh maps to `insert_overwrite_table`. Because Spark 4.1 implements refresh as truncate followed by append, the integration must select the overwrite path without invoking Hudi's destructive full-table truncate. Where to intercept that path remains an open implementation question; a general table-format replace API is not part of this RFC. Review Comment: 🤖 This is the central correctness question of the RFC and it's left open. As I read Spark 4.1's DatasetManager, the refresh issues a SQL `TRUNCATE TABLE` (which Hudi's HoodieAnalysis routes to TruncateHoodieTableCommand) followed by a plain append `saveAsTable`. Could you enumerate the candidate interception points and their trade-offs? E.g. (a) making Hudi's TRUNCATE a non-destructive replacecommit keeps the timeline but still leaves a window where readers see an empty table and breaks the 'existing snapshot unchanged on failure' row; (b) a no-op TRUNCATE + `insert_overwrite_table` on the append gives the atomicity you want but only if scoped to SDP, otherwise it changes TRUNCATE semantics for every Hudi SQL user (conflicts with Goal 6). Also worth noting that `HoodieInternalV2Table` already maps DSv2 `truncate()` to overwrite-table, so a DSv2 write path may be the cleaner hook. And what happens when a refresh legitimately produces zero rows? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,405 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +# RFC-112: Support Spark Declarative Pipelines + +## Proposers + +- @cshuo +- @danny0405 + +## Approvers + +- TBD + +## Status + +Issue: TBD + +## Abstract + +[Spark Declarative Pipelines (SDP)](https://spark.apache.org/docs/4.1.1/declarative-pipelines-programming-guide.html) lets users define tables and transformations while Spark builds the dependency graph and runs the batch and streaming flows. Hudi provides atomic commits, keyed updates, event ordering, deletes, incremental queries, and table services. + +This RFC makes Hudi a supported SDP source and target. It covers catalog-based table creation, materialized views, streaming tables, basic Hudi-to-Hudi incremental reads, and keyed upserts through Hudi's existing writer. + +SDP and Hudi retain their existing responsibilities: + +- SDP owns dataset declarations, dependency analysis, orchestration, flow checkpoints, retries, and execution order. +- Hudi owns table storage, commit atomicity, record reconciliation, incremental read semantics, and table services. + +Users can build the same pipelines today with Spark jobs, Hudi writers or Hudi Streamer, and an external orchestrator. This integration removes much of that glue code while keeping the outputs as ordinary Hudi tables. + +## Background + +### Spark Declarative Pipelines + +Apache Spark 4.1 introduced SDP for declarative batch and streaming ETL. SDP persists two main dataset types: + +- A **materialized view**, which has exactly one batch flow that computes a table. +- A **streaming table**, which has one or more streaming flows that append their output to a table. + +Users define pipelines in Python or SQL. SDP validates dataset references, builds the dataflow graph, runs independent flows in parallel, and stores Structured Streaming checkpoints in the pipeline storage location. + +In the Spark 4.1 implementation, a streaming table flow uses a Structured Streaming append-mode write to `DataStreamWriter.toTable`, and a batch table flow uses append-mode `DataFrameWriter.saveAsTable`. Before running a materialized-view batch flow, SDP truncates an existing target table. See Spark's [`FlowExecution`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala) and [`DatasetManager`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala). + +These table operations are the storage-format integration boundary; this RFC does not add a Hudi-specific execution engine. + +### Relevant Hudi capabilities + +Hudi already provides the required data-plane primitives: + +- `HoodieCatalog` can create and load Hudi tables through Spark's catalog APIs. +- The Hudi batch writer supports bulk insert, insert, upsert, insert overwrite, and table insert overwrite operations. An upsert can carry delete-marker records. +- `HoodieStreamingSink` writes Structured Streaming micro-batches and records checkpoint metadata used to avoid recommitting an already committed batch after retry. +- Hudi's streaming source and incremental query modes can expose records changed since a prior instant. + +The integration exposes these capabilities through SDP's existing table and flow abstractions. Hudi Streamer, the DataFrame writer, Spark SQL DML, and direct Structured Streaming APIs remain unchanged. + +### Current gaps + +`format="hudi"` is not yet an end-to-end supported SDP contract. Some operations may already work, but table creation, refresh, restart, upsert configuration, and Hudi-to-Hudi chaining are neither documented nor covered by an integration suite. + +Materialized-view refresh is the main correctness gap. SDP 4.1 truncates the target before appending the recomputed result, while Hudi's full-table `TRUNCATE TABLE` deletes the table path and reinitializes its properties. Using this path would discard the timeline and could expose an empty or partially rebuilt table if the refresh fails. + +The integration must also address: + +- SDP-created table properties must be persisted and reliably supplied to Hudi batch and streaming writers. +- Streaming restart must coordinate the SDP/Spark checkpoint with Hudi's sink checkpoint metadata. +- A Hudi source used through `readStream.table` must preserve incremental offset semantics. +- SDP must carry Hudi's persisted record-key, precombine, and record-merger settings into the existing upsert path. Deletes remain upsert input rows marked by `_hoodie_is_deleted`; Spark append output mode does not replace this contract. + +### Related work + +- Hudi's Spark 4.1 support ([issue #17663](https://github.com/apache/hudi/issues/17663), [PR #17674](https://github.com/apache/hudi/pull/17674)) provides the runtime baseline, but not an SDP contract. +- DataSource V2 read work ([RFC-98](../rfc-98/rfc-98.md), [issue #15023](https://github.com/apache/hudi/issues/15023)) may improve the catalog read path but is not required for the initial integration. +- The [SQL insert-overwrite discussion](https://github.com/apache/hudi/discussions/13815) shows why refresh semantics must be explicit rather than inferred from an append call. +- [Issue #13973](https://github.com/apache/hudi/issues/13973) shows that a latest-state incremental read cannot fully propagate hard deletes. Flows that require complete delete propagation can use Hudi's existing CDC incremental format. +- [Issue #5537](https://github.com/apache/hudi/issues/5537) tracks named-catalog support. The first version uses `HoodieCatalog` as `spark_catalog`. + +## Motivation + +### Representative use case + +A typical order pipeline looks like this: + +```text +MySQL/PostgreSQL CDC or Kafka + | + v + bronze_order_events (Hudi) + | + v + current_orders (Hudi) + | + v + daily_order_metrics (Hudi) +``` + +The bronze table stores incoming events. The current-state table uses a record key and precombine field to reconcile duplicates and out-of-order events, including delete-marker records. Downstream streaming tables consume changed records incrementally, while batch aggregates can be materialized as views. + +Today this usually means several Spark applications plus custom handling for table creation, write options, checkpoints, dependencies, and restarts. With SDP, the graph lives in one project and the persisted datasets remain regular Hudi tables that other engines can query. + +Related user request issues include: + +- #19281: [multi-dataset incremental processing](https://github.com/apache/hudi/issues/19281) +- #5189: [chained incremental Hudi tables](https://github.com/apache/hudi/issues/5189) +- #13973: [downstream delete propagation](https://github.com/apache/hudi/issues/13973). + +### User value + +1. **Declarative pipelines.** Users define datasets as SQL or DataFrame transformations; SDP manages dependencies, checkpoints, and Hudi writes. +2. **Managed incremental composition.** Hudi already provides incremental reads; SDP composes them into a dependency-aware pipeline, managing execution order, checkpoints, and recovery without separate Spark jobs or hand-written orchestration. +3. **Open outputs.** Each dataset remains a standard Hudi table that can be queried independently of SDP by any Hudi-compatible engine. + +SDP does not make arbitrary joins or aggregates incremental; performance benefits apply only to flows that can process changes incrementally. + +## Goals + +1. Create Hudi tables from SDP datasets through `HoodieCatalog`. +2. Write materialized views with Hudi `bulk_insert` and `insert_overwrite_table` semantics. +3. Write SDP streaming tables through Hudi's Structured Streaming sink. +4. Support a basic incremental Hudi-to-Hudi streaming-table chain. +5. Expose Hudi's existing keyed upsert path through SDP, including record-key, precombine, record-merger, and delete-marker handling. +6. Keep non-SDP Hudi behavior unchanged. + +## Non-Goals + +The first version excludes: +- A complete Hudi DataSource V2 read or streaming-write implementation. +- Cross-table transactions across multiple SDP targets. +- End-to-end exactly-once delivery across an arbitrary external source and Hudi. Guarantees remain scoped to Spark/Hudi micro-batch retries at one target. Review Comment: 🤖 SDP allows multiple `append_flow`s into one streaming table. Since this is a non-goal for v1, how will the integration detect and reject it? If two flows hit the same Hudi target, both the sink's batch-id dedup (same checkpoint identifier) and the lack of a lock provider would cause silent lost/skipped writes rather than a clear error. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,405 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +# RFC-112: Support Spark Declarative Pipelines + +## Proposers + +- @cshuo +- @danny0405 + +## Approvers + +- TBD + +## Status + +Issue: TBD + +## Abstract + +[Spark Declarative Pipelines (SDP)](https://spark.apache.org/docs/4.1.1/declarative-pipelines-programming-guide.html) lets users define tables and transformations while Spark builds the dependency graph and runs the batch and streaming flows. Hudi provides atomic commits, keyed updates, event ordering, deletes, incremental queries, and table services. + +This RFC makes Hudi a supported SDP source and target. It covers catalog-based table creation, materialized views, streaming tables, basic Hudi-to-Hudi incremental reads, and keyed upserts through Hudi's existing writer. + +SDP and Hudi retain their existing responsibilities: + +- SDP owns dataset declarations, dependency analysis, orchestration, flow checkpoints, retries, and execution order. +- Hudi owns table storage, commit atomicity, record reconciliation, incremental read semantics, and table services. + +Users can build the same pipelines today with Spark jobs, Hudi writers or Hudi Streamer, and an external orchestrator. This integration removes much of that glue code while keeping the outputs as ordinary Hudi tables. + +## Background + +### Spark Declarative Pipelines + +Apache Spark 4.1 introduced SDP for declarative batch and streaming ETL. SDP persists two main dataset types: + +- A **materialized view**, which has exactly one batch flow that computes a table. +- A **streaming table**, which has one or more streaming flows that append their output to a table. + +Users define pipelines in Python or SQL. SDP validates dataset references, builds the dataflow graph, runs independent flows in parallel, and stores Structured Streaming checkpoints in the pipeline storage location. + +In the Spark 4.1 implementation, a streaming table flow uses a Structured Streaming append-mode write to `DataStreamWriter.toTable`, and a batch table flow uses append-mode `DataFrameWriter.saveAsTable`. Before running a materialized-view batch flow, SDP truncates an existing target table. See Spark's [`FlowExecution`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala) and [`DatasetManager`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala). + +These table operations are the storage-format integration boundary; this RFC does not add a Hudi-specific execution engine. + +### Relevant Hudi capabilities + +Hudi already provides the required data-plane primitives: + +- `HoodieCatalog` can create and load Hudi tables through Spark's catalog APIs. +- The Hudi batch writer supports bulk insert, insert, upsert, insert overwrite, and table insert overwrite operations. An upsert can carry delete-marker records. +- `HoodieStreamingSink` writes Structured Streaming micro-batches and records checkpoint metadata used to avoid recommitting an already committed batch after retry. +- Hudi's streaming source and incremental query modes can expose records changed since a prior instant. + +The integration exposes these capabilities through SDP's existing table and flow abstractions. Hudi Streamer, the DataFrame writer, Spark SQL DML, and direct Structured Streaming APIs remain unchanged. + +### Current gaps + +`format="hudi"` is not yet an end-to-end supported SDP contract. Some operations may already work, but table creation, refresh, restart, upsert configuration, and Hudi-to-Hudi chaining are neither documented nor covered by an integration suite. + +Materialized-view refresh is the main correctness gap. SDP 4.1 truncates the target before appending the recomputed result, while Hudi's full-table `TRUNCATE TABLE` deletes the table path and reinitializes its properties. Using this path would discard the timeline and could expose an empty or partially rebuilt table if the refresh fails. + +The integration must also address: + +- SDP-created table properties must be persisted and reliably supplied to Hudi batch and streaming writers. +- Streaming restart must coordinate the SDP/Spark checkpoint with Hudi's sink checkpoint metadata. +- A Hudi source used through `readStream.table` must preserve incremental offset semantics. +- SDP must carry Hudi's persisted record-key, precombine, and record-merger settings into the existing upsert path. Deletes remain upsert input rows marked by `_hoodie_is_deleted`; Spark append output mode does not replace this contract. + +### Related work + +- Hudi's Spark 4.1 support ([issue #17663](https://github.com/apache/hudi/issues/17663), [PR #17674](https://github.com/apache/hudi/pull/17674)) provides the runtime baseline, but not an SDP contract. +- DataSource V2 read work ([RFC-98](../rfc-98/rfc-98.md), [issue #15023](https://github.com/apache/hudi/issues/15023)) may improve the catalog read path but is not required for the initial integration. +- The [SQL insert-overwrite discussion](https://github.com/apache/hudi/discussions/13815) shows why refresh semantics must be explicit rather than inferred from an append call. +- [Issue #13973](https://github.com/apache/hudi/issues/13973) shows that a latest-state incremental read cannot fully propagate hard deletes. Flows that require complete delete propagation can use Hudi's existing CDC incremental format. +- [Issue #5537](https://github.com/apache/hudi/issues/5537) tracks named-catalog support. The first version uses `HoodieCatalog` as `spark_catalog`. + +## Motivation + +### Representative use case + +A typical order pipeline looks like this: + +```text +MySQL/PostgreSQL CDC or Kafka + | + v + bronze_order_events (Hudi) + | + v + current_orders (Hudi) + | + v + daily_order_metrics (Hudi) +``` + +The bronze table stores incoming events. The current-state table uses a record key and precombine field to reconcile duplicates and out-of-order events, including delete-marker records. Downstream streaming tables consume changed records incrementally, while batch aggregates can be materialized as views. + +Today this usually means several Spark applications plus custom handling for table creation, write options, checkpoints, dependencies, and restarts. With SDP, the graph lives in one project and the persisted datasets remain regular Hudi tables that other engines can query. + +Related user request issues include: + +- #19281: [multi-dataset incremental processing](https://github.com/apache/hudi/issues/19281) +- #5189: [chained incremental Hudi tables](https://github.com/apache/hudi/issues/5189) +- #13973: [downstream delete propagation](https://github.com/apache/hudi/issues/13973). + +### User value + +1. **Declarative pipelines.** Users define datasets as SQL or DataFrame transformations; SDP manages dependencies, checkpoints, and Hudi writes. +2. **Managed incremental composition.** Hudi already provides incremental reads; SDP composes them into a dependency-aware pipeline, managing execution order, checkpoints, and recovery without separate Spark jobs or hand-written orchestration. +3. **Open outputs.** Each dataset remains a standard Hudi table that can be queried independently of SDP by any Hudi-compatible engine. + +SDP does not make arbitrary joins or aggregates incremental; performance benefits apply only to flows that can process changes incrementally. + +## Goals + +1. Create Hudi tables from SDP datasets through `HoodieCatalog`. +2. Write materialized views with Hudi `bulk_insert` and `insert_overwrite_table` semantics. +3. Write SDP streaming tables through Hudi's Structured Streaming sink. +4. Support a basic incremental Hudi-to-Hudi streaming-table chain. +5. Expose Hudi's existing keyed upsert path through SDP, including record-key, precombine, record-merger, and delete-marker handling. +6. Keep non-SDP Hudi behavior unchanged. + +## Non-Goals + +The first version excludes: +- A complete Hudi DataSource V2 read or streaming-write implementation. +- Cross-table transactions across multiple SDP targets. +- End-to-end exactly-once delivery across an arbitrary external source and Hudi. Guarantees remain scoped to Spark/Hudi micro-batch retries at one target. +- Multiple concurrent append flows into one Hudi target in the initial phase. +- Arbitrary named catalogs for `HoodieCatalog`; the first version uses Hudi as `spark_catalog`. + +## Proposed User Experience + +The examples below implement the representative order pipeline in dependency order. Property names will use the canonical Spark 4.1 SDP and Hudi configuration names. + +### Pipeline configuration + +An SDP project configures the Hudi extension and catalog like any other Hudi Spark application: + +```yaml +name: hudi_orders_pipeline +libraries: + - glob: + include: transformations/** +storage: s3://pipelines/checkpoints/hudi_orders_pipeline +catalog: spark_catalog +database: lakehouse +configuration: + spark.sql.extensions: org.apache.spark.sql.hudi.HoodieSparkSessionExtension + spark.sql.catalog.spark_catalog: org.apache.spark.sql.hudi.catalog.HoodieCatalog +``` + +The SQL below assumes that `order_events_source` is an external streaming source containing parsed Kafka or database CDC events. Each `CREATE ... AS SELECT` statement defines both the target dataset and its single flow. + +### Streaming table with an append flow + +```sql +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'insert' +) +AS +SELECT * +FROM STREAM order_events_source; +``` + +SDP owns the checkpoint location and trigger; Hudi commits each micro-batch. + +### Streaming table with keyed upserts + +```sql +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'upsert', + 'hoodie.datasource.write.recordkey.field' = 'order_id', + 'hoodie.datasource.write.precombine.field' = 'event_sequence', + 'hoodie.datasource.write.partitionpath.field' = 'event_date' +) +AS +SELECT + *, + operation = 'DELETE' AS _hoodie_is_deleted +FROM STREAM bronze_order_events; +``` + +Spark append output mode only controls how each micro-batch reaches the sink; it does not make the Hudi target append-only. `HoodieStreamingSink` still runs the configured Hudi operation, which defaults to `upsert`. + +For an existing Hudi table, SDP loads the record key, precombine field, partition fields, and record-merger settings from the table configuration. For a new SDP table, these settings are declared as table properties and persisted by Hudi. SDP does not implement its own deduplication, ordering, merge, or delete path. It validates the schema and configuration, then passes each micro-batch to the existing Hudi writer. + +### Batch materialized view + +```sql +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'COPY_ON_WRITE' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and may use `bulk_insert`; later runs replace the result with `insert_overwrite_table`. Refresh must not delete and recreate the table. + +## Semantics + +### Dataset-to-Hudi operation mapping + +| SDP dataset/flow | Spark 4.1 execution shape | Proposed Hudi semantics | +| ----------------------------------- | --------------------------------------------- | --------------------------------------------------------- | +| New materialized view | Batch append to newly created table | Initial bulk insert or insert | +| Existing materialized view | Full batch recomputation | Hudi `insert_overwrite_table`; no physical table deletion | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured at table level | +| Hudi upsert target | Append-mode micro-batches | Existing Hudi `upsert` | +| Hudi streaming source | Structured Streaming read with SDP checkpoint | Basic Hudi latest-state incremental source | + +### Materialized-view write mapping + +Initial materialization maps to Hudi `bulk_insert` or `insert`; refresh maps to `insert_overwrite_table`. Because Spark 4.1 implements refresh as truncate followed by append, the integration must select the overwrite path without invoking Hudi's destructive full-table truncate. Where to intercept that path remains an open implementation question; a general table-format replace API is not part of this RFC. + +### Streaming commit and restart behavior + +SDP supplies the checkpoint location and trigger. Hudi keeps its existing sink commit and retry behavior; no second checkpoint tree or commit protocol is introduced. The first version allows one streaming flow per Hudi target. Review Comment: 🤖 HoodieStreamingSink's dedup (`canSkipBatch`) compares the incoming Spark batchId against the last batchId stored in commit metadata under `hoodie.datasource.write.streaming.checkpoint.identifier`. If SDP resets the checkpoint — `--full-refresh` on a streaming table, or a user changing the pipeline `storage` path — batch ids restart from 0 and the sink would silently skip every micro-batch until it passes the old high-water mark. Could the RFC spell out how the checkpoint identifier is derived (per flow + checkpoint location?) and what full refresh of a *streaming* table maps to on the Hudi side, given it also goes through the truncate path? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,405 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +# RFC-112: Support Spark Declarative Pipelines + +## Proposers + +- @cshuo +- @danny0405 + +## Approvers + +- TBD + +## Status + +Issue: TBD + +## Abstract + +[Spark Declarative Pipelines (SDP)](https://spark.apache.org/docs/4.1.1/declarative-pipelines-programming-guide.html) lets users define tables and transformations while Spark builds the dependency graph and runs the batch and streaming flows. Hudi provides atomic commits, keyed updates, event ordering, deletes, incremental queries, and table services. + +This RFC makes Hudi a supported SDP source and target. It covers catalog-based table creation, materialized views, streaming tables, basic Hudi-to-Hudi incremental reads, and keyed upserts through Hudi's existing writer. + +SDP and Hudi retain their existing responsibilities: + +- SDP owns dataset declarations, dependency analysis, orchestration, flow checkpoints, retries, and execution order. +- Hudi owns table storage, commit atomicity, record reconciliation, incremental read semantics, and table services. + +Users can build the same pipelines today with Spark jobs, Hudi writers or Hudi Streamer, and an external orchestrator. This integration removes much of that glue code while keeping the outputs as ordinary Hudi tables. + +## Background + +### Spark Declarative Pipelines + +Apache Spark 4.1 introduced SDP for declarative batch and streaming ETL. SDP persists two main dataset types: + +- A **materialized view**, which has exactly one batch flow that computes a table. +- A **streaming table**, which has one or more streaming flows that append their output to a table. + +Users define pipelines in Python or SQL. SDP validates dataset references, builds the dataflow graph, runs independent flows in parallel, and stores Structured Streaming checkpoints in the pipeline storage location. + +In the Spark 4.1 implementation, a streaming table flow uses a Structured Streaming append-mode write to `DataStreamWriter.toTable`, and a batch table flow uses append-mode `DataFrameWriter.saveAsTable`. Before running a materialized-view batch flow, SDP truncates an existing target table. See Spark's [`FlowExecution`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala) and [`DatasetManager`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala). + +These table operations are the storage-format integration boundary; this RFC does not add a Hudi-specific execution engine. + +### Relevant Hudi capabilities + +Hudi already provides the required data-plane primitives: + +- `HoodieCatalog` can create and load Hudi tables through Spark's catalog APIs. +- The Hudi batch writer supports bulk insert, insert, upsert, insert overwrite, and table insert overwrite operations. An upsert can carry delete-marker records. +- `HoodieStreamingSink` writes Structured Streaming micro-batches and records checkpoint metadata used to avoid recommitting an already committed batch after retry. +- Hudi's streaming source and incremental query modes can expose records changed since a prior instant. + +The integration exposes these capabilities through SDP's existing table and flow abstractions. Hudi Streamer, the DataFrame writer, Spark SQL DML, and direct Structured Streaming APIs remain unchanged. + +### Current gaps + +`format="hudi"` is not yet an end-to-end supported SDP contract. Some operations may already work, but table creation, refresh, restart, upsert configuration, and Hudi-to-Hudi chaining are neither documented nor covered by an integration suite. + +Materialized-view refresh is the main correctness gap. SDP 4.1 truncates the target before appending the recomputed result, while Hudi's full-table `TRUNCATE TABLE` deletes the table path and reinitializes its properties. Using this path would discard the timeline and could expose an empty or partially rebuilt table if the refresh fails. + +The integration must also address: + +- SDP-created table properties must be persisted and reliably supplied to Hudi batch and streaming writers. +- Streaming restart must coordinate the SDP/Spark checkpoint with Hudi's sink checkpoint metadata. +- A Hudi source used through `readStream.table` must preserve incremental offset semantics. +- SDP must carry Hudi's persisted record-key, precombine, and record-merger settings into the existing upsert path. Deletes remain upsert input rows marked by `_hoodie_is_deleted`; Spark append output mode does not replace this contract. + +### Related work + +- Hudi's Spark 4.1 support ([issue #17663](https://github.com/apache/hudi/issues/17663), [PR #17674](https://github.com/apache/hudi/pull/17674)) provides the runtime baseline, but not an SDP contract. +- DataSource V2 read work ([RFC-98](../rfc-98/rfc-98.md), [issue #15023](https://github.com/apache/hudi/issues/15023)) may improve the catalog read path but is not required for the initial integration. +- The [SQL insert-overwrite discussion](https://github.com/apache/hudi/discussions/13815) shows why refresh semantics must be explicit rather than inferred from an append call. +- [Issue #13973](https://github.com/apache/hudi/issues/13973) shows that a latest-state incremental read cannot fully propagate hard deletes. Flows that require complete delete propagation can use Hudi's existing CDC incremental format. +- [Issue #5537](https://github.com/apache/hudi/issues/5537) tracks named-catalog support. The first version uses `HoodieCatalog` as `spark_catalog`. + +## Motivation + +### Representative use case + +A typical order pipeline looks like this: + +```text +MySQL/PostgreSQL CDC or Kafka + | + v + bronze_order_events (Hudi) + | + v + current_orders (Hudi) + | + v + daily_order_metrics (Hudi) +``` + +The bronze table stores incoming events. The current-state table uses a record key and precombine field to reconcile duplicates and out-of-order events, including delete-marker records. Downstream streaming tables consume changed records incrementally, while batch aggregates can be materialized as views. + +Today this usually means several Spark applications plus custom handling for table creation, write options, checkpoints, dependencies, and restarts. With SDP, the graph lives in one project and the persisted datasets remain regular Hudi tables that other engines can query. + +Related user request issues include: + +- #19281: [multi-dataset incremental processing](https://github.com/apache/hudi/issues/19281) +- #5189: [chained incremental Hudi tables](https://github.com/apache/hudi/issues/5189) +- #13973: [downstream delete propagation](https://github.com/apache/hudi/issues/13973). + +### User value + +1. **Declarative pipelines.** Users define datasets as SQL or DataFrame transformations; SDP manages dependencies, checkpoints, and Hudi writes. +2. **Managed incremental composition.** Hudi already provides incremental reads; SDP composes them into a dependency-aware pipeline, managing execution order, checkpoints, and recovery without separate Spark jobs or hand-written orchestration. +3. **Open outputs.** Each dataset remains a standard Hudi table that can be queried independently of SDP by any Hudi-compatible engine. + +SDP does not make arbitrary joins or aggregates incremental; performance benefits apply only to flows that can process changes incrementally. + +## Goals + +1. Create Hudi tables from SDP datasets through `HoodieCatalog`. +2. Write materialized views with Hudi `bulk_insert` and `insert_overwrite_table` semantics. +3. Write SDP streaming tables through Hudi's Structured Streaming sink. +4. Support a basic incremental Hudi-to-Hudi streaming-table chain. +5. Expose Hudi's existing keyed upsert path through SDP, including record-key, precombine, record-merger, and delete-marker handling. +6. Keep non-SDP Hudi behavior unchanged. + +## Non-Goals + +The first version excludes: +- A complete Hudi DataSource V2 read or streaming-write implementation. +- Cross-table transactions across multiple SDP targets. +- End-to-end exactly-once delivery across an arbitrary external source and Hudi. Guarantees remain scoped to Spark/Hudi micro-batch retries at one target. +- Multiple concurrent append flows into one Hudi target in the initial phase. +- Arbitrary named catalogs for `HoodieCatalog`; the first version uses Hudi as `spark_catalog`. + +## Proposed User Experience + +The examples below implement the representative order pipeline in dependency order. Property names will use the canonical Spark 4.1 SDP and Hudi configuration names. + +### Pipeline configuration + +An SDP project configures the Hudi extension and catalog like any other Hudi Spark application: + +```yaml +name: hudi_orders_pipeline +libraries: + - glob: + include: transformations/** +storage: s3://pipelines/checkpoints/hudi_orders_pipeline +catalog: spark_catalog +database: lakehouse +configuration: + spark.sql.extensions: org.apache.spark.sql.hudi.HoodieSparkSessionExtension + spark.sql.catalog.spark_catalog: org.apache.spark.sql.hudi.catalog.HoodieCatalog +``` + +The SQL below assumes that `order_events_source` is an external streaming source containing parsed Kafka or database CDC events. Each `CREATE ... AS SELECT` statement defines both the target dataset and its single flow. + +### Streaming table with an append flow + +```sql +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'insert' +) +AS +SELECT * +FROM STREAM order_events_source; +``` + +SDP owns the checkpoint location and trigger; Hudi commits each micro-batch. + +### Streaming table with keyed upserts + +```sql +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'upsert', + 'hoodie.datasource.write.recordkey.field' = 'order_id', + 'hoodie.datasource.write.precombine.field' = 'event_sequence', + 'hoodie.datasource.write.partitionpath.field' = 'event_date' +) +AS +SELECT + *, + operation = 'DELETE' AS _hoodie_is_deleted +FROM STREAM bronze_order_events; +``` + +Spark append output mode only controls how each micro-batch reaches the sink; it does not make the Hudi target append-only. `HoodieStreamingSink` still runs the configured Hudi operation, which defaults to `upsert`. + +For an existing Hudi table, SDP loads the record key, precombine field, partition fields, and record-merger settings from the table configuration. For a new SDP table, these settings are declared as table properties and persisted by Hudi. SDP does not implement its own deduplication, ordering, merge, or delete path. It validates the schema and configuration, then passes each micro-batch to the existing Hudi writer. + +### Batch materialized view + +```sql +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'COPY_ON_WRITE' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and may use `bulk_insert`; later runs replace the result with `insert_overwrite_table`. Refresh must not delete and recreate the table. + +## Semantics + +### Dataset-to-Hudi operation mapping + +| SDP dataset/flow | Spark 4.1 execution shape | Proposed Hudi semantics | +| ----------------------------------- | --------------------------------------------- | --------------------------------------------------------- | +| New materialized view | Batch append to newly created table | Initial bulk insert or insert | +| Existing materialized view | Full batch recomputation | Hudi `insert_overwrite_table`; no physical table deletion | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured at table level | +| Hudi upsert target | Append-mode micro-batches | Existing Hudi `upsert` | +| Hudi streaming source | Structured Streaming read with SDP checkpoint | Basic Hudi latest-state incremental source | Review Comment: 🤖 For `FROM STREAM current_orders`, how does the user supply Hudi reader options (start instant, `hoodie.datasource.query.incremental.format=cdc` for delete propagation, fallback-to-full-scan)? SDP SQL has no options clause on a `STREAM` reference, so this presumably has to be table properties or Python `readStream.option(...)`. Also, does `readStream.table()` on a `HoodieCatalog`-loaded table actually resolve to `HoodieStreamSource` today, or is that a Phase 3 unknown? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,405 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +# RFC-112: Support Spark Declarative Pipelines + +## Proposers + +- @cshuo +- @danny0405 + +## Approvers + +- TBD + +## Status + +Issue: TBD + +## Abstract + +[Spark Declarative Pipelines (SDP)](https://spark.apache.org/docs/4.1.1/declarative-pipelines-programming-guide.html) lets users define tables and transformations while Spark builds the dependency graph and runs the batch and streaming flows. Hudi provides atomic commits, keyed updates, event ordering, deletes, incremental queries, and table services. + +This RFC makes Hudi a supported SDP source and target. It covers catalog-based table creation, materialized views, streaming tables, basic Hudi-to-Hudi incremental reads, and keyed upserts through Hudi's existing writer. + +SDP and Hudi retain their existing responsibilities: + +- SDP owns dataset declarations, dependency analysis, orchestration, flow checkpoints, retries, and execution order. +- Hudi owns table storage, commit atomicity, record reconciliation, incremental read semantics, and table services. + +Users can build the same pipelines today with Spark jobs, Hudi writers or Hudi Streamer, and an external orchestrator. This integration removes much of that glue code while keeping the outputs as ordinary Hudi tables. + +## Background + +### Spark Declarative Pipelines + +Apache Spark 4.1 introduced SDP for declarative batch and streaming ETL. SDP persists two main dataset types: + +- A **materialized view**, which has exactly one batch flow that computes a table. +- A **streaming table**, which has one or more streaming flows that append their output to a table. + +Users define pipelines in Python or SQL. SDP validates dataset references, builds the dataflow graph, runs independent flows in parallel, and stores Structured Streaming checkpoints in the pipeline storage location. + +In the Spark 4.1 implementation, a streaming table flow uses a Structured Streaming append-mode write to `DataStreamWriter.toTable`, and a batch table flow uses append-mode `DataFrameWriter.saveAsTable`. Before running a materialized-view batch flow, SDP truncates an existing target table. See Spark's [`FlowExecution`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala) and [`DatasetManager`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala). + +These table operations are the storage-format integration boundary; this RFC does not add a Hudi-specific execution engine. + +### Relevant Hudi capabilities + +Hudi already provides the required data-plane primitives: + +- `HoodieCatalog` can create and load Hudi tables through Spark's catalog APIs. +- The Hudi batch writer supports bulk insert, insert, upsert, insert overwrite, and table insert overwrite operations. An upsert can carry delete-marker records. +- `HoodieStreamingSink` writes Structured Streaming micro-batches and records checkpoint metadata used to avoid recommitting an already committed batch after retry. +- Hudi's streaming source and incremental query modes can expose records changed since a prior instant. + +The integration exposes these capabilities through SDP's existing table and flow abstractions. Hudi Streamer, the DataFrame writer, Spark SQL DML, and direct Structured Streaming APIs remain unchanged. + +### Current gaps + +`format="hudi"` is not yet an end-to-end supported SDP contract. Some operations may already work, but table creation, refresh, restart, upsert configuration, and Hudi-to-Hudi chaining are neither documented nor covered by an integration suite. + +Materialized-view refresh is the main correctness gap. SDP 4.1 truncates the target before appending the recomputed result, while Hudi's full-table `TRUNCATE TABLE` deletes the table path and reinitializes its properties. Using this path would discard the timeline and could expose an empty or partially rebuilt table if the refresh fails. + +The integration must also address: + +- SDP-created table properties must be persisted and reliably supplied to Hudi batch and streaming writers. +- Streaming restart must coordinate the SDP/Spark checkpoint with Hudi's sink checkpoint metadata. +- A Hudi source used through `readStream.table` must preserve incremental offset semantics. +- SDP must carry Hudi's persisted record-key, precombine, and record-merger settings into the existing upsert path. Deletes remain upsert input rows marked by `_hoodie_is_deleted`; Spark append output mode does not replace this contract. + +### Related work + +- Hudi's Spark 4.1 support ([issue #17663](https://github.com/apache/hudi/issues/17663), [PR #17674](https://github.com/apache/hudi/pull/17674)) provides the runtime baseline, but not an SDP contract. +- DataSource V2 read work ([RFC-98](../rfc-98/rfc-98.md), [issue #15023](https://github.com/apache/hudi/issues/15023)) may improve the catalog read path but is not required for the initial integration. +- The [SQL insert-overwrite discussion](https://github.com/apache/hudi/discussions/13815) shows why refresh semantics must be explicit rather than inferred from an append call. +- [Issue #13973](https://github.com/apache/hudi/issues/13973) shows that a latest-state incremental read cannot fully propagate hard deletes. Flows that require complete delete propagation can use Hudi's existing CDC incremental format. +- [Issue #5537](https://github.com/apache/hudi/issues/5537) tracks named-catalog support. The first version uses `HoodieCatalog` as `spark_catalog`. + +## Motivation + +### Representative use case + +A typical order pipeline looks like this: + +```text +MySQL/PostgreSQL CDC or Kafka + | + v + bronze_order_events (Hudi) + | + v + current_orders (Hudi) + | + v + daily_order_metrics (Hudi) +``` + +The bronze table stores incoming events. The current-state table uses a record key and precombine field to reconcile duplicates and out-of-order events, including delete-marker records. Downstream streaming tables consume changed records incrementally, while batch aggregates can be materialized as views. + +Today this usually means several Spark applications plus custom handling for table creation, write options, checkpoints, dependencies, and restarts. With SDP, the graph lives in one project and the persisted datasets remain regular Hudi tables that other engines can query. + +Related user request issues include: + +- #19281: [multi-dataset incremental processing](https://github.com/apache/hudi/issues/19281) +- #5189: [chained incremental Hudi tables](https://github.com/apache/hudi/issues/5189) +- #13973: [downstream delete propagation](https://github.com/apache/hudi/issues/13973). + +### User value + +1. **Declarative pipelines.** Users define datasets as SQL or DataFrame transformations; SDP manages dependencies, checkpoints, and Hudi writes. +2. **Managed incremental composition.** Hudi already provides incremental reads; SDP composes them into a dependency-aware pipeline, managing execution order, checkpoints, and recovery without separate Spark jobs or hand-written orchestration. +3. **Open outputs.** Each dataset remains a standard Hudi table that can be queried independently of SDP by any Hudi-compatible engine. + +SDP does not make arbitrary joins or aggregates incremental; performance benefits apply only to flows that can process changes incrementally. + +## Goals + +1. Create Hudi tables from SDP datasets through `HoodieCatalog`. +2. Write materialized views with Hudi `bulk_insert` and `insert_overwrite_table` semantics. +3. Write SDP streaming tables through Hudi's Structured Streaming sink. +4. Support a basic incremental Hudi-to-Hudi streaming-table chain. +5. Expose Hudi's existing keyed upsert path through SDP, including record-key, precombine, record-merger, and delete-marker handling. +6. Keep non-SDP Hudi behavior unchanged. + +## Non-Goals + +The first version excludes: +- A complete Hudi DataSource V2 read or streaming-write implementation. +- Cross-table transactions across multiple SDP targets. +- End-to-end exactly-once delivery across an arbitrary external source and Hudi. Guarantees remain scoped to Spark/Hudi micro-batch retries at one target. +- Multiple concurrent append flows into one Hudi target in the initial phase. +- Arbitrary named catalogs for `HoodieCatalog`; the first version uses Hudi as `spark_catalog`. + +## Proposed User Experience + +The examples below implement the representative order pipeline in dependency order. Property names will use the canonical Spark 4.1 SDP and Hudi configuration names. + +### Pipeline configuration + +An SDP project configures the Hudi extension and catalog like any other Hudi Spark application: + +```yaml +name: hudi_orders_pipeline +libraries: + - glob: + include: transformations/** +storage: s3://pipelines/checkpoints/hudi_orders_pipeline +catalog: spark_catalog +database: lakehouse +configuration: + spark.sql.extensions: org.apache.spark.sql.hudi.HoodieSparkSessionExtension + spark.sql.catalog.spark_catalog: org.apache.spark.sql.hudi.catalog.HoodieCatalog +``` + +The SQL below assumes that `order_events_source` is an external streaming source containing parsed Kafka or database CDC events. Each `CREATE ... AS SELECT` statement defines both the target dataset and its single flow. + +### Streaming table with an append flow + +```sql +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'insert' +) +AS +SELECT * +FROM STREAM order_events_source; +``` + +SDP owns the checkpoint location and trigger; Hudi commits each micro-batch. + +### Streaming table with keyed upserts + +```sql +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'upsert', + 'hoodie.datasource.write.recordkey.field' = 'order_id', + 'hoodie.datasource.write.precombine.field' = 'event_sequence', + 'hoodie.datasource.write.partitionpath.field' = 'event_date' +) +AS +SELECT + *, + operation = 'DELETE' AS _hoodie_is_deleted +FROM STREAM bronze_order_events; +``` + +Spark append output mode only controls how each micro-batch reaches the sink; it does not make the Hudi target append-only. `HoodieStreamingSink` still runs the configured Hudi operation, which defaults to `upsert`. + +For an existing Hudi table, SDP loads the record key, precombine field, partition fields, and record-merger settings from the table configuration. For a new SDP table, these settings are declared as table properties and persisted by Hudi. SDP does not implement its own deduplication, ordering, merge, or delete path. It validates the schema and configuration, then passes each micro-batch to the existing Hudi writer. + +### Batch materialized view + +```sql +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'COPY_ON_WRITE' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and may use `bulk_insert`; later runs replace the result with `insert_overwrite_table`. Refresh must not delete and recreate the table. + +## Semantics + +### Dataset-to-Hudi operation mapping + +| SDP dataset/flow | Spark 4.1 execution shape | Proposed Hudi semantics | +| ----------------------------------- | --------------------------------------------- | --------------------------------------------------------- | +| New materialized view | Batch append to newly created table | Initial bulk insert or insert | +| Existing materialized view | Full batch recomputation | Hudi `insert_overwrite_table`; no physical table deletion | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured at table level | +| Hudi upsert target | Append-mode micro-batches | Existing Hudi `upsert` | +| Hudi streaming source | Structured Streaming read with SDP checkpoint | Basic Hudi latest-state incremental source | + +### Materialized-view write mapping + +Initial materialization maps to Hudi `bulk_insert` or `insert`; refresh maps to `insert_overwrite_table`. Because Spark 4.1 implements refresh as truncate followed by append, the integration must select the overwrite path without invoking Hudi's destructive full-table truncate. Where to intercept that path remains an open implementation question; a general table-format replace API is not part of this RFC. + +### Streaming commit and restart behavior + +SDP supplies the checkpoint location and trigger. Hudi keeps its existing sink commit and retry behavior; no second checkpoint tree or commit protocol is introduced. The first version allows one streaming flow per Hudi target. + +### Configuration ownership and precedence + +Configuration is split by ownership: + +- **Table identity:** table type, record key, partition fields, key generator, and record-merger settings. Hudi persists these values; SDP loads them for existing tables, and a pipeline run cannot silently change them. +- **Write behavior:** operation and writer tuning. These are resolved per write and checked against the persisted table configuration. +- **Flow execution:** checkpoint location, trigger, retry, and flow identity. SDP owns these values. + +Hudi options are supplied as SDP table properties and normalized to canonical names. Table properties take precedence over supported `spark.hoodie.*` session settings and Hudi defaults. + +Invalid properties should fail during SDP `dry-run` when possible, and otherwise before the first write. + +### Schema and partition evolution + +SDP follows Hudi's existing schema and partition evolution rules. + +- Hudi-supported additive schema changes may be accepted for streaming tables. +- Dropping, renaming, or changing columns follows Hudi's configured schema evolution behavior. +- Record-key, precombine, and partition columns must exist in the target schema. Review Comment: 🤖 Which component enforces this? SDP recomputes the MV schema each run and, I believe, issues ALTER TABLE to reconcile before writing. Is the check in HoodieCatalog.alterTable, in the writer's schema reconciliation (`hoodie.datasource.write.reconcile.schema` / schema-on-read), or both? Worth pinning down so the 'fails before replacing the prior snapshot' guarantee isn't dependent on a config the user may have set differently. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,405 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +# RFC-112: Support Spark Declarative Pipelines + +## Proposers + +- @cshuo +- @danny0405 + +## Approvers + +- TBD + +## Status + +Issue: TBD + +## Abstract + +[Spark Declarative Pipelines (SDP)](https://spark.apache.org/docs/4.1.1/declarative-pipelines-programming-guide.html) lets users define tables and transformations while Spark builds the dependency graph and runs the batch and streaming flows. Hudi provides atomic commits, keyed updates, event ordering, deletes, incremental queries, and table services. + +This RFC makes Hudi a supported SDP source and target. It covers catalog-based table creation, materialized views, streaming tables, basic Hudi-to-Hudi incremental reads, and keyed upserts through Hudi's existing writer. + +SDP and Hudi retain their existing responsibilities: + +- SDP owns dataset declarations, dependency analysis, orchestration, flow checkpoints, retries, and execution order. +- Hudi owns table storage, commit atomicity, record reconciliation, incremental read semantics, and table services. + +Users can build the same pipelines today with Spark jobs, Hudi writers or Hudi Streamer, and an external orchestrator. This integration removes much of that glue code while keeping the outputs as ordinary Hudi tables. + +## Background + +### Spark Declarative Pipelines + +Apache Spark 4.1 introduced SDP for declarative batch and streaming ETL. SDP persists two main dataset types: + +- A **materialized view**, which has exactly one batch flow that computes a table. +- A **streaming table**, which has one or more streaming flows that append their output to a table. + +Users define pipelines in Python or SQL. SDP validates dataset references, builds the dataflow graph, runs independent flows in parallel, and stores Structured Streaming checkpoints in the pipeline storage location. + +In the Spark 4.1 implementation, a streaming table flow uses a Structured Streaming append-mode write to `DataStreamWriter.toTable`, and a batch table flow uses append-mode `DataFrameWriter.saveAsTable`. Before running a materialized-view batch flow, SDP truncates an existing target table. See Spark's [`FlowExecution`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala) and [`DatasetManager`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala). + +These table operations are the storage-format integration boundary; this RFC does not add a Hudi-specific execution engine. + +### Relevant Hudi capabilities + +Hudi already provides the required data-plane primitives: + +- `HoodieCatalog` can create and load Hudi tables through Spark's catalog APIs. +- The Hudi batch writer supports bulk insert, insert, upsert, insert overwrite, and table insert overwrite operations. An upsert can carry delete-marker records. +- `HoodieStreamingSink` writes Structured Streaming micro-batches and records checkpoint metadata used to avoid recommitting an already committed batch after retry. +- Hudi's streaming source and incremental query modes can expose records changed since a prior instant. + +The integration exposes these capabilities through SDP's existing table and flow abstractions. Hudi Streamer, the DataFrame writer, Spark SQL DML, and direct Structured Streaming APIs remain unchanged. + +### Current gaps + +`format="hudi"` is not yet an end-to-end supported SDP contract. Some operations may already work, but table creation, refresh, restart, upsert configuration, and Hudi-to-Hudi chaining are neither documented nor covered by an integration suite. + +Materialized-view refresh is the main correctness gap. SDP 4.1 truncates the target before appending the recomputed result, while Hudi's full-table `TRUNCATE TABLE` deletes the table path and reinitializes its properties. Using this path would discard the timeline and could expose an empty or partially rebuilt table if the refresh fails. + +The integration must also address: + +- SDP-created table properties must be persisted and reliably supplied to Hudi batch and streaming writers. +- Streaming restart must coordinate the SDP/Spark checkpoint with Hudi's sink checkpoint metadata. +- A Hudi source used through `readStream.table` must preserve incremental offset semantics. +- SDP must carry Hudi's persisted record-key, precombine, and record-merger settings into the existing upsert path. Deletes remain upsert input rows marked by `_hoodie_is_deleted`; Spark append output mode does not replace this contract. + +### Related work + +- Hudi's Spark 4.1 support ([issue #17663](https://github.com/apache/hudi/issues/17663), [PR #17674](https://github.com/apache/hudi/pull/17674)) provides the runtime baseline, but not an SDP contract. +- DataSource V2 read work ([RFC-98](../rfc-98/rfc-98.md), [issue #15023](https://github.com/apache/hudi/issues/15023)) may improve the catalog read path but is not required for the initial integration. +- The [SQL insert-overwrite discussion](https://github.com/apache/hudi/discussions/13815) shows why refresh semantics must be explicit rather than inferred from an append call. +- [Issue #13973](https://github.com/apache/hudi/issues/13973) shows that a latest-state incremental read cannot fully propagate hard deletes. Flows that require complete delete propagation can use Hudi's existing CDC incremental format. +- [Issue #5537](https://github.com/apache/hudi/issues/5537) tracks named-catalog support. The first version uses `HoodieCatalog` as `spark_catalog`. + +## Motivation + +### Representative use case + +A typical order pipeline looks like this: + +```text +MySQL/PostgreSQL CDC or Kafka + | + v + bronze_order_events (Hudi) + | + v + current_orders (Hudi) + | + v + daily_order_metrics (Hudi) +``` + +The bronze table stores incoming events. The current-state table uses a record key and precombine field to reconcile duplicates and out-of-order events, including delete-marker records. Downstream streaming tables consume changed records incrementally, while batch aggregates can be materialized as views. + +Today this usually means several Spark applications plus custom handling for table creation, write options, checkpoints, dependencies, and restarts. With SDP, the graph lives in one project and the persisted datasets remain regular Hudi tables that other engines can query. + +Related user request issues include: + +- #19281: [multi-dataset incremental processing](https://github.com/apache/hudi/issues/19281) +- #5189: [chained incremental Hudi tables](https://github.com/apache/hudi/issues/5189) +- #13973: [downstream delete propagation](https://github.com/apache/hudi/issues/13973). + +### User value + +1. **Declarative pipelines.** Users define datasets as SQL or DataFrame transformations; SDP manages dependencies, checkpoints, and Hudi writes. +2. **Managed incremental composition.** Hudi already provides incremental reads; SDP composes them into a dependency-aware pipeline, managing execution order, checkpoints, and recovery without separate Spark jobs or hand-written orchestration. +3. **Open outputs.** Each dataset remains a standard Hudi table that can be queried independently of SDP by any Hudi-compatible engine. + +SDP does not make arbitrary joins or aggregates incremental; performance benefits apply only to flows that can process changes incrementally. + +## Goals + +1. Create Hudi tables from SDP datasets through `HoodieCatalog`. +2. Write materialized views with Hudi `bulk_insert` and `insert_overwrite_table` semantics. +3. Write SDP streaming tables through Hudi's Structured Streaming sink. +4. Support a basic incremental Hudi-to-Hudi streaming-table chain. +5. Expose Hudi's existing keyed upsert path through SDP, including record-key, precombine, record-merger, and delete-marker handling. +6. Keep non-SDP Hudi behavior unchanged. + +## Non-Goals + +The first version excludes: +- A complete Hudi DataSource V2 read or streaming-write implementation. +- Cross-table transactions across multiple SDP targets. +- End-to-end exactly-once delivery across an arbitrary external source and Hudi. Guarantees remain scoped to Spark/Hudi micro-batch retries at one target. +- Multiple concurrent append flows into one Hudi target in the initial phase. +- Arbitrary named catalogs for `HoodieCatalog`; the first version uses Hudi as `spark_catalog`. + +## Proposed User Experience + +The examples below implement the representative order pipeline in dependency order. Property names will use the canonical Spark 4.1 SDP and Hudi configuration names. + +### Pipeline configuration + +An SDP project configures the Hudi extension and catalog like any other Hudi Spark application: + +```yaml +name: hudi_orders_pipeline +libraries: + - glob: + include: transformations/** +storage: s3://pipelines/checkpoints/hudi_orders_pipeline +catalog: spark_catalog +database: lakehouse +configuration: + spark.sql.extensions: org.apache.spark.sql.hudi.HoodieSparkSessionExtension + spark.sql.catalog.spark_catalog: org.apache.spark.sql.hudi.catalog.HoodieCatalog +``` + +The SQL below assumes that `order_events_source` is an external streaming source containing parsed Kafka or database CDC events. Each `CREATE ... AS SELECT` statement defines both the target dataset and its single flow. + +### Streaming table with an append flow + +```sql +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'insert' +) +AS +SELECT * +FROM STREAM order_events_source; +``` + +SDP owns the checkpoint location and trigger; Hudi commits each micro-batch. + +### Streaming table with keyed upserts + +```sql +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'upsert', + 'hoodie.datasource.write.recordkey.field' = 'order_id', + 'hoodie.datasource.write.precombine.field' = 'event_sequence', + 'hoodie.datasource.write.partitionpath.field' = 'event_date' +) +AS +SELECT + *, + operation = 'DELETE' AS _hoodie_is_deleted +FROM STREAM bronze_order_events; +``` + +Spark append output mode only controls how each micro-batch reaches the sink; it does not make the Hudi target append-only. `HoodieStreamingSink` still runs the configured Hudi operation, which defaults to `upsert`. + +For an existing Hudi table, SDP loads the record key, precombine field, partition fields, and record-merger settings from the table configuration. For a new SDP table, these settings are declared as table properties and persisted by Hudi. SDP does not implement its own deduplication, ordering, merge, or delete path. It validates the schema and configuration, then passes each micro-batch to the existing Hudi writer. + +### Batch materialized view + +```sql +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'COPY_ON_WRITE' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and may use `bulk_insert`; later runs replace the result with `insert_overwrite_table`. Refresh must not delete and recreate the table. + +## Semantics + +### Dataset-to-Hudi operation mapping + +| SDP dataset/flow | Spark 4.1 execution shape | Proposed Hudi semantics | +| ----------------------------------- | --------------------------------------------- | --------------------------------------------------------- | +| New materialized view | Batch append to newly created table | Initial bulk insert or insert | +| Existing materialized view | Full batch recomputation | Hudi `insert_overwrite_table`; no physical table deletion | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured at table level | +| Hudi upsert target | Append-mode micro-batches | Existing Hudi `upsert` | +| Hudi streaming source | Structured Streaming read with SDP checkpoint | Basic Hudi latest-state incremental source | + +### Materialized-view write mapping + +Initial materialization maps to Hudi `bulk_insert` or `insert`; refresh maps to `insert_overwrite_table`. Because Spark 4.1 implements refresh as truncate followed by append, the integration must select the overwrite path without invoking Hudi's destructive full-table truncate. Where to intercept that path remains an open implementation question; a general table-format replace API is not part of this RFC. + +### Streaming commit and restart behavior + +SDP supplies the checkpoint location and trigger. Hudi keeps its existing sink commit and retry behavior; no second checkpoint tree or commit protocol is introduced. The first version allows one streaming flow per Hudi target. + +### Configuration ownership and precedence + +Configuration is split by ownership: + +- **Table identity:** table type, record key, partition fields, key generator, and record-merger settings. Hudi persists these values; SDP loads them for existing tables, and a pipeline run cannot silently change them. +- **Write behavior:** operation and writer tuning. These are resolved per write and checked against the persisted table configuration. +- **Flow execution:** checkpoint location, trigger, retry, and flow identity. SDP owns these values. + +Hudi options are supplied as SDP table properties and normalized to canonical names. Table properties take precedence over supported `spark.hoodie.*` session settings and Hudi defaults. + +Invalid properties should fail during SDP `dry-run` when possible, and otherwise before the first write. + +### Schema and partition evolution + +SDP follows Hudi's existing schema and partition evolution rules. + +- Hudi-supported additive schema changes may be accepted for streaming tables. +- Dropping, renaming, or changing columns follows Hudi's configured schema evolution behavior. +- Record-key, precombine, and partition columns must exist in the target schema. +- Partitioning changes fail; SDP does not turn them into an implicit table replacement. +- A materialized-view refresh that produces an incompatible schema fails before replacing the prior snapshot. + +## Implementation + +### 1. Compatibility harness + +Add an integration harness using the Spark 4.1 pipeline runner and matching Hudi bundle. It creates a real pipeline project, runs `spark-pipelines dry-run` and `spark-pipelines run`, and inspects the result through Spark SQL and Hudi timeline APIs. Direct DataFrame tests alone do not cover SDP's catalog and orchestration paths. Review Comment: 🤖 My understanding is that the `spark-pipelines` CLI in Spark 4.1 drives execution through Spark Connect (the PipelinesHandler lives in the Connect server). If so, that's a new runtime mode for Hudi: the bundle, `HoodieSparkSessionExtension`, and `HoodieCatalog` must be installed server-side, and the harness can't use driver-side Hudi APIs from the client. Could the RFC confirm the execution model and call out Spark Connect as an explicit compatibility requirement / risk? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,405 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +# RFC-112: Support Spark Declarative Pipelines + +## Proposers + +- @cshuo +- @danny0405 + +## Approvers + +- TBD + +## Status + +Issue: TBD + +## Abstract + +[Spark Declarative Pipelines (SDP)](https://spark.apache.org/docs/4.1.1/declarative-pipelines-programming-guide.html) lets users define tables and transformations while Spark builds the dependency graph and runs the batch and streaming flows. Hudi provides atomic commits, keyed updates, event ordering, deletes, incremental queries, and table services. + +This RFC makes Hudi a supported SDP source and target. It covers catalog-based table creation, materialized views, streaming tables, basic Hudi-to-Hudi incremental reads, and keyed upserts through Hudi's existing writer. + +SDP and Hudi retain their existing responsibilities: + +- SDP owns dataset declarations, dependency analysis, orchestration, flow checkpoints, retries, and execution order. +- Hudi owns table storage, commit atomicity, record reconciliation, incremental read semantics, and table services. + +Users can build the same pipelines today with Spark jobs, Hudi writers or Hudi Streamer, and an external orchestrator. This integration removes much of that glue code while keeping the outputs as ordinary Hudi tables. + +## Background + +### Spark Declarative Pipelines + +Apache Spark 4.1 introduced SDP for declarative batch and streaming ETL. SDP persists two main dataset types: + +- A **materialized view**, which has exactly one batch flow that computes a table. +- A **streaming table**, which has one or more streaming flows that append their output to a table. + +Users define pipelines in Python or SQL. SDP validates dataset references, builds the dataflow graph, runs independent flows in parallel, and stores Structured Streaming checkpoints in the pipeline storage location. + +In the Spark 4.1 implementation, a streaming table flow uses a Structured Streaming append-mode write to `DataStreamWriter.toTable`, and a batch table flow uses append-mode `DataFrameWriter.saveAsTable`. Before running a materialized-view batch flow, SDP truncates an existing target table. See Spark's [`FlowExecution`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/FlowExecution.scala) and [`DatasetManager`](https://github.com/apache/spark/blob/branch-4.1/sql/pipelines/src/main/scala/org/apache/spark/sql/pipelines/graph/DatasetManager.scala). + +These table operations are the storage-format integration boundary; this RFC does not add a Hudi-specific execution engine. + +### Relevant Hudi capabilities + +Hudi already provides the required data-plane primitives: + +- `HoodieCatalog` can create and load Hudi tables through Spark's catalog APIs. +- The Hudi batch writer supports bulk insert, insert, upsert, insert overwrite, and table insert overwrite operations. An upsert can carry delete-marker records. +- `HoodieStreamingSink` writes Structured Streaming micro-batches and records checkpoint metadata used to avoid recommitting an already committed batch after retry. +- Hudi's streaming source and incremental query modes can expose records changed since a prior instant. + +The integration exposes these capabilities through SDP's existing table and flow abstractions. Hudi Streamer, the DataFrame writer, Spark SQL DML, and direct Structured Streaming APIs remain unchanged. + +### Current gaps + +`format="hudi"` is not yet an end-to-end supported SDP contract. Some operations may already work, but table creation, refresh, restart, upsert configuration, and Hudi-to-Hudi chaining are neither documented nor covered by an integration suite. + +Materialized-view refresh is the main correctness gap. SDP 4.1 truncates the target before appending the recomputed result, while Hudi's full-table `TRUNCATE TABLE` deletes the table path and reinitializes its properties. Using this path would discard the timeline and could expose an empty or partially rebuilt table if the refresh fails. + +The integration must also address: + +- SDP-created table properties must be persisted and reliably supplied to Hudi batch and streaming writers. +- Streaming restart must coordinate the SDP/Spark checkpoint with Hudi's sink checkpoint metadata. +- A Hudi source used through `readStream.table` must preserve incremental offset semantics. +- SDP must carry Hudi's persisted record-key, precombine, and record-merger settings into the existing upsert path. Deletes remain upsert input rows marked by `_hoodie_is_deleted`; Spark append output mode does not replace this contract. + +### Related work + +- Hudi's Spark 4.1 support ([issue #17663](https://github.com/apache/hudi/issues/17663), [PR #17674](https://github.com/apache/hudi/pull/17674)) provides the runtime baseline, but not an SDP contract. +- DataSource V2 read work ([RFC-98](../rfc-98/rfc-98.md), [issue #15023](https://github.com/apache/hudi/issues/15023)) may improve the catalog read path but is not required for the initial integration. +- The [SQL insert-overwrite discussion](https://github.com/apache/hudi/discussions/13815) shows why refresh semantics must be explicit rather than inferred from an append call. +- [Issue #13973](https://github.com/apache/hudi/issues/13973) shows that a latest-state incremental read cannot fully propagate hard deletes. Flows that require complete delete propagation can use Hudi's existing CDC incremental format. +- [Issue #5537](https://github.com/apache/hudi/issues/5537) tracks named-catalog support. The first version uses `HoodieCatalog` as `spark_catalog`. + +## Motivation + +### Representative use case + +A typical order pipeline looks like this: + +```text +MySQL/PostgreSQL CDC or Kafka + | + v + bronze_order_events (Hudi) + | + v + current_orders (Hudi) + | + v + daily_order_metrics (Hudi) +``` + +The bronze table stores incoming events. The current-state table uses a record key and precombine field to reconcile duplicates and out-of-order events, including delete-marker records. Downstream streaming tables consume changed records incrementally, while batch aggregates can be materialized as views. + +Today this usually means several Spark applications plus custom handling for table creation, write options, checkpoints, dependencies, and restarts. With SDP, the graph lives in one project and the persisted datasets remain regular Hudi tables that other engines can query. + +Related user request issues include: + +- #19281: [multi-dataset incremental processing](https://github.com/apache/hudi/issues/19281) +- #5189: [chained incremental Hudi tables](https://github.com/apache/hudi/issues/5189) +- #13973: [downstream delete propagation](https://github.com/apache/hudi/issues/13973). + +### User value + +1. **Declarative pipelines.** Users define datasets as SQL or DataFrame transformations; SDP manages dependencies, checkpoints, and Hudi writes. +2. **Managed incremental composition.** Hudi already provides incremental reads; SDP composes them into a dependency-aware pipeline, managing execution order, checkpoints, and recovery without separate Spark jobs or hand-written orchestration. +3. **Open outputs.** Each dataset remains a standard Hudi table that can be queried independently of SDP by any Hudi-compatible engine. + +SDP does not make arbitrary joins or aggregates incremental; performance benefits apply only to flows that can process changes incrementally. + +## Goals + +1. Create Hudi tables from SDP datasets through `HoodieCatalog`. +2. Write materialized views with Hudi `bulk_insert` and `insert_overwrite_table` semantics. +3. Write SDP streaming tables through Hudi's Structured Streaming sink. +4. Support a basic incremental Hudi-to-Hudi streaming-table chain. +5. Expose Hudi's existing keyed upsert path through SDP, including record-key, precombine, record-merger, and delete-marker handling. +6. Keep non-SDP Hudi behavior unchanged. + +## Non-Goals + +The first version excludes: +- A complete Hudi DataSource V2 read or streaming-write implementation. +- Cross-table transactions across multiple SDP targets. +- End-to-end exactly-once delivery across an arbitrary external source and Hudi. Guarantees remain scoped to Spark/Hudi micro-batch retries at one target. +- Multiple concurrent append flows into one Hudi target in the initial phase. +- Arbitrary named catalogs for `HoodieCatalog`; the first version uses Hudi as `spark_catalog`. + +## Proposed User Experience + +The examples below implement the representative order pipeline in dependency order. Property names will use the canonical Spark 4.1 SDP and Hudi configuration names. + +### Pipeline configuration + +An SDP project configures the Hudi extension and catalog like any other Hudi Spark application: + +```yaml +name: hudi_orders_pipeline +libraries: + - glob: + include: transformations/** +storage: s3://pipelines/checkpoints/hudi_orders_pipeline +catalog: spark_catalog +database: lakehouse +configuration: + spark.sql.extensions: org.apache.spark.sql.hudi.HoodieSparkSessionExtension + spark.sql.catalog.spark_catalog: org.apache.spark.sql.hudi.catalog.HoodieCatalog +``` + +The SQL below assumes that `order_events_source` is an external streaming source containing parsed Kafka or database CDC events. Each `CREATE ... AS SELECT` statement defines both the target dataset and its single flow. + +### Streaming table with an append flow + +```sql +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'insert' +) +AS +SELECT * +FROM STREAM order_events_source; +``` + +SDP owns the checkpoint location and trigger; Hudi commits each micro-batch. + +### Streaming table with keyed upserts + +```sql +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'MERGE_ON_READ', + 'hoodie.datasource.write.operation' = 'upsert', + 'hoodie.datasource.write.recordkey.field' = 'order_id', + 'hoodie.datasource.write.precombine.field' = 'event_sequence', + 'hoodie.datasource.write.partitionpath.field' = 'event_date' +) +AS +SELECT + *, + operation = 'DELETE' AS _hoodie_is_deleted +FROM STREAM bronze_order_events; +``` + +Spark append output mode only controls how each micro-batch reaches the sink; it does not make the Hudi target append-only. `HoodieStreamingSink` still runs the configured Hudi operation, which defaults to `upsert`. + +For an existing Hudi table, SDP loads the record key, precombine field, partition fields, and record-merger settings from the table configuration. For a new SDP table, these settings are declared as table properties and persisted by Hudi. SDP does not implement its own deduplication, ordering, merge, or delete path. It validates the schema and configuration, then passes each micro-batch to the existing Hudi writer. + +### Batch materialized view + +```sql +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'hoodie.datasource.write.table.type' = 'COPY_ON_WRITE' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and may use `bulk_insert`; later runs replace the result with `insert_overwrite_table`. Refresh must not delete and recreate the table. + +## Semantics + +### Dataset-to-Hudi operation mapping + +| SDP dataset/flow | Spark 4.1 execution shape | Proposed Hudi semantics | +| ----------------------------------- | --------------------------------------------- | --------------------------------------------------------- | +| New materialized view | Batch append to newly created table | Initial bulk insert or insert | +| Existing materialized view | Full batch recomputation | Hudi `insert_overwrite_table`; no physical table deletion | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured at table level | +| Hudi upsert target | Append-mode micro-batches | Existing Hudi `upsert` | +| Hudi streaming source | Structured Streaming read with SDP checkpoint | Basic Hudi latest-state incremental source | + +### Materialized-view write mapping + +Initial materialization maps to Hudi `bulk_insert` or `insert`; refresh maps to `insert_overwrite_table`. Because Spark 4.1 implements refresh as truncate followed by append, the integration must select the overwrite path without invoking Hudi's destructive full-table truncate. Where to intercept that path remains an open implementation question; a general table-format replace API is not part of this RFC. + +### Streaming commit and restart behavior + +SDP supplies the checkpoint location and trigger. Hudi keeps its existing sink commit and retry behavior; no second checkpoint tree or commit protocol is introduced. The first version allows one streaming flow per Hudi target. + +### Configuration ownership and precedence + +Configuration is split by ownership: + +- **Table identity:** table type, record key, partition fields, key generator, and record-merger settings. Hudi persists these values; SDP loads them for existing tables, and a pipeline run cannot silently change them. +- **Write behavior:** operation and writer tuning. These are resolved per write and checked against the persisted table configuration. +- **Flow execution:** checkpoint location, trigger, retry, and flow identity. SDP owns these values. + +Hudi options are supplied as SDP table properties and normalized to canonical names. Table properties take precedence over supported `spark.hoodie.*` session settings and Hudi defaults. Review Comment: 🤖 Two questions on config ownership: (1) persisting write-behavior properties like `hoodie.datasource.write.operation` in TBLPROPERTIES means non-SDP writers (Spark SQL INSERT, DataFrame writer that reads catalog props) also inherit them — is that intended, or should write-behavior stay flow-scoped? (2) Hudi SQL DDL already has its own canonical table options (`primaryKey`, `preCombineField`, `type`) and a normalization layer in HoodieOptionConfig; could the RFC state that the SDP examples map onto that existing layer rather than introducing a second set of names? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,405 @@ +<!-- + Licensed to the Apache Software Foundation (ASF) under one or more + contributor license agreements. See the NOTICE file distributed with + this work for additional information regarding copyright ownership. + The ASF licenses this file to You under the Apache License, Version 2.0 + (the "License"); you may not use this file except in compliance with + the License. You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. +--> + +# RFC-112: Support Spark Declarative Pipelines + +## Proposers + +- @cshuo +- @danny0405 + +## Approvers + +- TBD + +## Status + +Issue: TBD + +## Abstract + +[Spark Declarative Pipelines (SDP)](https://spark.apache.org/docs/4.1.1/declarative-pipelines-programming-guide.html) lets users define tables and transformations while Spark builds the dependency graph and runs the batch and streaming flows. Hudi provides atomic commits, keyed updates, event ordering, deletes, incremental queries, and table services. + +This RFC makes Hudi a supported SDP source and target. It covers catalog-based table creation, materialized views, streaming tables, basic Hudi-to-Hudi incremental reads, and keyed upserts through Hudi's existing writer. + +SDP and Hudi retain their existing responsibilities: + +- SDP owns dataset declarations, dependency analysis, orchestration, flow checkpoints, retries, and execution order. +- Hudi owns table storage, commit atomicity, record reconciliation, incremental read semantics, and table services. Review Comment: 🤖 The RFC says Hudi retains table services but doesn't discuss how they run under SDP. E.g. a materialized view refreshed via insert_overwrite_table produces a replacecommit per run, and replaced file groups linger until clean — with frequent refreshes what cleaner/archival defaults are recommended? For streaming tables, does async compaction/clustering inside HoodieStreamingSink work when many flows share one Spark app and SDP controls the app lifecycle (does pipeline stop wait for in-flight table services)? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
