viirya commented on code in PR #6883: URL: https://github.com/apache/datafusion-comet/pull/6883#discussion_r4362179750
########## native/local/src/lib.rs: ########## @@ -0,0 +1,300 @@ +// 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. + +//! Query-scoped execution, independent of Spark tasks and JNI. +//! +//! A caller supplies a fresh, fully planned graph for each execution and receives one +//! result stream. The caller owns the Tokio runtime and must keep it running until +//! cleanup finishes. Spark admission and JNI live outside this crate. + +pub mod handle; + +use std::collections::VecDeque; +use std::panic::AssertUnwindSafe; +use std::pin::Pin; +use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard}; +use std::task::{Context, Poll}; + +use datafusion::arrow::datatypes::SchemaRef; +use datafusion::arrow::record_batch::RecordBatch; +use datafusion::common::{DataFusionError, Result}; +use datafusion::execution::TaskContext; +use datafusion::physical_plan::stream::RecordBatchStreamAdapter; +use datafusion::physical_plan::{ + execute_stream, ExecutionPlan, ExecutionPlanProperties, RecordBatchStream, + SendableRecordBatchStream, +}; +use futures::future::poll_fn; +use futures::{FutureExt, Stream, StreamExt}; +use tokio::task::JoinHandle; + +/// Batches each root partition may run ahead of the one being delivered, when results are +/// delivered in partition order. +const PARTITION_BUFFER: usize = 2; Review Comment: Thanks, confirmed. Your query at `local[8]` (median of 4 runs after a warmup, each mode in its own JVM, forward and reverse order, ms): | | Spark | Comet | Comet local | |---|---:|---:|---:| | before | 2668 / 3190 | 891 / 873 | 4511 / 4337 | | after (b55a395f9) | 3496 / 2864 | 1143 / 929 | 905 / 866 | The filter coalesces its output, so each partition yields only a handful of batches, and a partition waiting for its turn stopped after its first two. As you suggested, the started partitions now run ahead while the batches they hold together total less than `spark.driver.maxResultSize` (no limit when it is 0); past that, each holds at most two batches until the earlier ones are delivered, so the first partition always progresses. The bound on how many partitions start is unchanged. Two native tests cover it: a later partition runs to its end while the first delivers nothing, and with a zero limit it stops at two batches. `local-execution.md` describes the new behavior. The wide-projection collect from before is unaffected (5M rows × 8 longs: Comet 1111 / 979, local 547 / 496). -- 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]
