feat: add DataFile.create helper for building DataFile metadata (#6427)

## Summary
- Adds `DataFile.create(dataset, path, *, base_id=None)` classmethod
that reads a lance file's metadata and automatically constructs a
`DataFile` with correct field IDs, column indices, file version, and
file size
- Eliminates the need for manual `DataFile` construction when performing
`DataReplacement` operations
- Handles packed structs, structural file versions (v2.1+), subset
columns, and external base paths

Closes #6413

## Test plan
- [x] `test_data_file_create_basic` — verifies fields, column_indices,
version, file_size for a two-column file
- [x] `test_data_file_create_subset_columns` — single column from a
multi-column dataset
- [x] `test_data_file_create_end_to_end` — full DataReplacement
round-trip using the new helper
- [x] `test_data_file_create_unknown_column` — error on column not in
dataset schema
- [x] All existing `test_table_ops.py` tests still pass

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

---------

Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

Author: westonpace

python/python/lance/fragment.py

@@ -250,6 +250,36 @@ def field_ids(self) -> List[int]:
         )
         return self.fields
 
+    @classmethod
+    def create(
+        cls,
+        dataset: "LanceDataset",
+        path: str,
+        *,
+        base_id: Optional[int] = None,
+    ) -> "DataFile":
+        """Create a DataFile by reading metadata from an existing lance file.
+
+        This is a convenience method for creating DataFile metadata needed
+        for operations like DataReplacement. It opens the file, reads its
+        schema and version information, matches columns to the dataset's
+        schema to determine field IDs, and calculates column indices.
+
+        Parameters
+        ----------
+        dataset : LanceDataset
+            The dataset this file will belong to.
+        path : str
+            The path to the data file, relative to the dataset's data directory.
+        base_id : int, optional
+            The base path ID if the file is outside the dataset directory.
+
+        Returns
+        -------
+        DataFile
+        """
+        return _Fragment.create_data_file(dataset._ds, path, base_id=base_id)
+
 
 class LanceFragment(pa.dataset.Fragment):
     def __init__(

python/python/tests/test_table_ops.py

@@ -106,3 +106,95 @@ def test_replacement_after_index(tmp_path: str):
             ),
             read_version=ds3.version,
         )
+
+
+def test_data_file_create_basic(tmp_path: str):
+    """DataFile.create should read file metadata and produce correct fields/indices."""
+    table = pa.table({"a": range(10), "b": range(10, 20)})
+    ds = lance.write_dataset(table, tmp_path)
+
+    # Write a lance file with both columns
+    new_file_name = f"{uuid.uuid4()}.lance"
+    new_file_path = f"{tmp_path}/data/{new_file_name}"
+    with LanceFileWriter(new_file_path) as writer:
+        writer.write_batch(table)
+
+    df = DataFile.create(ds, new_file_name)
+
+    # Should have both field IDs from the dataset
+    frag = ds.get_fragments()[0]
+    expected_fields = frag.data_files()[0].fields
+    assert df.fields == expected_fields
+    assert df.column_indices == [0, 1]
+    assert df.file_major_version == int(stable_version().split(".")[0])
+    assert df.file_minor_version == int(stable_version().split(".")[1])
+    assert df.file_size_bytes is not None and df.file_size_bytes > 0
+
+
+def test_data_file_create_subset_columns(tmp_path: str):
+    """DataFile.create should work for a file with a subset of dataset columns."""
+    table = pa.table({"a": range(10), "b": range(10, 20)})
+    ds = lance.write_dataset(table, tmp_path)
+    ds.add_columns({"c": "a + b"})
+    ds = lance.dataset(tmp_path)
+
+    # Write a file with only column b
+    new_file_name = f"{uuid.uuid4()}.lance"
+    new_file_path = f"{tmp_path}/data/{new_file_name}"
+    with LanceFileWriter(new_file_path, pa.schema([("b", pa.int64())])) as writer:
+        writer.write_batch(pa.table({"b": range(100, 110)}))
+
+    df = DataFile.create(ds, new_file_name)
+
+    # Should only have b's field ID
+    frag = ds.get_fragments()[0]
+    all_fields = frag.data_files()[0].fields
+    # b is the second field in the original data file
+    b_field_id = all_fields[1]
+    assert df.fields == [b_field_id]
+    assert df.column_indices == [0]
+
+
+def test_data_file_create_end_to_end(tmp_path: str):
+    """DataFile.create should work end-to-end with DataReplacement."""
+    table = pa.table({"a": range(100)})
+    ds = lance.write_dataset(table, tmp_path)
+    ds.add_columns({"b": "a + 1"})
+    ds = lance.dataset(tmp_path)
+
+    # Write a replacement file for column b
+    new_file_name = f"{uuid.uuid4()}.lance"
+    new_file_path = f"{tmp_path}/data/{new_file_name}"
+    replacement_data = pa.table({"b": range(200, 300)})
+    with LanceFileWriter(new_file_path, pa.schema([("b", pa.int64())])) as writer:
+        writer.write_batch(replacement_data)
+
+    # Use DataFile.create instead of manual construction
+    df = DataFile.create(ds, new_file_name)
+
+    ds.commit(
+        ds.uri,
+        lance.LanceOperation.DataReplacement(
+            [lance.LanceOperation.DataReplacementGroup(0, df)]
+        ),
+        read_version=ds.version,
+    )
+
+    result = lance.dataset(tmp_path).to_table()
+    assert result.column("b").to_pylist() == list(range(200, 300))
+    assert result.column("a").to_pylist() == list(range(100))
+
+
+def test_data_file_create_unknown_column(tmp_path: str):
+    """DataFile.create should raise an error for a file with unknown columns."""
+    table = pa.table({"a": range(10)})
+    ds = lance.write_dataset(table, tmp_path)
+
+    # Write a file with a column not in the dataset
+    new_file_name = f"{uuid.uuid4()}.lance"
+    new_file_path = f"{tmp_path}/data/{new_file_name}"
+    with LanceFileWriter(new_file_path) as writer:
+        writer.write_batch(pa.table({"z": range(10)}))
+
+    with pytest.raises(Exception, match="z"):
+        DataFile.create(ds, new_file_name)

python/src/fragment.rs

@@ -96,6 +96,20 @@ impl FileFragment {
         Ok(PyLance(metadata))
     }
 
+    #[staticmethod]
+    #[pyo3(signature = (dataset, path, base_id=None))]
+    fn create_data_file(
+        dataset: &Dataset,
+        path: &str,
+        base_id: Option<u32>,
+    ) -> PyResult<PyLance<DataFile>> {
+        let ds = dataset.ds.clone();
+        let data_file = rt()
+            .block_on(None, ds.create_data_file(path, base_id))?
+            .infer_error()?;
+        Ok(PyLance(data_file))
+    }
+
     #[staticmethod]
     #[pyo3(signature = (dataset_uri, fragment_id, reader, **kwargs))]
     fn create(

rust/lance/src/dataset.rs

@@ -29,14 +29,17 @@ use lance_core::utils::tracing::{
 };
 use lance_datafusion::projection::ProjectionPlan;
 use lance_file::datatypes::populate_schema_dictionary;
-use lance_file::reader::FileReaderOptions;
+use lance_file::reader::{FileReader, FileReaderOptions};
 use lance_file::version::LanceFileVersion;
 use lance_index::{IndexType, progress::IndexBuildProgress};
 use lance_io::object_store::{
     LanceNamespaceStorageOptionsProvider, ObjectStore, ObjectStoreParams, StorageOptions,
     StorageOptionsAccessor, StorageOptionsProvider,
 };
-use lance_io::utils::{read_last_block, read_message, read_metadata_offset, read_struct};
+use lance_io::scheduler::{ScanScheduler, SchedulerConfig};
+use lance_io::utils::{
+    CachedFileSize, read_last_block, read_message, read_metadata_offset, read_struct,
+};
 use lance_namespace::LanceNamespace;
 use lance_table::format::{
     DataFile, DataStorageFormat, DeletionFile, Fragment, IndexMetadata, Manifest, RowIdMeta, pb,
@@ -57,6 +60,7 @@ use serde::{Deserialize, Serialize};
 use std::borrow::Cow;
 use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
 use std::fmt::Debug;
+use std::num::NonZero;
 use std::ops::Range;
 use std::pin::Pin;
 use std::sync::Arc;
@@ -1714,14 +1718,116 @@ impl Dataset {
     }
 
     pub(crate) fn data_file_dir(&self, data_file: &DataFile) -> Result<Path> {
-        match data_file.base_id.as_ref() {
+        self.data_file_dir_for_base(data_file.base_id)
+    }
+
+    /// Create a [`DataFile`] by reading metadata from an existing lance file.
+    ///
+    /// This reads the file's schema and version information, matches columns to
+    /// the dataset's schema to determine field IDs, and calculates column indices.
+    /// This is useful for constructing `DataFile` metadata needed for operations
+    /// like [`Operation::DataReplacement`].
+    ///
+    /// # Arguments
+    ///
+    /// * `path` - The path to the data file, relative to the dataset's data directory.
+    /// * `base_id` - The base path ID if the file is outside the dataset directory.
+    pub async fn create_data_file(&self, path: &str, base_id: Option<u32>) -> Result<DataFile> {
+        let data_dir = self.data_file_dir_for_base(base_id)?;
+        let filepath = data_dir.child(path);
+
+        // Get file size
+        let file_size = self.object_store().size(&filepath).await?;
+
+        // Read file metadata
+        let scheduler = ScanScheduler::new(
+            self.object_store.clone(),
+            SchedulerConfig::new(2 * 1024 * 1024 * 1024),
+        );
+        let file = scheduler
+            .open_file(&filepath, &CachedFileSize::new(file_size))
+            .await?;
+        let file_metadata = FileReader::read_all_metadata(&file).await?;
+
+        let file_version = LanceFileVersion::try_from_major_minor(
+            file_metadata.major_version as u32,
+            file_metadata.minor_version as u32,
+        )?;
+
+        // Get top-level column names from file schema in file order
+        let column_names: Vec<&str> = file_metadata
+            .file_schema
+            .fields
+            .iter()
+            .map(|f| f.name.as_str())
+            .collect();
+
+        // Project dataset schema by file column names to get dataset field IDs
+        let projected_ds_schema = self.schema().project(&column_names)?;
+
+        // Walk both schemas in parallel to build fields and column_indices
+        let is_structural = file_version >= LanceFileVersion::V2_1;
+        let ds_fields: Vec<_> = projected_ds_schema.fields_pre_order().collect();
+        let file_fields: Vec<_> = file_metadata.file_schema.fields_pre_order().collect();
+
+        if ds_fields.len() != file_fields.len() {
+            return Err(Error::invalid_input(format!(
+                "Schema mismatch: dataset projection has {} fields but file has {} fields",
+                ds_fields.len(),
+                file_fields.len()
+            )));
+        }
+
+        let mut fields = Vec::new();
+        let mut column_indices = Vec::new();
+        let mut curr_column_idx: i32 = 0;
+        let mut packed_struct_fields_num: usize = 0;
+
+        for (ds_field, file_field) in ds_fields.iter().zip(file_fields.iter()) {
+            if ds_field.name != file_field.name {
+                return Err(Error::invalid_input(format!(
+                    "Schema mismatch: expected field '{}' but file has '{}'",
+                    ds_field.name, file_field.name
+                )));
+            }
+
+            if packed_struct_fields_num > 0 {
+                packed_struct_fields_num -= 1;
+                continue;
+            }
+
+            if file_field.is_packed_struct() {
+                fields.push(ds_field.id);
+                column_indices.push(curr_column_idx);
+                curr_column_idx += 1;
+                packed_struct_fields_num = file_field.children.len();
+            } else if file_field.children.is_empty() || !is_structural {
+                fields.push(ds_field.id);
+                column_indices.push(curr_column_idx);
+                curr_column_idx += 1;
+            }
+        }
+
+        let file_size_nz = NonZero::new(file_size);
+        Ok(DataFile::new(
+            path,
+            fields,
+            column_indices,
+            file_metadata.major_version as u32,
+            file_metadata.minor_version as u32,
+            file_size_nz,
+            base_id,
+        ))
+    }
+
+    /// Resolve the data directory for a given base_id.
+    ///
+    /// If `base_id` is `None`, returns the default data directory.
+    fn data_file_dir_for_base(&self, base_id: Option<u32>) -> Result<Path> {
+        match base_id {
             Some(base_id) => {
-                let base_paths = &self.manifest.base_paths;
-                let base_path = base_paths.get(base_id).ok_or_else(|| {
-                    Error::invalid_input(format!(
-                        "base_path id {} not found for data_file {}",
-                        base_id, data_file.path
-                    ))
+                let base_path = self.manifest.base_paths.get(&base_id).ok_or_else(|| {
+                    Error::invalid_input(format!("base_path id {} not found", base_id))
                 })?;
                 let path = base_path.extract_path(self.session.store_registry())?;
                 if base_path.is_dataset_root {