hudi-agent commented on code in PR #19997: URL: https://github.com/apache/hudi/pull/19997#discussion_r4068864964
########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,429 @@ +<!-- + 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`. For an existing materialized view, SDP issues `TRUNCATE TABLE`, calls `catalog.alterTable` to reconcile schema and properties, then runs the batch flow. 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 needs end-to-end validation. SDP 4.1 truncates the target before reconciling its schema and writing the recomputed result. Hudi's current full-table `TRUNCATE TABLE` deletes the table path, including the timeline, and reinitializes its properties. This RFC reuses that behavior and documents its limitations; a timeline-preserving truncate is an independent Hudi improvement, not a prerequisite for SDP support. + +The integration must also address: + +- SDP table properties and flow-scoped write options must be resolved and passed to the Hudi writer. +- 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` or `insert`, following SDP's truncate-and-recompute refresh sequence. +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. +- Atomic materialized-view refresh across truncate, schema reconciliation, and the batch write. +- Changes to Hudi's full-table truncate semantics. Timeline-preserving truncate can be developed independently. +- 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. +- 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, using Hudi's existing SQL table options and `SET` for flow-scoped write options. + +### 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 +SET spark.hoodie.datasource.write.operation = insert; + +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'type' = 'mor' +) +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 +SET spark.hoodie.datasource.write.operation = upsert; + +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'type' = 'mor', + 'primaryKey' = 'order_id', + 'preCombineField' = 'event_sequence' +) +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 +SET spark.hoodie.datasource.write.operation = bulk_insert; + +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'type' = 'cow' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and writes the result using `bulk_insert`. Later runs truncate the target, reconcile its schema and properties, and write the recomputed result. With Hudi's current truncate implementation, each refresh removes the previous data and timeline history. + +## 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 | Hudi `bulk_insert` or `insert` | +| Existing materialized view | Truncate, alter, then batch append | Existing Hudi truncate, then `bulk_insert` or `insert` | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured per flow | +| 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 uses Hudi `bulk_insert` or `insert`. For an existing materialized view, Spark 4.1 executes `TRUNCATE TABLE`, reconciles schema and properties through `catalog.alterTable`, then writes the recomputed result through append-mode `saveAsTable`. Hudi uses the same configured write operation after truncate. An empty result leaves the target empty. + +These steps are not a single transaction. Readers may see an empty target during refresh. If schema reconciliation or the batch flow fails after truncate, the previous result is not restored, and completed catalog changes remain in effect. + +The initial integration reuses Hudi's current full-table truncate, which deletes old data files and timeline history. This can disrupt concurrent readers and invalidate existing incremental offsets. Timeline-preserving truncate is an independent Hudi improvement, not a prerequisite for SDP support. + +Materialized views are supported as batch outputs, but not as incremental sources in the initial scope. + +### Streaming commit and restart behavior + +SDP manages each flow's checkpoint location and trigger. Multiple append flows can write to the same Hudi target, each with a distinct Hudi checkpoint identifier that remains stable across restarts. Concurrent flows require Hudi's existing multi-writer configuration and an appropriate lock provider. Each flow commits and retries independently. + +Normal restarts reuse the same Spark checkpoint and Hudi checkpoint identifier. A streaming-table full refresh starts a fresh checkpoint and uses Hudi's current full-table truncate to remove target data, timeline history, and sink checkpoint metadata before rebuilding the table. Resetting the Spark checkpoint while retaining the Hudi target is outside this RFC's scope. + +### Configuration ownership and precedence + +Configuration is split by ownership: + +- **Table identity:** table type, record key, precombine field, 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, supplied through SQL `SET spark.hoodie.*` as flow-scoped write options. +- **Flow execution:** checkpoint location, trigger, retry, and flow identity. SDP owns these values; the Hudi integration maps each flow to a distinct sink checkpoint identifier. + +SDP captures the active `SET` values at each flow definition; a `SET` remains in effect for subsequent definitions until overridden. Flow settings override pipeline and Hudi defaults and must be compatible with the persisted table configuration. The integration strips the `spark.` prefix from write options and reuses `HoodieOptionConfig` to normalize table options. Flows writing to the same target must use compatible concurrency and lock settings. The sink checkpoint identifier is resolved per flow, not shared as a table property. + +Invalid properties should fail during SDP `dry-run` when possible, and otherwise before the first write. + +### Table services + +Table services follow the existing Hudi writer behavior, with no SDP-specific scheduling or retention defaults. Integration tests cover service shutdown, recovery, and resource cleanup when flows finish, stop, or restart, including multiple flows sharing a target. + +### 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. + +`HoodieCatalog.alterTable` validates the complete set of requested schema and property changes before applying supported changes. It accepts unchanged properties and rejects unsupported changes or changes to table identity. The Hudi writer validates the actual input schema against the persisted table configuration using its existing schema reconciliation rules. Review Comment: 🤖 Today `HoodieCatalog.alterTable` only handles `AddColumn`, `UpdateColumnType`, and `UpdateColumnComment` and throws `UnsupportedOperationException` for everything else (`SetProperty`, `RemoveProperty`, `DeleteColumn`, `RenameColumn`, ...). Two things make SDP's reconcile produce a non-empty change set on every run even when the user changed nothing: `HoodieInternalV2Table.schema()` returns `tableSchema` *with* the five `_hoodie_*` meta columns, and `properties()` returns `catalogProperties` rather than the declared `TBLPROPERTIES` keys. Could the RFC enumerate which `TableChange` kinds the integration must accept/ignore, and how the meta-column and property-key deltas are excluded from the diff so that a refresh with an unchanged definition is a no-op? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,429 @@ +<!-- + 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`. For an existing materialized view, SDP issues `TRUNCATE TABLE`, calls `catalog.alterTable` to reconcile schema and properties, then runs the batch flow. 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 needs end-to-end validation. SDP 4.1 truncates the target before reconciling its schema and writing the recomputed result. Hudi's current full-table `TRUNCATE TABLE` deletes the table path, including the timeline, and reinitializes its properties. This RFC reuses that behavior and documents its limitations; a timeline-preserving truncate is an independent Hudi improvement, not a prerequisite for SDP support. + +The integration must also address: + +- SDP table properties and flow-scoped write options must be resolved and passed to the Hudi writer. +- 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` or `insert`, following SDP's truncate-and-recompute refresh sequence. +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. +- Atomic materialized-view refresh across truncate, schema reconciliation, and the batch write. +- Changes to Hudi's full-table truncate semantics. Timeline-preserving truncate can be developed independently. +- 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. +- 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, using Hudi's existing SQL table options and `SET` for flow-scoped write options. + +### 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 +SET spark.hoodie.datasource.write.operation = insert; + +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'type' = 'mor' +) +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 +SET spark.hoodie.datasource.write.operation = upsert; + +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'type' = 'mor', + 'primaryKey' = 'order_id', + 'preCombineField' = 'event_sequence' +) +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 +SET spark.hoodie.datasource.write.operation = bulk_insert; + +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'type' = 'cow' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and writes the result using `bulk_insert`. Later runs truncate the target, reconcile its schema and properties, and write the recomputed result. With Hudi's current truncate implementation, each refresh removes the previous data and timeline history. + +## 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 | Hudi `bulk_insert` or `insert` | +| Existing materialized view | Truncate, alter, then batch append | Existing Hudi truncate, then `bulk_insert` or `insert` | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured per flow | +| 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 uses Hudi `bulk_insert` or `insert`. For an existing materialized view, Spark 4.1 executes `TRUNCATE TABLE`, reconciles schema and properties through `catalog.alterTable`, then writes the recomputed result through append-mode `saveAsTable`. Hudi uses the same configured write operation after truncate. An empty result leaves the target empty. + +These steps are not a single transaction. Readers may see an empty target during refresh. If schema reconciliation or the batch flow fails after truncate, the previous result is not restored, and completed catalog changes remain in effect. + +The initial integration reuses Hudi's current full-table truncate, which deletes old data files and timeline history. This can disrupt concurrent readers and invalidate existing incremental offsets. Timeline-preserving truncate is an independent Hudi improvement, not a prerequisite for SDP support. + +Materialized views are supported as batch outputs, but not as incremental sources in the initial scope. + +### Streaming commit and restart behavior + +SDP manages each flow's checkpoint location and trigger. Multiple append flows can write to the same Hudi target, each with a distinct Hudi checkpoint identifier that remains stable across restarts. Concurrent flows require Hudi's existing multi-writer configuration and an appropriate lock provider. Each flow commits and retries independently. + +Normal restarts reuse the same Spark checkpoint and Hudi checkpoint identifier. A streaming-table full refresh starts a fresh checkpoint and uses Hudi's current full-table truncate to remove target data, timeline history, and sink checkpoint metadata before rebuilding the table. Resetting the Spark checkpoint while retaining the Hudi target is outside this RFC's scope. Review Comment: 🤖 When a streaming table that is itself a source (e.g. `bronze_order_events` feeding `current_orders`) gets a full refresh, its path and timeline are deleted and rebuilt with fresh, later instants. The downstream flow's SDP checkpoint still holds a `HoodieSourceOffset` (completion time) from the old timeline; `HoodieStreamSourceV2.getBatch` would then read every post-truncate commit as a new increment, i.e. replay the entire rebuilt upstream into the downstream table (duplicates for insert targets, no delete propagation for rows that disappeared). This is the opposite of the 'fail rather than skip data' guarantee at line 340. Does the design require upstream full refresh to cascade to dependents, and/or should the Hudi source detect a timeline reset (e.g. table-creation marker newer than the offset) and error? <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,429 @@ +<!-- + 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`. For an existing materialized view, SDP issues `TRUNCATE TABLE`, calls `catalog.alterTable` to reconcile schema and properties, then runs the batch flow. 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 needs end-to-end validation. SDP 4.1 truncates the target before reconciling its schema and writing the recomputed result. Hudi's current full-table `TRUNCATE TABLE` deletes the table path, including the timeline, and reinitializes its properties. This RFC reuses that behavior and documents its limitations; a timeline-preserving truncate is an independent Hudi improvement, not a prerequisite for SDP support. + +The integration must also address: + +- SDP table properties and flow-scoped write options must be resolved and passed to the Hudi writer. +- 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` or `insert`, following SDP's truncate-and-recompute refresh sequence. +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. +- Atomic materialized-view refresh across truncate, schema reconciliation, and the batch write. +- Changes to Hudi's full-table truncate semantics. Timeline-preserving truncate can be developed independently. +- 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. +- 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, using Hudi's existing SQL table options and `SET` for flow-scoped write options. + +### 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 +SET spark.hoodie.datasource.write.operation = insert; + +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'type' = 'mor' +) +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 +SET spark.hoodie.datasource.write.operation = upsert; + +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'type' = 'mor', + 'primaryKey' = 'order_id', + 'preCombineField' = 'event_sequence' +) +AS +SELECT + *, Review Comment: 🤖 `SELECT *, ... FROM STREAM bronze_order_events` will carry the upstream `_hoodie_commit_time`/`_hoodie_record_key`/... columns into this flow's output schema, which is what SDP hands to `catalog.createTable` and later compares against the table during reconcile. The writer does drop meta columns on write (`HoodieSparkSqlWriter` drops `HOODIE_META_COLUMNS`), so the data path is fine, but the declared target schema now contains user columns that collide with Hudi's own meta fields. Should the integration strip them in `HoodieCatalog.createTable`, reject the definition, or require users to project explicitly? Line 338 defers this to documentation, but it looks like a design decision the catalog contract needs to make. <sub><i>⚠️ AI-generated; verify before applying. React 👍/👎 to flag quality.</i></sub> ########## rfc/rfc-112/rfc-112.md: ########## @@ -0,0 +1,429 @@ +<!-- + 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`. For an existing materialized view, SDP issues `TRUNCATE TABLE`, calls `catalog.alterTable` to reconcile schema and properties, then runs the batch flow. 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 needs end-to-end validation. SDP 4.1 truncates the target before reconciling its schema and writing the recomputed result. Hudi's current full-table `TRUNCATE TABLE` deletes the table path, including the timeline, and reinitializes its properties. This RFC reuses that behavior and documents its limitations; a timeline-preserving truncate is an independent Hudi improvement, not a prerequisite for SDP support. + +The integration must also address: + +- SDP table properties and flow-scoped write options must be resolved and passed to the Hudi writer. +- 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` or `insert`, following SDP's truncate-and-recompute refresh sequence. +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. +- Atomic materialized-view refresh across truncate, schema reconciliation, and the batch write. +- Changes to Hudi's full-table truncate semantics. Timeline-preserving truncate can be developed independently. +- 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. +- 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, using Hudi's existing SQL table options and `SET` for flow-scoped write options. + +### 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 +SET spark.hoodie.datasource.write.operation = insert; + +CREATE STREAMING TABLE bronze_order_events +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'type' = 'mor' +) +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 +SET spark.hoodie.datasource.write.operation = upsert; + +CREATE STREAMING TABLE current_orders +USING hudi +PARTITIONED BY (event_date) +TBLPROPERTIES ( + 'type' = 'mor', + 'primaryKey' = 'order_id', + 'preCombineField' = 'event_sequence' +) +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 +SET spark.hoodie.datasource.write.operation = bulk_insert; + +CREATE MATERIALIZED VIEW daily_order_metrics +USING hudi +PARTITIONED BY (order_date) +TBLPROPERTIES ( + 'type' = 'cow' +) +AS +SELECT + order_date, + state, + COUNT(*) AS order_count +FROM current_orders +GROUP BY order_date, state; +``` + +The first run creates the table and writes the result using `bulk_insert`. Later runs truncate the target, reconcile its schema and properties, and write the recomputed result. With Hudi's current truncate implementation, each refresh removes the previous data and timeline history. + +## 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 | Hudi `bulk_insert` or `insert` | +| Existing materialized view | Truncate, alter, then batch append | Existing Hudi truncate, then `bulk_insert` or `insert` | +| Streaming table with an append flow | Append-mode micro-batches | Hudi `insert` or `bulk_insert`, configured per flow | +| 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 uses Hudi `bulk_insert` or `insert`. For an existing materialized view, Spark 4.1 executes `TRUNCATE TABLE`, reconciles schema and properties through `catalog.alterTable`, then writes the recomputed result through append-mode `saveAsTable`. Hudi uses the same configured write operation after truncate. An empty result leaves the target empty. + +These steps are not a single transaction. Readers may see an empty target during refresh. If schema reconciliation or the batch flow fails after truncate, the previous result is not restored, and completed catalog changes remain in effect. + +The initial integration reuses Hudi's current full-table truncate, which deletes old data files and timeline history. This can disrupt concurrent readers and invalidate existing incremental offsets. Timeline-preserving truncate is an independent Hudi improvement, not a prerequisite for SDP support. + +Materialized views are supported as batch outputs, but not as incremental sources in the initial scope. + +### Streaming commit and restart behavior + +SDP manages each flow's checkpoint location and trigger. Multiple append flows can write to the same Hudi target, each with a distinct Hudi checkpoint identifier that remains stable across restarts. Concurrent flows require Hudi's existing multi-writer configuration and an appropriate lock provider. Each flow commits and retries independently. + +Normal restarts reuse the same Spark checkpoint and Hudi checkpoint identifier. A streaming-table full refresh starts a fresh checkpoint and uses Hudi's current full-table truncate to remove target data, timeline history, and sink checkpoint metadata before rebuilding the table. Resetting the Spark checkpoint while retaining the Hudi target is outside this RFC's scope. + +### Configuration ownership and precedence + +Configuration is split by ownership: + +- **Table identity:** table type, record key, precombine field, 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, supplied through SQL `SET spark.hoodie.*` as flow-scoped write options. +- **Flow execution:** checkpoint location, trigger, retry, and flow identity. SDP owns these values; the Hudi integration maps each flow to a distinct sink checkpoint identifier. + +SDP captures the active `SET` values at each flow definition; a `SET` remains in effect for subsequent definitions until overridden. Flow settings override pipeline and Hudi defaults and must be compatible with the persisted table configuration. The integration strips the `spark.` prefix from write options and reuses `HoodieOptionConfig` to normalize table options. Flows writing to the same target must use compatible concurrency and lock settings. The sink checkpoint identifier is resolved per flow, not shared as a table property. Review Comment: 🤖 For the batch/SQL path this already works today (`ProvidesHoodieConfig.combineOptions` reads `spark.hoodie.*` from `SQLConf`), but the streaming path doesn't: `DefaultSource.createSink` → `HoodieStreamingSink` only sees the `DataStreamWriter` options plus table storage properties, and `HoodieWriterUtils.parametersWithWriteDefaults` only normalizes `spark.hoodie.*` keys inside that map, never the session conf. Could you spell out the mechanism by which a flow-scoped `SET` reaches the sink (e.g. merging session hoodie confs in `createSink`), and whether that becomes a behavior change for non-SDP `writeStream` users whose session happens to carry `spark.hoodie.*` keys? Note `canSkipBatch` also keys off `OPERATION` from those same options, so if the operation doesn't arrive the dedup check silently assumes `upsert`. <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: 🤖 Checked against branch-4.1 and this holds: `streamRelationPrimary` accepts `optionsClause`, and since `HoodieInternalV2Table` is a `V2TableWithV1Fallback` without `MICRO_BATCH_READ`, `FindDataSourceTable` falls back to a V1 `StreamingRelation` built from `table.provider` with the `WITH` options merged in — so they should reach `DefaultSource.createSource` as-is. One caveat worth noting in the RFC: `generateDatasourceOptions` throws (unless the legacy flag is set) when a `WITH` key conflicts with the same key in the table's storage properties, so any hoodie reader option persisted on the table can't be overridden per-flow. -- 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]
