diff --git a/Cargo.lock b/Cargo.lock
index 5a0e38a53d31..7aca7dc3acc5 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -3183,6 +3183,7 @@ dependencies = [
  "polars-error",
  "polars-json",
  "polars-parquet",
+ "polars-schema",
  "polars-time",
  "polars-utils",
  "rayon",
diff --git a/crates/polars-core/src/schema.rs b/crates/polars-core/src/schema.rs
index b2ec55b528c7..d100cf91172f 100644
--- a/crates/polars-core/src/schema.rs
+++ b/crates/polars-core/src/schema.rs
@@ -90,97 +90,46 @@ impl SchemaExt for Schema {
     }
 }
 
-/// This trait exists to be unify the API of polars Schema and arrows Schema.
-pub trait IndexOfSchema: Debug {
-    /// Get the index of a column by name.
-    fn index_of(&self, name: &str) -> Option<usize>;
-
-    /// Get a vector of all column names.
-    fn get_names(&self) -> Vec<&PlSmallStr>;
-
-    fn get_names_str(&self) -> Vec<&str>;
-
-    fn get_names_owned(&self) -> Vec<PlSmallStr>;
-
-    fn try_index_of(&self, name: &str) -> PolarsResult<usize> {
-        self.index_of(name).ok_or_else(|| {
-            polars_err!(
-                ColumnNotFound:
-                "unable to find column {:?}; valid columns: {:?}", name, self.get_names(),
-            )
-        })
-    }
-}
-
-impl IndexOfSchema for Schema {
-    fn index_of(&self, name: &str) -> Option<usize> {
-        self.index_of(name)
-    }
-
-    fn get_names(&self) -> Vec<&PlSmallStr> {
-        self.iter_names().collect()
-    }
-
-    fn get_names_owned(&self) -> Vec<PlSmallStr> {
-        self.iter_names().cloned().collect()
-    }
+pub trait SchemaNamesAndDtypes {
+    const IS_ARROW: bool;
+    type DataType: Debug + Clone + Default + PartialEq;
 
-    fn get_names_str(&self) -> Vec<&str> {
-        self.iter_names().map(|x| x.as_str()).collect()
-    }
+    fn iter_names_and_dtypes(
+        &self,
+    ) -> impl ExactSizeIterator<Item = (&PlSmallStr, &Self::DataType)>;
 }
 
-impl IndexOfSchema for ArrowSchema {
-    fn index_of(&self, name: &str) -> Option<usize> {
-        self.iter_values().position(|f| f.name.as_str() == name)
-    }
-
-    fn get_names(&self) -> Vec<&PlSmallStr> {
-        self.iter_values().map(|f| &f.name).collect()
-    }
-
-    fn get_names_owned(&self) -> Vec<PlSmallStr> {
-        self.iter_values().map(|f| f.name.clone()).collect()
-    }
+impl SchemaNamesAndDtypes for ArrowSchema {
+    const IS_ARROW: bool = true;
+    type DataType = ArrowDataType;
 
-    fn get_names_str(&self) -> Vec<&str> {
-        self.iter_values().map(|f| f.name.as_str()).collect()
+    fn iter_names_and_dtypes(
+        &self,
+    ) -> impl ExactSizeIterator<Item = (&PlSmallStr, &Self::DataType)> {
+        self.iter_values().map(|x| (&x.name, &x.dtype))
     }
 }
 
-pub trait SchemaNamesAndDtypes {
-    const IS_ARROW: bool;
-    type DataType: Debug + PartialEq;
-
-    /// Get a vector of (name, dtype) pairs
-    fn get_names_and_dtypes(&'_ self) -> Vec<(&'_ str, Self::DataType)>;
-}
-
 impl SchemaNamesAndDtypes for Schema {
     const IS_ARROW: bool = false;
     type DataType = DataType;
 
-    fn get_names_and_dtypes(&'_ self) -> Vec<(&'_ str, Self::DataType)> {
+    fn iter_names_and_dtypes(
+        &self,
+    ) -> impl ExactSizeIterator<Item = (&PlSmallStr, &Self::DataType)> {
         self.iter()
-            .map(|(name, dtype)| (name.as_str(), dtype.clone()))
-            .collect()
-    }
-}
-
-impl SchemaNamesAndDtypes for ArrowSchema {
-    const IS_ARROW: bool = true;
-    type DataType = ArrowDataType;
-
-    fn get_names_and_dtypes(&'_ self) -> Vec<(&'_ str, Self::DataType)> {
-        self.iter_values()
-            .map(|x| (x.name.as_str(), x.dtype.clone()))
-            .collect()
     }
 }
 
-pub fn ensure_matching_schema<S: SchemaNamesAndDtypes>(lhs: &S, rhs: &S) -> PolarsResult<()> {
-    let lhs = lhs.get_names_and_dtypes();
-    let rhs = rhs.get_names_and_dtypes();
+pub fn ensure_matching_schema<D>(
+    lhs: &polars_schema::Schema<D>,
+    rhs: &polars_schema::Schema<D>,
+) -> PolarsResult<()>
+where
+    polars_schema::Schema<D>: SchemaNamesAndDtypes,
+{
+    let lhs = lhs.iter_names_and_dtypes();
+    let rhs = rhs.iter_names_and_dtypes();
 
     if lhs.len() != rhs.len() {
         polars_bail!(
@@ -190,7 +139,7 @@ pub fn ensure_matching_schema<S: SchemaNamesAndDtypes>(lhs: &S, rhs: &S) -> Pola
         );
     }
 
-    for (i, ((l_name, l_dtype), (r_name, r_dtype))) in lhs.iter().zip(&rhs).enumerate() {
+    for (i, ((l_name, l_dtype), (r_name, r_dtype))) in lhs.zip(rhs).enumerate() {
         if l_name != r_name {
             polars_bail!(
                 SchemaMismatch:
@@ -199,18 +148,20 @@ pub fn ensure_matching_schema<S: SchemaNamesAndDtypes>(lhs: &S, rhs: &S) -> Pola
             )
         }
         if l_dtype != r_dtype
-            && (!S::IS_ARROW
+            && (!polars_schema::Schema::<D>::IS_ARROW
                 || unsafe {
                     // For timezone normalization. Easier than writing out the entire PartialEq.
                     DataType::from_arrow(
-                        std::mem::transmute::<&<S as SchemaNamesAndDtypes>::DataType, &ArrowDataType>(
-                            l_dtype,
-                        ),
+                        std::mem::transmute::<
+                            &<polars_schema::Schema<D> as SchemaNamesAndDtypes>::DataType,
+                            &ArrowDataType,
+                        >(l_dtype),
                         true,
                     ) != DataType::from_arrow(
-                        std::mem::transmute::<&<S as SchemaNamesAndDtypes>::DataType, &ArrowDataType>(
-                            r_dtype,
-                        ),
+                        std::mem::transmute::<
+                            &<polars_schema::Schema<D> as SchemaNamesAndDtypes>::DataType,
+                            &ArrowDataType,
+                        >(r_dtype),
                         true,
                     )
                 })
diff --git a/crates/polars-io/Cargo.toml b/crates/polars-io/Cargo.toml
index 9eb2addc8be5..64259f78ad09 100644
--- a/crates/polars-io/Cargo.toml
+++ b/crates/polars-io/Cargo.toml
@@ -13,6 +13,7 @@ polars-core = { workspace = true }
 polars-error = { workspace = true }
 polars-json = { workspace = true, optional = true }
 polars-parquet = { workspace = true, optional = true }
+polars-schema = { workspace = true }
 polars-time = { workspace = true, features = [], optional = true }
 polars-utils = { workspace = true, features = ['mmap'] }
 
diff --git a/crates/polars-io/src/csv/read/options.rs b/crates/polars-io/src/csv/read/options.rs
index 7659565918ef..83b356fabde8 100644
--- a/crates/polars-io/src/csv/read/options.rs
+++ b/crates/polars-io/src/csv/read/options.rs
@@ -2,7 +2,7 @@ use std::path::PathBuf;
 use std::sync::Arc;
 
 use polars_core::datatypes::{DataType, Field};
-use polars_core::schema::{IndexOfSchema, Schema, SchemaRef};
+use polars_core::schema::{Schema, SchemaRef};
 use polars_error::PolarsResult;
 use polars_utils::pl_str::PlSmallStr;
 #[cfg(feature = "serde")]
diff --git a/crates/polars-io/src/csv/write/writer.rs b/crates/polars-io/src/csv/write/writer.rs
index 32c657b6e1a6..f3017ce189ec 100644
--- a/crates/polars-io/src/csv/write/writer.rs
+++ b/crates/polars-io/src/csv/write/writer.rs
@@ -2,7 +2,7 @@ use std::io::Write;
 use std::num::NonZeroUsize;
 
 use polars_core::frame::DataFrame;
-use polars_core::schema::{IndexOfSchema, Schema};
+use polars_core::schema::Schema;
 use polars_core::POOL;
 use polars_error::PolarsResult;
 
@@ -228,7 +228,11 @@ impl<W: Write> BatchedWriter<W> {
 
         if !self.has_written_header {
             self.has_written_header = true;
-            let names = self.schema.get_names_str();
+            let names = self
+                .schema
+                .iter_names()
+                .map(|x| x.as_str())
+                .collect::<Vec<_>>();
             write_header(&mut self.writer.buffer, &names, &self.writer.options)?;
         };
 
diff --git a/crates/polars-io/src/hive.rs b/crates/polars-io/src/hive.rs
index ddf1d8973b3e..b027e6d1d054 100644
--- a/crates/polars-io/src/hive.rs
+++ b/crates/polars-io/src/hive.rs
@@ -1,5 +1,4 @@
 use polars_core::frame::DataFrame;
-use polars_core::schema::IndexOfSchema;
 use polars_core::series::Series;
 
 /// Materializes hive partitions.
@@ -9,9 +8,9 @@ use polars_core::series::Series;
 /// # Safety
 ///
 /// num_rows equals the height of the df when the df height is non-zero.
-pub(crate) fn materialize_hive_partitions<S: IndexOfSchema>(
+pub(crate) fn materialize_hive_partitions<D>(
     df: &mut DataFrame,
-    reader_schema: &S,
+    reader_schema: &polars_schema::Schema<D>,
     hive_partition_columns: Option<&[Series]>,
     num_rows: usize,
 ) {
diff --git a/crates/polars-io/src/parquet/read/read_impl.rs b/crates/polars-io/src/parquet/read/read_impl.rs
index 89a799262629..e43c34ca2d70 100644
--- a/crates/polars-io/src/parquet/read/read_impl.rs
+++ b/crates/polars-io/src/parquet/read/read_impl.rs
@@ -475,7 +475,7 @@ fn rg_to_dfs_prefiltered(
                 // We first add the columns with the live columns at the start. Then, we do a
                 // projections that puts the columns at the right spot.
                 df._add_columns(rg_columns, &rearranged_schema)?;
-                let df = df.select(schema.get_names_owned())?;
+                let df = df.select(schema.iter_names_cloned())?;
 
                 PolarsResult::Ok(Some(df))
             })
diff --git a/crates/polars-io/src/utils/other.rs b/crates/polars-io/src/utils/other.rs
index 7267a6616924..1984e6ad480e 100644
--- a/crates/polars-io/src/utils/other.rs
+++ b/crates/polars-io/src/utils/other.rs
@@ -111,26 +111,10 @@ pub(crate) fn columns_to_projection(
     schema: &ArrowSchema,
 ) -> PolarsResult<Vec<usize>> {
     let mut prj = Vec::with_capacity(columns.len());
-    if columns.len() > 100 {
-        let mut column_names = PlHashMap::with_capacity(schema.len());
-        schema.iter_values().enumerate().for_each(|(i, c)| {
-            column_names.insert(c.name.as_str(), i);
-        });
-
-        for column in columns.iter() {
-            let Some(&i) = column_names.get(column.as_str()) else {
-                polars_bail!(
-                    ColumnNotFound:
-                    "unable to find column {:?}; valid columns: {:?}", column, schema.get_names(),
-                );
-            };
-            prj.push(i);
-        }
-    } else {
-        for column in columns.iter() {
-            let i = schema.try_index_of(column)?;
-            prj.push(i);
-        }
+
+    for column in columns {
+        let i = schema.try_index_of(column)?;
+        prj.push(i);
     }
 
     Ok(prj)
diff --git a/crates/polars-mem-engine/src/executors/projection_simple.rs b/crates/polars-mem-engine/src/executors/projection_simple.rs
index f88ad62c8956..c3102d3b7222 100644
--- a/crates/polars-mem-engine/src/executors/projection_simple.rs
+++ b/crates/polars-mem-engine/src/executors/projection_simple.rs
@@ -15,7 +15,7 @@ impl ProjectionSimple {
 impl Executor for ProjectionSimple {
     fn execute(&mut self, state: &mut ExecutionState) -> PolarsResult<DataFrame> {
         state.should_stop()?;
-        let columns = self.columns.get_names_owned();
+        let columns = self.columns.iter_names_cloned().collect::<Vec<_>>();
 
         let profile_name = if state.has_node_timer() {
             let name = comma_delimited("simple-projection".to_string(), columns.as_slice());
diff --git a/crates/polars-mem-engine/src/planner/lp.rs b/crates/polars-mem-engine/src/planner/lp.rs
index 45487f7b7024..e1b53bea2151 100644
--- a/crates/polars-mem-engine/src/planner/lp.rs
+++ b/crates/polars-mem-engine/src/planner/lp.rs
@@ -430,7 +430,7 @@ fn create_physical_plan_impl(
                 .transpose()?;
             Ok(Box::new(executors::DataFrameExec {
                 df,
-                projection: output_schema.map(|s| s.iter_names().cloned().collect()),
+                projection: output_schema.map(|s| s.iter_names_cloned().collect()),
                 filter: selection,
                 predicate_has_windows: state.has_windows,
             }))
diff --git a/crates/polars-pipe/src/executors/operators/reproject.rs b/crates/polars-pipe/src/executors/operators/reproject.rs
index 0c176b134af8..a4f6010bef79 100644
--- a/crates/polars-pipe/src/executors/operators/reproject.rs
+++ b/crates/polars-pipe/src/executors/operators/reproject.rs
@@ -1,6 +1,5 @@
 use polars_core::error::PolarsResult;
 use polars_core::frame::DataFrame;
-use polars_core::prelude::IndexOfSchema;
 use polars_core::schema::Schema;
 
 use crate::operators::DataChunk;
@@ -15,12 +14,9 @@ pub(crate) fn reproject_chunk(
         // the positions for subsequent calls
         let chunk_schema = chunk.data.schema();
 
-        let check_duplicates = false;
-        let out = chunk.data._select_with_schema_impl(
-            schema.get_names_owned().as_slice(),
-            &chunk_schema,
-            check_duplicates,
-        )?;
+        let out = chunk
+            .data
+            .select_with_schema_unchecked(schema.iter_names_cloned(), &chunk_schema)?;
 
         *positions = out
             .get_columns()
diff --git a/crates/polars-pipe/src/executors/sinks/reproject.rs b/crates/polars-pipe/src/executors/sinks/reproject.rs
index ecba66f188e4..bd9553b75f97 100644
--- a/crates/polars-pipe/src/executors/sinks/reproject.rs
+++ b/crates/polars-pipe/src/executors/sinks/reproject.rs
@@ -1,6 +1,5 @@
 use std::any::Any;
 
-use polars_core::prelude::IndexOfSchema;
 use polars_core::schema::SchemaRef;
 
 use crate::executors::sources::ReProjectSource;
@@ -41,7 +40,7 @@ impl Sink for ReProjectSink {
     fn finalize(&mut self, context: &PExecutionContext) -> PolarsResult<FinalizedSink> {
         Ok(match self.sink.finalize(context)? {
             FinalizedSink::Finished(df) => {
-                FinalizedSink::Finished(df._select_impl(self.schema.get_names_owned().as_slice())?)
+                FinalizedSink::Finished(df.select(self.schema.iter_names_cloned())?)
             },
             FinalizedSink::Source(source) => {
                 FinalizedSink::Source(Box::new(ReProjectSource::new(self.schema.clone(), source)))
diff --git a/crates/polars-pipe/src/pipeline/convert.rs b/crates/polars-pipe/src/pipeline/convert.rs
index 368fc91b17ef..0a6a8946feba 100644
--- a/crates/polars-pipe/src/pipeline/convert.rs
+++ b/crates/polars-pipe/src/pipeline/convert.rs
@@ -67,7 +67,7 @@ where
                 }
                 // projection is free
                 if let Some(schema) = output_schema {
-                    let columns = schema.iter_names().cloned().collect::<Vec<_>>();
+                    let columns = schema.iter_names_cloned().collect::<Vec<_>>();
                     df = df._select_impl_unchecked(&columns)?;
                 }
             }
@@ -590,7 +590,7 @@ where
     let op = match lp_arena.get(node) {
         SimpleProjection { input, columns, .. } => {
             let input_schema = lp_arena.get(*input).schema(lp_arena);
-            let columns = columns.iter_names().cloned().collect();
+            let columns = columns.iter_names_cloned().collect();
             let op = operators::SimpleProjectionOperator::new(columns, input_schema.into_owned());
             Box::new(op) as Box<dyn Operator>
         },
diff --git a/crates/polars-plan/src/plans/conversion/dsl_to_ir.rs b/crates/polars-plan/src/plans/conversion/dsl_to_ir.rs
index 084779a68a28..5ab23be19a14 100644
--- a/crates/polars-plan/src/plans/conversion/dsl_to_ir.rs
+++ b/crates/polars-plan/src/plans/conversion/dsl_to_ir.rs
@@ -1073,14 +1073,14 @@ pub(crate) fn maybe_init_projection_excluding_hive(
 
     let (first_hive_name, _) = hive_schema.get_at_index(0)?;
 
+    // TODO: Optimize this
     let names = match reader_schema {
-        Either::Left(ref v) => {
-            let names = v.get_names_owned();
-            names.contains(first_hive_name).then_some(names)
-        },
+        Either::Left(ref v) => v
+            .contains(first_hive_name.as_str())
+            .then(|| v.iter_names_cloned().collect::<Vec<_>>()),
         Either::Right(ref v) => v
             .contains(first_hive_name.as_str())
-            .then(|| v.get_names_owned()),
+            .then(|| v.iter_names_cloned().collect()),
     };
 
     let names = names?;
diff --git a/crates/polars-plan/src/plans/optimizer/cache_states.rs b/crates/polars-plan/src/plans/optimizer/cache_states.rs
index 4a2a2d87e75b..da13d047d43f 100644
--- a/crates/polars-plan/src/plans/optimizer/cache_states.rs
+++ b/crates/polars-plan/src/plans/optimizer/cache_states.rs
@@ -15,7 +15,7 @@ fn get_upper_projections(
     // During projection pushdown all accumulated.
     match parent {
         SimpleProjection { columns, .. } => {
-            let iter = columns.iter_names().cloned();
+            let iter = columns.iter_names_cloned();
             names_scratch.extend(iter);
             *found_required_columns = true;
             false
@@ -264,7 +264,7 @@ pub(super) fn set_cache_states(
                     // all columns
                     if !found_required_columns {
                         let schema = lp.schema(lp_arena);
-                        v.names_union.extend(schema.iter_names().cloned());
+                        v.names_union.extend(schema.iter_names_cloned());
                     }
                 }
                 frame.cache_id = Some(*id);
diff --git a/crates/polars-plan/src/plans/optimizer/projection_pushdown/mod.rs b/crates/polars-plan/src/plans/optimizer/projection_pushdown/mod.rs
index 20e0d0d28633..61c86e789d95 100644
--- a/crates/polars-plan/src/plans/optimizer/projection_pushdown/mod.rs
+++ b/crates/polars-plan/src/plans/optimizer/projection_pushdown/mod.rs
@@ -346,7 +346,7 @@ impl ProjectionPushDown {
                 expr_arena,
             ),
             SimpleProjection { columns, input, .. } => {
-                let exprs = names_to_expr_irs(columns.iter_names().cloned(), expr_arena);
+                let exprs = names_to_expr_irs(columns.iter_names_cloned(), expr_arena);
                 process_projection(
                     self,
                     input,
diff --git a/crates/polars-schema/src/schema.rs b/crates/polars-schema/src/schema.rs
index b927a88f3b63..3f03bdffde24 100644
--- a/crates/polars-schema/src/schema.rs
+++ b/crates/polars-schema/src/schema.rs
@@ -14,10 +14,7 @@ pub struct Schema<D> {
 
 impl<D: Eq> Eq for Schema<D> {}
 
-impl<D> Schema<D>
-where
-    D: Clone + Default,
-{
+impl<D> Schema<D> {
     pub fn with_capacity(capacity: usize) -> Self {
         let fields = PlIndexMap::with_capacity(capacity);
         Self { fields }
@@ -59,42 +56,6 @@ where
         self.fields.insert(key, value)
     }
 
-    /// Create a new schema from this one, inserting a field with `name` and `dtype` at the given `index`.
-    ///
-    /// If a field named `name` already exists, it is updated with the new dtype. Regardless, the field named `name` is
-    /// always moved to the given index. Valid indices range from `0` (front of the schema) to `self.len()` (after the
-    /// end of the schema).
-    ///
-    /// For a mutating version that doesn't clone, see [`insert_at_index`][Self::insert_at_index].
-    ///
-    /// Runtime: **O(m * n)** where `m` is the (average) length of the field names and `n` is the number of fields in
-    /// the schema. This method clones every field in the schema.
-    ///
-    /// Returns: `Ok(new_schema)` if `index <= self.len()`, else `Err(PolarsError)`
-    pub fn new_inserting_at_index(
-        &self,
-        index: usize,
-        name: PlSmallStr,
-        field: D,
-    ) -> PolarsResult<Self> {
-        polars_ensure!(
-            index <= self.len(),
-            OutOfBounds:
-                "index {} is out of bounds for schema with length {} (the max index allowed is self.len())",
-                    index,
-                    self.len()
-        );
-
-        let mut new = Self::default();
-        let mut iter = self.fields.iter().filter_map(|(fld_name, dtype)| {
-            (fld_name != &name).then_some((fld_name.clone(), dtype.clone()))
-        });
-        new.fields.extend(iter.by_ref().take(index));
-        new.fields.insert(name.clone(), field);
-        new.fields.extend(iter);
-        Ok(new)
-    }
-
     /// Insert a field with `name` and `dtype` at the given `index` into this schema.
     ///
     /// If a field named `name` already exists, it is updated with the new dtype. Regardless, the field named `name` is
@@ -275,21 +236,6 @@ where
         self.fields.extend(other.fields)
     }
 
-    /// Merge borrowed `other` into `self`.
-    ///
-    /// Merging logic:
-    /// - Fields that occur in `self` but not `other` are unmodified
-    /// - Fields that occur in `other` but not `self` are appended, in order, to the end of `self`
-    /// - Fields that occur in both `self` and `other` are updated with the dtype from `other`, but keep their original
-    ///   index
-    pub fn merge_from_ref(&mut self, other: &Self) {
-        self.fields.extend(
-            other
-                .iter()
-                .map(|(column, field)| (column.clone(), field.clone())),
-        )
-    }
-
     /// Iterates over the `(&name, &dtype)` pairs in this schema.
     ///
     /// For an owned version, use [`iter_fields`][Self::iter_fields], which clones the data to iterate owned `Field`s
@@ -306,6 +252,10 @@ where
         self.fields.iter().map(|(name, _dtype)| name)
     }
 
+    pub fn iter_names_cloned(&self) -> impl '_ + ExactSizeIterator<Item = PlSmallStr> {
+        self.iter_names().cloned()
+    }
+
     /// Iterates over references to the dtypes in this schema.
     pub fn iter_values(&self) -> impl '_ + ExactSizeIterator<Item = &D> {
         self.fields.iter().map(|(_name, dtype)| dtype)
@@ -324,6 +274,74 @@ where
         self.fields.get_index_of(name)
     }
 
+    pub fn try_index_of(&self, name: &str) -> PolarsResult<usize> {
+        let Some(i) = self.fields.get_index_of(name) else {
+            polars_bail!(
+                ColumnNotFound:
+                "unable to find column {:?}; valid columns: {:?}",
+                name, self.iter_names().collect::<Vec<_>>(),
+            )
+        };
+
+        Ok(i)
+    }
+}
+
+impl<D> Schema<D>
+where
+    D: Clone + Default,
+{
+    /// Create a new schema from this one, inserting a field with `name` and `dtype` at the given `index`.
+    ///
+    /// If a field named `name` already exists, it is updated with the new dtype. Regardless, the field named `name` is
+    /// always moved to the given index. Valid indices range from `0` (front of the schema) to `self.len()` (after the
+    /// end of the schema).
+    ///
+    /// For a mutating version that doesn't clone, see [`insert_at_index`][Self::insert_at_index].
+    ///
+    /// Runtime: **O(m * n)** where `m` is the (average) length of the field names and `n` is the number of fields in
+    /// the schema. This method clones every field in the schema.
+    ///
+    /// Returns: `Ok(new_schema)` if `index <= self.len()`, else `Err(PolarsError)`
+    pub fn new_inserting_at_index(
+        &self,
+        index: usize,
+        name: PlSmallStr,
+        field: D,
+    ) -> PolarsResult<Self> {
+        polars_ensure!(
+            index <= self.len(),
+            OutOfBounds:
+                "index {} is out of bounds for schema with length {} (the max index allowed is self.len())",
+                    index,
+                    self.len()
+        );
+
+        let mut new = Self::default();
+        let mut iter = self.fields.iter().filter_map(|(fld_name, dtype)| {
+            (fld_name != &name).then_some((fld_name.clone(), dtype.clone()))
+        });
+        new.fields.extend(iter.by_ref().take(index));
+        new.fields.insert(name.clone(), field);
+        new.fields.extend(iter);
+        Ok(new)
+    }
+
+    /// Merge borrowed `other` into `self`.
+    ///
+    /// Merging logic:
+    /// - Fields that occur in `self` but not `other` are unmodified
+    /// - Fields that occur in `other` but not `self` are appended, in order, to the end of `self`
+    /// - Fields that occur in both `self` and `other` are updated with the dtype from `other`, but keep their original
+    ///   index
+    pub fn merge_from_ref(&mut self, other: &Self) {
+        self.fields.extend(
+            other
+                .iter()
+                .map(|(column, field)| (column.clone(), field.clone())),
+        )
+    }
+
     /// Generates another schema with just the specified columns selected from this one.
     pub fn try_project<I>(&self, columns: I) -> PolarsResult<Self>
     where
@@ -421,7 +439,6 @@ impl<D> From<PlIndexMap<PlSmallStr, D>> for Schema<D> {
 impl<F, D> FromIterator<F> for Schema<D>
 where
     F: Into<(PlSmallStr, D)>,
-    D: Clone,
 {
     fn from_iter<I: IntoIterator<Item = F>>(iter: I) -> Self {
         let fields = PlIndexMap::from_iter(iter.into_iter().map(|x| x.into()));
diff --git a/crates/polars-stream/src/physical_plan/lower_ir.rs b/crates/polars-stream/src/physical_plan/lower_ir.rs
index b993ea6ac557..65044a717213 100644
--- a/crates/polars-stream/src/physical_plan/lower_ir.rs
+++ b/crates/polars-stream/src/physical_plan/lower_ir.rs
@@ -1,7 +1,7 @@
 use std::sync::Arc;
 
 use polars_core::prelude::{InitHashMaps, PlHashMap, PlIndexMap};
-use polars_core::schema::{IndexOfSchema, Schema};
+use polars_core::schema::Schema;
 use polars_error::{polars_err, PolarsResult};
 use polars_plan::plans::expr_ir::{ExprIR, OutputName};
 use polars_plan::plans::{AExpr, IR};
@@ -26,7 +26,7 @@ pub fn lower_ir(
     let output_schema = IR::schema_with_cache(node, ir_arena, schema_cache);
     let node_kind = match ir_node {
         IR::SimpleProjection { input, columns } => {
-            let columns = columns.get_names_owned();
+            let columns = columns.iter_names_cloned().collect::<Vec<_>>();
             let phys_input = lower_ir(
                 *input,
                 ir_arena,
@@ -200,7 +200,7 @@ pub fn lower_ir(
                     let phys_input = phys_sm.insert(PhysNode::new(schema, node_kind));
                     node_kind = PhysNodeKind::SimpleProjection {
                         input: phys_input,
-                        columns: projection_schema.get_names_owned(),
+                        columns: projection_schema.iter_names_cloned().collect::<Vec<_>>(),
                     };
                     schema = projection_schema.clone();
                 }
diff --git a/crates/polars/tests/it/io/csv.rs b/crates/polars/tests/it/io/csv.rs
index 604ef2782959..992754436c60 100644
--- a/crates/polars/tests/it/io/csv.rs
+++ b/crates/polars/tests/it/io/csv.rs
@@ -520,7 +520,7 @@ fn test_empty_bytes_to_dataframe() {
 
     let result = CsvReadOptions::default()
         .with_has_header(false)
-        .with_columns(Some(schema.iter_names().cloned().collect()))
+        .with_columns(Some(schema.iter_names_cloned().collect()))
         .with_schema(Some(Arc::new(schema)))
         .into_reader_with_file_handle(file)
         .finish();