feat(iceberg): support refreshable batch iceberg table

e2e_test/iceberg/test_case/pure_slt/refresh_iceberg_table.slt

@@ -0,0 +1,129 @@
+control substitution on
+
+statement ok
+create secret my_secret with (
+  backend = 'meta'
+) as 'hummockadmin';
+
+statement ok
+create connection my_conn
+with (
+    type = 'iceberg',
+    warehouse.path = 's3://hummock001/iceberg_connection',
+    s3.access.key = secret my_secret,
+    s3.secret.key = secret my_secret,
+    s3.endpoint = 'http://127.0.0.1:9301',
+    s3.region = 'us-west-2',
+    catalog.type = 'storage',
+);
+
+statement ok
+set iceberg_engine_connection = 'public.my_conn';
+
+statement ok
+create table t_refresh (id int primary key, name varchar, foo varchar) with (commit_checkpoint_interval = 10) engine = iceberg;
+
+statement ok
+insert into t_refresh values(1, 'xxx', 'a') , (2, 'yyy', 'b'), (3, 'zzz', 'c'), (4, 'www', 'd'), (5, 'zzz', 'c');
+
+statement ok
+flush;
+
+statement ok
+delete from t_refresh where id = 4;
+
+statement ok
+flush;
+
+sleep 10s
+
+query ?
+select count(*) = 1 from rw_iceberg_files where source_name = '__iceberg_source_t_refresh' and content = 'PositionDeletes';
+----
+t
+
+
+statement ok
+create table iceberg_batch_table ( primary key (id) ) with (
+  connector = 'iceberg',
+  warehouse.path = 's3://hummock001/iceberg_connection',
+  s3.access.key = secret my_secret,
+  s3.secret.key = secret my_secret,
+  s3.endpoint = 'http://127.0.0.1:9301',
+  s3.region = 'us-west-2',
+  catalog.type = 'storage',
+  table.name = 't_refresh',
+  database.name = 'public',
+  refresh_mode = 'MANUAL_TRIGGER'
+);
+
+
+statement ok
+refresh table iceberg_batch_table;
+
+sleep 10s
+
+query I
+select id from t_refresh order by id;
+----
+1
+2
+3
+5
+
+query I
+select id from iceberg_batch_table order by id;
+----
+1
+2
+3
+5
+
+# ==== end PositionDeletes
+
+statement ok
+delete from t_refresh where name = 'zzz' and foo = 'c';
+
+statement ok
+flush;
+
+sleep 10s
+
+query ?
+select count(*) = 1 from rw_iceberg_files where source_name = '__iceberg_source_t_refresh' and content = 'EqualityDeletes';
+----
+t
+
+
+statement ok
+refresh table iceberg_batch_table;
+
+sleep 10s
+
+query I
+select id from t_refresh order by id;
+----
+1
+2
+
+
+query I
+select id from iceberg_batch_table order by id;
+----
+1
+2
+
+
+# Cleanup
+
+statement ok
+drop table iceberg_batch_table;
+
+statement ok
+drop table t_refresh;
+
+statement ok
+drop connection my_conn;
+
+statement ok
+drop secret my_secret;

proto/catalog.proto

@@ -149,6 +149,9 @@ message Source {
   // Used for secret connect options.
   map<string, secret.SecretRef> secret_refs = 19;
 
+  // Source refresh mode
+  plan_common.SourceRefreshMode refresh_mode = 20;
+
   // Per-source catalog version, used by schema change.
   uint64 version = 100;
 

proto/plan_common.proto

@@ -300,3 +300,13 @@ enum AdditionalColumnType {
   ADDITIONAL_COLUMN_TYPE_NORMAL = 7;
   ADDITIONAL_COLUMN_TYPE_PAYLOAD = 8;
 }
+
+message SourceRefreshMode {
+  message SourceRefreshModeStreaming {}
+  message SourceRefreshModeManualTrigger {}
+
+  oneof refresh_mode {
+    SourceRefreshModeStreaming streaming = 1;
+    SourceRefreshModeManualTrigger manual_trigger = 2;
+  }
+}

proto/stream_plan.proto

@@ -276,6 +276,7 @@ message StreamSource {
   map<string, secret.SecretRef> secret_refs = 10;
   // Downstream columns are used by list node to know which columns are needed.
   optional Columns downstream_columns = 11;
+  plan_common.SourceRefreshMode refresh_mode = 12;
 }
 
 // copy contents from StreamSource to prevent compatibility issues in the future
@@ -292,6 +293,7 @@ message StreamFsFetch {
   // Source rate limit
   optional uint32 rate_limit = 9;
   map<string, secret.SecretRef> secret_refs = 10;
+  plan_common.SourceRefreshMode refresh_mode = 11;
 }
 
 // The executor only for receiving barrier from the meta service. It always resides in the leaves

src/batch/executors/src/executor/iceberg_scan.rs

@@ -22,7 +22,8 @@ use risingwave_common::catalog::{
 use risingwave_common::types::{DataType, ScalarImpl};
 use risingwave_connector::WithOptionsSecResolved;
 use risingwave_connector::source::iceberg::{
-    IcebergFileScanTask, IcebergProperties, IcebergScanOpts, IcebergSplit, scan_task_to_chunk,
+    IcebergFileScanTask, IcebergProperties, IcebergScanOpts, IcebergSplit,
+    scan_task_to_chunk_with_deletes,
 };
 use risingwave_connector::source::{ConnectorProperties, SplitImpl, SplitMetaData};
 use risingwave_expr::expr::LiteralExpression;
@@ -109,13 +110,14 @@ impl IcebergScanExecutor {
 
         for data_file_scan_task in data_file_scan_tasks {
             #[for_await]
-            for chunk in scan_task_to_chunk(
+            for chunk in scan_task_to_chunk_with_deletes(
                 table.clone(),
                 data_file_scan_task,
                 IcebergScanOpts {
                     chunk_size: self.chunk_size,
                     need_seq_num: self.need_seq_num,
                     need_file_path_and_pos: self.need_file_path_and_pos,
+                    handle_delete_files: false,
                 },
                 self.metrics.as_ref().map(|m| m.iceberg_scan_metrics()),
             ) {