forbid for iceberg table engine

Author: wenym1

src/frontend/planner_test/src/lib.rs

@@ -883,6 +883,7 @@ impl TestCase {
                     None,
                     false,
                     None,
+                    true,
                 ) {
                     Ok(sink_plan) => {
                         ret.sink_plan = Some(explain_plan(&sink_plan.into()));

src/frontend/src/handler/create_sink.rs

@@ -387,6 +387,8 @@ pub async fn gen_sink_plan(
         }
     }
 
+    let allow_snapshot_backfill = target_table_catalog.is_none() && !is_iceberg_engine_internal;
+
     let sink_plan = plan_root.gen_sink_plan(
         sink_table_name,
         definition,
@@ -400,6 +402,7 @@ pub async fn gen_sink_plan(
         partition_info,
         user_specified_columns,
         auto_refresh_schema_from_table,
+        allow_snapshot_backfill,
     )?;
 
     let sink_desc = sink_plan.sink_desc().clone();

src/frontend/src/optimizer/mod.rs

@@ -1121,7 +1121,7 @@ impl LogicalPlanRoot {
     }
 
     /// Optimize and generate a create sink plan.
-    #[allow(clippy::too_many_arguments)]
+    #[expect(clippy::too_many_arguments)]
     pub fn gen_sink_plan(
         self,
         sink_name: String,
@@ -1136,10 +1136,15 @@ impl LogicalPlanRoot {
         partition_info: Option<PartitionComputeInfo>,
         user_specified_columns: bool,
         auto_refresh_schema_from_table: Option<Arc<TableCatalog>>,
+        allow_snapshot_backfill: bool,
     ) -> Result<StreamSink> {
         let backfill_type = if without_backfill {
             BackfillType::UpstreamOnly
-        } else if target_table.is_none() && self.should_use_snapshot_backfill() {
+        } else if allow_snapshot_backfill && self.should_use_snapshot_backfill() {
+            assert!(
+                target_table.is_none(),
+                "should not allow snapshot backfill for sink-into-table"
+            );
             // Snapshot backfill on sink-into-table is not allowed
             BackfillType::SnapshotBackfill
         } else if self.should_use_arrangement_backfill() {