perf: intern DataFile fields/column_indices to reduce manifest memory (#6477)

## Summary

- Change `DataFile.fields` and `DataFile.column_indices` from `Vec<i32>`
to `Arc<[i32]>` so that fragments with identical field lists share a
single heap allocation
- Add `DataFileFieldInterner` that deduplicates these slices during
manifest deserialization
- In homogeneous tables (the common case), every fragment carries the
same field list, so at 20M fragments this saves **~2.4 GB** of redundant
heap allocations

## Motivation

When dataset manifests grow large (>1 GB with millions of fragments),
opening the dataset becomes very expensive in terms of memory. Each
`DataFile` previously owned its own `Vec<i32>` for `fields` and
`column_indices`, even though in most tables every fragment has the
exact same field list. This PR deduplicates those allocations at
deserialization time.

### Per-fragment memory breakdown (before)

| Field | Size per fragment |
|-------|------------------|
| `fields: Vec<i32>` (10 fields) | ~64 bytes |
| `column_indices: Vec<i32>` (10 cols) | ~64 bytes |
| **Total redundant** | **~128 bytes x 20M = ~2.4 GB** |

### After this change

With interning, all 20M fragments share a single `Arc<[i32]>` allocation
(~80 bytes total instead of 2.4 GB).

## Changes

- **`lance-table/src/format/fragment.rs`** — Core struct change
(`Vec<i32>` → `Arc<[i32]>`), custom `Serialize`/`Deserialize` impls, and
`DataFileFieldInterner`
- **`lance-table/src/format/manifest.rs`** — Use interner during
manifest deserialization
- **`lance/src/dataset/fragment.rs`**, **`merge_insert.rs`**,
**`io/commit.rs`** — Tombstoning and field-remapping rebuilt as new
`Arc<[i32]>` instead of in-place mutation
- **`python/src/fragment.rs`**, **`java/lance-jni/src/fragment.rs`** —
FFI boundary conversions
- Various test files — Updated struct literals and assertions

## Compatibility

- No format change — protobuf schema is unchanged
- Serde JSON output is identical (custom impl serializes `Arc<[i32]>` as
`[i32]`)
- All public API signatures that take `Vec<i32>` (e.g.,
`DataFile::new()`, `Fragment::add_file()`) still accept `Vec<i32>` and
convert internally

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: beinan

java/lance-jni/src/fragment.rs

@@ -536,8 +536,8 @@ impl IntoJava for &FragmentUpdateResult {
 impl IntoJava for &DataFile {
     fn into_java<'a>(self, env: &mut JNIEnv<'a>) -> Result<JObject<'a>> {
         let path = env.new_string(self.path.clone())?.into();
-        let fields = JLance(self.fields.clone()).into_java(env)?;
-        let column_indices = JLance(self.column_indices.clone()).into_java(env)?;
+        let fields = JLance(self.fields.to_vec()).into_java(env)?;
+        let column_indices = JLance(self.column_indices.to_vec()).into_java(env)?;
         let file_size_bytes = match self.file_size_bytes.get() {
             Some(f) => JLance(u64::from(f) as i64).into_java(env)?,
             None => JObject::null(),
@@ -775,8 +775,8 @@ impl FromJObjectWithEnv<DataFile> for JObject<'_> {
         let base_id = get_base_id(env, self)?;
         Ok(DataFile {
             path,
-            fields,
-            column_indices,
+            fields: fields.into(),
+            column_indices: column_indices.into(),
             file_major_version,
             file_minor_version,
             file_size_bytes,

python/src/debug.rs

@@ -61,8 +61,8 @@ impl PrettyPrintableFragment {
                 let schema = schema.project_by_ids(&file.fields, false);
                 PrettyPrintableDataFile {
                     path: file.path.clone(),
-                    fields: file.fields.clone(),
-                    column_indices: file.column_indices.clone(),
+                    fields: file.fields.to_vec(),
+                    column_indices: file.column_indices.to_vec(),
                     schema,
                     major_version: file.file_major_version,
                     minor_version: file.file_minor_version,

python/src/fragment.rs

@@ -846,10 +846,12 @@ impl FromPyObject<'_> for PyLance<DataFile> {
     fn extract_bound(ob: &pyo3::Bound<'_, PyAny>) -> PyResult<Self> {
         let file_size_bytes: Option<u64> = ob.getattr("file_size_bytes")?.extract()?;
         let file_size_bytes = CachedFileSize::new(file_size_bytes.unwrap_or(0));
+        let fields: Vec<i32> = ob.getattr("fields")?.extract()?;
+        let column_indices: Vec<i32> = ob.getattr("column_indices")?.extract()?;
         Ok(Self(DataFile {
             path: ob.getattr("path")?.extract()?,
-            fields: ob.getattr("fields")?.extract()?,
-            column_indices: ob.getattr("column_indices")?.extract()?,
+            fields: fields.into(),
+            column_indices: column_indices.into(),
             file_major_version: ob.getattr("file_major_version")?.extract()?,
             file_minor_version: ob.getattr("file_minor_version")?.extract()?,
             file_size_bytes,
@@ -872,8 +874,8 @@ impl<'py> IntoPyObject<'py> for PyLance<&DataFile> {
         let file_size_bytes = self.0.file_size_bytes.get().map(u64::from);
         cls.call1((
             &self.0.path,
-            self.0.fields.clone(),
-            self.0.column_indices.clone(),
+            self.0.fields.to_vec(),
+            self.0.column_indices.to_vec(),
             self.0.file_major_version,
             self.0.file_minor_version,
             file_size_bytes,

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

@@ -1,15 +1,17 @@
 // SPDX-License-Identifier: Apache-2.0
 // SPDX-FileCopyrightText: Copyright The Lance Authors
 
+use std::collections::HashMap;
 use std::num::NonZero;
+use std::sync::Arc;
 
 use deepsize::DeepSizeOf;
 use lance_core::Error;
 use lance_file::format::{MAJOR_VERSION, MINOR_VERSION};
 use lance_file::version::LanceFileVersion;
 use lance_io::utils::CachedFileSize;
 use object_store::path::Path;
-use serde::{Deserialize, Serialize};
+use serde::{Deserialize, Deserializer, Serialize, Serializer};
 
 use crate::format::pb;
 
@@ -22,12 +24,16 @@ use lance_core::error::Result;
 /// Lance Data File
 ///
 /// A data file is one piece of file storing data.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, DeepSizeOf)]
+#[derive(Debug, Clone, PartialEq, Eq, DeepSizeOf)]
 pub struct DataFile {
     /// Relative path of the data file to dataset root.
     pub path: String,
     /// The ids of fields in this file.
-    pub fields: Vec<i32>,
+    ///
+    /// When identical across many fragments (common case), multiple `DataFile`
+    /// instances share a single heap allocation via `Arc`, significantly
+    /// reducing manifest memory for large tables.
+    pub fields: Arc<[i32]>,
     /// The offsets of the fields listed in `fields`, empty in v1 files
     ///
     /// Note that -1 is a possibility and it indices that the field has
@@ -37,13 +43,10 @@ pub struct DataFile {
     /// `column_indices`; such columns are ignored by field-id–based projection.
     /// For example, some fields, such as blob fields, occupy multiple
     /// columns in the file but only have a single field id.
-    #[serde(default)]
-    pub column_indices: Vec<i32>,
+    pub column_indices: Arc<[i32]>,
     /// The major version of the file format used to write this file.
-    #[serde(default)]
     pub file_major_version: u32,
     /// The minor version of the file format used to write this file.
-    #[serde(default)]
     pub file_minor_version: u32,
 
     /// The size of the file in bytes, if known.
@@ -53,6 +56,52 @@ pub struct DataFile {
     pub base_id: Option<u32>,
 }
 
+// Custom Serialize: convert Arc<[i32]> to slice for transparent JSON output
+impl Serialize for DataFile {
+    fn serialize<S: Serializer>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error> {
+        use serde::ser::SerializeStruct;
+        let mut s = serializer.serialize_struct("DataFile", 7)?;
+        s.serialize_field("path", &self.path)?;
+        s.serialize_field("fields", self.fields.as_ref())?;
+        s.serialize_field("column_indices", self.column_indices.as_ref())?;
+        s.serialize_field("file_major_version", &self.file_major_version)?;
+        s.serialize_field("file_minor_version", &self.file_minor_version)?;
+        s.serialize_field("file_size_bytes", &self.file_size_bytes)?;
+        s.serialize_field("base_id", &self.base_id)?;
+        s.end()
+    }
+}
+
+// Custom Deserialize: read Vec<i32> and convert to Arc<[i32]>
+impl<'de> Deserialize<'de> for DataFile {
+    fn deserialize<D: Deserializer<'de>>(deserializer: D) -> std::result::Result<Self, D::Error> {
+        #[derive(Deserialize)]
+        struct DataFileHelper {
+            path: String,
+            fields: Vec<i32>,
+            #[serde(default)]
+            column_indices: Vec<i32>,
+            #[serde(default)]
+            file_major_version: u32,
+            #[serde(default)]
+            file_minor_version: u32,
+            file_size_bytes: CachedFileSize,
+            base_id: Option<u32>,
+        }
+
+        let helper = DataFileHelper::deserialize(deserializer)?;
+        Ok(Self {
+            path: helper.path,
+            fields: Arc::from(helper.fields),
+            column_indices: Arc::from(helper.column_indices),
+            file_major_version: helper.file_major_version,
+            file_minor_version: helper.file_minor_version,
+            file_size_bytes: helper.file_size_bytes,
+            base_id: helper.base_id,
+        })
+    }
+}
+
 impl DataFile {
     pub fn new(
         path: impl Into<String>,
@@ -65,8 +114,8 @@ impl DataFile {
     ) -> Self {
         Self {
             path: path.into(),
-            fields,
-            column_indices,
+            fields: Arc::from(fields),
+            column_indices: Arc::from(column_indices),
             file_major_version,
             file_minor_version,
             file_size_bytes: file_size_bytes.into(),
@@ -82,8 +131,8 @@ impl DataFile {
     ) -> Self {
         Self {
             path: path.into(),
-            fields: vec![],
-            column_indices: vec![],
+            fields: Arc::from([]),
+            column_indices: Arc::from([]),
             file_major_version,
             file_minor_version,
             file_size_bytes: Default::default(),
@@ -158,8 +207,8 @@ impl From<&DataFile> for pb::DataFile {
     fn from(df: &DataFile) -> Self {
         Self {
             path: df.path.clone(),
-            fields: df.fields.clone(),
-            column_indices: df.column_indices.clone(),
+            fields: df.fields.to_vec(),
+            column_indices: df.column_indices.to_vec(),
             file_major_version: df.file_major_version,
             file_minor_version: df.file_minor_version,
             file_size_bytes: df.file_size_bytes.get().map_or(0, |v| v.get()),
@@ -174,14 +223,77 @@ impl TryFrom<pb::DataFile> for DataFile {
     fn try_from(proto: pb::DataFile) -> Result<Self> {
         Ok(Self {
             path: proto.path,
-            fields: proto.fields,
-            column_indices: proto.column_indices,
+            fields: Arc::from(proto.fields),
+            column_indices: Arc::from(proto.column_indices),
+            file_major_version: proto.file_major_version,
+            file_minor_version: proto.file_minor_version,
+            file_size_bytes: CachedFileSize::new(proto.file_size_bytes),
+            base_id: proto.base_id,
+        })
+    }
+}
+
+/// Interns `fields` and `column_indices` slices so that `DataFile` instances
+/// with identical content share a single `Arc<[i32]>` allocation.
+///
+/// At 20M fragments the deduplication typically saves multiple GB of heap
+/// because every fragment in a homogeneous table carries the same field list.
+#[derive(Default)]
+pub struct DataFileFieldInterner {
+    fields: HashMap<Vec<i32>, Arc<[i32]>>,
+    column_indices: HashMap<Vec<i32>, Arc<[i32]>>,
+}
+
+impl DataFileFieldInterner {
+    /// Intern a `Vec<i32>`: if the same content was seen before, return a
+    /// clone of the existing `Arc`; otherwise store and return a new one.
+    fn intern(cache: &mut HashMap<Vec<i32>, Arc<[i32]>>, v: Vec<i32>) -> Arc<[i32]> {
+        cache
+            .entry(v)
+            .or_insert_with_key(|k| Arc::from(k.as_slice()))
+            .clone()
+    }
+
+    /// Convert a protobuf `DataFile`, interning `fields` and `column_indices`.
+    pub fn intern_data_file(&mut self, proto: pb::DataFile) -> Result<DataFile> {
+        Ok(DataFile {
+            path: proto.path,
+            fields: Self::intern(&mut self.fields, proto.fields),
+            column_indices: Self::intern(&mut self.column_indices, proto.column_indices),
             file_major_version: proto.file_major_version,
             file_minor_version: proto.file_minor_version,
             file_size_bytes: CachedFileSize::new(proto.file_size_bytes),
             base_id: proto.base_id,
         })
     }
+
+    /// Convert a protobuf `DataFragment`, interning fields within its data files.
+    pub fn intern_fragment(&mut self, p: pb::DataFragment) -> Result<Fragment> {
+        let physical_rows = if p.physical_rows > 0 {
+            Some(p.physical_rows as usize)
+        } else {
+            None
+        };
+        Ok(Fragment {
+            id: p.id,
+            files: p
+                .files
+                .into_iter()
+                .map(|f| self.intern_data_file(f))
+                .collect::<Result<_>>()?,
+            deletion_file: p.deletion_file.map(DeletionFile::try_from).transpose()?,
+            row_id_meta: p.row_id_sequence.map(RowIdMeta::try_from).transpose()?,
+            physical_rows,
+            last_updated_at_version_meta: p
+                .last_updated_at_version_sequence
+                .map(RowDatasetVersionMeta::try_from)
+                .transpose()?,
+            created_at_version_meta: p
+                .created_at_version_sequence
+                .map(RowDatasetVersionMeta::try_from)
+                .transpose()?,
+        })
+    }
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, DeepSizeOf)]
@@ -624,9 +736,9 @@ mod tests {
     fn data_file_validate_allows_extra_columns() {
         let data_file = DataFile {
             path: "foo.lance".to_string(),
-            fields: vec![1, 2],
+            fields: Arc::from([1, 2]),
             // One extra column without a field id mapping
-            column_indices: vec![0, 1, 2],
+            column_indices: Arc::from([0, 1, 2]),
             file_major_version: MAJOR_VERSION as u32,
             file_minor_version: MINOR_VERSION as u32,
             file_size_bytes: Default::default(),

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

@@ -17,6 +17,7 @@ use std::sync::Arc;
 
 use super::Fragment;
 use crate::feature_flags::{FLAG_STABLE_ROW_IDS, has_deprecated_v2_feature_flag};
+use crate::format::fragment::DataFileFieldInterner;
 use crate::format::pb;
 use lance_core::cache::LanceCache;
 use lance_core::datatypes::Schema;
@@ -426,7 +427,7 @@ impl Manifest {
         let fragment_max_id = self
             .fragments
             .iter()
-            .flat_map(|f| f.files.iter().flat_map(|file| file.fields.as_slice()))
+            .flat_map(|f| f.files.iter().flat_map(|file| file.fields.iter()))
             .max()
             .copied();
         let fragment_max_id = fragment_max_id.unwrap_or(-1);
@@ -863,10 +864,11 @@ impl TryFrom<pb::Manifest> for Manifest {
             }),
             _ => None,
         };
+        let mut interner = DataFileFieldInterner::default();
         let fragments = Arc::new(
             p.fragments
                 .into_iter()
-                .map(Fragment::try_from)
+                .map(|f| interner.intern_fragment(f))
                 .collect::<Result<Vec<_>>>()?,
         );
         let fragment_offsets = compute_fragment_offsets(fragments.as_slice());

rust/lance/src/dataset/fragment.rs

@@ -1209,7 +1209,7 @@ impl FileFragment {
         let mut seen_fields = HashSet::new();
         for data_file in &self.metadata.files {
             let last = -1;
-            for field_id in &data_file.fields {
+            for field_id in data_file.fields.iter() {
                 if *field_id <= last {
                     return Err(Error::corrupt_file(
                         self.dataset
@@ -1666,12 +1666,19 @@ impl FileFragment {
         // Mark fields in updated data files as obsolete ("tombstone").
         let updated_fields = updated_fragment.files.last().unwrap().fields.clone();
         for data_file in &mut updated_fragment.files.iter_mut().rev().skip(1) {
-            for field in &mut data_file.fields {
-                if updated_fields.contains(field) {
-                    // Tombstone these fields
-                    *field = -2;
-                }
-            }
+            let new_fields: Arc<[i32]> = data_file
+                .fields
+                .iter()
+                .map(|field| {
+                    if updated_fields.contains(field) {
+                        -2 // Tombstone
+                    } else {
+                        *field
+                    }
+                })
+                .collect::<Vec<_>>()
+                .into();
+            data_file.fields = new_fields;
         }
         // Remove data files that have become entirely tombstoned.
         updated_fragment