fix(cache): validate cache_index schema collisions in worker dat… (#3216)

* fix(data-cache): validate cache_index schema collisions in worker datasource

Signed-off-by: Hitanshi Goklani <hitanshigoklani33@gmail.com>

* Update pkg/data_cache/src/worker/worker_datasource.rs

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
Signed-off-by: Hitanshi7556 <hitanshigoklani33@gmail.com>

---------

Signed-off-by: Hitanshi Goklani <hitanshigoklani33@gmail.com>
Signed-off-by: Hitanshi7556 <hitanshigoklani33@gmail.com>
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: Hitanshi7556

pkg/data_cache/src/worker/worker_datasource.rs

@@ -111,7 +111,18 @@ impl WorkerDataSource {
 
         let fields = schema.fields().clone();
         let mut builder = SchemaBuilder::from(&fields);
-        builder.push(Field::new(CACHE_INDEX_COLUMN, DataType::UInt64, false)); // TODO:// validate name collision
+
+        // Validate that source schema does not already contain cache_index column
+        if schema.field_with_name(CACHE_INDEX_COLUMN).is_ok() {
+            return Err(format!(
+                "Source schema already contains a column named '{}'. This conflicts with the cache index column added by the data cache worker.",
+                CACHE_INDEX_COLUMN
+            )
+            .into());
+        }
+
+        builder.push(Field::new(CACHE_INDEX_COLUMN, DataType::UInt64, false));
+
         let output_schema = Arc::new(Schema::new(builder.finish().fields));
         Ok(Self {
             file_urls,
@@ -513,3 +524,41 @@ async fn filter_and_create_stream(
         ))))),
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use arrow_schema::{DataType, Field, Schema};
+
+    #[test]
+    fn fails_when_source_schema_contains_cache_index() {
+        // Build a schema that already has a cache_index column
+        let schema = Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new(CACHE_INDEX_COLUMN, DataType::UInt64, false),
+        ]);
+
+        // The check we added: if cache_index already exists, is_ok() returns true
+        let collision = schema.field_with_name(CACHE_INDEX_COLUMN).is_ok();
+        assert!(
+            collision,
+            "Should detect cache_index collision in source schema"
+        );
+    }
+
+    #[test]
+    fn passes_when_source_schema_has_no_cache_index() {
+        // Build a schema with no cache_index column
+        let schema = Schema::new(vec![
+            Field::new("id", DataType::Int32, false),
+            Field::new("name", DataType::Utf8, false),
+        ]);
+
+        // No collision — field_with_name returns Err, so is_ok() is false
+        let collision = schema.field_with_name(CACHE_INDEX_COLUMN).is_ok();
+        assert!(
+            !collision,
+            "Should not detect cache_index collision when column is absent"
+        );
+    }
+}