refactor drop method

Author: xanderbailey

crates/catalog/glue/src/catalog.rs

@@ -677,12 +677,7 @@ impl Catalog for GlueCatalog {
     async fn purge_table(&self, table: &TableIdent) -> Result<()> {
         let table_info = self.load_table(table).await?;
         self.drop_table(table).await?;
-        iceberg::drop_table_data(
-            table_info.file_io(),
-            table_info.metadata(),
-            table_info.metadata_location(),
-        )
-        .await
+        iceberg::drop_table_data(table_info).await
     }
 
     /// Asynchronously checks the existence of a specified table

crates/catalog/hms/src/catalog.rs

@@ -624,12 +624,7 @@ impl Catalog for HmsCatalog {
     async fn purge_table(&self, table: &TableIdent) -> Result<()> {
         let table_info = self.load_table(table).await?;
         self.drop_table(table).await?;
-        iceberg::drop_table_data(
-            table_info.file_io(),
-            table_info.metadata(),
-            table_info.metadata_location(),
-        )
-        .await
+        iceberg::drop_table_data(table_info).await
     }
 
     /// Asynchronously checks the existence of a specified table

crates/catalog/sql/src/catalog.rs

@@ -774,12 +774,7 @@ impl Catalog for SqlCatalog {
     async fn purge_table(&self, table: &TableIdent) -> Result<()> {
         let table_info = self.load_table(table).await?;
         self.drop_table(table).await?;
-        iceberg::drop_table_data(
-            table_info.file_io(),
-            table_info.metadata(),
-            table_info.metadata_location(),
-        )
-        .await
+        iceberg::drop_table_data(table_info).await
     }
 
     async fn load_table(&self, identifier: &TableIdent) -> Result<Table> {

crates/iceberg/src/catalog/memory/catalog.rs

@@ -343,12 +343,7 @@ impl Catalog for MemoryCatalog {
     async fn purge_table(&self, table_ident: &TableIdent) -> Result<()> {
         let table_info = self.load_table(table_ident).await?;
         self.drop_table(table_ident).await?;
-        crate::catalog::utils::drop_table_data(
-            table_info.file_io(),
-            table_info.metadata(),
-            table_info.metadata_location(),
-        )
-        .await
+        crate::catalog::utils::drop_table_data(table_info).await
     }
 
     /// Check if a table exists in the catalog.

crates/iceberg/src/catalog/utils.rs

@@ -23,7 +23,7 @@ use futures::{TryStreamExt, stream};
 
 use crate::Result;
 use crate::io::FileIO;
-use crate::spec::{ManifestListReader, TableMetadata, TableMetadataRef};
+use crate::table::Table;
 
 const DELETE_CONCURRENCY: usize = 10;
 
@@ -36,22 +36,16 @@ const DELETE_CONCURRENCY: usize = 10;
 /// Data files within manifests are only deleted if the `gc.enabled` table
 /// property is `true` (the default), to avoid corrupting other tables that
 /// may share the same data files.
-pub async fn drop_table_data(
-    io: &FileIO,
-    metadata: &TableMetadata,
-    metadata_location: Option<&str>,
-) -> Result<()> {
+pub async fn drop_table_data(table_info: Table) -> Result<()> {
     let mut manifest_lists_to_delete: HashSet<String> = HashSet::new();
     let mut manifests_to_delete: HashSet<String> = HashSet::new();
 
-    let metadata_ref = TableMetadataRef::new(metadata.clone());
+    let metadata = table_info.metadata_ref();
+    let io = table_info.file_io();
     // Load all manifest lists concurrently
     let results: Vec<_> =
         futures::future::try_join_all(metadata.snapshots().map(|snapshot| async {
-            let manifest_list =
-                ManifestListReader::new(snapshot.clone(), io.clone(), metadata_ref.clone())
-                    .load()
-                    .await?;
+            let manifest_list = table_info.manifest_list_reader(snapshot).load().await?;
             Ok::<_, crate::Error>((snapshot.manifest_list().to_string(), manifest_list))
         }))
         .await?;
@@ -101,7 +95,7 @@ pub async fn drop_table_data(
         .await?;
 
     // Delete the current metadata file
-    if let Some(location) = metadata_location {
+    if let Some(location) = table_info.metadata_location() {
         io.delete(location).await?;
     }
 

crates/iceberg/src/spec/manifest_list.rs

@@ -92,7 +92,7 @@ impl ManifestList {
 
 /// A manifest list reader that encapsulates the logic for loading and parsing a [`ManifestList`]
 /// from a snapshot.
-pub struct ManifestListReader {
+pub(crate) struct ManifestListReader {
     snapshot: SnapshotRef,
     file_io: FileIO,
     table_metadata: TableMetadataRef,

crates/iceberg/src/table.rs

@@ -265,7 +265,7 @@ impl Table {
     }
 
     /// Creates a [`ManifestListReader`] for the given snapshot.
-    pub fn manifest_list_reader(&self, snapshot: &SnapshotRef) -> ManifestListReader {
+    pub(crate) fn manifest_list_reader(&self, snapshot: &SnapshotRef) -> ManifestListReader {
         ManifestListReader::new(
             snapshot.clone(),
             self.file_io.clone(),