lint fix

Author: anoop-narang

src/table_deletions.rs

@@ -240,7 +240,7 @@ impl TableProvider for TableDeletionsTable {
                         .map(|&i| self.output_schema.field(i).clone())
                         .collect();
                     Arc::new(Schema::new(fields))
-                }
+                },
                 None => self.output_schema.clone(),
             };
             return Ok(Arc::new(EmptyExec::new(output_schema)));
@@ -320,15 +320,17 @@ impl DeletedRowsExec {
 impl DisplayAs for DeletedRowsExec {
     fn fmt_as(&self, t: DisplayFormatType, f: &mut std::fmt::Formatter) -> std::fmt::Result {
         match t {
-            DisplayFormatType::Default | DisplayFormatType::Verbose | DisplayFormatType::TreeRender => {
+            DisplayFormatType::Default
+            | DisplayFormatType::Verbose
+            | DisplayFormatType::TreeRender => {
                 write!(
                     f,
                     "DeletedRowsExec: snapshot_id={}, full_delete={}, has_previous={}",
                     self.snapshot_id,
                     self.current_delete_scan.is_none(),
                     self.previous_delete_scan.is_some()
                 )
-            }
+            },
         }
     }
 }
@@ -365,28 +367,31 @@ impl ExecutionPlan for DeletedRowsExec {
         let mut idx = 0;
 
         let current = if self.current_delete_scan.is_some() {
-            let c = children.get(idx).cloned().ok_or_else(|| {
-                DataFusionError::Internal("Missing current delete child".into())
-            })?;
+            let c = children
+                .get(idx)
+                .cloned()
+                .ok_or_else(|| DataFusionError::Internal("Missing current delete child".into()))?;
             idx += 1;
             Some(c)
         } else {
             None
         };
 
         let previous = if self.previous_delete_scan.is_some() {
-            let p = children.get(idx).cloned().ok_or_else(|| {
-                DataFusionError::Internal("Missing previous delete child".into())
-            })?;
+            let p = children
+                .get(idx)
+                .cloned()
+                .ok_or_else(|| DataFusionError::Internal("Missing previous delete child".into()))?;
             idx += 1;
             Some(p)
         } else {
             None
         };
 
-        let data = children.get(idx).cloned().ok_or_else(|| {
-            DataFusionError::Internal("Missing data file child".into())
-        })?;
+        let data = children
+            .get(idx)
+            .cloned()
+            .ok_or_else(|| DataFusionError::Internal("Missing data file child".into()))?;
 
         Ok(Arc::new(DeletedRowsExec::new(
             current,
@@ -569,7 +574,10 @@ impl DeletedRowsStream {
 
         // Append CDC columns
         let num_output_rows = keep_indices.len();
-        columns.push(Arc::new(Int64Array::from(vec![self.snapshot_id; num_output_rows])));
+        columns.push(Arc::new(Int64Array::from(vec![
+            self.snapshot_id;
+            num_output_rows
+        ])));
         columns.push(Arc::new(StringArray::from(vec!["delete"; num_output_rows])));
 
         RecordBatch::try_new(self.output_schema.clone(), columns)
@@ -590,7 +598,7 @@ impl Stream for DeletedRowsStream {
                         Poll::Ready(Some(Ok(batch))) => {
                             let positions = Self::extract_positions(&batch);
                             self.current_positions.extend(positions);
-                        }
+                        },
                         Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(e))),
                         Poll::Ready(None) => {
                             if self.previous_delete_stream.is_some() {
@@ -599,44 +607,44 @@ impl Stream for DeletedRowsStream {
                                 self.compute_deleted_positions();
                                 self.state = StreamState::ReadingData;
                             }
-                        }
+                        },
                         Poll::Pending => return Poll::Pending,
                     }
-                }
+                },
                 StreamState::ReadingPreviousDelete => {
                     let prev = self.previous_delete_stream.as_mut().unwrap();
                     match Pin::new(prev).poll_next(cx) {
                         Poll::Ready(Some(Ok(batch))) => {
                             let positions = Self::extract_positions(&batch);
                             self.previous_positions.extend(positions);
-                        }
+                        },
                         Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(e))),
                         Poll::Ready(None) => {
                             self.compute_deleted_positions();
                             self.state = StreamState::ReadingData;
-                        }
+                        },
                         Poll::Pending => return Poll::Pending,
                     }
-                }
+                },
                 StreamState::ReadingData => {
                     match Pin::new(&mut self.data_stream).poll_next(cx) {
                         Poll::Ready(Some(Ok(batch))) => {
                             match self.filter_batch(&batch)? {
                                 Some(filtered) => return Poll::Ready(Some(Ok(filtered))),
                                 None => continue, // No deleted rows in this batch
                             }
-                        }
+                        },
                         Poll::Ready(Some(Err(e))) => return Poll::Ready(Some(Err(e))),
                         Poll::Ready(None) => {
                             self.state = StreamState::Done;
                             return Poll::Ready(None);
-                        }
+                        },
                         Poll::Pending => return Poll::Pending,
                     }
-                }
+                },
                 StreamState::Done => {
                     return Poll::Ready(None);
-                }
+                },
             }
         }
     }

src/table_functions.rs

@@ -229,7 +229,9 @@ pub struct DucklakeTableDeletionsFunction {
 
 impl DucklakeTableDeletionsFunction {
     pub fn new(provider: Arc<dyn MetadataProvider>) -> Self {
-        Self { provider }
+        Self {
+            provider,
+        }
     }
 
     fn parse_table_name(table_name: &str) -> (&str, &str) {
@@ -260,7 +262,7 @@ impl TableFunctionImpl for DucklakeTableDeletionsFunction {
                     "First argument to ducklake_table_deletions() must be a string literal \
                      (e.g., 'main.users' or 'users')"
                 );
-            }
+            },
         };
 
         // Parse start_snapshot argument
@@ -271,7 +273,7 @@ impl TableFunctionImpl for DucklakeTableDeletionsFunction {
                 return plan_err!(
                     "Second argument to ducklake_table_deletions() must be an integer (start_snapshot)"
                 );
-            }
+            },
         };
 
         // Parse end_snapshot argument
@@ -282,7 +284,7 @@ impl TableFunctionImpl for DucklakeTableDeletionsFunction {
                 return plan_err!(
                     "Third argument to ducklake_table_deletions() must be an integer (end_snapshot)"
                 );
-            }
+            },
         };
 
         // Validate snapshot range