This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23530-21986cf76d35cab60a6b7d0c334689486ddb2492 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit 3a29d6bd8cc9ac2bf5efee9f070dcdeea9f97b32 Author: Pepijn Van Eeckhoudt <[email protected]> AuthorDate: Mon Jul 13 18:02:12 2026 +0200 Add minimal genarator-like stream implementation (#23530) ## Which issue does this PR close? None, related to PR #23407 ## Rationale for this change Manually writing `Stream` implementations can be quite tedious since it requires implementing the state machine of the stream yourself. This often results in hard to read code. The same problem exists with manual `Future` implementations and `async` was added to the Rust language to mitigate exactly this problem. Unfortunately there's no language level support yet to help with writing streams/generators. There are quite a few projects that attempt to fill this gap: - Tokio's [async_stream](https://docs.rs/async-stream/latest/async_stream/) provides a proc macro that recognises a `yield` keyword. This is a good implementation, but proc macros don't play nice with code formatting, increase compile time, and in this particular case prevent decomposition into smaller functions. - [async_fn_stream](https://docs.rs/async-fn-stream/latest/async_fn_stream/) provides an implementation of the same concept, but without the proc macro. Unfortunately this implementation chooses a heavier mechanism to communicate generated values back to the stream via a `SmallVec`. - [genawaiter](https://docs.rs/genawaiter/latest/genawaiter/sync/index.html) is a more general purpose generator library, but this can also be used to implement `Stream`s. This project does miss some of the ergonomics provided by the other two in the form of `try_` variants that help in implementing fallible streams. Since none of these variants seems like the ideal candidate, the best option might be to have a custom implementation of the concept tailored to the needs of the DataFusion project. This PR provides an initial draft of exactly that. ## What changes are included in this PR? - Adds `async_stream` and `async_try_stream` functions that create Stream implementations based on an async generator function. The initial implementation was inspired by the macro expansion produced by `async_stream`. The code was then adapted further taking inspiration from the two other libraries. Specifically, the `Emitter` terminology was taken from `async_fn_stream` and the `Arc<Mutex<Option<T>>>` value transfer mechanism was taken from `genawaiter`. `async_stream` uses a very light weight thread-local storage based mechanism to handle value transfer, but this felt too risky to use when the emitter is exposed. It makes sense in the Tokio implementation since the proc macro can manage control flow in a stricter way. I wasn't entirely sure if taking some inspiration from has code licensing implications. I applied ASLv2 for now on the added code. ## Are these changes tested? Test code was mainly adapted from the tokio implementation. ## Are there any user-facing changes? Yes, new public function `async_stream` and `async_try_stream` --------- Co-authored-by: Raz Luvaton <[email protected]> --- Cargo.lock | 1 + Cargo.toml | 1 + datafusion/execution/Cargo.toml | 1 + datafusion/execution/src/async_stream.rs | 729 +++++++++++++++++++++++++++++++ datafusion/execution/src/lib.rs | 3 + datafusion/physical-plan/Cargo.toml | 2 +- 6 files changed, 736 insertions(+), 1 deletion(-) diff --git a/Cargo.lock b/Cargo.lock index 45d7e5b15e..49e298d316 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2120,6 +2120,7 @@ dependencies = [ "object_store", "parking_lot", "parquet", + "pin-project-lite", "rand 0.9.4", "tempfile", "tokio", diff --git a/Cargo.toml b/Cargo.toml index a59a9d69c4..72676c8263 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -187,6 +187,7 @@ pbjson = { version = "0.9.0" } pbjson-types = "0.9" percent-encoding = "2.3" pin-project = "1" +pin-project-lite = "^0.2.7" # Should match arrow-flight's version of prost. prost = "0.14.1" rand = "0.9" diff --git a/datafusion/execution/Cargo.toml b/datafusion/execution/Cargo.toml index 0aa2739e35..c9d4acd364 100644 --- a/datafusion/execution/Cargo.toml +++ b/datafusion/execution/Cargo.toml @@ -65,6 +65,7 @@ log = { workspace = true } object_store = { workspace = true, features = ["fs"] } parking_lot = { workspace = true } parquet = { workspace = true, optional = true } +pin-project-lite = { workspace = true } rand = { workspace = true } tempfile = { workspace = true } tokio = { workspace = true } diff --git a/datafusion/execution/src/async_stream.rs b/datafusion/execution/src/async_stream.rs new file mode 100644 index 0000000000..a84984d192 --- /dev/null +++ b/datafusion/execution/src/async_stream.rs @@ -0,0 +1,729 @@ +// 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. + +use futures::Stream; +use futures::future::FusedFuture; +use futures::stream::FusedStream; +use parking_lot::Mutex; +use pin_project_lite::pin_project; +use std::ops::DerefMut; +use std::pin::Pin; +use std::sync::Arc; +use std::task::{Context, Poll}; + +/// Creates a [`Stream`] from an async generator function. +/// +/// The `generator` closure receives an [`Emitter<T>`] and runs as an async +/// block. Each `emitter.emit(value).await` call suspends the generator and +/// produces the next item in the stream. The stream ends when the generator +/// future resolves. +/// +/// # Example +/// +/// ``` +/// use datafusion_execution::async_stream; +/// use futures::StreamExt; +/// +/// # #[tokio::main(flavor = "current_thread")] +/// # async fn main() { +/// let stream = async_stream(|mut emitter| async move { +/// for i in 0_i32..3 { +/// emitter.emit(i).await; +/// } +/// }); +/// +/// let values: Vec<i32> = stream.collect().await; +/// assert_eq!(values, vec![0, 1, 2]); +/// # } +/// ``` +pub fn async_stream<T, F: Future<Output = ()>>( + generator: impl FnOnce(Emitter<T>) -> F, +) -> impl FusedStream<Item = T> { + let (emitter, receiver) = tx_rx(); + AsyncStream::new(receiver, generator(emitter)) +} + +/// Creates a fallible [`Stream`] from an async generator function. +/// +/// The `generator` closure receives a [`TryEmitter<T, E>`] and runs as an +/// async block that returns `Result<(), E>`. Each `emitter.emit(value).await` +/// call suspends the generator and produces `Ok(value)` as the next stream +/// item. The `?` operator can be used inside the generator to short-circuit on +/// errors: the error is emitted as the final `Err(e)` item and the stream +/// ends. The stream also ends when the generator future resolves to `Ok(())`. +/// +/// # Example +/// +/// ``` +/// use datafusion_execution::async_try_stream; +/// use futures::StreamExt; +/// +/// # #[tokio::main(flavor = "current_thread")] +/// # async fn main() { +/// let stream = async_try_stream(|mut emitter| async move { +/// emitter.emit(1_i32).await; +/// emitter.emit(2_i32).await; +/// Err::<(), _>("something went wrong")?; +/// emitter.emit(3_i32).await; // never reached +/// Ok(()) +/// }); +/// +/// let values: Vec<Result<i32, &str>> = stream.collect().await; +/// assert_eq!(values, vec![Ok(1), Ok(2), Err("something went wrong")]); +/// # } +/// ``` +pub fn async_try_stream<T, E, F: Future<Output = Result<(), E>>>( + generator: impl FnOnce(TryEmitter<T, E>) -> F, +) -> impl FusedStream<Item = Result<T, E>> { + let (try_emitter, mut emitter, receiver) = try_tx_rx::<T, E>(); + AsyncStream::new(receiver, async move { + if let Err(e) = generator(try_emitter).await { + emitter.emit(Err(e)).await + } + }) +} + +/// Creates an `Emitter`/`Receiver` pair +fn tx_rx<T>() -> (Emitter<T>, Receiver<T>) { + let slot = Arc::new(Mutex::new(None)); + ( + Emitter { + slot: Arc::clone(&slot), + }, + Receiver { slot }, + ) +} + +/// Creates an `TryEmitter`/`Emitter`/`Receiver` triplet +#[expect( + clippy::type_complexity, + reason = "three-element tuple is clearer than an alias here" +)] +fn try_tx_rx<T, E>() -> ( + TryEmitter<T, E>, + Emitter<Result<T, E>>, + Receiver<Result<T, E>>, +) { + let slot = Arc::new(Mutex::new(None)); + ( + TryEmitter { + slot: Arc::clone(&slot), + }, + Emitter { + slot: Arc::clone(&slot), + }, + Receiver { slot }, + ) +} + +/// Value slot shared between [`Emitter`] and [`Receiver`]. +/// Use `Arc<Mutex>` to ensure the created `Stream` implementations +/// are both `Send` and `Sync`. +type SlotRef<T> = Arc<Mutex<Option<T>>>; + +/// A handle for emitting values from an [`async_stream`] generator. +/// +/// The generator closure receives an `Emitter<T>` as its argument. +pub struct Emitter<T> { + slot: SlotRef<T>, +} + +/// A handle for emitting values from an [`async_try_stream`] generator. +/// +/// The generator closure receives a `TryEmitter<T, E>` as its argument. +pub struct TryEmitter<T, E> { + slot: SlotRef<Result<T, E>>, +} + +struct Receiver<T> { + slot: SlotRef<T>, +} + +impl<T> Emitter<T> { + /// Returns a `Future` that emits `value` as the next stream item. + /// + /// The returned future **must be awaited immediately**. On its first poll it + /// yields `Poll::Pending`, handing control back to the stream consumer so it + /// can observe the emitted value. On the next poll (triggered by the + /// consumer calling `poll_next` again) it completes with `Poll::Ready(())`, + /// resuming the generator. + /// + /// # Panics + /// + /// Panics if `emit` is called a second time before the previous future has + /// been awaited, because doing so would silently overwrite the unconsumed + /// value. + pub fn emit(&mut self, value: T) -> impl FusedFuture<Output = ()> { + let mut guard = self.slot.lock(); + match guard.deref_mut() { + Some(_) => panic!("Misuse: await was not called after calling emit"), + slot => *slot = Some(value), + } + + Emit { done: false } + } +} + +impl<T, E> TryEmitter<T, E> { + /// Emits `Ok(value)` as the next stream item and suspends the generator. + /// + /// Behaves identically to [`Emitter::emit`]: the returned future must be + /// awaited immediately and yields `Poll::Pending` on its first poll to + /// transfer control to the stream consumer. + /// + /// # Panics + /// + /// Panics if called before the previous emit future has been awaited. + pub fn emit(&mut self, value: T) -> impl FusedFuture<Output = ()> { + let mut guard = self.slot.lock(); + match guard.deref_mut() { + Some(_) => panic!("Misuse: await was not called after calling emit"), + slot => *slot = Some(Ok::<T, E>(value)), + } + + Emit { done: false } + } +} + +struct Emit { + done: bool, +} + +impl FusedFuture for Emit { + fn is_terminated(&self) -> bool { + self.done + } +} + +impl Future for Emit { + type Output = (); + + fn poll(mut self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<()> { + if !self.done { + self.done = true; + // Poll::Pending causes the generator to yield, returning control back to the + // calling Stream + Poll::Pending + } else { + Poll::Ready(()) + } + } +} + +pin_project! { + struct AsyncStream<T, U> { + rx: Receiver<T>, + done: bool, + #[pin] + generator: U, + } +} + +impl<T, U> AsyncStream<T, U> { + fn new(rx: Receiver<T>, generator: U) -> AsyncStream<T, U> { + AsyncStream { + rx, + done: false, + generator, + } + } +} + +impl<T, U> FusedStream for AsyncStream<T, U> +where + U: Future<Output = ()>, +{ + fn is_terminated(&self) -> bool { + self.done + } +} + +impl<T, U> Stream for AsyncStream<T, U> +where + U: Future<Output = ()>, +{ + type Item = T; + + fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> { + let this = self.project(); + + if *this.done { + return Poll::Ready(None); + } + + // The `Option::take` call below ensures the next time poll is called the slot is + // already set to None + debug_assert!(this.rx.slot.lock().is_none()); + let res = this.generator.poll(cx); + *this.done = res.is_ready(); + + match this.rx.slot.lock().take() { + // Generator filled slot -> return next stream item + Some(v) => Poll::Ready(Some(v)), + // Generator did not fill slot and completed -> return None to indicate end of stream + None if *this.done => Poll::Ready(None), + // Generator did not fill slot and not completed -> return Pending since some Future + // other than Emit returned Pending. + None => Poll::Pending, + } + } + + fn size_hint(&self) -> (usize, Option<usize>) { + if self.done { (0, Some(0)) } else { (0, None) } + } +} + +#[cfg(test)] +mod test { + use crate::async_stream::Emitter; + use crate::{async_stream, async_try_stream}; + use futures::stream::FusedStream; + use futures::{Stream, StreamExt, pin_mut}; + use std::assert_matches; + use std::pin::Pin; + use std::sync::Arc; + use std::sync::atomic::{AtomicUsize, Ordering}; + use std::task::{Context, Poll}; + use tokio::sync::mpsc; + + #[tokio::test] + async fn noop_stream() { + let s = async_stream(|_: Emitter<()>| async {}); + pin_mut!(s); + + assert_eq!(s.next().await, None); + } + + #[tokio::test] + async fn empty_stream() { + let mut ran = false; + + { + let r = &mut ran; + let s = async_stream(|_: Emitter<()>| async { + *r = true; + println!("hello world!"); + }); + pin_mut!(s); + + assert_eq!(s.next().await, None); + } + + assert!(ran); + } + + #[tokio::test] + async fn emit_single_value() { + let s = async_stream(|mut emitter| async move { + emitter.emit("hello").await; + }); + + let values: Vec<_> = s.collect().await; + + assert_eq!(1, values.len()); + assert_eq!("hello", values[0]); + } + + #[tokio::test] + async fn fused() { + let s = async_stream(|mut emitter| async move { + emitter.emit("hello").await; + }); + pin_mut!(s); + + assert!(!s.is_terminated()); + assert_eq!(s.next().await, Some("hello")); + assert_eq!(s.next().await, None); + + assert!(s.is_terminated()); + // This should return None from now on + assert_eq!(s.next().await, None); + } + + #[tokio::test] + async fn emit_multi_value() { + let s = async_stream(|mut emitter| async move { + emitter.emit("hello").await; + emitter.emit("world").await; + emitter.emit("dizzy").await; + }); + + let values: Vec<_> = s.collect().await; + + assert_eq!(3, values.len()); + assert_eq!("hello", values[0]); + assert_eq!("world", values[1]); + assert_eq!("dizzy", values[2]); + } + + #[tokio::test] + #[should_panic = "await was not called after calling emit"] + async fn emit_without_await() { + let s = async_stream(|mut emitter| async move { + #[expect(clippy::let_underscore_future)] + { + let _ = emitter.emit("hello"); + let _ = emitter.emit("world"); + } + }); + + let _: Vec<_> = s.collect().await; + } + + #[tokio::test] + async fn unit_emit_in_select() { + use tokio::select; + + async fn do_stuff_async() {} + + let s = async_stream(|mut emitter| async move { + select! { + _ = do_stuff_async() => emitter.emit(()).await, + else => emitter.emit(()).await, + } + }); + + let values: Vec<_> = s.collect().await; + assert_eq!(values.len(), 1); + } + + #[tokio::test] + async fn emit_with_select() { + use tokio::select; + + async fn do_stuff_async() {} + async fn more_async_work() {} + + let s = async_stream(|mut emitter| async move { + select! { + _ = do_stuff_async() => emitter.emit("hey").await, + _ = more_async_work() => emitter.emit("hey").await, + else => emitter.emit("hey").await, + } + }); + + let values: Vec<_> = s.collect().await; + assert_eq!(values, vec!["hey"]); + } + + #[tokio::test] + async fn return_stream() { + fn build_stream() -> impl Stream<Item = u32> { + async_stream(|mut emitter| async move { + emitter.emit(1).await; + emitter.emit(2).await; + emitter.emit(3).await; + }) + } + + let s = build_stream(); + + let values: Vec<_> = s.collect().await; + assert_eq!(3, values.len()); + assert_eq!(1, values[0]); + assert_eq!(2, values[1]); + assert_eq!(3, values[2]); + } + + #[tokio::test] + async fn consume_channel() { + let (tx, mut rx) = mpsc::channel(10); + + let s = async_stream(|mut emitter| async move { + while let Some(v) = rx.recv().await { + emitter.emit(v).await; + } + }); + + pin_mut!(s); + + for i in 0..3 { + assert_matches!(tx.send(i).await, Ok(_)); + assert_eq!(Some(i), s.next().await); + } + + drop(tx); + assert_eq!(None, s.next().await); + } + + #[tokio::test] + async fn borrow_self() { + struct Data(String); + + impl Data { + fn stream(&self) -> impl Stream<Item = &str> + '_ { + async_stream(move |mut emitter| async move { + emitter.emit(&self.0[..]).await; + }) + } + } + + let data = Data("hello".to_string()); + let s = data.stream(); + pin_mut!(s); + + assert_eq!(Some("hello"), s.next().await); + } + + #[tokio::test] + async fn stream_in_stream() { + let s = async_stream(|mut emitter| async move { + let s = async_stream(|mut inner_emitter| async move { + for i in 0..3 { + inner_emitter.emit(i).await; + } + }); + + pin_mut!(s); + while let Some(v) = s.next().await { + emitter.emit(v).await; + } + }); + + let values: Vec<_> = s.collect().await; + assert_eq!(3, values.len()); + } + + // Demonstrates that capturing an outer Emitter<T> inside an inner async_stream with a + // different item type is no longer undefined behaviour: the outer emitter writes to its own + // typed slot, so the inner stream never sees any values. The outer stream receives the + // "foo" strings instead because they land in its slot. + #[tokio::test] + async fn stream_in_stream_misuse() { + let s = async_stream(|mut emitter| async move { + let s = async_stream(|_inner_emitter: Emitter<i32>| async move { + for _i in 0..3 { + emitter.emit("foo").await; + } + }); + + pin_mut!(s); + while let Some(v) = s.next().await { + println!("{v}"); + } + }); + + let values: Vec<_> = s.collect().await; + assert_eq!(3, values.len()); + } + + #[tokio::test] + async fn emit_non_unpin_value() { + let s: Vec<_> = async_stream(|mut emitter| async move { + for i in 0..3 { + emitter.emit(async move { i }).await; + } + }) + .buffered(1) + .collect() + .await; + + assert_eq!(s, vec![0, 1, 2]); + } + + #[tokio::test] + async fn should_not_call_handler_function_if_not_polled() { + let _ = async_stream(|_: Emitter<()>| async move { + panic!("should not be called"); + }); + } + + #[tokio::test] + async fn should_not_continue_until_next_poll() { + let s = async_stream(|mut emitter| async move { + emitter.emit("hey").await; + panic!("make sure poll based and not push based"); + }); + pin_mut!(s); + let _ = s.next().await; + } + + #[test] + fn inner_try_stream() { + use tokio::select; + + async fn do_stuff_async() {} + + let _ = async_stream(|mut emitter| async move { + select! { + _ = do_stuff_async() => { + let another_s = async_try_stream(|mut inner_emitter| async move { + inner_emitter.emit(()).await; + Ok(()) + }); + let _: Result<(), ()> = Box::pin(another_s).next().await.unwrap(); + }, + else => {}, + } + emitter.emit(()).await; + }); + } + + #[tokio::test] + async fn single_err() { + let s = async_try_stream(|mut emitter| async move { + if true { + Err("hello")?; + } else { + emitter.emit("world").await; + } + + unreachable!(); + }); + + let values: Vec<_> = s.collect().await; + assert_eq!(1, values.len()); + assert_eq!(Err("hello"), values[0]); + } + + #[tokio::test] + async fn emit_then_err() { + let s = async_try_stream(|mut emitter| async move { + emitter.emit("hello").await; + Err("world")?; + unreachable!(); + }); + + let values: Vec<_> = s.collect().await; + assert_eq!(2, values.len()); + assert_eq!(Ok("hello"), values[0]); + assert_eq!(Err("world"), values[1]); + } + + #[tokio::test] + async fn convert_err() { + struct ErrorA(u8); + #[derive(PartialEq, Debug)] + struct ErrorB(u8); + impl From<ErrorA> for ErrorB { + fn from(a: ErrorA) -> ErrorB { + ErrorB(a.0) + } + } + + fn test() -> impl Stream<Item = Result<&'static str, ErrorB>> { + async_try_stream(|mut emitter| async move { + if true { + Err(ErrorA(1))?; + } else { + Err(ErrorB(2))?; + } + emitter.emit("unreachable").await; + Ok(()) + }) + } + + let values: Vec<_> = test().collect().await; + assert_eq!(1, values.len()); + assert_eq!(Err(ErrorB(1)), values[0]); + } + + #[tokio::test] + async fn multi_try() { + fn test() -> impl Stream<Item = Result<i32, String>> { + async_try_stream(|mut emitter| async move { + let a = Ok::<_, String>(Ok::<_, String>(123))??; + for _ in 1..10 { + emitter.emit(a).await; + } + Ok(()) + }) + } + let values: Vec<_> = test().collect().await; + assert_eq!(9, values.len()); + assert_eq!( + std::iter::repeat_n(123, 9).map(Ok).collect::<Vec<_>>(), + values + ); + } + + use pin_project_lite::pin_project; + + pin_project! { + struct MyStream<T: Stream> { + #[pin] + input: T, + } + } + + impl<T: Stream> Stream for MyStream<T> { + type Item = T::Item; + + fn poll_next( + self: Pin<&mut Self>, + cx: &mut Context<'_>, + ) -> Poll<Option<Self::Item>> { + let this = self.project(); + this.input.poll_next(cx) + } + } + + #[tokio::test] + async fn emit_does_not_hold_on_value() { + let waker = futures::task::noop_waker_ref(); + let mut cx = Context::from_waker(waker); + + let run = Arc::<AtomicUsize>::new(AtomicUsize::new(0)); + let moved = Arc::clone(&run); + let s = async_stream(|mut emitter| async move { + for _ in 0..2 { + let before = moved.fetch_add(1, Ordering::SeqCst); + emitter.emit(before).await; + } + }); + + let mut my_stream = Box::pin(MyStream { input: s }); + + #[derive(Debug, PartialEq)] + struct Item { + before: usize, + result: Poll<Option<usize>>, + after: usize, + } + + let mut results = vec![]; + + assert_eq!(run.load(Ordering::SeqCst), 0); + + while run.load(Ordering::SeqCst) < 2 { + let before = run.load(Ordering::SeqCst); + let result = my_stream.poll_next_unpin(&mut cx); + let after = run.load(Ordering::SeqCst); + results.push(Item { + before, + result, + after, + }); + } + + assert_eq!( + results, + vec![ + Item { + before: 0, + result: Poll::Ready(Some(0)), + after: 1, + }, + Item { + before: 1, + result: Poll::Ready(Some(1)), + after: 2, + } + ] + ); + } +} diff --git a/datafusion/execution/src/lib.rs b/datafusion/execution/src/lib.rs index 5c646066ed..5af7064f1c 100644 --- a/datafusion/execution/src/lib.rs +++ b/datafusion/execution/src/lib.rs @@ -27,6 +27,7 @@ //! DataFusion execution configuration and runtime structures +mod async_stream; pub mod cache; pub mod config; pub mod disk_manager; @@ -38,12 +39,14 @@ pub mod runtime_env; pub mod spill_file; mod stream; mod task; + pub mod registry { pub use datafusion_expr::registry::{ FunctionRegistry, MemoryFunctionRegistry, SerializerRegistry, }; } +pub use async_stream::{Emitter, TryEmitter, async_stream, async_try_stream}; pub use disk_manager::DiskManager; pub use registry::FunctionRegistry; pub use spill_file::{SpillFile, SpillWriter, TempFileFactory}; diff --git a/datafusion/physical-plan/Cargo.toml b/datafusion/physical-plan/Cargo.toml index c43ae81003..543726bb63 100644 --- a/datafusion/physical-plan/Cargo.toml +++ b/datafusion/physical-plan/Cargo.toml @@ -85,7 +85,7 @@ itertools = { workspace = true, features = ["use_std"] } log = { workspace = true } num-traits = { workspace = true } parking_lot = { workspace = true } -pin-project-lite = "^0.2.7" +pin-project-lite = { workspace = true } serde_json = { workspace = true, features = ["preserve_order"] } tokio = { workspace = true } --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
