Preserve projection for inline scan

datafusion/core/tests/execution/logical_plan.rs

@@ -19,13 +19,16 @@
 //! create them and depend on them. Test executable semantics of logical plans.
 
 use arrow::array::Int64Array;
-use arrow::datatypes::{DataType, Field};
+use arrow::datatypes::{DataType, Field, Schema};
+use datafusion::datasource::{provider_as_source, ViewTable};
 use datafusion::execution::session_state::SessionStateBuilder;
-use datafusion_common::{Column, DFSchema, Result, ScalarValue, Spans};
+use datafusion_common::{Column, DFSchema, DFSchemaRef, Result, ScalarValue, Spans};
 use datafusion_execution::TaskContext;
 use datafusion_expr::expr::{AggregateFunction, AggregateFunctionParams};
 use datafusion_expr::logical_plan::{LogicalPlan, Values};
-use datafusion_expr::{Aggregate, AggregateUDF, Expr};
+use datafusion_expr::{
+    Aggregate, AggregateUDF, EmptyRelation, Expr, LogicalPlanBuilder, UNNAMED_TABLE,
+};
 use datafusion_functions_aggregate::count::Count;
 use datafusion_physical_plan::collect;
 use std::collections::HashMap;
@@ -96,3 +99,32 @@ where
     };
     element
 }
+
+#[test]
+fn inline_scan_projection_test() -> Result<()> {
+    let name = UNNAMED_TABLE;
+    let column = "a";
+
+    let schema = Schema::new(vec![
+        Field::new("a", DataType::Int32, false),
+        Field::new("b", DataType::Int32, false),
+    ]);
+    let projection = vec![schema.index_of(column)?];
+
+    let provider = ViewTable::new(
+        LogicalPlan::EmptyRelation(EmptyRelation {
+            produce_one_row: false,
+            schema: DFSchemaRef::new(DFSchema::try_from(schema)?),
+        }),
+        None,
+    );
+    let source = provider_as_source(Arc::new(provider));
+
+    let plan = LogicalPlanBuilder::scan(name, source, Some(projection))?.build()?;
+    assert_eq!(
+        plan.schema().field_names(),
+        vec![format!("{name}.{column}")]
+    );
+
+    Ok(())
+}

datafusion/expr/src/logical_plan/builder.rs

@@ -498,7 +498,7 @@ impl LogicalPlanBuilder {
             TableScan::try_new(table_name, table_source, projection, filters, fetch)?;
 
         // Inline TableScan
-        if table_scan.filters.is_empty() {
+        if table_scan.projection.is_none() && table_scan.filters.is_empty() {
             if let Some(p) = table_scan.source.get_logical_plan() {
                 let sub_plan = p.into_owned();
                 // Ensures that the reference to the inlined table remains the