chore: converting more "uri" to actual Url 🔥

In this commit I am intentionally introducing the convention in our Rust
APIs:

  * anything named `uri` is expected to take a `AsRef<str>` type which
    is expected to turn into a `Url`
  * anything named `url` is expected to take a `Url` type.

As such I have renamed a number of our APIs which previously had been
converted to use `Url` but were referring to them as `uri`

Signed-off-by: R. Tyler Croy <[email protected]>

Author: rtyler

crates/aws/src/logstore/dynamodb_logstore.rs

@@ -183,10 +183,6 @@ impl LogStore for S3DynamoDbLogStore {
         "S3DynamoDbLogStore".into()
     }
 
-    fn root_uri(&self) -> String {
-        self.table_path.clone()
-    }
-
     async fn refresh(&self) -> DeltaResult<()> {
         let entry = self
             .lock_client

crates/benchmarks/src/merge.rs

@@ -443,7 +443,7 @@ pub async fn prepare_source_and_table(
 
     let batches = parquet_df.collect().await?;
     let fields: Vec<StructField> = delta_schema.fields().cloned().collect();
-    let table = DeltaOps::try_from_uri(temp_table_url)
+    let table = DeltaOps::try_from_url(temp_table_url)
         .await?
         .create()
         .with_columns(fields)

crates/benchmarks/src/smoke.rs

@@ -35,7 +35,7 @@ pub async fn run_smoke_once(table_url: &Url, params: &SmokeParams) -> DeltaResul
         ],
     )?;
 
-    let table = DeltaOps::try_from_uri(table_url.clone())
+    let table = DeltaOps::try_from_url(table_url.clone())
         .await?
         .write(vec![batch])
         .with_save_mode(SaveMode::Overwrite)

crates/benchmarks/src/tpcds_queries.rs

@@ -71,7 +71,7 @@ pub async fn register_tpcds_tables(
 
         let batches = parquet_df.collect().await?;
         let fields: Vec<StructField> = delta_schema.fields().cloned().collect();
-        let table = DeltaOps::try_from_uri(temp_table_url)
+        let table = DeltaOps::try_from_url(temp_table_url)
             .await?
             .create()
             .with_columns(fields)

crates/catalog-unity/src/datafusion.rs

@@ -237,7 +237,7 @@ impl SchemaProvider for UnitySchemaProvider {
                 })?;
                 let table_url = ensure_table_uri(&table.storage_location)
                     .map_err(|e| DataFusionError::External(Box::new(e)))?;
-                let table = DeltaTableBuilder::from_uri(table_url)
+                let table = DeltaTableBuilder::from_url(table_url)
                     .map_err(|e| DataFusionError::External(Box::new(e)))?
                     .with_storage_options(new_storage_opts)
                     .load()

crates/catalog-unity/src/lib.rs

@@ -860,7 +860,7 @@ impl ObjectStoreFactory for UnityCatalogFactory {
         // TODO(roeap): we should not have to go through the table here.
         // ideally we just create the right storage ...
         let table_url = ensure_table_uri(&table_path)?;
-        let mut builder = DeltaTableBuilder::from_uri(table_url)?;
+        let mut builder = DeltaTableBuilder::from_url(table_url)?;
 
         if let Some(runtime) = &config.runtime {
             builder = builder.with_io_runtime(runtime.clone());

crates/core/PERF.adoc

@@ -0,0 +1,47 @@
+
+= Build performance
+
+== Nightly
+
+=== threads=8
+
+
+[source]
+----
+RUSTFLAGS="-Z threads=8" cargo +nightly build  54.90s user 12.40s system 90% cpu 1:14.54 total
+----
+
+=== basic
+
+[source]
+----
+cargo +nightly build  45.63s user 9.93s system 70% cpu 1:18.59 total
+----
+
+
+=== threads=8 with datafusion
+
+[source]
+----
+RUSTFLAGS="-Z threads=8" cargo +nightly build --features datafusion  109.25s user 17.27s system 72% cpu 2:53.74 total
+----
+
+
+=== no threads
+
+[source]
+----
+cargo +nightly build --features datafusion  80.02s user 13.46s system 41% cpu 3:47.94 total
+----
+
+
+[source]
+----
+RUSTFLAGS="-Zcodegen-backend=cranelift" cargo +nightly build --features   58.19s user 12.43s system 38% cpu 3:03.59 total
+----
+
+[source]
+----
+RUSTFLAGS="-Zthreads=8 -Zcodegen-backend=cranelift" cargo +nightly build    78.85s user 14.81s system 51% cpu 3:02.71 total
+
+----

crates/core/src/delta_datafusion/mod.rs

@@ -692,13 +692,12 @@ impl PhysicalExtensionCodec for DeltaPhysicalCodec {
     ) -> Result<Arc<dyn ExecutionPlan>, DataFusionError> {
         let wire: DeltaScanWire = serde_json::from_reader(buf)
             .map_err(|_| DataFusionError::Internal("Unable to decode DeltaScan".to_string()))?;
-        let delta_scan = DeltaScan {
-            table_uri: wire.table_uri,
-            parquet_scan: (*inputs)[0].clone(),
-            config: wire.config,
-            logical_schema: wire.logical_schema,
-            metrics: ExecutionPlanMetricsSet::new(),
-        };
+        let delta_scan = DeltaScan::new(
+            &wire.table_url,
+            wire.config,
+            (*inputs)[0].clone(),
+            wire.logical_schema,
+        );
         Ok(Arc::new(delta_scan))
     }
 
@@ -712,11 +711,7 @@ impl PhysicalExtensionCodec for DeltaPhysicalCodec {
             .downcast_ref::<DeltaScan>()
             .ok_or_else(|| DataFusionError::Internal("Not a delta scan!".to_string()))?;
 
-        let wire = DeltaScanWire {
-            table_uri: delta_scan.table_uri.to_owned(),
-            config: delta_scan.config.clone(),
-            logical_schema: delta_scan.logical_schema.clone(),
-        };
+        let wire = DeltaScanWire::from(delta_scan);
         serde_json::to_writer(buf, &wire)
             .map_err(|_| DataFusionError::Internal("Unable to encode delta scan!".to_string()))?;
         Ok(())
@@ -867,7 +862,7 @@ mod tests {
     };
     use serde_json::json;
     use std::fmt::{self, Debug, Display, Formatter};
-    use std::ops::{Deref, Range};
+    use std::ops::Range;
     use tokio::sync::mpsc::{UnboundedReceiver, UnboundedSender, unbounded_channel};
 
     use super::*;
@@ -1140,13 +1135,12 @@ mod tests {
             Field::new("a", ArrowDataType::Utf8, false),
             Field::new("b", ArrowDataType::Int32, false),
         ]));
-        let exec_plan = Arc::from(DeltaScan {
-            table_uri: "s3://my_bucket/this/is/some/path".to_string(),
-            parquet_scan: Arc::from(EmptyExec::new(schema.clone())),
-            config: DeltaScanConfig::default(),
-            logical_schema: schema.clone(),
-            metrics: ExecutionPlanMetricsSet::new(),
-        });
+        let exec_plan = Arc::from(DeltaScan::new(
+            &Url::parse("s3://my_bucket/this/is/some/path").unwrap(),
+            DeltaScanConfig::default(),
+            Arc::from(EmptyExec::new(schema.clone())),
+            schema.clone(),
+        ));
         let proto: protobuf::PhysicalPlanNode =
             protobuf::PhysicalPlanNode::try_from_physical_plan(exec_plan.clone(), &codec)
                 .expect("to proto");

crates/core/src/delta_datafusion/table_provider.rs

@@ -48,6 +48,7 @@ use futures::StreamExt as _;
 use itertools::Itertools;
 use object_store::ObjectMeta;
 use serde::{Deserialize, Serialize};
+use url::Url;
 
 use crate::delta_datafusion::schema_adapter::DeltaSchemaAdapterFactory;
 use crate::delta_datafusion::{
@@ -60,8 +61,8 @@ use crate::kernel::{Action, Add, EagerSnapshot, Remove};
 use crate::operations::write::WriterStatsConfig;
 use crate::operations::write::writer::{DeltaWriter, WriterConfig};
 use crate::protocol::{DeltaOperation, SaveMode};
-use crate::{DeltaResult, DeltaTableError, logstore::LogStoreRef};
-use crate::{DeltaTable, ensure_table_uri};
+use crate::table::normalize_table_url;
+use crate::{DeltaResult, DeltaTable, DeltaTableError, logstore::LogStoreRef};
 
 const PATH_COLUMN: &str = "__delta_rs_path";
 
@@ -690,7 +691,7 @@ impl<'a> DeltaScanBuilder<'a> {
             .add(files_pruned);
 
         Ok(DeltaScan {
-            table_uri: ensure_table_uri(self.log_store.root_uri())?.as_str().into(),
+            table_url: self.log_store.root_url().clone(),
             parquet_scan: DataSourceExec::from_data_source(file_scan_config),
             config,
             logical_schema,
@@ -883,23 +884,52 @@ impl TableProvider for DeltaTableProvider {
 /// A wrapper for parquet scans
 #[derive(Debug)]
 pub struct DeltaScan {
-    /// The URL of the ObjectStore root
-    pub table_uri: String,
+    /// The normalized [Url] of the ObjectStore root
+    table_url: Url,
     /// Column that contains an index that maps to the original metadata Add
-    pub config: DeltaScanConfig,
+    pub(crate) config: DeltaScanConfig,
     /// The parquet scan to wrap
-    pub parquet_scan: Arc<dyn ExecutionPlan>,
+    pub(crate) parquet_scan: Arc<dyn ExecutionPlan>,
     /// The schema of the table to be used when evaluating expressions
-    pub logical_schema: Arc<Schema>,
+    pub(crate) logical_schema: Arc<Schema>,
     /// Metrics for scan reported via DataFusion
-    pub(super) metrics: ExecutionPlanMetricsSet,
+    metrics: ExecutionPlanMetricsSet,
+}
+
+impl DeltaScan {
+    pub(crate) fn new(
+        table_url: &Url,
+        config: DeltaScanConfig,
+        parquet_scan: Arc<dyn ExecutionPlan>,
+        logical_schema: Arc<Schema>,
+    ) -> Self {
+        Self {
+            table_url: normalize_table_url(table_url),
+            metrics: ExecutionPlanMetricsSet::new(),
+            config,
+            parquet_scan,
+            logical_schema,
+        }
+    }
 }
 
+#[non_exhaustive]
 #[derive(Debug, Serialize, Deserialize)]
 pub(super) struct DeltaScanWire {
-    pub table_uri: String,
-    pub config: DeltaScanConfig,
-    pub logical_schema: Arc<Schema>,
+    /// This [Url] should have already been passed through [normalize_table_url]
+    pub(crate) table_url: Url,
+    pub(crate) config: DeltaScanConfig,
+    pub(crate) logical_schema: Arc<Schema>,
+}
+
+impl From<&DeltaScan> for DeltaScanWire {
+    fn from(scan: &DeltaScan) -> Self {
+        Self {
+            table_url: scan.table_url.clone(),
+            config: scan.config.clone(),
+            logical_schema: scan.logical_schema.clone(),
+        }
+    }
 }
 
 impl DisplayAs for DeltaScan {
@@ -940,7 +970,7 @@ impl ExecutionPlan for DeltaScan {
             )));
         }
         Ok(Arc::new(DeltaScan {
-            table_uri: self.table_uri.clone(),
+            table_url: self.table_url.clone(),
             config: self.config.clone(),
             parquet_scan: children[0].clone(),
             logical_schema: self.logical_schema.clone(),
@@ -955,7 +985,7 @@ impl ExecutionPlan for DeltaScan {
     ) -> Result<Option<Arc<dyn ExecutionPlan>>> {
         if let Some(parquet_scan) = self.parquet_scan.repartitioned(target_partitions, config)? {
             Ok(Some(Arc::new(DeltaScan {
-                table_uri: self.table_uri.clone(),
+                table_url: self.table_url.clone(),
                 config: self.config.clone(),
                 parquet_scan,
                 logical_schema: self.logical_schema.clone(),

crates/core/src/kernel/transaction/application.rs

@@ -19,7 +19,7 @@ mod tests {
         let tmp_path = std::fs::canonicalize(tmp_dir.path()).unwrap();
 
         let batch = get_record_batch(None, false);
-        let table = DeltaOps::try_from_uri(ensure_table_uri(tmp_path.to_str().unwrap()).unwrap())
+        let table = DeltaOps::try_from_url(ensure_table_uri(tmp_path.to_str().unwrap()).unwrap())
             .await
             .unwrap()
             .write(vec![batch.clone()])
@@ -55,7 +55,7 @@ mod tests {
         // Test Txn Id can be read from existing table
 
         let mut table2 =
-            DeltaTableBuilder::from_uri(ensure_table_uri(tmp_path.to_str().unwrap()).unwrap())
+            DeltaTableBuilder::from_url(ensure_table_uri(tmp_path.to_str().unwrap()).unwrap())
                 .unwrap()
                 .load()
                 .await
@@ -123,7 +123,7 @@ mod tests {
         // Create a checkpoint and then load
         checkpoints::create_checkpoint(&table, None).await.unwrap();
         let table3 =
-            DeltaTableBuilder::from_uri(ensure_table_uri(tmp_path.to_str().unwrap()).unwrap())
+            DeltaTableBuilder::from_url(ensure_table_uri(tmp_path.to_str().unwrap()).unwrap())
                 .unwrap()
                 .load()
                 .await