Merge branch 'main' into fix-update-totals

Author: dannycjones

.github/workflows/ci.yml

@@ -195,7 +195,7 @@ jobs:
 
       - name: Install cargo-nextest
         if: matrix.test-suite.name == 'default'
-        uses: taiki-e/install-action@d9be7d8cda89035c9c843f78bd44d4f72d8403d4 # v2.79.7
+        uses: taiki-e/install-action@e49978b799e49ff429d162b7a30601a569ab6538 # v2.81.1
         with:
           tool: cargo-nextest
 

.github/workflows/ci_typos.yml

@@ -47,4 +47,4 @@ jobs:
         with:
           persist-credentials: false
       - name: Check typos
-        uses: crate-ci/typos@7b04f660f4ee4f048d18fd341887cf28dfbedfe2 # v1.46.3
+        uses: crate-ci/typos@f8a58b6b53f2279f71eb605f03a4ae4d10608f45 # v1.47.0

.github/workflows/public-api.yml

@@ -52,7 +52,7 @@ jobs:
           repo-token: ${{ secrets.GITHUB_TOKEN }}
 
       - name: Install cargo-public-api
-        uses: taiki-e/install-action@d9be7d8cda89035c9c843f78bd44d4f72d8403d4 # v2.79.7
+        uses: taiki-e/install-action@e49978b799e49ff429d162b7a30601a569ab6538 # v2.81.1
         with:
           tool: cargo-public-api
 

Cargo.lock

@@ -40,7 +40,7 @@ use iceberg_storage_opendal::OpenDalStorageFactory;
 use crate::error::{from_aws_build_error, from_aws_sdk_error};
 use crate::utils::{
     convert_to_database, convert_to_glue_table, convert_to_namespace, create_sdk_config,
-    get_default_table_location, get_metadata_location, validate_namespace,
+    get_default_table_location, get_metadata_location, is_iceberg_table, validate_namespace,
 };
 use crate::{
     AWS_ACCESS_KEY_ID, AWS_REGION_NAME, AWS_SECRET_ACCESS_KEY, AWS_SESSION_TOKEN, with_catalog_id,
@@ -504,9 +504,21 @@ impl Catalog for GlueCatalog {
         }
 
         let db_name = validate_namespace(namespace)?;
-        let table_list = self.list_tables(namespace).await?;
 
-        if !table_list.is_empty() {
+        // Check for ANY Glue table in the database, not just Iceberg tables.
+        // Glue's `delete_database` will fail if any table (Iceberg or not) is
+        // still present, and `list_tables` only returns Iceberg tables, so we
+        // query Glue directly here.
+        let builder = self
+            .client
+            .0
+            .get_tables()
+            .database_name(&db_name)
+            .max_results(1);
+        let builder = with_catalog_id!(builder, self.config);
+        let resp = builder.send().await.map_err(from_aws_sdk_error)?;
+
+        if !resp.table_list().is_empty() {
             return Err(Error::new(
                 ErrorKind::DataInvalid,
                 format!("Database with name: {} is not empty", &db_name),
@@ -521,12 +533,16 @@ impl Catalog for GlueCatalog {
         Ok(())
     }
 
-    /// Asynchronously lists all tables within a specified namespace.
+    /// Asynchronously lists all Iceberg tables within a specified namespace.
+    ///
+    /// Glue databases may contain a mix of Iceberg and non-Iceberg tables
+    /// (e.g. plain Hive tables). Only tables whose `table_type` parameter is
+    /// set to `ICEBERG` (case-insensitive) are returned
     ///
     /// # Returns
     /// A `Result<Vec<TableIdent>>`, which is:
     /// - `Ok(vec![...])` containing a vector of `TableIdent` instances, each
-    /// representing a table within the specified namespace.
+    /// representing an Iceberg table within the specified namespace.
     /// - `Err(...)` if an error occurs during namespace validation or while
     /// querying the database.
     async fn list_tables(&self, namespace: &NamespaceIdent) -> Result<Vec<TableIdent>> {
@@ -551,6 +567,7 @@ impl Catalog for GlueCatalog {
             let tables: Vec<_> = resp
                 .table_list()
                 .iter()
+                .filter(|tbl| is_iceberg_table(&tbl.parameters))
                 .map(|tbl| TableIdent::new(namespace.clone(), tbl.name().to_string()))
                 .collect();
 

bindings/python/uv.lock

@@ -226,6 +226,17 @@ pub(crate) fn get_default_table_location(
     }
 }
 
+/// Returns `true` if the given Glue table is an Iceberg table.
+///
+/// Iceberg tables are identified by the `table_type=ICEBERG` parameter
+/// (case-insensitive).
+pub(crate) fn is_iceberg_table(parameters: &Option<HashMap<String, String>>) -> bool {
+    parameters
+        .as_ref()
+        .and_then(|p| p.get(TABLE_TYPE))
+        .is_some_and(|v| v.eq_ignore_ascii_case(ICEBERG))
+}
+
 /// Get metadata location from `GlueTable` parameters
 pub(crate) fn get_metadata_location(
     parameters: &Option<HashMap<String, String>>,
@@ -302,6 +313,36 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_is_iceberg_table() {
+        // table_type=ICEBERG -> iceberg
+        let params_table_type = Some(HashMap::from([(
+            TABLE_TYPE.to_string(),
+            ICEBERG.to_string(),
+        )]));
+        assert!(is_iceberg_table(&params_table_type));
+
+        // table_type is case-insensitive
+        let params_table_type_lower = Some(HashMap::from([(
+            TABLE_TYPE.to_string(),
+            "iceberg".to_string(),
+        )]));
+        assert!(is_iceberg_table(&params_table_type_lower));
+
+        // Plain Hive table -> not iceberg
+        let params_hive = Some(HashMap::from([(
+            TABLE_TYPE.to_string(),
+            "EXTERNAL_TABLE".to_string(),
+        )]));
+        assert!(!is_iceberg_table(&params_hive));
+
+        // No parameters at all -> not iceberg
+        assert!(!is_iceberg_table(&None));
+
+        // Empty parameters -> not iceberg
+        assert!(!is_iceberg_table(&Some(HashMap::new())));
+    }
+
     #[test]
     fn test_convert_to_glue_table() -> Result<()> {
         let table_name = "my_table".to_string();

crates/catalog/glue/src/catalog.rs

@@ -257,7 +257,11 @@ impl SqlCatalog {
                 "StorageFactory must be provided for SqlCatalog. Use `with_storage_factory` to configure it.",
             )
         })?;
-        let fileio = FileIOBuilder::new(factory).build();
+        // Forward catalog props so storage-backend keys reach the FileIO.
+        // Unrecognized keys are ignored by backends.
+        let fileio = FileIOBuilder::new(factory)
+            .with_props(config.props.clone())
+            .build();
 
         install_default_drivers();
         let max_connections: u32 = config
@@ -598,7 +602,7 @@ impl Catalog for SqlCatalog {
             let mut tx = self.connection.begin().await.map_err(from_sqlx_error)?;
             let update_stmt = format!(
                 "UPDATE {NAMESPACE_TABLE_NAME} SET {NAMESPACE_FIELD_PROPERTY_VALUE} = ?
-                 WHERE {CATALOG_FIELD_CATALOG_NAME} = ? 
+                 WHERE {CATALOG_FIELD_CATALOG_NAME} = ?
                  AND {NAMESPACE_FIELD_NAME} = ?
                  AND {NAMESPACE_FIELD_PROPERTY_KEY} = ?"
             );
@@ -689,7 +693,7 @@ impl Catalog for SqlCatalog {
                          WHERE {CATALOG_FIELD_TABLE_NAMESPACE} = ?
                           AND {CATALOG_FIELD_CATALOG_NAME} = ?
                           AND (
-                                {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}' 
+                                {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
                                 OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
                           )",
                     ),
@@ -728,7 +732,7 @@ impl Catalog for SqlCatalog {
                       AND {CATALOG_FIELD_CATALOG_NAME} = ?
                       AND {CATALOG_FIELD_TABLE_NAME} = ?
                       AND (
-                        {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}' 
+                        {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
                         OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
                       )"
                 ),
@@ -755,7 +759,7 @@ impl Catalog for SqlCatalog {
                   AND {CATALOG_FIELD_TABLE_NAME} = ?
                   AND {CATALOG_FIELD_TABLE_NAMESPACE} = ?
                   AND (
-                    {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}' 
+                    {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
                     OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
                   )"
             ),
@@ -791,7 +795,7 @@ impl Catalog for SqlCatalog {
                       AND {CATALOG_FIELD_TABLE_NAME} = ?
                       AND {CATALOG_FIELD_TABLE_NAMESPACE} = ?
                       AND (
-                        {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}' 
+                        {CATALOG_FIELD_RECORD_TYPE} = '{CATALOG_FIELD_TABLE_RECORD_TYPE}'
                         OR {CATALOG_FIELD_RECORD_TYPE} IS NULL
                       )"
                 ),
@@ -1180,6 +1184,33 @@ mod tests {
         new_sql_catalog(warehouse_loc.clone(), Some("iceberg")).await;
     }
 
+    // Regression test: storage-backend props set on the catalog must reach
+    // the FileIO; otherwise authenticated backends fail with 401s on writes.
+    #[tokio::test]
+    async fn test_storage_props_propagate_to_file_io() {
+        let sql_lite_uri = format!("sqlite:{}", temp_path());
+        sqlx::Sqlite::create_database(&sql_lite_uri).await.unwrap();
+        let warehouse_location = temp_path();
+
+        let catalog = SqlCatalogBuilder::default()
+            .with_storage_factory(Arc::new(LocalFsStorageFactory))
+            .load(
+                "iceberg",
+                HashMap::from_iter([
+                    (SQL_CATALOG_PROP_URI.to_string(), sql_lite_uri),
+                    (SQL_CATALOG_PROP_WAREHOUSE.to_string(), warehouse_location),
+                    ("s3.region".to_string(), "us-east-1".to_string()),
+                    ("hf.token".to_string(), "hf_test_token".to_string()),
+                ]),
+            )
+            .await
+            .unwrap();
+
+        let props = catalog.fileio.config().props();
+        assert_eq!(props.get("s3.region"), Some(&"us-east-1".to_string()));
+        assert_eq!(props.get("hf.token"), Some(&"hf_test_token".to_string()));
+    }
+
     #[tokio::test]
     async fn test_builder_method() {
         let sql_lite_uri = format!("sqlite:{}", temp_path());