akshaychitneni commented on code in PR #1993: URL: https://github.com/apache/datafusion-ballista/pull/1993#discussion_r3941386803
########## ballista/scheduler/src/checkpoint.rs: ########## @@ -0,0 +1,595 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Resolution of lazy checkpoints into separate jobs. +//! +//! A `BallistaCheckpointNode` in a submitted logical plan means "materialise +//! everything below me before evaluating anything above me". This module turns +//! that into two jobs at planning time: a `CopyTo` job that writes the subtree +//! to the checkpoint location, and the caller's job, rewritten to scan it. +//! +//! Nothing here reaches physical planning: by the time the plan is handed to +//! `create_physical_plan` every checkpoint node has been replaced by a scan. + +use async_trait::async_trait; +use ballista_core::error::{BallistaError, Result}; +use ballista_core::extension::BallistaCheckpointNode; +use datafusion::common::DFSchema; +use datafusion::common::tree_node::{TreeNode, TreeNodeRecursion}; +use datafusion::datasource::file_format::DefaultFileType; +use datafusion::datasource::file_format::parquet::ParquetFormatFactory; +use datafusion::datasource::listing::ListingTableUrl; +use datafusion::logical_expr::{Expr, LogicalPlanBuilder, UserDefinedLogicalNodeCore}; +use datafusion::logical_expr::{LogicalPlan, dml::CopyTo}; +use datafusion::prelude::{ParquetReadOptions, SessionContext}; +use futures::StreamExt; +use futures::future::BoxFuture; +use std::sync::Arc; + +/// Runs a plan to completion as its own job. +/// +/// Implemented by `SchedulerServer`; the indirection keeps this module free of +/// the event loop and lets tests substitute a local runner. +#[async_trait] +pub(crate) trait CheckpointMaterializer: Send + Sync { Review Comment: I think .cache() (#1829) needs this same materialize-then-substitute approach. I think a generic name here like `Materializer`, and materialize() returning the output's partition locations (not just Result<()>) so cache can register/pin where the output landed, would let it reuse this instead of re-deriving. I can propose the generalization as a follow-up so it doesn't grow this PR's scope. cc @sandugood @milenkovicm -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
