fix(datafusion): reject non-append insert operations

Author: u70b3

crates/integrations/datafusion/src/table/mod.rs

@@ -159,11 +159,11 @@ impl TableProvider for IcebergTableProvider {
         &self,
         state: &dyn Session,
         input: Arc<dyn ExecutionPlan>,
-        insert_op: InsertOp,
+        _insert_op: InsertOp,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
-        if insert_op != InsertOp::Append {
+        if _insert_op != InsertOp::Append {
             return Err(DataFusionError::NotImplemented(format!(
-                "IcebergTableProvider supports only append inserts, got {insert_op}"
+                "IcebergTableProvider supports only append inserts, got {_insert_op}"
             )));
         }
 
@@ -717,57 +717,39 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn test_catalog_backed_provider_rejects_overwrite_insert() {
+    async fn test_catalog_backed_provider_rejects_non_append_op() {
         use datafusion::physical_plan::empty::EmptyExec;
 
         let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await;
         let provider = IcebergTableProvider::try_new(catalog, namespace, table_name)
             .await
             .unwrap();
         let ctx = SessionContext::new();
-        let input = Arc::new(EmptyExec::new(provider.schema())) as Arc<dyn ExecutionPlan>;
-
-        let error = provider
-            .insert_into(&ctx.state(), input, InsertOp::Overwrite)
-            .await
-            .expect_err("overwrite inserts should be rejected");
 
-        assert!(
-            matches!(
-                error,
-                DataFusionError::NotImplemented(ref message)
-                    if message
-                        == "IcebergTableProvider supports only append inserts, got Insert Overwrite"
+        for (insert_op, expected_message) in [
+            (
+                InsertOp::Overwrite,
+                "IcebergTableProvider supports only append inserts, got Insert Overwrite",
             ),
-            "unexpected error: {error}"
-        );
-    }
-
-    #[tokio::test]
-    async fn test_catalog_backed_provider_rejects_replace_insert() {
-        use datafusion::physical_plan::empty::EmptyExec;
-
-        let (catalog, namespace, table_name, _temp_dir) = get_test_catalog_and_table().await;
-        let provider = IcebergTableProvider::try_new(catalog, namespace, table_name)
-            .await
-            .unwrap();
-        let ctx = SessionContext::new();
-        let input = Arc::new(EmptyExec::new(provider.schema())) as Arc<dyn ExecutionPlan>;
-
-        let error = provider
-            .insert_into(&ctx.state(), input, InsertOp::Replace)
-            .await
-            .expect_err("replace inserts should be rejected");
-
-        assert!(
-            matches!(
-                error,
-                DataFusionError::NotImplemented(ref message)
-                    if message
-                        == "IcebergTableProvider supports only append inserts, got Replace Into"
+            (
+                InsertOp::Replace,
+                "IcebergTableProvider supports only append inserts, got Replace Into",
             ),
-            "unexpected error: {error}"
-        );
+        ] {
+            let input = Arc::new(EmptyExec::new(provider.schema())) as Arc<dyn ExecutionPlan>;
+            let error = provider
+                .insert_into(&ctx.state(), input, insert_op)
+                .await
+                .expect_err("non-append inserts should be rejected");
+
+            assert!(
+                matches!(
+                    error,
+                    DataFusionError::NotImplemented(ref message) if message == expected_message
+                ),
+                "unexpected error: {error}"
+            );
+        }
     }
 
     #[tokio::test]