Make FileScanConfig decoding context aware

Author: thinkharderdev

ballista/rust/core/src/serde/physical_plan/mod.rs

@@ -121,19 +121,19 @@ impl AsExecutionPlan for PhysicalPlanNode {
                 Ok(Arc::new(FilterExec::try_new(predicate, input)?))
             }
             PhysicalPlanType::CsvScan(scan) => Ok(Arc::new(CsvExec::new(
-                scan.base_conf.as_ref().unwrap().try_into()?,
+                decode_scan_config(scan.base_conf.as_ref().unwrap(), &ctx)?,
                 scan.has_header,
                 str_to_byte(&scan.delimiter)?,
             ))),
             PhysicalPlanType::ParquetScan(scan) => {
                 Ok(Arc::new(ParquetExec::new(
-                    scan.base_conf.as_ref().unwrap().try_into()?,
+                    decode_scan_config(scan.base_conf.as_ref().unwrap(), &ctx)?,
                     // TODO predicate should be de-serialized
                     None,
                 )))
             }
             PhysicalPlanType::AvroScan(scan) => Ok(Arc::new(AvroExec::new(
-                scan.base_conf.as_ref().unwrap().try_into()?,
+                decode_scan_config(scan.base_conf.as_ref().unwrap(), &ctx)?,
             ))),
             PhysicalPlanType::CoalesceBatches(coalesce_batches) => {
                 let input: Arc<dyn ExecutionPlan> =

ballista/rust/executor/src/execution_loop.rs

@@ -15,7 +15,6 @@
 // specific language governing permissions and limitations
 // under the License.
 
-
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::mpsc::{Receiver, Sender, TryRecvError};
 use std::{sync::Arc, time::Duration};

ballista/rust/executor/src/executor.rs

@@ -28,8 +28,7 @@ use datafusion::error::DataFusionError;
 use datafusion::execution::runtime_env::RuntimeEnv;
 use datafusion::physical_plan::display::DisplayableExecutionPlan;
 use datafusion::physical_plan::{ExecutionPlan, Partitioning};
-use datafusion::prelude::{ExecutionContext,ExecutionConfig};
-
+use datafusion::prelude::{ExecutionConfig, ExecutionContext};
 
 /// Ballista executor
 pub struct Executor<T: 'static + AsExecutionPlan> {