fix: make sure percent-encoded file paths are properly decoded when c…

…oming from parquet checkpoints, to prevent tombstone and file paths mismatch (e.g. file path is read from checkpoint while tombstone path is read from JSON)

crates/deltalake-core/src/kernel/actions/mod.rs

@@ -8,7 +8,7 @@ use std::collections::HashMap;
 use serde::{Deserialize, Serialize};
 
 pub(crate) mod schemas;
-mod serde_path;
+pub(crate) mod serde_path;
 pub(crate) mod types;
 
 pub use types::*;

crates/deltalake-core/src/kernel/actions/serde_path.rs

@@ -54,7 +54,7 @@ fn encode_path(path: &str) -> String {
     percent_encode(path.as_bytes(), INVALID).to_string()
 }
 
-fn decode_path(path: &str) -> Result<String, Utf8Error> {
+pub fn decode_path(path: &str) -> Result<String, Utf8Error> {
     Ok(percent_decode_str(path).decode_utf8()?.to_string())
 }
 

crates/deltalake-core/src/protocol/checkpoints.rs

@@ -70,6 +70,13 @@ impl From<CheckpointError> for ProtocolError {
     }
 }
 
+use core::str::Utf8Error;
+impl From<Utf8Error> for ProtocolError {
+    fn from(value: Utf8Error) -> Self {
+        Self::Generic(value.to_string())
+    }
+}
+
 /// The record batch size for checkpoint parquet file
 pub const CHECKPOINT_RECORD_BATCH_SIZE: usize = 5000;
 

crates/deltalake-core/src/protocol/parquet_read/mod.rs

@@ -6,6 +6,7 @@ use num_traits::cast::ToPrimitive;
 use parquet::record::{Field, ListAccessor, MapAccessor, RowAccessor};
 use serde_json::json;
 
+use crate::kernel::serde_path::decode_path;
 use crate::kernel::{
     Action, Add, AddCDCFile, DeletionVectorDescriptor, Metadata, Protocol, Remove, StorageType, Txn,
 };
@@ -119,10 +120,13 @@ impl Add {
         for (i, (name, _)) in record.get_column_iter().enumerate() {
             match name.as_str() {
                 "path" => {
-                    re.path = record
-                        .get_string(i)
-                        .map_err(|_| gen_action_type_error("add", "path", "string"))?
-                        .clone();
+                    re.path = decode_path(
+                        record
+                            .get_string(i)
+                            .map_err(|_| gen_action_type_error("add", "path", "string"))?
+                            .clone()
+                            .as_str(),
+                    )?;
                 }
                 "size" => {
                     re.size = record
@@ -515,10 +519,13 @@ impl Remove {
         for (i, (name, _)) in record.get_column_iter().enumerate() {
             match name.as_str() {
                 "path" => {
-                    re.path = record
-                        .get_string(i)
-                        .map_err(|_| gen_action_type_error("remove", "path", "string"))?
-                        .clone();
+                    re.path = decode_path(
+                        record
+                            .get_string(i)
+                            .map_err(|_| gen_action_type_error("remove", "path", "string"))?
+                            .clone()
+                            .as_str(),
+                    )?;
                 }
                 "dataChange" => {
                     re.data_change = record