feat: add metrics, add x500 x10000 slt

Signed-off-by: MrCroxx <[email protected]>

e2e_test/backfill/adaptive-rate-limit/amplification-10000.slt

@@ -0,0 +1,92 @@
+statement ok
+create table fact(v1 int);
+
+statement ok
+create table dim(v1 int);
+
+statement ok
+create table dim2(v1 int);
+
+statement ok
+insert into fact select 1 from generate_series(1, 1000000);
+
+statement ok
+insert into dim select 1 from generate_series(1, 100);
+
+statement ok
+insert into dim2 select 1 from generate_series(1, 100);
+
+statement ok
+flush;
+
+statement ok
+set background_ddl = true;
+
+statement ok
+create sink s1 as select fact.v1 from fact join dim on fact.v1 = dim.v1 join dim2 on fact.v1 = dim2.v1 with (connector = 'blackhole');
+
+statement ok
+set background_ddl = false;
+
+# Let at least 16 barriers pass through
+# Then we have 1 * 2^16 = 65536
+
+skipif in-memory
+sleep 10s
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+# statement ok
+# drop sink s1;
+
+# statement ok
+# drop table fact;
+
+# statement ok
+# drop table dim;

e2e_test/backfill/adaptive-rate-limit/amplification-500.slt

@@ -0,0 +1,89 @@
+statement ok
+create table fact(v1 int);
+
+statement ok
+create table dim(v1 int);
+
+statement ok
+create table dim2(v1 int);
+
+statement ok
+insert into fact select 1 from generate_series(1, 1000000);
+
+statement ok
+insert into dim select 1 from generate_series(1, 500);
+
+statement ok
+flush;
+
+statement ok
+set background_ddl = true;
+
+statement ok
+create materialized view m1 as select fact.v1 from fact join dim on fact.v1 = dim.v1;
+
+statement ok
+set background_ddl = false;
+
+# Let at least 16 barriers pass through
+# Then we have 1 * 2^16 = 65536
+
+skipif in-memory
+sleep 10s
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+statement ok
+flush;
+
+# statement ok
+# drop sink s1;
+
+# statement ok
+# drop table fact;
+
+# statement ok
+# drop table dim;

grafana/risingwave-dev-dashboard.dashboard.py

@@ -916,6 +916,17 @@ def section_streaming(outer_panels):
                         ),
                     ],
                 ),
+                panels.timeseries_count(
+                    "Backfill Snapshot Read Adaptive Rate Limit",
+                    "",
+                    [
+                        panels.target(
+                            f"{metric('backfill_rate_limit_bytes')}",
+                            "table_id={{table_id}} @ {{%s}}"
+                            % NODE_LABEL,
+                        )
+                    ],
+                ),
                 panels.timeseries_count(
                     "Source Upstream Status",
                     "Monitor each source upstream, 0 means the upstream is not normal, 1 means the source is ready.",

src/common/src/config.rs

@@ -2119,7 +2119,7 @@ pub mod default {
         }
 
         pub fn backfill_adaptive_rate_limit_min() -> f64 {
-            100.0
+            10.0
         }
 
         pub fn backfill_adaptive_rate_limit_max() -> f64 {

src/config/example.toml

@@ -146,7 +146,7 @@ stream_enable_actor_tokio_metrics = false
 stream_exchange_connection_pool_size = 1
 stream_enable_auto_schema_change = true
 stream_enable_shared_source = true
-stream_backfill_adaptive_rate_limit_min = 100.0
+stream_backfill_adaptive_rate_limit_min = 10.0
 stream_backfill_adaptive_rate_limit_max = 100000.0
 stream_backfill_adaptive_rate_limit_init = 1000.0
 stream_backfill_adaptive_rate_limit_step_min = 10.0