fix(query): propagate runtime provider errors

Signed-off-by: discord9 <discord9@163.com>

src/datanode/src/datanode.rs

@@ -61,8 +61,8 @@ use tokio::sync::Notify;
 use crate::config::{DatanodeOptions, RegionEngineConfig, StorageConfig};
 use crate::error::{
     self, BuildDatanodeSnafu, BuildMetricEngineSnafu, BuildMitoEngineSnafu, CreateDirSnafu,
-    GetMetadataSnafu, MissingCacheSnafu, MissingNodeIdSnafu, OpenLogStoreSnafu, Result,
-    ShutdownInstanceSnafu, ShutdownServerSnafu, StartServerSnafu,
+    DataFusionSnafu, GetMetadataSnafu, MissingCacheSnafu, MissingNodeIdSnafu, OpenLogStoreSnafu,
+    Result, ShutdownInstanceSnafu, ShutdownServerSnafu, StartServerSnafu,
 };
 use crate::event_listener::{
     NoopRegionServerEventListener, RegionServerEventListenerRef, RegionServerEventReceiver,
@@ -432,7 +432,7 @@ impl DatanodeBuilder {
     ) -> Result<RegionServer> {
         let opts: &DatanodeOptions = &self.opts;
 
-        let query_engine_factory = QueryEngineFactory::new_with_plugins(
+        let query_engine_factory = QueryEngineFactory::try_new_with_plugins(
             // query engine in datanode only executes plan with resolved table source.
             DummyCatalogManager::arc(),
             None,
@@ -443,7 +443,8 @@ impl DatanodeBuilder {
             false,
             self.plugins.clone(),
             opts.query.clone(),
-        );
+        )
+        .context(DataFusionSnafu)?;
         let query_engine = query_engine_factory.query_engine();
 
         let table_provider_factory = self

src/flow/src/server.rs

@@ -57,8 +57,8 @@ use crate::adapter::flownode_impl::{FlowDualEngine, FlowDualEngineRef};
 use crate::adapter::{FlowStreamingEngineRef, create_worker};
 use crate::batching_mode::engine::BatchingEngine;
 use crate::error::{
-    CacheRequiredSnafu, ExternalSnafu, ListFlowsSnafu, ParseAddrSnafu, ShutdownServerSnafu,
-    StartServerSnafu, UnexpectedSnafu, to_status_with_last_err,
+    CacheRequiredSnafu, DatafusionSnafu, ExternalSnafu, ListFlowsSnafu, ParseAddrSnafu,
+    ShutdownServerSnafu, StartServerSnafu, UnexpectedSnafu, to_status_with_last_err,
 };
 use crate::heartbeat::HeartbeatTask;
 use crate::metrics::{METRIC_FLOW_PROCESSING_TIME, METRIC_FLOW_ROWS};
@@ -384,7 +384,7 @@ impl FlownodeBuilder {
 
     pub async fn build(mut self) -> Result<FlownodeInstance, Error> {
         // TODO(discord9): does this query engine need those?
-        let query_engine_factory = QueryEngineFactory::new_with_plugins(
+        let query_engine_factory = QueryEngineFactory::try_new_with_plugins(
             // query engine in flownode is only used for translate plan with resolved table source.
             self.catalog_manager.clone(),
             None,
@@ -395,7 +395,10 @@ impl FlownodeBuilder {
             false,
             Default::default(),
             self.opts.query.clone(),
-        );
+        )
+        .context(DatafusionSnafu {
+            context: "Failed to build query engine",
+        })?;
         let manager = Arc::new(
             self.build_manager(query_engine_factory.query_engine())
                 .await?,

src/frontend/src/instance/builder.rs

@@ -45,7 +45,7 @@ use query::QueryEngineFactory;
 use query::region_query::RegionQueryHandlerFactoryRef;
 use snafu::{OptionExt, ResultExt};
 
-use crate::error::{self, ExternalSnafu, Result};
+use crate::error::{self, DataFusionSnafu, ExternalSnafu, Result};
 use crate::events::EventHandlerImpl;
 use crate::frontend::FrontendOptions;
 use crate::heartbeat::frontend_peer_addr;
@@ -242,7 +242,7 @@ impl FrontendBuilder {
 
         let mut query_options = self.options.query.clone();
         query_options.enable_per_region_metrics = self.options.logging.enable_per_region_metrics;
-        let query_engine = QueryEngineFactory::new_with_plugins(
+        let query_engine = QueryEngineFactory::try_new_with_plugins(
             self.catalog_manager.clone(),
             Some(partition_manager.clone()),
             Some(region_query_handler.clone()),
@@ -253,6 +253,7 @@ impl FrontendBuilder {
             plugins.clone(),
             query_options,
         )
+        .context(DataFusionSnafu)?
         .query_engine();
 
         let frontend_peer_addr = frontend_peer_addr(&self.options);