diff --git a/crates/deltalake-core/src/operations/mod.rs b/crates/deltalake-core/src/operations/mod.rs index abf9753648..4b50b32d4b 100644 --- a/crates/deltalake-core/src/operations/mod.rs +++ b/crates/deltalake-core/src/operations/mod.rs @@ -7,6 +7,7 @@ //! with a [data stream][datafusion::physical_plan::SendableRecordBatchStream], //! if the operation returns data as well. +use std::collections::{HashMap}; use self::create::CreateBuilder; use self::filesystem_check::FileSystemCheckBuilder; use self::vacuum::VacuumBuilder; @@ -77,6 +78,18 @@ impl DeltaOps { Err(err) => Err(err), } } + + + /// try from uri with storage options + pub async fn try_from_uri_with_storage_options(uri: impl AsRef, storage_options: HashMap) -> DeltaResult { + let mut table = DeltaTableBuilder::from_uri(uri).with_storage_options(storage_options).build()?; + // We allow for uninitialized locations, since we may want to create the table + match table.load().await { + Ok(_) => Ok(table.into()), + Err(DeltaTableError::NotATable(_)) => Ok(table.into()), + Err(err) => Err(err), + } + } /// Create a new [`DeltaOps`] instance, backed by an un-initialized in memory table /// diff --git a/python/src/lib.rs b/python/src/lib.rs index 9dfe90c9dc..f3619da144 100644 --- a/python/src/lib.rs +++ b/python/src/lib.rs @@ -1157,13 +1157,10 @@ fn write_to_deltalake( // let schema: StructType = (&schema.0).try_into().map_err(PythonError::from)?; - let options = storage_options.clone().unwrap_or_default(); - let table = DeltaTableBuilder::from_uri(table_uri).with_storage_options(options); - - let table = table.build().map_err(PythonError::from)?; + let table = rt()?.block_on(DeltaOps::try_from_uri_with_storage_options(&table_uri, options)).map_err(PythonError::from)?; - let builder = DeltaOps(table) + let builder = table .write(batches) .with_save_mode(mode) .with_write_batch_size(max_rows_per_group as usize)