fix the error handling

datafusion/core/src/catalog/dynamic_file_schema.rs

@@ -75,24 +75,31 @@ impl SchemaProvider for DynamicFileSchemaProvider {
     }
 
     async fn table(&self, name: &str) -> Result<Option<Arc<dyn TableProvider>>> {
-        let inner_table = self.inner.table(name).await?;
-        if inner_table.is_some() {
-            return Ok(inner_table);
+        if let Ok(Some(inner_table)) = self.inner.table(name).await {
+            return Ok(Some(inner_table));
         }
+
         let optimized_url = substitute_tilde(name.to_owned());
-        let table_url = ListingTableUrl::parse(optimized_url.as_str())?;
+        let Ok(table_url) = ListingTableUrl::parse(optimized_url.as_str()) else {
+            return Ok(None);
+        };
+
         let state = &self
             .state_store
             .get_state()
             .upgrade()
             .ok_or_else(|| plan_datafusion_err!("locking error"))?
             .read()
             .clone();
-        let cfg = ListingTableConfig::new(table_url.clone())
+        if let Ok(cfg) = ListingTableConfig::new(table_url.clone())
             .infer(state)
-            .await?;
-
-        Ok(Some(Arc::new(ListingTable::try_new(cfg)?)))
+            .await
+        {
+            ListingTable::try_new(cfg)
+                .map(|table| Some(Arc::new(table) as Arc<dyn TableProvider>))
+        } else {
+            Ok(None)
+        }
     }
 
     fn deregister_table(&self, name: &str) -> Result<Option<Arc<dyn TableProvider>>> {

datafusion/core/src/execution/context/mod.rs

@@ -307,18 +307,23 @@ impl SessionContext {
     /// Creates a new `SessionContext` using the provided [`SessionState`]
     pub fn new_with_state(state: SessionState) -> Self {
         let state_ref = Arc::new(RwLock::new(state.clone()));
-        state
+
+        if let Ok(provider) = state
             // provide a fake table reference to get the default schema provider.
             .schema_for_ref(TableReference::full(
                 state.config_options().catalog.default_catalog.as_str(),
                 state.config_options().catalog.default_schema.as_str(),
                 UNNAMED_TABLE,
             ))
-            .unwrap()
-            .as_any()
-            .downcast_ref::<DynamicFileSchemaProvider>()
-            .unwrap()
-            .with_state(Arc::downgrade(&state_ref));
+        {
+            if let Some(provider) = provider
+                .as_any()
+                .downcast_ref::<DynamicFileSchemaProvider>()
+            {
+                provider.with_state(Arc::downgrade(&state_ref));
+            }
+        }
+
         Self {
             session_id: state_ref.clone().read().session_id().to_string(),
             session_start_time: Utc::now(),