fix

Author: BugenZhao

Cargo.lock

@@ -72,7 +72,7 @@ impl Debug for GrpcExchangeSource {
 }
 
 impl ExchangeSource for GrpcExchangeSource {
-    type TakeDataFuture<'a> = impl Future<Output = Result<Option<DataChunk>>>;
+    type TakeDataFuture<'a> = impl Future<Output = Result<Option<DataChunk>>> + 'a;
 
     fn take_data(&mut self) -> Self::TakeDataFuture<'_> {
         async {

src/batch/src/execution/grpc_exchange.rs

@@ -52,7 +52,7 @@ impl Debug for LocalExchangeSource {
 }
 
 impl ExchangeSource for LocalExchangeSource {
-    type TakeDataFuture<'a> = impl Future<Output = Result<Option<DataChunk>>>;
+    type TakeDataFuture<'a> = impl Future<Output = Result<Option<DataChunk>>> + 'a;
 
     fn take_data(&mut self) -> Self::TakeDataFuture<'_> {
         async {

src/batch/src/execution/local_exchange.rs

@@ -251,7 +251,7 @@ impl FakeExchangeSource {
 }
 
 impl ExchangeSource for FakeExchangeSource {
-    type TakeDataFuture<'a> = impl Future<Output = Result<Option<DataChunk>>>;
+    type TakeDataFuture<'a> = impl Future<Output = Result<Option<DataChunk>>> + 'a;
 
     fn take_data(&mut self) -> Self::TakeDataFuture<'_> {
         async {

src/batch/src/executor/test_utils.rs

@@ -42,7 +42,7 @@ impl Debug for BroadcastSender {
 }
 
 impl ChanSender for BroadcastSender {
-    type SendFuture<'a> = impl Future<Output = BatchResult<()>>;
+    type SendFuture<'a> = impl Future<Output = BatchResult<()>> + 'a;
 
     fn send(&mut self, chunk: Option<DataChunk>) -> Self::SendFuture<'_> {
         async move {
@@ -65,7 +65,7 @@ pub struct BroadcastReceiver {
 }
 
 impl ChanReceiver for BroadcastReceiver {
-    type RecvFuture<'a> = impl Future<Output = Result<Option<DataChunkInChannel>>>;
+    type RecvFuture<'a> = impl Future<Output = Result<Option<DataChunkInChannel>>> + 'a;
 
     fn recv(&mut self) -> Self::RecvFuture<'_> {
         async move {

src/batch/src/task/broadcast_channel.rs

@@ -39,7 +39,7 @@ pub struct FifoReceiver {
 }
 
 impl ChanSender for FifoSender {
-    type SendFuture<'a> = impl Future<Output = BatchResult<()>>;
+    type SendFuture<'a> = impl Future<Output = BatchResult<()>> + 'a;
 
     fn send(&mut self, chunk: Option<DataChunk>) -> Self::SendFuture<'_> {
         async {
@@ -52,7 +52,7 @@ impl ChanSender for FifoSender {
 }
 
 impl ChanReceiver for FifoReceiver {
-    type RecvFuture<'a> = impl Future<Output = Result<Option<DataChunkInChannel>>>;
+    type RecvFuture<'a> = impl Future<Output = Result<Option<DataChunkInChannel>>> + 'a;
 
     fn recv(&mut self) -> Self::RecvFuture<'_> {
         async move {

src/batch/src/task/fifo_channel.rs

@@ -105,7 +105,7 @@ fn generate_new_data_chunks(
 }
 
 impl ChanSender for HashShuffleSender {
-    type SendFuture<'a> = impl Future<Output = BatchResult<()>>;
+    type SendFuture<'a> = impl Future<Output = BatchResult<()>> + 'a;
 
     fn send(&mut self, chunk: Option<DataChunk>) -> Self::SendFuture<'_> {
         async move {
@@ -150,7 +150,7 @@ impl HashShuffleSender {
 }
 
 impl ChanReceiver for HashShuffleReceiver {
-    type RecvFuture<'a> = impl Future<Output = Result<Option<DataChunkInChannel>>>;
+    type RecvFuture<'a> = impl Future<Output = Result<Option<DataChunkInChannel>>> + 'a;
 
     fn recv(&mut self) -> Self::RecvFuture<'_> {
         async move {

src/batch/src/task/hash_shuffle_channel.rs

@@ -302,10 +302,8 @@ impl<I: HummockIterator> UnorderedMergeIteratorInner<I> {
         // before returning.
 
         // TODO(chi): workaround for Rust toolchain 2022-10-16, removed boxed() later
-
-        let f = unsafe_boxed_static_future(node.iter.next());
-
-        match f.await {
+        
+        match unsafe_boxed_static_future(node.iter.next()).await {
             Ok(_) => {}
             Err(e) => {
                 // If the iterator returns error, we should clear the heap, so that this