Creates a new [`SplitProjection`] by splitting a projection into simple file column indices and a remainder projection that is applied after reading the file. In other words: we get a `Vec ` projection that is meant to be applied on top of `file_schema` and a remainder projection that is applied to the result of that first projection. Here `file_schema` is expected to be the *logical* sche
(logical_file_schema: &Schema, projection: &ProjectionExprs)
| 187 | /// Partition columns are always expected to be at the end of the table schema. |
| 188 | /// Note that `file_schema` is *not* the physical schema of the file. |
| 189 | pub fn new(logical_file_schema: &Schema, projection: &ProjectionExprs) -> Self { |
| 190 | let num_file_schema_columns = logical_file_schema.fields().len(); |
| 191 | |
| 192 | // Collect all unique columns and classify as file or partition |
| 193 | let mut file_columns = Vec::new(); |
| 194 | let mut partition_columns = Vec::new(); |
| 195 | let mut all_columns = std::collections::HashMap::new(); |
| 196 | |
| 197 | // Extract all unique column references (index -> name) |
| 198 | for proj_expr in projection { |
| 199 | proj_expr |
| 200 | .expr |
| 201 | .apply(|expr| { |
| 202 | if let Some(column) = expr.downcast_ref::<Column>() { |
| 203 | all_columns |
| 204 | .entry(column.index()) |
| 205 | .or_insert_with(|| column.name().to_string()); |
| 206 | } |
| 207 | Ok(datafusion_common::tree_node::TreeNodeRecursion::Continue) |
| 208 | }) |
| 209 | .expect("infallible apply"); |
| 210 | } |
| 211 | |
| 212 | // Sort by index and classify into file vs partition columns |
| 213 | let mut sorted_columns: Vec<_> = all_columns |
| 214 | .into_iter() |
| 215 | .map(|(idx, name)| (name, idx)) |
| 216 | .collect(); |
| 217 | sorted_columns.sort_by_key(|(_, idx)| *idx); |
| 218 | |
| 219 | // Separate file and partition columns, assigning final indices |
| 220 | // Pre-create all remapped columns to avoid duplicate Arc'd expressions |
| 221 | let mut column_mapping = std::collections::HashMap::new(); |
| 222 | let mut file_idx = 0; |
| 223 | let mut partition_idx = 0; |
| 224 | |
| 225 | for (name, original_index) in sorted_columns { |
| 226 | let new_index = if original_index < num_file_schema_columns { |
| 227 | // File column: gets index [0..num_file_columns) |
| 228 | file_columns.push(original_index); |
| 229 | let idx = file_idx; |
| 230 | file_idx += 1; |
| 231 | idx |
| 232 | } else { |
| 233 | // Partition column: gets index [num_file_columns..) |
| 234 | partition_columns.push(original_index); |
| 235 | let idx = file_idx + partition_idx; |
| 236 | partition_idx += 1; |
| 237 | idx |
| 238 | }; |
| 239 | |
| 240 | // Pre-create the remapped column so all references can share the same Arc |
| 241 | let new_column: Arc<dyn datafusion_physical_plan::PhysicalExpr> = |
| 242 | Arc::new(Column::new(&name, new_index)); |
| 243 | column_mapping.insert(original_index, new_column); |
| 244 | } |
| 245 | |
| 246 | // Single tree transformation: remap all column references using pre-created columns |
nothing calls this directly
no test coverage detected