feat: add periodic refresh table jobs & refactor ProgressTracker

ci/workflows/pull-request.yml

@@ -537,7 +537,7 @@ steps:
       - docker-compose#v5.5.0: *docker-compose
       # Only upload zipped files, otherwise the logs is too much.
       - ./ci/plugins/upload-failure-logs-zipped
-    timeout_in_minutes: 25
+    timeout_in_minutes: 30
     retry: *auto-retry
 
   - label: "end-to-end test (deterministic simulation)"

e2e_test/iceberg/test_case/pure_slt/periodcal_refresh_iceberg_table.slt

@@ -0,0 +1,186 @@
+control substitution on
+
+statement ok
+create secret IF NOT EXISTS my_secret with (
+  backend = 'meta'
+) as 'hummockadmin';
+
+statement ok
+create connection IF NOT EXISTS 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;
+
+
+statement error
+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_interval_sec = 'abcdef'
+);
+----
+db error: ERROR: Failed to run the query
+
+Caused by:
+  Invalid Parameter Value: `refresh_interval_sec` must be a positive integer and larger than 0, but got: abcdef (error: invalid digit found in string)
+
+
+statement error
+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_interval_sec = '-10'
+);
+----
+db error: ERROR: Failed to run the query
+
+Caused by:
+  Invalid Parameter Value: `refresh_interval_sec` must be larger than 0, but got: -10
+
+
+statement error
+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_interval_sec = '15'
+);
+----
+db error: ERROR: Failed to run the query
+
+Caused by:
+  Invalid Parameter Value: `refresh_interval_sec` is not allowed when `refresh_mode` is not 'FULL_RELOAD'
+
+
+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 = 'FULL_RELOAD',
+  refresh_interval_sec = '5'
+);
+
+query ?
+select current_status, trigger_interval_secs from rw_refresh_table_state;
+----
+IDLE 5
+
+
+# wait for refresh
+sleep 20s
+
+
+query ? retry 3 backoff 1s
+select id from t_refresh order by id;
+----
+1
+2
+3
+5
+
+
+query ? retry 3 backoff 1s
+select id from iceberg_batch_table order by id;
+----
+1
+2
+3
+5
+
+
+statement ok
+insert into t_refresh values (6, 'aaa', 'a') , (7, 'bbb', 'b') , (8, 'ccc', 'c');
+
+statement ok
+flush;
+
+sleep 20s
+
+query ? retry 3 backoff 1s
+select id from t_refresh order by id;
+----
+1
+2
+3
+5
+6
+7
+8
+
+
+query ? retry 3 backoff 1s
+select id from iceberg_batch_table order by id;
+----
+1
+2
+3
+5
+6
+7
+8
+
+
+# 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;

e2e_test/iceberg/test_case/pure_slt/refresh_iceberg_table.slt

@@ -37,7 +37,7 @@ flush;
 
 sleep 10s
 
-query ?
+query ? retry 5 backoff 1s
 select count(*) = 1 from rw_iceberg_files where source_name = '__iceberg_source_t_refresh' and content = 'PositionDeletes';
 ----
 t
@@ -68,13 +68,25 @@ create table iceberg_batch_table ( primary key (id) ) with (
   catalog.type = 'storage',
   table.name = 't_refresh',
   database.name = 'public',
-  refresh_mode = 'FULL_RECOMPUTE'
+  refresh_mode = 'FULL_RELOAD'
 );
 
 
+query ?
+select current_status from rw_refresh_table_state;
+----
+IDLE
+
+
 statement ok
 refresh table iceberg_batch_table;
 
+query ?
+select current_status from rw_refresh_table_state;
+----
+REFRESHING
+
+
 sleep 10s
 
 query I
@@ -112,16 +124,23 @@ t
 statement ok
 refresh table iceberg_batch_table;
 
-sleep 10s
+sleep 15s
 
-query I
+query I retry 3 backoff 1s
 select id from t_refresh order by id;
 ----
 1
 2
 
 
-query I
+# make sure the refresh finished
+query ?
+select current_status from rw_refresh_table_state;
+----
+IDLE
+
+
+query I retry 3 backoff 1s
 select id from iceberg_batch_table order by id;
 ----
 1

grafana/dashboard/dev/__init__.py

@@ -38,6 +38,7 @@ def section(func: Callable[[Panels], list]):
 from . import kinesis_metrics as _
 from . import system_params as _
 from . import vector_search as _
+from . import refresh_manager as _
 
 def generate_panels(panels: Panels):
     return [x for s in sections for x in s(panels)]

grafana/dashboard/dev/refresh_manager.py

@@ -0,0 +1,53 @@
+from ..common import *
+from . import section
+
+@section
+def _(outer_panels: Panels):
+    panels = outer_panels.sub_panel()
+    return [
+        outer_panels.row_collapsed(
+            "Refresh Manager",
+            [
+                panels.timeseries_latency(
+                    "Refresh Job Duration",
+                    "Time taken to complete a refresh job",
+                    [
+                        panels.target(
+                            f"{metric('meta_refresh_job_duration')}",
+                            "{{table_id}} - {{status}}"
+                        )
+                    ]
+                ),
+                panels.timeseries_count(
+                    "Refresh Job Finish Rate",
+                    "Number of finished refresh jobs",
+                    [
+                        panels.target(
+                            f"{metric('meta_refresh_job_finish_cnt')}",
+                            "{{status}}"
+                        )
+                    ]
+                ),
+                panels.timeseries_count(
+                    "Refresh Cron Job Triggers",
+                    "Number of cron refresh jobs triggered",
+                    [
+                        panels.target(
+                            f"rate({metric('meta_refresh_cron_job_trigger_cnt')}[$__rate_interval])",
+                            "{{table_id}}"
+                        )
+                    ]
+                ),
+                panels.timeseries_count(
+                    "Refresh Cron Job Misses",
+                    "Number of skipped cron refresh jobs",
+                    [
+                        panels.target(
+                            f"rate({metric('meta_refresh_cron_job_miss_cnt')}[$__rate_interval])",
+                            f"{{table_id}}"
+                        )
+                    ]
+                ),
+            ]
+        )
+    ]

proto/catalog.proto

@@ -549,8 +549,8 @@ message Table {
 
   optional CdcTableType cdc_table_type = 47;
 
-  // Current refresh state of the table for refreshable tables
-  optional RefreshState refresh_state = 48;
+  reserved "refresh_state";
+  reserved 48;
 
   // Per-table catalog version, used by schema change. `None` for internal
   // tables and tests. Not to be confused with the global catalog version for

proto/meta.proto

@@ -406,6 +406,18 @@ message ListCdcProgressResponse {
   map<uint32, CdcProgress> cdc_progress = 1;
 }
 
+message ListRefreshTableStatesRequest {}
+
+message ListRefreshTableStatesResponse {
+  message RefreshTableState {
+    uint32 table_id = 1;
+    string current_status = 2;
+    optional string last_trigger_time = 3;
+    optional int64 trigger_interval_secs = 4;
+  }
+  repeated RefreshTableState states = 1;
+}
+
 service StreamManagerService {
   rpc Flush(FlushRequest) returns (FlushResponse);
   rpc Pause(PauseRequest) returns (PauseResponse);
@@ -429,6 +441,7 @@ service StreamManagerService {
   rpc SetSyncLogStoreAligned(SetSyncLogStoreAlignedRequest) returns (SetSyncLogStoreAlignedResponse);
   rpc ListCdcProgress(ListCdcProgressRequest) returns (ListCdcProgressResponse);
   rpc ListUnmigratedTables(ListUnmigratedTablesRequest) returns (ListUnmigratedTablesResponse);
+  rpc ListRefreshTableStates(ListRefreshTableStatesRequest) returns (ListRefreshTableStatesResponse);
 }
 
 // Below for cluster service.

proto/plan_common.proto

@@ -307,10 +307,12 @@
 
 message SourceRefreshMode {
   message SourceRefreshModeStreaming {}
-  message SourceRefreshModeFullRecompute {}
+  message SourceRefreshModeFullReload {
+    optional int64 refresh_interval_sec = 1;
+  }
 
   oneof refresh_mode {
     SourceRefreshModeStreaming streaming = 1;
-    SourceRefreshModeFullRecompute full_recompute = 2;
+    SourceRefreshModeFullReload full_reload = 2;
   }
 }

risedev.yml

@@ -342,6 +342,7 @@ profile:
       - use: frontend
 
   ci-iceberg-test:
+    config-path: src/config/ci.toml
     steps:
       - use: minio
       - use: mysql