[HSTACK] Register log store on next::DeltaScan::scan_with_args

Author: ccciudatu

crates/core/src/delta_datafusion/table_provider/next/mod.rs

@@ -40,12 +40,14 @@ use datafusion::{
     physical_plan::ExecutionPlan,
 };
 use datafusion::catalog::{ScanArgs, ScanResult};
+use datafusion::error::DataFusionError;
 use delta_kernel::table_configuration::TableConfiguration;
 use serde::{Deserialize, Serialize};
-
+use url::Url;
+use crate::delta_datafusion::engine::AsObjectStoreUrl;
 pub use self::scan::{DeltaScanExec, DeltaNextPhysicalCodec};
 pub(crate) use self::scan::KernelScanPlan;
-use crate::delta_datafusion::DeltaScanConfig;
+use crate::delta_datafusion::{DeltaScanConfig, DeltaSessionExt};
 use crate::delta_datafusion::engine::DataFusionEngine;
 use crate::delta_datafusion::table_provider::TableProviderBuilder;
 use crate::kernel::{EagerSnapshot, Snapshot};
@@ -54,6 +56,7 @@ mod scan;
 
 /// Default column name for the file id column we add to files read from disk.
 pub(crate) use crate::delta_datafusion::file_id::FILE_ID_COLUMN_DEFAULT;
+use crate::logstore::{object_store_factories, StorageConfig};
 
 #[derive(Clone, Debug, Serialize, Deserialize)]
 pub enum SnapshotWrapper {
@@ -219,6 +222,16 @@ impl TableProvider for DeltaScan {
 
     async fn scan_with_args<'a>(&self, state: &dyn Session, args: ScanArgs<'a>) -> Result<ScanResult> {
         let engine = DataFusionEngine::new_from_session(state);
+        let table_uri = self.snapshot.table_configuration().table_root();
+        if state.runtime_env().object_store(table_uri.as_object_store_url()).is_err() {
+            let url_key = Url::parse(&format!("{}://", table_uri.scheme()))
+                .map_err(|e| DataFusionError::External(Box::new(e)))?;
+            if let Some(entry) = object_store_factories().get(&url_key) {
+                let storage_config = StorageConfig::parse_options(&self.snapshot.snapshot().load_config().options)?;
+                let (store, _) = entry.value().parse_url_opts(table_uri, &storage_config)?;
+                state.runtime_env().register_object_store(table_uri, store);
+            }
+        }
 
         // Filter out file_id column from projection if present
         let file_id_idx = self