fix(datafusion): reject non-append insert operations (#2714)

## Which issue does this PR close?

- Closes #2711.

## What changes are included in this PR?

Reject DataFusion `InsertOp::Overwrite` and `InsertOp::Replace` before
constructing an Iceberg write plan.

Previously, `IcebergTableProvider::insert_into` ignored the requested
insert operation while the commit executor always used `fast_append`.
Unsupported overwrite or replace requests could therefore be silently
executed with append semantics.

This patch keeps the existing append path unchanged and returns
`DataFusionError::NotImplemented` for unsupported insert operations. It
preserves the existing `_insert_op` parameter name to avoid changing the
generated public API snapshot.

## Are these changes tested?

Yes. Added unit tests verifying that:

- `InsertOp::Overwrite` returns `DataFusionError::NotImplemented`.
- `InsertOp::Replace` returns `DataFusionError::NotImplemented`.
- Existing append-insert behavior remains unchanged.

Local verification:

- `cargo test -p iceberg-datafusion test_catalog_backed_provider_rejects
-- --nocapture`
- `cargo fmt --all -- --check`
- `git diff --check`

Author: u70b3

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

@@ -161,6 +161,12 @@ impl TableProvider for IcebergTableProvider {
         input: Arc<dyn ExecutionPlan>,
         _insert_op: InsertOp,
     ) -> DFResult<Arc<dyn ExecutionPlan>> {
+        if _insert_op != InsertOp::Append {
+            return Err(DataFusionError::NotImplemented(format!(
+                "IcebergTableProvider supports only append inserts, got {_insert_op}"
+            )));
+        }
+
         // Load fresh table metadata from catalog
         let table = self
             .catalog
@@ -710,6 +716,42 @@ mod tests {
         false
     }
 
+    #[tokio::test]
+    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();
+
+        for (insert_op, expected_message) in [
+            (
+                InsertOp::Overwrite,
+                "IcebergTableProvider supports only append inserts, got Insert Overwrite",
+            ),
+            (
+                InsertOp::Replace,
+                "IcebergTableProvider supports only append inserts, got Replace Into",
+            ),
+        ] {
+            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]
     async fn test_insert_plan_fanout_enabled_no_sort() {
         use datafusion::datasource::TableProvider;