fix(cache): drop schema-id from the manifest-list cache key

crates/iceberg/src/io/object_cache.rs

@@ -22,7 +22,7 @@ use crate::encryption::EncryptionManager;
 use crate::io::FileIO;
 use crate::spec::{
     FormatVersion, Manifest, ManifestFile, ManifestList, ManifestListReader, ManifestReader,
-    SchemaId, SnapshotRef, TableMetadataRef,
+    SnapshotRef, TableMetadataRef,
 };
 use crate::{Error, ErrorKind, Result};
 
@@ -36,7 +36,7 @@ pub(crate) enum CachedItem {
 
 #[derive(Clone, Debug, Hash, Eq, PartialEq)]
 pub(crate) enum CachedObjectKey {
-    ManifestList((String, FormatVersion, SchemaId)),
+    ManifestList((String, FormatVersion)),
     // 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
@@ -159,7 +159,6 @@ impl ObjectCache {
         let key = CachedObjectKey::ManifestList((
             snapshot.manifest_list().to_string(),
             table_metadata.format_version,
-            snapshot.schema_id().unwrap(),
         ));
         let cache_entry = self
             .cache
@@ -214,6 +213,7 @@ impl ObjectCache {
 
 #[cfg(test)]
 mod tests {
+    use std::collections::HashMap;
     use std::fs;
 
     use minijinja::value::Value;
@@ -226,7 +226,8 @@ mod tests {
     use crate::io::{FileIO, OutputFile};
     use crate::spec::{
         DataContentType, DataFileBuilder, DataFileFormat, Literal, ManifestEntry,
-        ManifestListWriter, ManifestStatus, ManifestWriterBuilder, Struct, TableMetadata,
+        ManifestListWriter, ManifestStatus, ManifestWriterBuilder, Operation, Snapshot, Struct,
+        Summary, TableMetadata,
     };
     use crate::table::Table;
     use crate::test_utils::test_runtime;
@@ -444,6 +445,51 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn test_get_manifest_list_with_no_schema_id() {
+        let mut fixture = TableTestFixture::new();
+        fixture.setup_manifest_files().await;
+
+        let current_snapshot = fixture.table.metadata().current_snapshot().unwrap();
+
+        // The spec marks `schema-id` optional in every version (v1-v3), so a
+        // snapshot may omit it; fetching its manifest list must not depend on the
+        // schema-id being present.
+        let snapshot_without_schema_id: SnapshotRef = Snapshot::builder()
+            .with_snapshot_id(current_snapshot.snapshot_id())
+            .with_sequence_number(current_snapshot.sequence_number())
+            .with_timestamp_ms(current_snapshot.timestamp_ms())
+            .with_manifest_list(current_snapshot.manifest_list())
+            .with_summary(Summary {
+                operation: Operation::Append,
+                additional_properties: HashMap::new(),
+            })
+            .build()
+            .into();
+        assert!(snapshot_without_schema_id.schema_id().is_none());
+        assert!(current_snapshot.schema_id().is_some());
+
+        let object_cache = ObjectCache::new(fixture.table.file_io().clone(), None);
+
+        // Cold miss: the schema-id-less snapshot populates the cache.
+        let inserted = object_cache
+            .get_manifest_list(&snapshot_without_schema_id, &fixture.table.metadata_ref())
+            .await
+            .unwrap();
+        assert_eq!(inserted.entries().len(), 1);
+
+        // Warm hit: the original snapshot carries a schema-id but points at the same
+        // manifest-list location, so it returns the same cached entry.
+        let cached = object_cache
+            .get_manifest_list(current_snapshot, &fixture.table.metadata_ref())
+            .await
+            .unwrap();
+        assert!(
+            Arc::ptr_eq(&inserted, &cached),
+            "snapshots with and without schema-id at one location must share a cache entry"
+        );
+    }
+
     #[tokio::test]
     async fn test_get_manifest_keys_on_first_row_id() {
         use crate::spec::{NestedField, PartitionSpec, PrimitiveType, Schema, SchemaRef, Type};