nsivabalan opened a new pull request, #20378: URL: https://github.com/apache/hudi/pull/20378
### Describe the issue this Pull Request addresses **WIP / RFC-112 groundwork.** While validating Spark 4.1 Declarative Pipelines against Hudi (see the review on #19997), two existing behaviours combined into a data-safety problem: 1. Full-table `TRUNCATE TABLE` deletes the base path and re-initializes `hoodie.properties` (`TruncateHoodieTableCommand`). History, savepoints, the metadata table and every incremental/streaming reader offset are gone, and the operation is a non-transactional filesystem delete. 2. `HoodieStreamingSink` records its last committed batch id in commit metadata under `hoodie.datasource.write.streaming.checkpoint.identifier` (default `default_single_writer`) and skips any incoming batch id ≤ that value. The only reason this did not lose data after SDP's `--full-refresh` (fresh Structured Streaming checkpoint → batch ids restart at 0) was that (1) wiped the timeline first. The moment truncate stops destroying history, the sink silently skips every batch until the old high-water mark is passed. The two therefore have to change together. ### Summary and Changelog - `TRUNCATE TABLE t` (no partition spec) now issues a single `replacecommit` (`insert_overwrite_table` of an empty dataset) that replaces every file group. Timeline, savepoints, MDT and reader offsets stay valid; the truncation is visible as an instant. `TRUNCATE TABLE t PARTITION (...)` is unchanged (already a `delete_partition` replacecommit). - New advanced config `hoodie.sql.truncate.purge.enabled` (default `false`) restores the previous delete-and-reinit behaviour for users who want a hard reset. - `HoodieStreamingSink` scopes the default checkpoint identifier to the query's `checkpointLocation` (`default_single_writer@<checkpointLocation>`) when no explicit identifier is configured. A stream resumed from the same checkpoint keeps deduplicating replayed batches; a stream started from a fresh checkpoint is no longer matched against another stream's high-water mark. An explicit identifier behaves exactly as before. Logic lives in `HoodieStreamingSink.resolveCheckpointIdentifier`. The sink also forwards the resolved identifier in the write options, because the bulk-insert row-writer path does not go through the sink's pre-commit hook and builds the checkpoint entry on its own in `DataSourceUtils.getExtraMetadata` (found by `testStructuredStreamingWithBulkInsert`). - Tests: `TestTruncateTable` (timeline retained + trailing replacecommit + writes after truncate, for COW/MOR × partitioned/non-partitioned; purge mode resets the timeline) and `TestStructuredStreaming` (dedup scoped to checkpoint location; resolver unit cases; existing default-identifier test adapted). Validated end to end against stock Spark 4.1.1 `spark-pipelines` (Connect) with a Kafka → Hudi streaming table: `--full-refresh` now yields `N commits + replacecommit + N commits`, batch ids 0..N re-land under the new identifier, row counts identical before/after. ### Impact - User-facing behaviour change for `TRUNCATE TABLE`: history is retained instead of purged, and the table directory is not removed. Readers see one more instant; storage is reclaimed by the cleaner instead of immediately. Opt back in with `hoodie.sql.truncate.purge.enabled=true`. - Commit metadata: the `_hudi_streaming_sink_checkpoint` map key changes for streams that rely on the default identifier. Existing tables keep their old entries; after upgrade the first restart of such a stream will not find a matching entry and will not deduplicate a batch that was committed to Hudi but not yet acknowledged in the Spark checkpoint at the moment of upgrade (a crash exactly at the upgrade boundary; upserts are idempotent, inserts would duplicate that one batch). Open question below. - No storage format, timeline schema or public API changes. ### Risk Level medium. Truncate now goes through the regular write path (`insert_overwrite_table` with an empty input) for COW and MOR; covered by the new tests plus the SDP end-to-end run. The identifier change is covered by tests; the upgrade-boundary window is documented above. ### Open questions for reviewers - Should the sink also write the legacy `default_single_writer` entry for one release to close the upgrade-boundary window, at the cost of re-introducing the skip after a checkpoint reset on tables that have only legacy entries? - `hoodie.sql.truncate.purge.enabled` is read from the session conf only; should it also be accepted as a table/SQL option? - Title carries `!` because `TRUNCATE` semantics change; happy to drop it if we consider "history retained" non-breaking. - Truncate on a table with pending compaction/clustering now inherits whatever `INSERT OVERWRITE TABLE` does today for those file groups; worth confirming that is acceptable (it was not a question when truncate was a filesystem delete). - Still to add before un-drafting: tests for truncating an empty table and truncating twice in a row. ### Documentation Update - New config `hoodie.sql.truncate.purge.enabled` (description in `DataSourceWriteOptions`). - SQL DDL page: note that `TRUNCATE TABLE` is a replacecommit and how to opt into purge. ### Contributor's checklist - [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [x] Enough context is provided in the sections above - [x] Adequate tests were added if applicable cc @danny0405 @cshuo 🤖 Generated with [Claude Code](https://claude.com/claude-code) -- 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]
