feat: support incremental scan between 2 snapshots

[xxchan]

Related code in Java

• https://github.com/apache/iceberg/blob/9a98de04103755ad4833ac6c3918c39022d54a99/core/src/main/java/org/apache/iceberg/util/SnapshotUtil.java#L208-L226
• https://github.com/apache/iceberg/blob/9a98de04103755ad4833ac6c3918c39022d54a99/api/src/main/java/org/apache/iceberg/IncrementalScan.java
• https://github.com/apache/iceberg/blob/6ec3de390d3fa6e797c6975b1eaaea41719db0fe/core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java#L68-L98

[chenzl25]

I think simply checking manifest status Added is not enough, since compaction will add an manifest entry with Added as well, however, we don't want to consume a compaction snapshot. We should detect whether a snapshot is a compaction snapshot. If the snapshot is a compaction, we should skip it. For compaction snapshot, I think we can just check if there is a manifest entry having ManifestStatus::Deleted.

[xxchan]

We have filtered Append snapshots above:

let append_snapshots =
        ancestors_between(table_metadata, latest_snapshot_id, oldest_snapshot_id)
            .filter(|snapshot| matches!(snapshot.summary().operation, Operation::Append))

The definition of snapshot operation is:
(so compaction snapshot should be Replace)

/// The operation field is used by some operations, like snapshot expiration, to skip processing certain snapshots.
pub enum Operation {
    /// Only data files were added and no files were removed.
    Append,
    /// Data and delete files were added and removed without changing table data;
    /// i.e., compaction, changing the data file format, or relocating data files.
    Replace,
    /// Data and delete files were added and removed in a logical overwrite operation.
    Overwrite,
    /// Data files were removed and their contents logically deleted and/or delete files were added to delete rows.
    Delete,
}

[chenzl25]

cc @Li0k

[xxchan]

Found a problem: Seems delete files are also included in Append snapshot. 🤔

Encountered an entry for a delete file in a data file manifest: 

ManifestEntry { status: Added, snapshot_id: Some(4746599406424318107), sequence_number: Some(4), file_sequence_number: Some(4), data_file: DataFile { content: EqualityDeletes, file_path: "s3://hummock001/iceberg_connection/public/t/data/11-00077-eq-del-0195a491-109c-75b1-a93c-e29d371c2d0b.parquet", file_format: Parquet, partition: Struct { fields: [], null_bitmap: BitVec<usize, bitvec::order::Lsb0> { addr: 0x8, head: 000000, bits: 0, capacity: 0 } [] }, record_count: 2, file_size_in_bytes: 680, column_sizes: {1: 82}, value_counts: {1: 2}, null_value_counts: {1: 0}, nan_value_counts: {}, lower_bounds: {1: Datum { type: Long, literal: Long(524235685507891200) }}, upper_bounds: {1: Datum { type: Long, literal: Long(524235873517572096) }}, key_metadata: None, split_offsets: [4], equality_ids: [1], sort_order_id: None } }

The java code uses a ignoreDeleted to further filter the manifest.
https://github.com/apache/iceberg/blob/6ec3de390d3fa6e797c6975b1eaaea41719db0fe/core/src/main/java/org/apache/iceberg/BaseIncrementalAppendScan.java#L86

[chenzl25]

/// Only data files were added and no files were removed.
Append,

Indeed, by definition, there should be no delete files in an append snapshot. However, RisingWave's iceberg sink would include delete files to the append snapshot. Shall we change the snapshot operation to Overwrite if it is not an append-only commit? But once we change it to OverWrite, it means we can't read any incremental data between the snapshots. It seems we should add the added data of overwrite snapshot to the incremental read as well.

[xxchan]

Some facts I found:

• Changelog scan with deletes are not supported yet: Support changelog scan for table with delete files apache/iceberg#10935
• Changelog scan is not used by flink

[xxchan]

Oh, it's used in SparkChangelogTable https://iceberg.apache.org/docs/latest/spark-procedures/?h=changes#carry-over-rows

[chenzl25]

cc @ZENOTME If we commit delete files to the iceberg, it seems we should change the iceberg snapshot operation to OverWrite instead of Append.

[chenzl25]

Oh, it's used in SparkChangelogTable https://iceberg.apache.org/docs/latest/spark-procedures/?h=changes#carry-over-rows

I think we should handle OverWrite as well. Ignoring delete files is acceptable.

[xxchan]

+1. kind of like "force-append-only"

[chenzl25]

LGTM!

[oven-yang]

I was looking over the code, and I’m wondering if there might be a chance for the logic to get stuck here when the channel’s buffer size is smaller than the number of entries.
It seems like the channel could block once the buffer fills up, and since the consumers only start after all entries are sent, it might cause a hang.
I could be missing something, though—do you think this could be an issue, or is there a detail I might not be seeing? I’d really appreciate your thoughts!

[xxchan]

Hi, curious how did you find here 👀

[oven-yang]

I came across your PR while browsing through this repo, and it caught my attention because I'm looking for a way to implement incremental ingestion for an Iceberg table using Rust. Great work on it!

[xxchan]

Thanks, I think your insight is correct here. I've refined implementation in #36