From 964b6dfca2b099f029d216d4f5e66f55df05d507 Mon Sep 17 00:00:00 2001
From: yangjie01 <yangjie01@baidu.com>
Date: Wed, 15 Jul 2026 11:35:50 +0800
Subject: [PATCH] fix(dataset): reject all system column names on write

The write-path reserved-name guard only rejected three of the five
system column names (`_rowid`, `_rowaddr`, `_rowoffset`), missing
`_row_created_at_version` and `_row_last_updated_at_version`. Those two
were added by the row-tracking feature and the guard was never extended.

System columns are virtual: they are produced at read time and never
stored in the physical data. `Projection::to_schema` appends `_rowid`,
`_rowaddr`, and the two row-version columns to a projection, and
`scanner.rs`'s `filterable_schema` sets all of those flags on every
filtered scan. So a user data column literally named
`_row_created_at_version` passed ingest, then collided with the appended
system field the next time the dataset was scanned with a filter, hitting
`to_schema`'s `extend(...).unwrap()` and panicking.

Replace the hardcoded three-name check with `is_system_column`, which
covers all five names, so the collision is rejected at the write boundary
with a clear error instead of surfacing as a later panic. Add a
parameterized test covering every system column name.
---
 rust/lance/src/dataset/write/insert.rs | 41 ++++++++++++++++++++++++--
 1 file changed, 38 insertions(+), 3 deletions(-)

diff --git a/rust/lance/src/dataset/write/insert.rs b/rust/lance/src/dataset/write/insert.rs
index 6e1db342f9c..144cc2bac6e 100644
--- a/rust/lance/src/dataset/write/insert.rs
+++ b/rust/lance/src/dataset/write/insert.rs
@@ -8,8 +8,8 @@ use arrow_array::{RecordBatch, RecordBatchIterator};
 use datafusion::execution::SendableRecordBatchStream;
 use humantime::format_duration;
 use lance_core::datatypes::{NullabilityComparison, Schema, SchemaCompareOptions};
+use lance_core::is_system_column;
 use lance_core::utils::tracing::{DATASET_WRITING_EVENT, TRACE_DATASET_EVENTS};
-use lance_core::{ROW_ADDR, ROW_ID, ROW_OFFSET};
 use lance_datafusion::utils::StreamingWriteSource;
 use lance_file::version::LanceFileVersion;
 use lance_io::object_store::ObjectStore;
@@ -328,9 +328,12 @@ impl<'a> InsertBuilder<'a> {
             normalized_data_schema.check_compatible(dataset.schema(), &schema_cmp_opts)?;
         }
 
-        // Make sure we aren't using any reserved column names
+        // The system columns (`_rowid`, `_rowaddr`, `_rowoffset`, and the row-version
+        // columns) are virtual: they're injected into scan results at read time and
+        // never stored. A stored column sharing one of these names would collide with
+        // the system column on read, so reject it at write time.
         for field in data_schema.fields.iter() {
-            if field.name == ROW_ID || field.name == ROW_ADDR || field.name == ROW_OFFSET {
+            if is_system_column(&field.name) {
                 return Err(Error::invalid_input_source(
                     format!(
                         "The column {} is a reserved name and cannot be used in a Lance dataset",
@@ -506,6 +509,38 @@ mod test {
         );
     }
 
+    #[rstest::rstest]
+    #[case::row_id("_rowid")]
+    #[case::row_addr("_rowaddr")]
+    #[case::row_offset("_rowoffset")]
+    #[case::row_created_at_version("_row_created_at_version")]
+    #[case::row_last_updated_at_version("_row_last_updated_at_version")]
+    #[tokio::test]
+    async fn rejects_reserved_system_column_names(#[case] reserved_name: &str) {
+        // Every system column name must be rejected on write. The row-version
+        // columns (`_row_created_at_version`, `_row_last_updated_at_version`) are
+        // computed at read time and appended by `Projection::to_schema`; a user
+        // data column sharing one of those names would otherwise pass ingest and
+        // later collide with the appended field.
+        let schema = Arc::new(Schema::new(vec![Field::new(
+            reserved_name,
+            DataType::Int32,
+            false,
+        )]));
+        let batch = RecordBatch::try_new(schema.clone(), vec![Arc::new(Int32Array::from(vec![1]))])
+            .unwrap();
+
+        let result = InsertBuilder::new("memory://")
+            .execute_stream(RecordBatchIterator::new(vec![Ok(batch)], schema.clone()))
+            .await;
+
+        let err = result.expect_err("writing a reserved system column name should fail");
+        assert!(
+            err.to_string().contains("reserved name"),
+            "unexpected error for {reserved_name}: {err}"
+        );
+    }
+
     #[tokio::test]
     async fn allow_overwrite_to_v2_2_without_blob_upgrade() {
         let schema = Arc::new(Schema::new(vec![Field::new("id", DataType::Int32, false)]));