[Data] Add Iceberg-backend in Checkpointing

[WinkerDu]

Description

Implement Iceberg-backend Ray data checkpoint for this PR #59870

Related issues

Relates to #59870

Additional information

1. Introduce new CheckpointBackend ICEBERG
2. Introduce new param as catalog_kwargs when initializing CheckpointConfig

More usage examples can refers for python/ray/data/tests/test_iceberg_checkpoint.py

[gemini-code-assist]

Code Review

This pull request introduces an Iceberg backend for Ray Data checkpointing, which is a significant feature enhancement. The changes include adding a new ICEBERG checkpoint backend type, implementing the IcebergCheckpointWriter, and updating CheckpointLoader and BatchBasedCheckpointFilter to handle Iceberg-specific logic. The configuration has also been updated to accept catalog_kwargs. The addition of an end-to-end test in test_iceberg_checkpoint.py is excellent for ensuring correctness. Overall, the implementation is well-structured. I've identified one critical issue with exception handling and one opportunity to reduce code duplication for better maintainability.

[WinkerDu]

@owenowenisme Please take a review, thx :)

[TheR1sing3un]

@owenowenisme Please take a review, thx :)

I'm not sure if there are some deviations in my understanding. I think this issue: #59869 aims to apply the checkpoint mechanism that uses iceberg as the pipeline sink. In other words, it means that the checkpoint capability can be utilized in the read and write operations of more data sources. Please confirm. cc @owenowenisme

[WinkerDu]

@TheR1sing3un The existing Ray Data checkpoint mechanism already supports using file-based checkpoints for the Iceberg read/write interfaces. You can refer to the unit tests in this PR for details. Therefore, my understanding is that this issue aims to implement an Iceberg-based checkpoint mechanism.
@owenowenisme Please confirm.

[owenowenisme]

@WinkerDu Correct. The current checkpoint mechanism only supports file-based data sinks. For others, such as Iceberg, we need a different approach because Iceberg requires an explicit commit process.

[WinkerDu]

@WinkerDu Correct. The current checkpoint mechanism only supports file-based data sinks. For others, such as Iceberg, we need a different approach because Iceberg requires an explicit commit process.

@owenowenisme Is the implementation of this patch consistent with the expected behavior?

1. Introduce new CheckpointBackend ICEBERG
2. Introduce new param as catalog_kwargs when initializing CheckpointConfig

[xinyuangui2]

Could you briefly talk about the design in the description? Thanks

[WinkerDu]

Could you briefly talk about the design in the description? Thanks

@xinyuangui2 Sure.

1. Overview

The primary objective of this commit is to introduce Iceberg Backend support for Ray Data's Checkpoint mechanism.

Previously, Ray Data Checkpoints primarily supported file-based storage (e.g., Parquet files). Introducing Iceberg as a backend leverages its table format features (such as ACID transactions and metadata management), making checkpoint state management more robust and standardized

2. Core Flowchart

graph TD
    subgraph Configuration [Configuration Phase]
        A[User Configures CheckpointConfig] -->|backend=ICEBERG| B(CheckpointConfig Object)
        B -->|Contains| C{catalog_kwargs}
        C -->|Passes To| D[Writer & Filter]
    end

Loading

graph TD
    subgraph Write_Phase [Write Phase - IcebergCheckpointWriter]
        D[Task Processes Block] --> E{Checkpoint Enabled?}
        E -- Yes --> F[IcebergCheckpointWriter.write_block_checkpoint]
        F -->|First Time| G[Load or Create Iceberg Table]
        G -->|Schema: id_col int64| H[(Iceberg Table)]
        F -->|Extract ID Column| I[Build ID Block]
        I --> J[IcebergDatasink.write]
        J -->|Immediate Commit| H
    end

Loading

graph TD
    subgraph Read_Resume_Phase [Read/Resume Phase - CheckpointFilter]
        K[Job Start/Resume] --> L[BatchBasedCheckpointFilter]
        L --> M[CheckpointLoader.load_checkpoint]
        M -->|backend=ICEBERG| N[pyiceberg.catalog.load_table]
        N --> O[Scan & Select ID Column]
        O --> P[Convert to Ray Dataset]
        Q[Input Data Source] --> R[Filter Logic]
        P --> R
        R -->|Filter Processed IDs| S[Unprocessed Data]
    end

Loading

3. Detailed Steps
   Step 1: Configuration Support

• Action : Enable ICEBERG backend and catalog_kwargs in CheckpointConfig .
• File : interfaces.py
• Logic : Added CheckpointBackend.ICEBERG enum and catalog_kwargs field. Modified backend inference to skip filesystem detection for Iceberg, delegating storage handling to the Iceberg SDK.

Step 2: Checkpoint Writing (Iceberg Table)

• Action : Implement writing processed IDs to an Iceberg table.
• File : checkpoint_writer.py
• Logic : Implemented IcebergCheckpointWriter. It initializes the Iceberg table (auto-creating with an ID column if missing) and uses IcebergDatasink to write and commit ID blocks directly from workers.

Step 3: Checkpoint Reading & Filtering

• Action : Load IDs from Iceberg to filter input data.
• File : checkpoint_filter.py
• Logic : Updated CheckpointLoader to scan the Iceberg table (ID column only) via pyiceberg, converting it to a Ray Dataset for filtering. Added drop_table logic for checkpoint deletion.

Step 4: Testing & Stability

• Action : Verify end-to-end flow and ensure thread safety.
• Files : test_iceberg_checkpoint.py, iceberg_datasource.py
• Logic : Added a full test cycle (Write -> Checkpoint -> New Data -> Filter). Fixed potential side-effects by safely copying kwargs in IcebergDatasource.

[cursor]

Cursor Bugbot has reviewed your changes and found 1 potential issue.

[cursor]

New ICEBERG backend requires use of deprecated parameter

Medium Severity

The newly added CheckpointBackend.ICEBERG can only be activated by passing override_backend=CheckpointBackend.ICEBERG, but override_backend emits a deprecation warning stating it "will be removed in August 2025." There is no alternative path (e.g., auto-inference from catalog_kwargs) to select the ICEBERG backend without triggering this warning. This means every user of the new Iceberg checkpoint feature sees a confusing deprecation notice for a newly introduced capability, and if override_backend is actually removed, the ICEBERG backend becomes unusable.

Additional Locations (1)

• python/ray/data/checkpoint/interfaces.py#L101-L107

 

[WinkerDu]

close this pr for wrong understanding for #59870 . Explicit Iceberg commit process with checkpoint support can be seen in #61753