fix: remove literal Series from projection state

crates/polars-arrow/src/legacy/utils.rs

@@ -1,3 +1,5 @@
+use std::borrow::Borrow;
+
 use crate::array::PrimitiveArray;
 use crate::bitmap::MutableBitmap;
 use crate::datatypes::ArrowDataType;
@@ -57,6 +59,15 @@ pub trait CustomIterTools: Iterator {
         }
         Some(start)
     }
+
+    fn contains<Q>(&mut self, query: &Q) -> bool
+    where
+        Self: Sized,
+        Self::Item: Borrow<Q>,
+        Q: PartialEq,
+    {
+        self.any(|x| x.borrow() == query)
+    }
 }
 
 pub trait CustomIterToolsSized: Iterator + Sized {}

crates/polars-plan/src/logical_plan/optimizer/projection_pushdown/mod.rs

@@ -134,9 +134,9 @@ fn update_scan_schema(
     let mut new_schema = Schema::with_capacity(acc_projections.len());
     let mut new_cols = Vec::with_capacity(acc_projections.len());
     for node in acc_projections.iter() {
-        for name in aexpr_to_leaf_names(*node, expr_arena) {
+        for name in aexpr_to_leaf_names_iter(*node, expr_arena) {
             let item = schema.get_full(&name).ok_or_else(|| {
-                polars_err!(ComputeError: "column '{}' not available in schema {:?}", name, schema)
+                polars_err!(ComputeError: "column '{}' not available in 'DataFrame' with {:?}", name, schema)
             })?;
             new_cols.push(item);
         }

crates/polars-plan/src/logical_plan/optimizer/projection_pushdown/projection.rs

@@ -23,13 +23,33 @@ fn check_double_projection(
     acc_projections: &mut Vec<Node>,
     projected_names: &mut PlHashSet<Arc<str>>,
 ) {
+    // Factor out the pruning function
+    fn prune_projections_by_name(
+        acc_projections: &mut Vec<Node>,
+        name: &str,
+        expr_arena: &Arena<AExpr>,
+    ) {
+        acc_projections.retain(|expr| {
+            !aexpr_to_leaf_names_iter(*expr, expr_arena).any(|q| q.as_ref() == name)
+        });
+    }
+
     for (_, ae) in (&*expr_arena).iter(*expr) {
-        if let AExpr::Alias(_, name) = ae {
-            if projected_names.remove(name) {
-                acc_projections
-                    .retain(|expr| !aexpr_to_leaf_names(*expr, expr_arena).contains(name));
-            }
-        }
+        match ae {
+            // Series literals come from another source so should not be pushed down.
+            AExpr::Literal(LiteralValue::Series(s)) => {
+                let name = s.name();
+                if projected_names.remove(name) {
+                    prune_projections_by_name(acc_projections, name, expr_arena)
+                }
+            },
+            AExpr::Alias(_, name) => {
+                if projected_names.remove(name) {
+                    prune_projections_by_name(acc_projections, name.as_ref(), expr_arena)
+                }
+            },
+            _ => {},
+        };
     }
 }
 

py-polars/tests/unit/test_projections.py

@@ -389,3 +389,22 @@ def test_rolling_key_projected_13617() -> None:
     assert r'DF ["idx", "value"]; PROJECT 2/2 COLUMNS' in plan
     out = ldf.collect(projection_pushdown=True)
     assert out.to_dict(as_series=False) == {"value": [["a"], ["b"]]}
+
+
+def test_projection_drop_with_series_lit_() -> None:
+    df = pl.DataFrame({"b": [1, 6, 8, 7]})
+    df2 = pl.DataFrame({"a": [1, 2, 4, 4], "b": [True, True, True, False]})
+
+    q = (
+        df2.lazy()
+        .select(
+            *["a", "b"], pl.lit("b").alias("b_name"), df.get_column("b").alias("b_old")
+        )
+        .filter(pl.col("b").not_())
+        .drop("b")
+    )
+    assert q.collect().to_dict(as_series=False) == {
+        "a": [4],
+        "b_name": ["b"],
+        "b_old": [7],
+    }