GeezerVison

-            replaced: Some(write_dv("old.puffin", old, old.len() as u64)),
+            replaced: (!old.is_empty()).then(|| write_dv("old.puffin", old, old.len() as u64)),
...
+    #[tokio::test]
+    async fn new_deletes_read_without_old_vector_selects_all_positions() {
+        let selected = newly_deleted(&[], &[0, 2, 1 << 33], 3).await.unwrap();
+        assert_eq!(selected, Some(vec![0, 2, 1 << 33]));
+        assert_eq!(newly_deleted(&[], &[], 0).await.unwrap(), None);
+    }

+    #[test]
+    fn classifies_delete_files() {
+        let file = |content, format| {
+            DataFileBuilder::default().file_path("d".to_string()).file_format(format)
+                .content(content).record_count(1).partition(Struct::empty())
+                .file_size_in_bytes(1).build().unwrap()
+        };
+        let pos_v2 = file(DataContentType::PositionDeletes, DataFileFormat::Parquet);
+        let eq = file(DataContentType::EqualityDeletes, DataFileFormat::Parquet);
+        let dv = file(DataContentType::PositionDeletes, DataFileFormat::Puffin);
+        assert!(is_deletion_vector(&dv));
+        assert!(!is_deletion_vector(&pos_v2) && !is_deletion_vector(&eq));
+        assert!(v2_delete_file_error(&eq).to_string().contains("equality delete file ('d')"));
+        assert!(v2_delete_file_error(&pos_v2).to_string().contains("V2 position delete file ('d')"));
+    }

┌───────────────────────┬─────────┬──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│ Point in your comment │ Status  │                                                           Evidence                                                           │
├───────────────────────┼─────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ a. Replays become     │ Not     │ The only change in IcebergInputEndpointInner::push_changed_files is where it reports health. It still pushes                 │
│ double deletes        │ fixed   │ QueueEntry::Rollback and re-reads every file, and IcebergInputEndpointInner::drain_batch_stream is unchanged. I compared the │
│                       │         │  code; I did not run it.                                                                                                     │
├───────────────────────┼─────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Errors that can never │ Not     │ input::is_permanent_read_error still matches only FeatureUnsupported, and IcebergReaderConfig::max_retries() still defaults  │
│  succeed are retried  │ fixed   │ to u32::MAX.                                                                                                                 │
│ forever               │         │                                                                                                                              │
├───────────────────────┼─────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ b. Never reports a    │ Fixed   │ The new IcebergInputEndpointInner::retry marks the connector unhealthy on every failed attempt. push_changed_files now does  │
│ problem               │         │ too; before, it did only when it gave up. The follow loop marks the connector healthy once a snapshot's changes are read.    │
├───────────────────────┼─────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ b. Never gives up;    │ Not     │ Errors are still turned into strings before the check. I tested a missing deletion vector with default settings: after 8 s   │
│ discards the error    │ fixed   │ it was still retrying (4 attempts, 4 unhealthy reports). The error DataInvalid => Failed to open file … No such file or      │
│ kind                  │         │ directory is not treated as permanent.                                                                                       │
├───────────────────────┼─────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ c. Manifest reads     │ Fixed   │ read_manifest_list and read_manifest now go through retry. Manifest, deletion-vector, file-size and data-file reads now      │
│ never retried         │         │ follow one retry policy.                                                                                                     │
└───────────────────────┴─────────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

from my friend:
by itself it does not stop repeated rows, and a full fix costs a lot of memory

A retry reads all files of the snapshot again, also the files that it already
queued. A full fix must remember which files it queued and retry only the
others. It must also hold each file in memory until it reads the file to the
end. That is cheap for a new deletion vector, because it keeps only the deleted
rows. But a removed file can be large, and holding it costs as much memory as
the file. Inserts and copy-on-write deletes on main already repeat rows the same
way, so the follow-up will fix all of them, with a test that makes a read fail.

I'd hold this PR: the fresh review found a new high-severity bug, I reproduced it through the connector, and a second finding comes with it. The two fixes you asked for landed (a missing file now stops at once), but the new check has a flaw of its own.

Your table (comment 6636088929), re-checked at 2e3a140c9

The only code change since my last review: input::is_permanent_read_error now also matches "datainvalid" and "notfound". The author also added a test and one docs sentence.

┌───────────────────────┬─────────────────────┬──────────────────────────────────────────────────────────────────────────────────────────────────────────────────┐
│         Point         │       Status        │                                                     Evidence                                                     │
├───────────────────────┼─────────────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ a. Replays become     │ Not fixed; the      │ His main argument is right: on main, a retry already re-queues inserts and copy-on-write deletes; the            │
│ double deletes        │ author wants a      │ QueueEntry::Rollback doc comment says so. His cost argument is wrong: a per-file count of rows already queued    │
│                       │ follow-up           │ would let a retry skip them without holding any file in memory (my inference; I haven't built it).               │
├───────────────────────┼─────────────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ Errors that can never │                     │ A missing deletion vector stops after 1 attempt with 1 unhealthy report, on the local filesystem, opendal's      │
│  succeed are retried  │ Fixed               │ filesystem and S3 (MinIO). I ran all three.                                                                      │
│ forever               │                     │                                                                                                                  │
├───────────────────────┼─────────────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ b. Never reports a    │ Fixed (unchanged)   │                                                                                                                  │
│ problem               │                     │                                                                                                                  │
├───────────────────────┼─────────────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ b. Never gives up;    │                     │                                                                                                                  │
│ discards the error    │ Partly fixed        │ It gives up now, but it still decides by matching words anywhere in the message, file path included (finding 3). │
│ kind                  │                     │                                                                                                                  │
├───────────────────────┼─────────────────────┼──────────────────────────────────────────────────────────────────────────────────────────────────────────────────┤
│ c. Manifest reads     │ Fixed (unchanged)   │                                                                                                                  │
│ never retried         │                     │                                                                                                                  │
└───────────────────────┴─────────────────────┴──────────────────────────────────────────────────────────────────────────────────────────────────────────────────┘

New findings

1. High: deletion vectors left behind by Iceberg Java 1.9 or earlier cause double deletes.
- Cause: before 1.10.0, Iceberg Java left a data file's deletion vector in place when a copy-on-write change removed the file. The spec rule ("When removing a data file, writers must also remove any deletion vector that applies to that data file") and Java's removeDanglingDeletesFor both arrived in 1.10.0. I checked the apache/iceberg 1.9.0, 1.9.2 and 1.10.0 tags.
- Effect: input::pair_deletion_vectors finds no removed vector for the file (/Users/leonid/projects/feldera.pr7590/crates/iceberg/src/input.rs:669). The connector then deletes every row of the file, including rows the vector had already deleted.
- Reproduced: I ran the PR's own e2e test on its own fixture, written by Spark 3.5.7 with Iceberg 1.9.2 instead of 1.10.0. The connector emits weight −1 for ids 50, 51, 52, 60, 61, 62, 70, 71 and 72. The read-count check still passes (104 = 104); only the row comparison catches it.
- Keyed tables: a delete removes the key's current value (per add_input_map's documented semantics). So the live rows u52, u62 and u72 would disappear from Feldera. This is inference; I didn't run a keyed table.

2. Medium: a valid later commit stops follow mode with a fatal error that blames the writer.
- Cause: Java drops a leftover vector (dropDeleteFilesOlderThan, present in both 1.9 and 1.10) once every live data file is newer than it. So an ordinary later DELETE removes the vector, and input::pair_deletion_vectors treats that as a spec violation (/Users/leonid/projects/feldera.pr7590/crates/iceberg/src/input.rs:692).
- Reproduced on the REST catalog:
  - Iceberg 1.9.2 runs a merge-on-read DELETE, then a copy-on-write DELETE.
  - Iceberg 1.10.0 then appends and runs a copy-on-write DELETE of another row.
  - The connector stops with: FATAL error … removes the deletion vector of data file '…' without replacing it or removing the file; the table breaks the Iceberg spec … Fix the writer and rewrite the table …
- Why the message misleads: the 1.10 writer is following the spec, and the writer that left the vector behind was following the spec of its time. Telling the user to fix the writer doesn't help.

An independent reviewer that I ran in parallel found 1 and 2; I reproduced both myself.

3. Medium-low: the new error check matches words inside file paths.
- input::is_permanent_read_error (/Users/leonid/projects/feldera.pr7590/crates/iceberg/src/input.rs:383) searches the whole message, which contains the path twice.
- Reproduced: I pointed opendal's S3 client at a closed port and read three paths:
  - code=Timeout: still retrying after 30 s.
  - code=NotFound or result=DataInvalid: gave up after the first attempt (7 s).
- Partition values are user data, so a table partitioned by an error code fails on its first temporary error.
- The local-filesystem storage labels every I/O error DataInvalid, so temporary local errors are now permanent too. I found this by reading local_fs.rs; it affects only metadata_location tables.
- Fix: check iceberg::Error::kind() and the source error's kind, not the formatted text.

4. Low: docs and tests.
- There's no changelog entry, though the PR marks the V2 rejection as a breaking change.
- The README's local catalog, tabulario/iceberg-rest:1.6.0, cannot run the new V3 test: Spark fails with "Unsupported format version: v3 (supported: v2)". The image CI uses works.
- The fixture pins Iceberg 1.10.0, so the tests cannot catch findings 1 and 2.
- The fixture's comment that the reader fills created_at from the partition is wrong (/Users/leonid/projects/feldera.pr7590/crates/iceberg/src/test/create_v3_tables.py:82). Spark writes created_at into every file, and iceberg-rust uses the partition value only when the column is missing. So the partition values the PR passes for deletion-vector reads are never exercised.
- The max_retries doc still describes snapshot reads only.

Tests I ran

┌─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┬──────────────────────┐
│                                                                  Test                                                                   │        Result        │
├─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼──────────────────────┤
│ feldera-iceberg unit tests                                                                                                              │ 32/32 pass           │
├─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼──────────────────────┤
│ rest_follow_v3_reads_only_changed_rows and rest_follow_rejects_v2_delete_files, against CI's catalog image                              │ both pass            │
│ (apache/iceberg-rest-fixture:1.9.1) with MinIO                                                                                          │                      │
├─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼──────────────────────┤
│ Same V3 test, table written by Iceberg 1.9.2                                                                                            │ fails (finding 1)    │
├─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┼──────────────────────┤
│ Same harness, 1.9.2 writes then 1.10.0 writes                                                                                           │ fatal error (finding │
│                                                                                                                                         │  2)                  │
└─────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────┴──────────────────────┘

On GitHub, still only the main check has run on this commit.

Possible fixes for 1 and 2 (the author's choice)

- Index the live vectors. IcebergInputEndpointInner::find_v2_delete_file already reads every delete manifest once. That same scan could build a map from data-file path to live deletion vector, updated per snapshot. The connector would apply the live vector when it reads a removed file, and ignore a removed vector whose data file is no longer live.
- Cheaper partial fix for 2. Ignore an unmatched removed vector whose sequence number is below the smallest min_sequence_number among the snapshot's data manifests, which mirrors the rule Java uses to drop it. Such a vector can't apply to any live file. This is my inference; I haven't built it.
- At the least, document that the table's writers must be Iceberg Java 1.10 or later.