chore: trailing slashes on the tablePath in S3DynamoDbLogStore are load-bearing

For better or worse, the trailing slashes on the `tablePath` DynamoDB
items are semantically important and a trailing slash coming from
deltalake-core causes lookup failures with multiple writers between
Spark and Rust.

This change ensures that the S3DynamoDbLogStore is always removing the
trailing slash should it exist.

Signed-off-by: R. Tyler Croy <rtyler@brokenco.de>

Author: rtyler

crates/aws/Cargo.toml

@@ -54,6 +54,7 @@ chrono = { workspace = true }
 deltalake-core = { path = "../core" }
 deltalake-test = { path = "../test" }
 pretty_env_logger = "0.5.0"
+pretty_assertions = "*"
 rand = "0.8"
 serde_json = { workspace = true }
 serial_test = "3"

crates/aws/src/lib.rs

@@ -151,6 +151,15 @@ pub struct DynamoDbLockClient {
     config: DynamoDbConfig,
 }
 
+#[cfg(test)]
+impl Default for DynamoDbLockClient {
+    fn default() -> Self {
+        let sdk_config = aws_config::SdkConfig::builder().build();
+        Self::try_new(&sdk_config, None, None, None, None, None, None, None, None)
+            .expect("Failed to create a default DynamoDbLockClient for testing purpose")
+    }
+}
+
 impl std::fmt::Debug for DynamoDbLockClient {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> Result<(), std::fmt::Error> {
         write!(f, "DynamoDbLockClient(config: {:?})", self.config)
@@ -323,19 +332,6 @@ impl DynamoDbLockClient {
         &self.config
     }
 
-    fn get_primary_key(&self, version: i64, table_path: &str) -> HashMap<String, AttributeValue> {
-        HashMap::from([
-            (
-                constants::ATTR_TABLE_PATH.to_owned(),
-                string_attr(table_path),
-            ),
-            (
-                constants::ATTR_FILE_NAME.to_owned(),
-                string_attr(format!("{version:020}.json")),
-            ),
-        ])
-    }
-
     /// Read a log entry from DynamoDb.
     pub async fn get_commit_entry(
         &self,
@@ -349,7 +345,7 @@ impl DynamoDbLockClient {
                         .get_item()
                         .consistent_read(true)
                         .table_name(&self.config.lock_table_name)
-                        .set_key(Some(self.get_primary_key(version, table_path)))
+                        .set_key(Some(get_primary_key(version, table_path)))
                         .send()
                         .await
                 },
@@ -442,7 +438,9 @@ impl DynamoDbLockClient {
                         .key_condition_expression(format!("{} = :tn", constants::ATTR_TABLE_PATH))
                         .set_expression_attribute_values(Some(HashMap::from([(
                             ":tn".into(),
-                            string_attr(table_path),
+                            // NOTE: the lack of trailing slashes is a load-bearing implementation
+                            // detail between the Delta/Spark and delta-rs S3DynamoDbLogStore
+                            string_attr(table_path.trim_end_matches('/')),
                         )])))
                         .send()
                         .await
@@ -488,7 +486,7 @@ impl DynamoDbLockClient {
                         .dynamodb_client
                         .update_item()
                         .table_name(self.get_lock_table_name())
-                        .set_key(Some(self.get_primary_key(version, table_path)))
+                        .set_key(Some(get_primary_key(version, table_path)))
                         .update_expression("SET complete = :c, expireTime = :e".to_owned())
                         .set_expression_attribute_values(Some(HashMap::from([
                             (":c".to_owned(), string_attr("true")),
@@ -535,7 +533,7 @@ impl DynamoDbLockClient {
                     .dynamodb_client
                     .delete_item()
                     .table_name(self.get_lock_table_name())
-                    .set_key(Some(self.get_primary_key(version, table_path)))
+                    .set_key(Some(get_primary_key(version, table_path)))
                     .set_expression_attribute_values(Some(HashMap::from([(
                         ":f".into(),
                         string_attr("false"),
@@ -631,21 +629,32 @@ fn epoch_to_system_time(s: u64) -> SystemTime {
     SystemTime::UNIX_EPOCH + Duration::from_secs(s)
 }
 
-fn create_value_map(
-    commit_entry: &CommitEntry,
-    table_path: &str,
-) -> HashMap<String, AttributeValue> {
-    // cut off `_delta_log` part: temp_path in DynamoDb is relative to `_delta_log` not table root.
-    let temp_path = Path::from_iter(commit_entry.temp_path.parts().skip(1));
-    let mut value_map = HashMap::from([
+/// Return the primary key as a [HashMap] for looking up log entries in the DynamoDb table
+///
+/// The `table_path` needs to be sent into DynamoDB without a trailing slash for the [Url] since
+/// that is a load-bearing part of the contract with Delta/Spark's implementation.
+fn get_primary_key(version: i64, table_path: &str) -> HashMap<String, AttributeValue> {
+    HashMap::from([
         (
             constants::ATTR_TABLE_PATH.to_owned(),
-            string_attr(table_path),
+            string_attr(table_path.trim_end_matches('/')),
         ),
         (
             constants::ATTR_FILE_NAME.to_owned(),
-            string_attr(format!("{:020}.json", commit_entry.version)),
+            string_attr(format!("{version:020}.json")),
         ),
+    ])
+}
+
+fn create_value_map(
+    commit_entry: &CommitEntry,
+    table_path: &str,
+) -> HashMap<String, AttributeValue> {
+    // cut off `_delta_log` part: temp_path in DynamoDb is relative to `_delta_log` not table root.
+    let temp_path = Path::from_iter(commit_entry.temp_path.parts().skip(1));
+    let mut value_map = get_primary_key(commit_entry.version, table_path);
+
+    value_map.extend(HashMap::from([
         (constants::ATTR_TEMP_PATH.to_owned(), string_attr(temp_path)),
         (
             constants::ATTR_COMPLETE.to_owned(),
@@ -655,7 +664,7 @@ fn create_value_map(
                 "false"
             }),
         ),
-    ]);
+    ]));
     commit_entry.expire_time.as_ref().map(|t| {
         value_map.insert(
             constants::ATTR_EXPIRE_TIME.to_owned(),
@@ -763,6 +772,8 @@ mod tests {
     use super::*;
     use aws_sdk_sts::config::ProvideCredentials;
 
+    use pretty_assertions::assert_eq;
+
     use object_store::memory::InMemory;
     use serial_test::serial;
 
@@ -773,6 +784,27 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_get_primary_key() -> DeltaResult<()> {
+        let version = 0;
+        let expected = HashMap::from([
+            (
+                constants::ATTR_TABLE_PATH.to_owned(),
+                // NOTE: the lack of a trailing slash is important for compatibility with the
+                // Delta/Spark S3DynamoDbLogStore
+                string_attr("s3://bucket/table"),
+            ),
+            (
+                constants::ATTR_FILE_NAME.to_owned(),
+                string_attr(format!("{version:020}.json")),
+            ),
+        ]);
+
+        assert_eq!(expected, get_primary_key(version, "s3://bucket/table"));
+        assert_eq!(expected, get_primary_key(version, "s3://bucket/table/"));
+        Ok(())
+    }
+
     #[test]
     fn commit_entry_roundtrip_test() -> Result<(), LockClientError> {
         let system_time = SystemTime::UNIX_EPOCH