feat: update iceberg writer and add more metrics

Cargo.toml

@@ -118,7 +118,7 @@ criterion = { version = "0.5", features = ["async_futures"] }
 tonic = { package = "madsim-tonic", version = "0.4.1" }
 tonic-build = { package = "madsim-tonic-build", version = "0.4.2" }
 prost = { version = "0.12" }
-icelake = { git = "https://github.com/icelake-io/icelake", rev = "5cdcdffd24f4624a0a43f92c5f368988169a799b", features = [
+icelake = { git = "https://github.com/icelake-io/icelake", rev = "3f7b53ba5b563524212c25810345d1314678e7fc", features = [
   "prometheus",
 ] }
 arrow-array = "49"

e2e_test/iceberg/test_case/cdc/load.slt

@@ -35,6 +35,9 @@ CREATE SINK s1 AS select * from products WITH (
     primary_key = 'id'
 );
 
+statement ok
+flush;
+
 query I
 select count(*) from products;
 ----

grafana/risingwave-dev-dashboard.dashboard.py

@@ -3398,32 +3398,62 @@ def section_iceberg_metrics(outer_panels):
             "Iceberg Sink Metrics",
             [
                 panels.timeseries_count(
-                    "Write Qps Of Iceberg File Appender",
-                    "iceberg file appender write qps",
+                    "Write Qps Of Iceberg Writer",
+                    "iceberg write qps",
                     [
                         panels.target(
-                            f"{metric('iceberg_file_appender_write_qps')}",
+                            f"{metric('iceberg_write_qps')}",
                            "{{executor_id}} @ {{sink_id}}",
                         ),
                     ]
                 ),
                 panels.timeseries_latency(
-                    "Write latency Of Iceberg File Appender",
+                    "Write Latency Of Iceberg Writer",
                     "",
                     [
                         *quantile(
                             lambda quantile, legend: panels.target(
-                                f"histogram_quantile({quantile}, sum(rate({metric('iceberg_file_appender_write_latency_bucket')}[$__rate_interval])) by (le, sink_id))",
+                                f"histogram_quantile({quantile}, sum(rate({metric('iceberg_write_latency_bucket')}[$__rate_interval])) by (le, sink_id))",
                                 f"p{legend}" + " @ {{sink_id}}",
                             ),
                             [50, 99, "max"],
                         ),
                         panels.target(
-                            f"sum by(le, sink_id)(rate({metric('iceberg_file_appender_write_latency_sum')}[$__rate_interval])) / sum by(le, type, job, instance) (rate({metric('iceberg_file_appender_write_latency_count')}[$__rate_interval]))",
+                            f"sum by(le, sink_id)(rate({metric('iceberg_write_latency_sum')}[$__rate_interval])) / sum by(le, type, job, instance) (rate({metric('iceberg_write_latency_count')}[$__rate_interval]))",
                             "avg @ {{sink_id}}",
                         ),
                     ],
                 ),
+                panels.timeseries_count(
+                    "Iceberg rolling unfushed data file",
+                    "",
+                    [
+                        panels.target(
+                            f"{metric('iceberg_rolling_unfushed_data_file')}",
+                           "{{executor_id}} @ {{sink_id}}",
+                        ),
+                    ]
+                ),
+                panels.timeseries_count(
+                    "Iceberg position delete cache num",
+                    "",
+                    [
+                        panels.target(
+                            f"{metric('iceberg_position_delete_cache_num')}",
+                           "{{executor_id}} @ {{sink_id}}",
+                        ),
+                    ]
+                ),
+                panels.timeseries_count(
+                    "Iceberg partition num",
+                    "",
+                    [
+                        panels.target(
+                            f"{metric('iceberg_partition_num')}",
+                           "{{executor_id}} @ {{sink_id}}",
+                        ),
+                    ]
+                ),
             ]
         )
     ]