Narrow the case when not swapping

Author: blaginin

datafusion/datasource/src/file_scan_config.rs

@@ -267,24 +267,32 @@ impl DataSource for FileScanConfig {
         // If there is any non-column or alias-carrier expression, Projection should not be removed.
         // This process can be moved into CsvExec, but it would be an overlap of their responsibility.
 
-        Ok((all_alias_free_columns(projection.expr())
-            && self.table_partition_cols.is_empty())
-        .then(|| {
-            let file_scan = self.clone();
-            let source = Arc::clone(&file_scan.file_source);
-            let new_projections = new_projections_for_columns(
-                projection,
-                &file_scan
-                    .projection
-                    .clone()
-                    .unwrap_or((0..self.file_schema.fields().len()).collect()),
-            );
-            file_scan
-                // Assign projected statistics to source
-                .with_projection(Some(new_projections))
-                .with_source(source)
-                .build() as _
-        }))
+        let partitioned_columns_in_proj = projection.expr().iter().any(|(expr, _)| {
+            expr.as_any()
+                .downcast_ref::<Column>()
+                .map(|expr| expr.index() >= self.file_schema.fields().len())
+                .unwrap_or(false)
+        });
+
+        Ok(
+            (all_alias_free_columns(projection.expr()) && !partitioned_columns_in_proj)
+                .then(|| {
+                    let file_scan = self.clone();
+                    let source = Arc::clone(&file_scan.file_source);
+                    let new_projections = new_projections_for_columns(
+                        projection,
+                        &file_scan
+                            .projection
+                            .clone()
+                            .unwrap_or((0..self.file_schema.fields().len()).collect()),
+                    );
+                    file_scan
+                        // Assign projected statistics to source
+                        .with_projection(Some(new_projections))
+                        .with_source(source)
+                        .build() as _
+                }),
+        )
     }
 }