feat: add free-form metadata map to DataFragment

docs/src/format/table/transaction.md

@@ -410,7 +410,7 @@ are rebaseable conflicts with Update:
 
 ### UpdateConfig
 
-Modifies table configuration, table metadata, schema metadata, or field metadata without changing data.
+Modifies table configuration, table metadata, schema metadata, field metadata, or fragment metadata without changing data.
 
 <details>
 <summary>UpdateConfig protobuf message</summary>
@@ -423,11 +423,11 @@ Modifies table configuration, table metadata, schema metadata, or field metadata
 
 #### UpdateConfig Compatibility
 
-An UpdateConfig operation only modifies table config and tends to be compatible with other operations. Here
+An UpdateConfig operation only modifies metadata and tends to be compatible with other operations. Here
 are the operations that conflict with UpdateConfig:
 
 - Overwrite
-- UpdateConfig (only if the two operations modify the same config)
+- UpdateConfig (only if the two operations modify the same config or the same fragment/field/schema metadata)
 
 ### DataReplacement
 

protos/table.proto

@@ -346,6 +346,10 @@ message DataFragment {
   // deletion tombstones. To compute the current number of rows, subtract
   // `deletion_file.num_deleted_rows` from this value.
   uint64 physical_rows = 4;
+
+  // Free-form key-value metadata for the fragment.
+  // Intended for lightweight metadata such as creation time, process ID, etc.
+  map<string, string> metadata = 11;
 }
 
 message DataFile {

protos/transaction.proto

@@ -287,12 +287,13 @@ message Transaction {
   }
 
   // An operation that updates the table config, table metadata, schema metadata,
-  // or field metadata.
+  // field metadata, or fragment metadata.
   message UpdateConfig {
     UpdateMap config_updates = 6;
     UpdateMap table_metadata_updates = 7;
     UpdateMap schema_metadata_updates = 8;
     map<int32, UpdateMap> field_metadata_updates = 9;
+    map<uint64, UpdateMap> fragment_metadata_updates = 10;
 
     // Deprecated -------------------------------
     map<string, string> upsert_values = 1;

rust/lance-table/benches/manifest_intern.rs

@@ -72,6 +72,7 @@ fn make_uniform_pb_fragments(n: u64, num_fields: usize) -> Vec<pb::DataFragment>
                     version_bytes.clone(),
                 ),
             ),
+            metadata: Default::default(),
         })
         .collect()
 }
@@ -148,6 +149,7 @@ fn make_diverse_pb_fragments(
                         version_payloads[version_idx].clone(),
                     ),
                 ),
+                metadata: Default::default(),
             }
         })
         .collect()

rust/lance-table/src/format/fragment.rs

@@ -380,6 +380,7 @@ impl DataFileFieldInterner {
             physical_rows,
             last_updated_at_version_meta,
             created_at_version_meta,
+            metadata: p.metadata,
         })
     }
 }
@@ -503,6 +504,10 @@ pub struct Fragment {
     /// Created at version metadata
     #[serde(skip_serializing_if = "Option::is_none")]
     pub created_at_version_meta: Option<RowDatasetVersionMeta>,
+
+    /// Free-form key-value metadata for the fragment.
+    #[serde(default, skip_serializing_if = "HashMap::is_empty")]
+    pub metadata: HashMap<String, String>,
 }
 
 impl Fragment {
@@ -515,6 +520,7 @@ impl Fragment {
             physical_rows: None,
             last_updated_at_version_meta: None,
             created_at_version_meta: None,
+            metadata: HashMap::new(),
         }
     }
 
@@ -554,6 +560,7 @@ impl Fragment {
             row_id_meta: None,
             last_updated_at_version_meta: None,
             created_at_version_meta: None,
+            metadata: HashMap::new(),
         }
     }
 
@@ -680,6 +687,7 @@ impl TryFrom<pb::DataFragment> for Fragment {
                 .created_at_version_sequence
                 .map(RowDatasetVersionMeta::try_from)
                 .transpose()?,
+            metadata: p.metadata,
         })
     }
 }
@@ -721,6 +729,7 @@ impl From<&Fragment> for pb::DataFragment {
             physical_rows: f.physical_rows.unwrap_or_default() as u64,
             last_updated_at_version_sequence,
             created_at_version_sequence,
+            metadata: f.metadata.clone(),
         }
     }
 }
@@ -786,6 +795,43 @@ mod tests {
         assert_eq!(fragment, fragment2);
     }
 
+    #[test]
+    fn test_roundtrip_fragment_with_metadata() {
+        let mut fragment = Fragment::new(1);
+        let schema = ArrowSchema::new(vec![ArrowField::new("x", DataType::Float16, true)]);
+        fragment.add_file_legacy("data.lance", &Schema::try_from(&schema).unwrap());
+        fragment
+            .metadata
+            .insert("created_at".into(), "2026-05-22T00:00:00Z".into());
+        fragment
+            .metadata
+            .insert("process_id".into(), "pid-42".into());
+
+        let proto = pb::DataFragment::from(&fragment);
+        assert_eq!(proto.metadata.len(), 2);
+        assert_eq!(
+            proto.metadata.get("created_at").unwrap(),
+            "2026-05-22T00:00:00Z"
+        );
+
+        let fragment2 = Fragment::try_from(proto).unwrap();
+        assert_eq!(fragment, fragment2);
+    }
+
+    #[test]
+    fn test_roundtrip_fragment_empty_metadata() {
+        let mut fragment = Fragment::new(2);
+        let schema = ArrowSchema::new(vec![ArrowField::new("x", DataType::Float16, true)]);
+        fragment.add_file_legacy("data.lance", &Schema::try_from(&schema).unwrap());
+
+        let proto = pb::DataFragment::from(&fragment);
+        assert!(proto.metadata.is_empty());
+
+        let fragment2 = Fragment::try_from(proto).unwrap();
+        assert_eq!(fragment, fragment2);
+        assert!(fragment2.metadata.is_empty());
+    }
+
     #[test]
     fn test_to_json() {
         let mut fragment = Fragment::new(123);

rust/lance-table/src/format/manifest.rs

@@ -1321,6 +1321,7 @@ mod tests {
                 physical_rows: None,
                 created_at_version_meta: None,
                 last_updated_at_version_meta: None,
+                metadata: HashMap::new(),
             },
             Fragment {
                 id: 1,
@@ -1333,6 +1334,7 @@ mod tests {
                 physical_rows: None,
                 created_at_version_meta: None,
                 last_updated_at_version_meta: None,
+                metadata: HashMap::new(),
             },
         ];
 

rust/lance/src/dataset.rs

@@ -3288,6 +3288,7 @@ impl Dataset {
                 table_metadata_updates: None,
                 schema_metadata_updates: None,
                 field_metadata_updates,
+                fragment_metadata_updates: HashMap::new(),
             },
         )
         .await

rust/lance/src/dataset/metadata.rs

@@ -80,18 +80,21 @@ impl<'a> std::future::IntoFuture for UpdateMetadataBuilder<'a> {
                     table_metadata_updates: None,
                     schema_metadata_updates: None,
                     field_metadata_updates: HashMap::new(),
+                    fragment_metadata_updates: HashMap::new(),
                 },
                 MetadataType::TableMetadata => Operation::UpdateConfig {
                     config_updates: None,
                     table_metadata_updates: Some(update_map),
                     schema_metadata_updates: None,
                     field_metadata_updates: HashMap::new(),
+                    fragment_metadata_updates: HashMap::new(),
                 },
                 MetadataType::SchemaMetadata => Operation::UpdateConfig {
                     config_updates: None,
                     table_metadata_updates: None,
                     schema_metadata_updates: Some(update_map),
                     field_metadata_updates: HashMap::new(),
+                    fragment_metadata_updates: HashMap::new(),
                 },
             };
 
@@ -167,6 +170,7 @@ impl<'a> std::future::IntoFuture for UpdateFieldMetadataBuilder<'a> {
                     table_metadata_updates: None,
                     schema_metadata_updates: None,
                     field_metadata_updates: self.field_metadata_updates,
+                    fragment_metadata_updates: HashMap::new(),
                 },
             )
             .await?;

rust/lance/src/dataset/optimize.rs

@@ -1691,6 +1691,7 @@ mod tests {
             physical_rows: Some(0),
             last_updated_at_version_meta: None,
             created_at_version_meta: None,
+            metadata: HashMap::new(),
         };
         let single_bin = CandidateBin {
             fragments: vec![fragment.clone()],

rust/lance/src/dataset/schema_evolution.rs

@@ -1046,6 +1046,7 @@ mod test {
                         physical_rows: Some(50),
                         last_updated_at_version_meta: None,
                         created_at_version_meta: None,
+                        metadata: HashMap::new(),
                     }))
                 } else {
                     Ok(None)