Repository: pola-rs/polars
File: crates/polars-stream/src/physical_plan/to_graph.rs (Rust)

Code before:
                // If validate_schema is false, the schema of the morsels may not match the
                // configured schema. In this case we set this to `None` and the reader will
                // retrieve the schema from the first morsel.
                output_schema: validate_schema.then(|| output_schema.clone()),
                get_batch_state: Some(GetBatchState::from(get_batch_fn)),
                execution_state: None,
                verbose: config::verbose(),
            };

            let file_reader_builder = Arc::new(BatchFnReaderBuilder {
                name,
                reader: std::sync::Mutex::new(Some(reader)),
                execution_state: Default::default(),
            }) as Arc<dyn FileReaderBuilder>;

            // Give multiscan a single scan source. (It doesn't actually read from this).
            let sources = ScanSources::Paths(Arc::from([PlPath::from_str("python-scan-0")]));
            let cloud_options = None;
            let final_output_schema = output_schema.clone();
            let file_projection_builder = ProjectionBuilder::new(output_schema.clone(), None);
            let row_index = None;
            let pre_slice = None;
            let predicate = None;
            let hive_parts = None;
            let include_file_paths = None;
            let missing_columns_policy = MissingColumnsPolicy::Insert;
            let forbid_extra_columns = Some(ForbidExtraColumns::Plain(output_schema));
            let cast_columns_policy = CastColumnsPolicy::ERROR_ON_MISMATCH;
            let deletion_files = None;
            let verbose = config::verbose();

            ctx.graph.add_node(
                nodes::io_sources::multi_file_reader::MultiFileReader::new(Arc::new(
                    MultiFileReaderConfig {
                        sources,
                        file_reader_builder,
                        cloud_options,
                        final_output_schema,
                        file_projection_builder,
                        row_index,
                        pre_slice,
                        predicate,
                        hive_parts,
                        include_file_paths,
                        missing_columns_policy,
                        forbid_extra_columns,
                        cast_columns_policy,
                        deletion_files,
                        // Initialized later
                        num_pipelines: RelaxedCell::new_usize(0),
                        n_readers_pre_init: RelaxedCell::new_usize(0),

Review comment (suggestion):
Insert missing columns to pass:

```
test_predicates.py::test_predicate_pushdown_map_elements_io_plugin_22860 - polars.exceptions.ColumnNotFoundError: did not find column x, consider passing `missing_columns='insert'`
```

Note that `forbid_extra_columns` below is also `None` to ignore extra columns outside the output schema.


Code after:
                // configured schema. In this case we set this to `None` and the reader will
                // retrieve the schema from the first morsel.
                output_schema: validate_schema.then(|| output_schema.clone()),
                get_batch_state: Some(GetBatchState::from(get_batch_fn)),
                execution_state: None,
                verbose: config::verbose(),
            };

            let file_reader_builder = Arc::new(BatchFnReaderBuilder {
                name,
                reader: std::sync::Mutex::new(Some(reader)),
                execution_state: Default::default(),
            }) as Arc<dyn FileReaderBuilder>;

            // Give multiscan a single scan source. (It doesn't actually read from this).
            let sources = ScanSources::Paths(Arc::from([PlPath::from_str("python-scan-0")]));
            let cloud_options = None;
            let final_output_schema = output_schema.clone();
            let file_projection_builder = ProjectionBuilder::new(output_schema.clone(), None);
            let row_index = None;
            let pre_slice = None;
            let predicate = None;
            let hive_parts = None;
            let include_file_paths = None;
            let missing_columns_policy = MissingColumnsPolicy::Raise;
            let forbid_extra_columns = None;
            let cast_columns_policy = CastColumnsPolicy::ERROR_ON_MISMATCH;
            let deletion_files = None;
            let verbose = config::verbose();

            ctx.graph.add_node(
                nodes::io_sources::multi_file_reader::MultiFileReader::new(Arc::new(
                    MultiFileReaderConfig {
                        sources,
                        file_reader_builder,
                        cloud_options,
                        final_output_schema,
                        file_projection_builder,
                        row_index,
                        pre_slice,
                        predicate,
                        hive_parts,
                        include_file_paths,
                        missing_columns_policy,
                        forbid_extra_columns,
                        cast_columns_policy,
                        deletion_files,
                        // Initialized later
                        num_pipelines: RelaxedCell::new_usize(0),
                        n_readers_pre_init: RelaxedCell::new_usize(0),
                        max_concurrent_scans: RelaxedCell::new_usize(0),