Fix Result handling after get_actions async removal

- Added ? operator to handle Result<Vec<Action>, DeltaTableError>
- Removed .await calls since get_actions is now synchronous
- Fixed pattern matching from &&Action to Action in filter predicates

Signed-off-by: Yousef Saad <[email protected]>

Author: yousefsaad12

crates/core/benches/json_parsing.rs

@@ -59,7 +59,7 @@ fn bench_simple_actions(c: &mut Criterion) {
     group.bench_function("get_actions", |b| {
         b.iter(|| {
             rt.block_on(async {
-                let result = get_actions(0, &commit_log).await;
+                let result = get_actions(0, &commit_log);
                 black_box(result.unwrap().len())
             })
         });
@@ -80,7 +80,7 @@ fn bench_with_stats(c: &mut Criterion) {
     group.bench_function("get_actions", |b| {
         b.iter(|| {
             rt.block_on(async {
-                let result = get_actions(0, &commit_log).await;
+                let result = get_actions(0, &commit_log);
                 black_box(result.unwrap().len())
             })
         });
@@ -101,7 +101,7 @@ fn bench_full_complexity(c: &mut Criterion) {
     group.bench_function("get_actions", |b| {
         b.iter(|| {
             rt.block_on(async {
-                let result = get_actions(0, &commit_log).await;
+                let result = get_actions(0, &commit_log);
                 black_box(result.unwrap().len())
             })
         });

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

@@ -244,7 +244,7 @@ impl WinningCommitSummary {
         let commit_log_bytes = log_store.read_commit_entry(winning_commit_version).await?;
         match commit_log_bytes {
             Some(bytes) => {
-                let actions = get_actions(winning_commit_version, &bytes).await?;
+                let actions = get_actions(winning_commit_version, &bytes)?; // ← ADD ? HERE
                 let commit_info = actions
                     .iter()
                     .find(|action| matches!(action, Action::CommitInfo(_)))

crates/core/src/logstore/mod.rs

@@ -309,7 +309,7 @@ pub trait LogStore: Send + Sync + AsAny {
             Err(err) => Err(err),
         }?;
 
-        let actions = crate::logstore::get_actions(next_version, &commit_log_bytes).await;
+        let actions = crate::logstore::get_actions(next_version, &commit_log_bytes);
         Ok(PeekCommit::New(next_version, actions?))
     }
 
@@ -564,7 +564,7 @@ pub fn to_uri(root: &Url, location: &Path) -> String {
 }
 
 /// Reads a commit and gets list of actions
-pub async fn get_actions(
+pub fn get_actions(
     version: i64,
     commit_log_bytes: &bytes::Bytes,
 ) -> Result<Vec<Action>, DeltaTableError> {
@@ -1006,7 +1006,7 @@ mod datafusion_tests {
 {"invalid json without closing brace"#,
         );
 
-        let result = get_actions(0, &malformed_json).await;
+        let result = get_actions(0, &malformed_json);
 
         match result {
             Err(DeltaTableError::InvalidJsonLog {

crates/core/src/operations/load_cdf.rs

@@ -128,7 +128,7 @@ impl CdfLoadBuilder {
         let ts = self.starting_timestamp.unwrap_or(DateTime::UNIX_EPOCH);
         for v in 0..self.snapshot.version() {
             if let Ok(Some(bytes)) = self.log_store.read_commit_entry(v).await {
-                if let Ok(actions) = get_actions(v, &bytes).await {
+                if let Ok(actions) = get_actions(v, &bytes) {
                     if actions.iter().any(|action| {
                         matches!(action, Action::CommitInfo(CommitInfo {
                             timestamp: Some(t), ..
@@ -209,7 +209,7 @@ impl CdfLoadBuilder {
             .ok_or(DeltaTableError::InvalidVersion(latest_version))?;
 
         let latest_version_actions: Vec<Action> =
-            get_actions(latest_version, &latest_snapshot_bytes).await?;
+            get_actions(latest_version, &latest_snapshot_bytes)?;
         let latest_version_commit = latest_version_actions
             .iter()
             .find(|a| matches!(a, Action::CommitInfo(_)));
@@ -240,7 +240,7 @@ impl CdfLoadBuilder {
                 .await?
                 .ok_or(DeltaTableError::InvalidVersion(version));
 
-            let version_actions: Vec<Action> = get_actions(version, &snapshot_bytes?).await?;
+            let version_actions: Vec<Action> = get_actions(version, &snapshot_bytes?)?;
 
             let mut ts = 0;
             let mut cdc_actions = vec![];
@@ -953,7 +953,7 @@ pub(crate) mod tests {
             .read_commit_entry(2)
             .await?
             .expect("failed to get snapshot bytes");
-        let version_actions = get_actions(2, &snapshot_bytes).await?;
+        let version_actions = get_actions(2, &snapshot_bytes)?;
 
         let cdc_actions = version_actions
             .iter()

crates/core/src/operations/merge/mod.rs

@@ -2008,9 +2008,7 @@ mod tests {
             .await
             .unwrap()
             .expect("failed to get snapshot bytes");
-        let actions = crate::logstore::get_actions(2, &snapshot_bytes)
-            .await
-            .unwrap();
+        let actions = crate::logstore::get_actions(2, &snapshot_bytes).unwrap();
 
         let schema_actions = actions
             .iter()
@@ -2084,9 +2082,7 @@ mod tests {
             .await
             .unwrap()
             .expect("failed to get snapshot bytes");
-        let actions = crate::logstore::get_actions(2, &snapshot_bytes)
-            .await
-            .unwrap();
+        let actions = crate::logstore::get_actions(2, &snapshot_bytes).unwrap();
 
         let schema_actions = actions
             .iter()
@@ -2195,9 +2191,7 @@ mod tests {
             .await
             .unwrap()
             .expect("failed to get snapshot bytes");
-        let actions = crate::logstore::get_actions(2, &snapshot_bytes)
-            .await
-            .unwrap();
+        let actions = crate::logstore::get_actions(2, &snapshot_bytes).unwrap();
 
         let schema_actions = actions
             .iter()

crates/core/src/operations/write/mod.rs

@@ -1681,7 +1681,7 @@ mod tests {
             .read_commit_entry(2)
             .await?
             .expect("failed to get snapshot bytes");
-        let version_actions = get_actions(2, &snapshot_bytes).await?;
+        let version_actions = get_actions(2, &snapshot_bytes)?;
 
         let cdc_actions = version_actions
             .iter()
@@ -1755,7 +1755,7 @@ mod tests {
             .read_commit_entry(2)
             .await?
             .expect("failed to get snapshot bytes");
-        let version_actions = get_actions(2, &snapshot_bytes).await?;
+        let version_actions = get_actions(2, &snapshot_bytes)?;
 
         let cdc_actions = version_actions
             .iter()
@@ -1856,7 +1856,7 @@ mod tests {
             .read_commit_entry(2)
             .await?
             .expect("failed to get snapshot bytes");
-        let version_actions = get_actions(2, &snapshot_bytes).await?;
+        let version_actions = get_actions(2, &snapshot_bytes)?;
 
         let cdc_actions = version_actions
             .iter()