chore: improve the unwrap error (#18571)

BohuTANG · web-flow · commit d0f0e4bed46d · 2025-08-20T16:48:15.000+08:00
diff --git a/src/query/pipeline/core/src/finished_chain.rs b/src/query/pipeline/core/src/finished_chain.rs
@@ -186,7 +186,7 @@ impl FinishedCallbackChain {
 
     fn log_states(apply_states: &[ApplyState]) {
         let mut message = String::new();
-        writeln!(&mut message, "Executor apply finished callback state:").unwrap();
+        let _ = writeln!(&mut message, "Executor apply finished callback state:");
         for apply_state in apply_states {
             let execute_state = match apply_state.successfully {
                 true => "\u{2705}",
@@ -201,7 +201,7 @@ impl FinishedCallbackChain {
                 },
             };
 
-            writeln!(
+            let _ = writeln!(
                 &mut message,
                 "├──{}:{:?} - {}{}:{}:{}",
                 execute_state,
@@ -210,8 +210,7 @@ impl FinishedCallbackChain {
                 apply_state.location.file(),
                 apply_state.location.line(),
                 apply_state.location.column()
-            )
-            .unwrap();
+            );
         }
 
         info!("{}", message);
diff --git a/src/query/pipeline/core/src/pipeline.rs b/src/query/pipeline/core/src/pipeline.rs
@@ -140,11 +140,9 @@ impl Pipeline {
     }
 
     pub fn finalize(mut self, root_scope: Option<Arc<PlanScope>>) -> Pipeline {
-        if root_scope.is_none() {
+        let Some(root_scope) = root_scope else {
             return self;
-        }
-
-        let root_scope = root_scope.unwrap();
+        };
 
         for node in self.graph.node_weights_mut() {
             let Some(scope) = node.scope.as_mut() else {
@@ -598,7 +596,10 @@ impl Pipeline {
             if let Some((source, target)) = other.graph.edge_endpoints(index) {
                 let source = NodeIndex::new(offset + source.index());
                 let target = NodeIndex::new(offset + target.index());
-                let edge_weight = other.graph.edge_weight(index).unwrap();
+                let edge_weight = other
+                    .graph
+                    .edge_weight(index)
+                    .expect("Edge weight must exist for valid edge index");
                 self.graph.add_edge(source, target, edge_weight.clone());
             }
         }
diff --git a/src/query/pipeline/core/src/processors/duplicate_processor.rs b/src/query/pipeline/core/src/processors/duplicate_processor.rs
@@ -81,7 +81,11 @@ impl Processor for DuplicateProcessor {
 
         self.input.set_need_data();
         if self.input.has_data() {
-            let block = self.input.pull_data().unwrap();
+            let block = self.input.pull_data().ok_or_else(|| {
+                databend_common_exception::ErrorCode::Internal(
+                    "Failed to pull data from input port when data is available",
+                )
+            })?;
             for output in self.outputs.iter() {
                 if !output.is_finished() {
                     output.push_data(block.clone());
diff --git a/src/query/pipeline/core/src/processors/port_trigger.rs b/src/query/pipeline/core/src/processors/port_trigger.rs
@@ -39,10 +39,20 @@ pub enum DirectedEdge {
 }
 
 impl DirectedEdge {
-    pub fn get_source<N, E>(&self, graph: &StableGraph<N, E>) -> NodeIndex {
+    pub fn get_source<N, E>(&self, graph: &StableGraph<N, E>) -> Result<NodeIndex> {
         match self {
-            DirectedEdge::Source(edge_index) => graph.edge_endpoints(*edge_index).unwrap().0,
-            DirectedEdge::Target(edge_index) => graph.edge_endpoints(*edge_index).unwrap().1,
+            DirectedEdge::Source(edge_index) => graph
+                .edge_endpoints(*edge_index)
+                .map(|(source, _)| source)
+                .ok_or_else(|| {
+                    ErrorCode::Internal(format!("Edge not found in graph: {:?}", edge_index))
+                }),
+            DirectedEdge::Target(edge_index) => graph
+                .edge_endpoints(*edge_index)
+                .map(|(_, target)| target)
+                .ok_or_else(|| {
+                    ErrorCode::Internal(format!("Edge not found in graph: {:?}", edge_index))
+                }),
         }
     }
 
@@ -105,13 +115,20 @@ impl UpdateList {
     /// # Safety
     ///
     /// Must be thread safe call. In other words, it needs to be called in single thread or in mutex guard.
-    pub unsafe fn create_trigger(self: &Arc<Self>, edge_index: EdgeIndex) -> *mut UpdateTrigger {
+    pub unsafe fn create_trigger(
+        self: &Arc<Self>,
+        edge_index: EdgeIndex,
+    ) -> Result<*mut UpdateTrigger> {
         let inner = &mut *self.inner.get();
         let update_trigger = UpdateTrigger::create(edge_index, self.inner.get());
         inner
             .updated_triggers
             .push(Arc::new(UnsafeCell::new(update_trigger)));
-        inner.updated_triggers.last().unwrap().get()
+        inner
+            .updated_triggers
+            .last()
+            .map(|trigger| trigger.get())
+            .ok_or_else(|| ErrorCode::Internal("Failed to get last trigger after push"))
     }
 }
 
diff --git a/src/query/pipeline/core/src/processors/resize_processor.rs b/src/query/pipeline/core/src/processors/resize_processor.rs
@@ -112,7 +112,11 @@ impl Processor for ResizeProcessor {
         }
 
         while !self.waiting_outputs.is_empty() && !self.waiting_inputs.is_empty() {
-            let output_index = self.waiting_outputs.pop_front().unwrap();
+            let output_index = self.waiting_outputs.pop_front().ok_or_else(|| {
+                databend_common_exception::ErrorCode::Internal(
+                    "Waiting outputs queue should not be empty",
+                )
+            })?;
 
             // Port is finished when waiting.
             if self.outputs[output_index].port.is_finished() {
@@ -124,11 +128,18 @@ impl Processor for ResizeProcessor {
                 continue;
             }
 
-            let input_index = self.waiting_inputs.pop_front().unwrap();
-
-            self.outputs[output_index]
-                .port
-                .push_data(self.inputs[input_index].port.pull_data().unwrap());
+            let input_index = self.waiting_inputs.pop_front().ok_or_else(|| {
+                databend_common_exception::ErrorCode::Internal(
+                    "Waiting inputs queue should not be empty",
+                )
+            })?;
+
+            let data = self.inputs[input_index].port.pull_data().ok_or_else(|| {
+                databend_common_exception::ErrorCode::Internal(
+                    "Failed to pull data from input port",
+                )
+            })?;
+            self.outputs[output_index].port.push_data(data);
             self.inputs[input_index].status = PortStatus::Idle;
             self.outputs[output_index].status = PortStatus::Idle;
 
diff --git a/src/query/pipeline/core/src/processors/sequence_group.rs b/src/query/pipeline/core/src/processors/sequence_group.rs
@@ -116,7 +116,11 @@ impl Processor for SequenceGroupProcessor {
                 }
             } else if input.port.has_data() {
                 if self.ignore_output {
-                    let _ignore = input.port.pull_data().unwrap()?;
+                    let _ignore = input.port.pull_data().ok_or_else(|| {
+                        databend_common_exception::ErrorCode::Internal(
+                            "Failed to pull data from input port when ignoring output",
+                        )
+                    })??;
                     input.status = PortStatus::Idle;
 
                     if input.port.is_finished() {
@@ -135,7 +139,11 @@ impl Processor for SequenceGroupProcessor {
         }
 
         while !self.waiting_outputs.is_empty() && !self.waiting_inputs.is_empty() {
-            let output_index = self.waiting_outputs.pop_front().unwrap();
+            let output_index = self.waiting_outputs.pop_front().ok_or_else(|| {
+                databend_common_exception::ErrorCode::Internal(
+                    "Waiting outputs queue should not be empty",
+                )
+            })?;
 
             // Port is finished when waiting.
             if self.outputs[output_index].port.is_finished() {
@@ -147,11 +155,18 @@ impl Processor for SequenceGroupProcessor {
                 continue;
             }
 
-            let input_index = self.waiting_inputs.pop_front().unwrap();
-
-            self.outputs[output_index]
-                .port
-                .push_data(self.inputs[input_index].port.pull_data().unwrap());
+            let input_index = self.waiting_inputs.pop_front().ok_or_else(|| {
+                databend_common_exception::ErrorCode::Internal(
+                    "Waiting inputs queue should not be empty",
+                )
+            })?;
+
+            let data = self.inputs[input_index].port.pull_data().ok_or_else(|| {
+                databend_common_exception::ErrorCode::Internal(
+                    "Failed to pull data from input port",
+                )
+            })?;
+            self.outputs[output_index].port.push_data(data);
             self.inputs[input_index].status = PortStatus::Idle;
             self.outputs[output_index].status = PortStatus::Idle;
 
diff --git a/src/query/pipeline/core/src/processors/shuffle_processor.rs b/src/query/pipeline/core/src/processors/shuffle_processor.rs
@@ -131,7 +131,12 @@ impl Processor for ShuffleProcessor {
         }
 
         if input.has_data() {
-            output.push_data(input.pull_data().unwrap());
+            let data = input.pull_data().ok_or_else(|| {
+                databend_common_exception::ErrorCode::Internal(
+                    "Failed to pull data from input port in shuffle processor",
+                )
+            })?;
+            output.push_data(data);
             return Ok(Event::NeedConsume);
         }
 
@@ -230,7 +235,12 @@ impl<T: Exchange> Processor for PartitionProcessor<T> {
         }
 
         if self.input.has_data() {
-            self.input_data = Some(self.input.pull_data().unwrap()?);
+            let data = self.input.pull_data().ok_or_else(|| {
+                databend_common_exception::ErrorCode::Internal(
+                    "Failed to pull data from input port in partition processor",
+                )
+            })??;
+            self.input_data = Some(data);
             return Ok(Event::Sync);
         }
 
@@ -327,12 +337,22 @@ impl<T: Exchange> Processor for MergePartitionProcessor<T> {
                 match T::STRATEGY {
                     MultiwayStrategy::Random => {
                         if self.output.can_push() {
-                            self.output.push_data(Ok(input.pull_data().unwrap()?));
+                            let data = input.pull_data().ok_or_else(|| {
+                                databend_common_exception::ErrorCode::Internal(
+                                    "Failed to pull data from input port in multiway random strategy"
+                                )
+                            })??;
+                            self.output.push_data(Ok(data));
                         }
                     }
                     MultiwayStrategy::Custom => {
                         if self.inputs_data[index].is_none() {
-                            self.inputs_data[index] = Some(input.pull_data().unwrap()?);
+                            let data = input.pull_data().ok_or_else(|| {
+                                databend_common_exception::ErrorCode::Internal(
+                                    "Failed to pull data from input port in multiway custom strategy"
+                                )
+                            })??;
+                            self.inputs_data[index] = Some(data);
                         }
                     }
                 }
diff --git a/src/query/pipeline/sinks/src/async_mpsc_sink.rs b/src/query/pipeline/sinks/src/async_mpsc_sink.rs
@@ -140,7 +140,11 @@ impl<T: AsyncMpscSink + 'static> Processor for AsyncMpscSinker<T> {
                         Ok(Event::Async)
                     }
                 } else {
-                    let block = input.pull_data().unwrap()?;
+                    let block = input.pull_data().ok_or_else(|| {
+                        databend_common_exception::ErrorCode::Internal(
+                            "Failed to pull data from input port in async mpsc sink",
+                        )
+                    })??;
                     self.input_data = Some(block);
                     Ok(Event::Async)
                 }
diff --git a/src/query/pipeline/sinks/src/async_sink.rs b/src/query/pipeline/sinks/src/async_sink.rs
@@ -135,7 +135,12 @@ impl<T: AsyncSink + 'static> Processor for AsyncSinker<T> {
         match self.input.has_data() {
             true => {
                 // Wake up upstream while executing async work
-                self.input_data = Some(self.input.pull_data().unwrap()?);
+                let data = self.input.pull_data().ok_or_else(|| {
+                    databend_common_exception::ErrorCode::Internal(
+                        "Failed to pull data from input port in async sink",
+                    )
+                })??;
+                self.input_data = Some(data);
                 self.input.set_need_data();
                 Ok(Event::Async)
             }
@@ -150,12 +155,37 @@ impl<T: AsyncSink + 'static> Processor for AsyncSinker<T> {
     async fn async_process(&mut self) -> Result<()> {
         if !self.called_on_start {
             self.called_on_start = true;
-            self.inner.as_mut().unwrap().on_start().await?;
+            self.inner
+                .as_mut()
+                .ok_or_else(|| {
+                    databend_common_exception::ErrorCode::Internal(
+                        "AsyncSink inner is None when calling on_start",
+                    )
+                })?
+                .on_start()
+                .await?;
         } else if let Some(data_block) = self.input_data.take() {
-            self.finished = self.inner.as_mut().unwrap().consume(data_block).await?;
+            self.finished = self
+                .inner
+                .as_mut()
+                .ok_or_else(|| {
+                    databend_common_exception::ErrorCode::Internal(
+                        "AsyncSink inner is None when consuming data",
+                    )
+                })?
+                .consume(data_block)
+                .await?;
         } else if !self.called_on_finish {
             self.called_on_finish = true;
-            self.inner.as_mut().unwrap().on_finish().await?;
+            self.inner
+                .as_mut()
+                .ok_or_else(|| {
+                    databend_common_exception::ErrorCode::Internal(
+                        "AsyncSink inner is None when calling on_finish",
+                    )
+                })?
+                .on_finish()
+                .await?;
         }
 
         Ok(())
diff --git a/src/query/pipeline/sinks/src/sync_mpsc_sink.rs b/src/query/pipeline/sinks/src/sync_mpsc_sink.rs
@@ -138,7 +138,11 @@ impl<T: SyncMpscSink + 'static> Processor for SyncMpscSinker<T> {
                         Ok(Event::Sync)
                     }
                 } else {
-                    let block = input.pull_data().unwrap()?;
+                    let block = input.pull_data().ok_or_else(|| {
+                        databend_common_exception::ErrorCode::Internal(
+                            "Failed to pull data from input port in sync mpsc sink",
+                        )
+                    })??;
                     self.input_data = Some(block);
                     Ok(Event::Sync)
                 }
diff --git a/src/query/pipeline/sinks/src/sync_sink.rs b/src/query/pipeline/sinks/src/sync_sink.rs
@@ -97,7 +97,12 @@ impl<T: Sink + 'static> Processor for Sinker<T> {
 
         match self.input.has_data() {
             true => {
-                self.input_data = Some(self.input.pull_data().unwrap()?);
+                let data = self.input.pull_data().ok_or_else(|| {
+                    databend_common_exception::ErrorCode::Internal(
+                        "Failed to pull data from input port in sync sink",
+                    )
+                })??;
+                self.input_data = Some(data);
                 Ok(Event::Sync)
             }
             false => {
diff --git a/src/query/pipeline/sinks/src/sync_sink_sender.rs b/src/query/pipeline/sinks/src/sync_sink_sender.rs
@@ -38,7 +38,12 @@ impl Sink for SyncSenderSink {
     const NAME: &'static str = "SyncSenderSink";
 
     fn consume(&mut self, data_block: DataBlock) -> Result<()> {
-        self.sender.blocking_send(Ok(data_block)).unwrap();
+        self.sender.blocking_send(Ok(data_block)).map_err(|e| {
+            databend_common_exception::ErrorCode::Internal(format!(
+                "Failed to send data block in sync sender sink: {}",
+                e
+            ))
+        })?;
         Ok(())
     }
 }
diff --git a/src/query/service/src/pipelines/executor/executor_graph.rs b/src/query/service/src/pipelines/executor/executor_graph.rs
@@ -158,7 +158,9 @@ impl Node {
     }
 
     pub unsafe fn create_trigger(&self, index: EdgeIndex) -> *mut UpdateTrigger {
-        self.updated_list.create_trigger(index)
+        self.updated_list
+            .create_trigger(index)
+            .expect("Failed to create trigger")
     }
 }
 
diff --git a/src/query/service/src/schedulers/fragments/plan_fragment.rs b/src/query/service/src/schedulers/fragments/plan_fragment.rs
diff --git a/src/query/service/src/servers/http/middleware/session_header.rs b/src/query/service/src/servers/http/middleware/session_header.rs
diff --git a/src/query/settings/src/settings.rs b/src/query/settings/src/settings.rs
diff --git a/tests/suites/1_stateful/09_http_handler/09_0007_session.py b/tests/suites/1_stateful/09_http_handler/09_0007_session.py

Original file line number	Diff line number	Diff line change
`@@ -131,7 +131,12 @@ impl Processor for ShuffleProcessor {`
`131`	`131`	`}`
`132`	`132`
`133`	`133`	`if input.has_data() {`
`134`		`- output.push_data(input.pull_data().unwrap());`
	`134`	`+ let data = input.pull_data().ok_or_else(\|\| {`
	`135`	`+ databend_common_exception::ErrorCode::Internal(`
	`136`	`+ "Failed to pull data from input port in shuffle processor",`
	`137`	`+ )`
	`138`	`+ })?;`
	`139`	`+ output.push_data(data);`
`135`	`140`	`return Ok(Event::NeedConsume);`
`136`	`141`	`}`
`137`	`142`
`@@ -230,7 +235,12 @@ impl<T: Exchange> Processor for PartitionProcessor<T> {`
`230`	`235`	`}`
`231`	`236`
`232`	`237`	`if self.input.has_data() {`
`233`		`- self.input_data = Some(self.input.pull_data().unwrap()?);`
	`238`	`+ let data = self.input.pull_data().ok_or_else(\|\| {`
	`239`	`+ databend_common_exception::ErrorCode::Internal(`
	`240`	`+ "Failed to pull data from input port in partition processor",`
	`241`	`+ )`
	`242`	`+ })??;`
	`243`	`+ self.input_data = Some(data);`
`234`	`244`	`return Ok(Event::Sync);`
`235`	`245`	`}`
`236`	`246`
`@@ -327,12 +337,22 @@ impl<T: Exchange> Processor for MergePartitionProcessor<T> {`
`327`	`337`	`match T::STRATEGY {`
`328`	`338`	`MultiwayStrategy::Random => {`
`329`	`339`	`if self.output.can_push() {`
`330`		`- self.output.push_data(Ok(input.pull_data().unwrap()?));`
	`340`	`+ let data = input.pull_data().ok_or_else(\|\| {`
	`341`	`+ databend_common_exception::ErrorCode::Internal(`
	`342`	`+ "Failed to pull data from input port in multiway random strategy"`
	`343`	`+ )`
	`344`	`+ })??;`
	`345`	`+ self.output.push_data(Ok(data));`
`331`	`346`	`}`
`332`	`347`	`}`
`333`	`348`	`MultiwayStrategy::Custom => {`
`334`	`349`	`if self.inputs_data[index].is_none() {`
`335`		`- self.inputs_data[index] = Some(input.pull_data().unwrap()?);`
	`350`	`+ let data = input.pull_data().ok_or_else(\|\| {`
	`351`	`+ databend_common_exception::ErrorCode::Internal(`
	`352`	`+ "Failed to pull data from input port in multiway custom strategy"`
	`353`	`+ )`
	`354`	`+ })??;`
	`355`	`+ self.inputs_data[index] = Some(data);`
`336`	`356`	`}`
`337`	`357`	`}`
`338`	`358`	`}`
Original file line number	Diff line number	Diff line change
`@@ -140,7 +140,11 @@ impl<T: AsyncMpscSink + 'static> Processor for AsyncMpscSinker<T> {`
`140`	`140`	`Ok(Event::Async)`
`141`	`141`	`}`
`142`	`142`	`} else {`
`143`		`- let block = input.pull_data().unwrap()?;`
	`143`	`+ let block = input.pull_data().ok_or_else(\|\| {`
	`144`	`+ databend_common_exception::ErrorCode::Internal(`
	`145`	`+ "Failed to pull data from input port in async mpsc sink",`
	`146`	`+ )`
	`147`	`+ })??;`
`144`	`148`	`self.input_data = Some(block);`
`145`	`149`	`Ok(Event::Async)`
`146`	`150`	`}`
Original file line number	Diff line number	Diff line change
`@@ -138,7 +138,11 @@ impl<T: SyncMpscSink + 'static> Processor for SyncMpscSinker<T> {`
`138`	`138`	`Ok(Event::Sync)`
`139`	`139`	`}`
`140`	`140`	`} else {`
`141`		`- let block = input.pull_data().unwrap()?;`
	`141`	`+ let block = input.pull_data().ok_or_else(\|\| {`
	`142`	`+ databend_common_exception::ErrorCode::Internal(`
	`143`	`+ "Failed to pull data from input port in sync mpsc sink",`
	`144`	`+ )`
	`145`	`+ })??;`
`142`	`146`	`self.input_data = Some(block);`
`143`	`147`	`Ok(Event::Sync)`
`144`	`148`	`}`