feat: inherit data-file first_row_id on manifest read

crates/iceberg/src/io/object_cache.rs

@@ -37,7 +37,11 @@ pub(crate) enum CachedItem {
 #[derive(Clone, Debug, Hash, Eq, PartialEq)]
 pub(crate) enum CachedObjectKey {
     ManifestList((String, FormatVersion, SchemaId)),
-    Manifest(String),
+    // The manifest-level `first_row_id` is part of the key because the parsed
+    // manifest inherits it onto its entries: the same physical manifest can be
+    // referenced with different offsets across snapshots and branches, so it
+    // cannot be shared under the path alone.
+    Manifest((String, Option<u64>)),
 }
 
 /// Caches metadata objects deserialized from immutable files
@@ -105,7 +109,10 @@ impl ObjectCache {
                 .map(Arc::new);
         }
 
-        let key = CachedObjectKey::Manifest(manifest_file.manifest_path.clone());
+        let key = CachedObjectKey::Manifest((
+            manifest_file.manifest_path.clone(),
+            manifest_file.first_row_id,
+        ));
 
         let cache_entry = self
             .cache
@@ -434,4 +441,73 @@ mod tests {
             "1.parquet"
         );
     }
+
+    #[tokio::test]
+    async fn test_get_manifest_keys_on_first_row_id() {
+        use crate::spec::{NestedField, PartitionSpec, PrimitiveType, Schema, SchemaRef, Type};
+
+        let io = FileIO::new_with_memory();
+        let path = "memory:///metadata/first_row_id_manifest.avro";
+
+        let schema: SchemaRef = Schema::builder()
+            .with_fields(vec![
+                NestedField::optional(1, "id", Type::Primitive(PrimitiveType::Long)).into(),
+            ])
+            .build()
+            .unwrap()
+            .into();
+        let partition_spec = PartitionSpec::builder(schema.clone())
+            .with_spec_id(0)
+            .build()
+            .unwrap();
+
+        let mut writer = ManifestWriterBuilder::new(
+            io.new_output(path).unwrap(),
+            Some(1),
+            schema,
+            partition_spec,
+        )
+        .build_v3_data();
+        writer
+            .add_entry(
+                ManifestEntry::builder()
+                    .status(ManifestStatus::Added)
+                    .data_file(
+                        DataFileBuilder::default()
+                            .content(DataContentType::Data)
+                            .file_path(path.to_string())
+                            .file_format(DataFileFormat::Parquet)
+                            .file_size_in_bytes(100)
+                            .record_count(1)
+                            .build()
+                            .unwrap(),
+                    )
+                    .build(),
+            )
+            .unwrap();
+        let manifest_file = writer.write_manifest_file().await.unwrap();
+
+        // Two manifest-list references to the same physical manifest carrying
+        // different offsets, as time travel or two branches would produce.
+        let mut manifest_file_a = manifest_file.clone();
+        manifest_file_a.first_row_id = Some(1000);
+        let mut manifest_file_b = manifest_file;
+        manifest_file_b.first_row_id = Some(2000);
+
+        let object_cache = ObjectCache::new(io, None);
+
+        let manifest_a = object_cache.get_manifest(&manifest_file_a).await.unwrap();
+        let manifest_b = object_cache.get_manifest(&manifest_file_b).await.unwrap();
+
+        // A cache keyed on path alone would serve manifest_a's inherited ids for
+        // the second read; keying on first_row_id keeps them distinct.
+        assert_eq!(
+            manifest_a.entries()[0].data_file().first_row_id(),
+            Some(1000)
+        );
+        assert_eq!(
+            manifest_b.entries()[0].data_file().first_row_id(),
+            Some(2000)
+        );
+    }
 }