feat: add written_bytes_since_open column to region_statistics table (#6904)

WenyXu · web-flow · commit 658d07bfc8f0 · 2025-09-05T07:27:30.000Z
* feat: add `write_bytes` column to `region_statistics` table

Signed-off-by: WenyXu &lt;wenymedia@gmail.com&gt;

* chore: update comments

Signed-off-by: WenyXu &lt;wenymedia@gmail.com&gt;

* chore: rename `write_bytes` to `written_bytes`

Signed-off-by: WenyXu &lt;wenymedia@gmail.com&gt;

* chore: rename `written_bytes` to `written_bytes_since_open`

Signed-off-by: WenyXu &lt;wenymedia@gmail.com&gt;

* chore: apply suggestions

Signed-off-by: WenyXu &lt;wenymedia@gmail.com&gt;

---------

Signed-off-by: WenyXu &lt;wenymedia@gmail.com&gt;
diff --git a/src/catalog/src/system_schema/information_schema/region_statistics.rs b/src/catalog/src/system_schema/information_schema/region_statistics.rs
@@ -40,6 +40,7 @@ const REGION_ID: &str = "region_id";
 const TABLE_ID: &str = "table_id";
 const REGION_NUMBER: &str = "region_number";
 const REGION_ROWS: &str = "region_rows";
+const WRITTEN_BYTES: &str = "written_bytes_since_open";
 const DISK_SIZE: &str = "disk_size";
 const MEMTABLE_SIZE: &str = "memtable_size";
 const MANIFEST_SIZE: &str = "manifest_size";
@@ -57,6 +58,7 @@ const INIT_CAPACITY: usize = 42;
 /// - `table_id`: The table id.
 /// - `region_number`: The region number.
 /// - `region_rows`: The number of rows in region.
+/// - `written_bytes_since_open`: The total bytes written of the region since region opened.
 /// - `memtable_size`: The memtable size in bytes.
 /// - `disk_size`: The approximate disk size in bytes.
 /// - `manifest_size`: The manifest size in bytes.
@@ -84,6 +86,7 @@ impl InformationSchemaRegionStatistics {
             ColumnSchema::new(TABLE_ID, ConcreteDataType::uint32_datatype(), false),
             ColumnSchema::new(REGION_NUMBER, ConcreteDataType::uint32_datatype(), false),
             ColumnSchema::new(REGION_ROWS, ConcreteDataType::uint64_datatype(), true),
+            ColumnSchema::new(WRITTEN_BYTES, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(DISK_SIZE, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(MEMTABLE_SIZE, ConcreteDataType::uint64_datatype(), true),
             ColumnSchema::new(MANIFEST_SIZE, ConcreteDataType::uint64_datatype(), true),
@@ -147,6 +150,7 @@ struct InformationSchemaRegionStatisticsBuilder {
     table_ids: UInt32VectorBuilder,
     region_numbers: UInt32VectorBuilder,
     region_rows: UInt64VectorBuilder,
+    written_bytes: UInt64VectorBuilder,
     disk_sizes: UInt64VectorBuilder,
     memtable_sizes: UInt64VectorBuilder,
     manifest_sizes: UInt64VectorBuilder,
@@ -166,6 +170,7 @@ impl InformationSchemaRegionStatisticsBuilder {
             table_ids: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
             region_numbers: UInt32VectorBuilder::with_capacity(INIT_CAPACITY),
             region_rows: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
+            written_bytes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             disk_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             memtable_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
             manifest_sizes: UInt64VectorBuilder::with_capacity(INIT_CAPACITY),
@@ -197,6 +202,7 @@ impl InformationSchemaRegionStatisticsBuilder {
             (TABLE_ID, &Value::from(region_stat.id.table_id())),
             (REGION_NUMBER, &Value::from(region_stat.id.region_number())),
             (REGION_ROWS, &Value::from(region_stat.num_rows)),
+            (WRITTEN_BYTES, &Value::from(region_stat.written_bytes)),
             (DISK_SIZE, &Value::from(region_stat.approximate_bytes)),
             (MEMTABLE_SIZE, &Value::from(region_stat.memtable_size)),
             (MANIFEST_SIZE, &Value::from(region_stat.manifest_size)),
@@ -216,6 +222,7 @@ impl InformationSchemaRegionStatisticsBuilder {
         self.region_numbers
             .push(Some(region_stat.id.region_number()));
         self.region_rows.push(Some(region_stat.num_rows));
+        self.written_bytes.push(Some(region_stat.written_bytes));
         self.disk_sizes.push(Some(region_stat.approximate_bytes));
         self.memtable_sizes.push(Some(region_stat.memtable_size));
         self.manifest_sizes.push(Some(region_stat.manifest_size));
@@ -232,6 +239,7 @@ impl InformationSchemaRegionStatisticsBuilder {
             Arc::new(self.table_ids.finish()),
             Arc::new(self.region_numbers.finish()),
             Arc::new(self.region_rows.finish()),
+            Arc::new(self.written_bytes.finish()),
             Arc::new(self.disk_sizes.finish()),
             Arc::new(self.memtable_sizes.finish()),
             Arc::new(self.manifest_sizes.finish()),
diff --git a/src/cmd/src/standalone.rs b/src/cmd/src/standalone.rs
@@ -834,7 +834,7 @@ impl InformationExtension for StandaloneInformationExtension {
                     region_manifest: region_stat.manifest.into(),
                     data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
                     metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
-                    write_bytes: 0,
+                    written_bytes: region_stat.written_bytes,
                 }
             })
             .collect::<Vec<_>>();
diff --git a/src/common/meta/src/datanode.rs b/src/common/meta/src/datanode.rs
@@ -102,8 +102,8 @@ pub struct RegionStat {
     pub index_size: u64,
     /// The manifest infoof the region.
     pub region_manifest: RegionManifestInfo,
-    /// The write bytes.
-    pub write_bytes: u64,
+    /// The total bytes written of the region since region opened.
+    pub written_bytes: u64,
     /// The latest entry id of topic used by data.
     /// **Only used by remote WAL prune.**
     pub data_topic_latest_entry_id: u64,
@@ -306,7 +306,7 @@ impl From<&api::v1::meta::RegionStat> for RegionStat {
             sst_num: region_stat.sst_num,
             index_size: region_stat.index_size,
             region_manifest: region_stat.manifest.into(),
-            write_bytes: region_stat.write_bytes,
+            written_bytes: region_stat.written_bytes,
             data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,
             metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,
         }
diff --git a/src/meta-srv/src/handler/collect_leader_region_handler.rs b/src/meta-srv/src/handler/collect_leader_region_handler.rs
@@ -87,7 +87,7 @@ mod tests {
             index_size: 0,
             data_topic_latest_entry_id: 0,
             metadata_topic_latest_entry_id: 0,
-            write_bytes: 0,
+            written_bytes: 0,
         }
     }
 
diff --git a/src/meta-srv/src/handler/failure_handler.rs b/src/meta-srv/src/handler/failure_handler.rs
@@ -105,7 +105,7 @@ mod tests {
                 },
                 data_topic_latest_entry_id: 0,
                 metadata_topic_latest_entry_id: 0,
-                write_bytes: 0,
+                written_bytes: 0,
             }
         }
         acc.stat = Some(Stat {
diff --git a/src/meta-srv/src/handler/persist_stats_handler.rs b/src/meta-srv/src/handler/persist_stats_handler.rs
@@ -51,14 +51,14 @@ const DEFAULT_CONTEXT: InserterContext = InserterContext {
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 struct PersistedRegionStat {
     region_id: RegionId,
-    write_bytess: u64,
+    written_bytes: u64,
 }
 
 impl From<&RegionStat> for PersistedRegionStat {
     fn from(stat: &RegionStat) -> Self {
         Self {
             region_id: stat.id,
-            write_bytess: stat.write_bytes,
+            written_bytes: stat.written_bytes,
         }
     }
 }
@@ -94,8 +94,8 @@ fn compute_persist_region_stat(
     let write_bytes_delta = persisted_region_stat
         .and_then(|persisted_region_stat| {
             region_stat
-                .write_bytes
-                .checked_sub(persisted_region_stat.write_bytess)
+                .written_bytes
+                .checked_sub(persisted_region_stat.written_bytes)
         })
         .unwrap_or_default();
 
@@ -273,7 +273,7 @@ mod tests {
     fn create_test_region_stat(
         table_id: u32,
         region_number: u32,
-        write_bytes: u64,
+        written_bytes: u64,
         engine: &str,
     ) -> RegionStat {
         let region_id = RegionId::new(table_id, region_number);
@@ -294,7 +294,7 @@ mod tests {
                 manifest_version: 1,
                 flushed_entry_id: 100,
             },
-            write_bytes,
+            written_bytes,
             data_topic_latest_entry_id: 200,
             metadata_topic_latest_entry_id: 200,
         }
@@ -326,7 +326,7 @@ mod tests {
         let timestamp_millis = 1640995260000; // 2022-01-01 00:01:00 UTC
         let persisted_stat = PersistedRegionStat {
             region_id: region_stat.id,
-            write_bytess: 1000, // Previous write bytes
+            written_bytes: 1000, // Previous write bytes
         };
         let result = compute_persist_region_stat(
             &region_stat,
@@ -354,7 +354,7 @@ mod tests {
         let timestamp_millis = 1640995320000; // 2022-01-01 00:02:00 UTC
         let persisted_stat = PersistedRegionStat {
             region_id: region_stat.id,
-            write_bytess: 1200, // Previous write bytes (higher than current)
+            written_bytes: 1200, // Previous write bytes (higher than current)
         };
         let result = compute_persist_region_stat(
             &region_stat,
@@ -382,7 +382,7 @@ mod tests {
         let timestamp_millis = 1640995380000; // 2022-01-01 00:03:00 UTC
         let persisted_stat = PersistedRegionStat {
             region_id: region_stat.id,
-            write_bytess: 2000, // Same as current write bytes
+            written_bytes: 2000, // Same as current write bytes
         };
         let result = compute_persist_region_stat(
             &region_stat,
@@ -410,7 +410,7 @@ mod tests {
         let timestamp_millis = 1640995620000; // 2022-01-01 00:07:00 UTC
         let persisted_stat = PersistedRegionStat {
             region_id: region_stat.id,
-            write_bytess: 1000, // Higher than current, would cause underflow
+            written_bytes: 1000, // Higher than current, would cause underflow
         };
         let result = compute_persist_region_stat(
             &region_stat,
@@ -507,7 +507,7 @@ mod tests {
             region_id,
             PersistedRegionStat {
                 region_id,
-                write_bytess: 500,
+                written_bytes: 500,
             },
         );
         let (expected_row, expected_persisted_region_stat) = to_persisted_if_leader(
diff --git a/src/meta-srv/src/handler/region_lease_handler.rs b/src/meta-srv/src/handler/region_lease_handler.rs
@@ -167,7 +167,7 @@ mod test {
             },
             data_topic_latest_entry_id: 0,
             metadata_topic_latest_entry_id: 0,
-            write_bytes: 0,
+            written_bytes: 0,
         }
     }
 
diff --git a/src/meta-srv/src/selector/weight_compute.rs b/src/meta-srv/src/selector/weight_compute.rs
@@ -198,7 +198,7 @@ mod tests {
                 },
                 data_topic_latest_entry_id: 0,
                 metadata_topic_latest_entry_id: 0,
-                write_bytes: 0,
+                written_bytes: 0,
             }],
             ..Default::default()
         }
@@ -227,7 +227,7 @@ mod tests {
                 },
                 data_topic_latest_entry_id: 0,
                 metadata_topic_latest_entry_id: 0,
-                write_bytes: 0,
+                written_bytes: 0,
             }],
             ..Default::default()
         }
@@ -256,7 +256,7 @@ mod tests {
                 },
                 data_topic_latest_entry_id: 0,
                 metadata_topic_latest_entry_id: 0,
-                write_bytes: 0,
+                written_bytes: 0,
             }],
             ..Default::default()
         }
diff --git a/src/metric-engine/src/utils.rs b/src/metric-engine/src/utils.rs
@@ -62,7 +62,7 @@ pub fn get_region_statistic(mito: &MitoEngine, region_id: RegionId) -> Option<Re
                 metadata_flushed_entry_id: metadata_stat.manifest.data_flushed_entry_id(),
                 metadata_manifest_version: metadata_stat.manifest.data_manifest_version(),
             },
-            write_bytes: metadata_stat.write_bytes + data_stat.write_bytes,
+            written_bytes: metadata_stat.written_bytes + data_stat.written_bytes,
             data_topic_latest_entry_id: data_stat.data_topic_latest_entry_id,
             metadata_topic_latest_entry_id: metadata_stat.metadata_topic_latest_entry_id,
         }),
diff --git a/src/mito2/src/engine/basic_test.rs b/src/mito2/src/engine/basic_test.rs
@@ -93,7 +93,7 @@ async fn test_write_to_region() {
     };
     put_rows(&engine, region_id, rows).await;
     let region = engine.get_region(region_id).unwrap();
-    assert!(region.write_bytes.load(Ordering::Relaxed) > 0);
+    assert!(region.written_bytes.load(Ordering::Relaxed) > 0);
 }
 
 #[apply(multiple_log_store_factories)]
@@ -167,7 +167,7 @@ async fn test_region_replay(factory: Option<LogStoreFactory>) {
 
     // The replay won't update the write bytes rate meter.
     let region = engine.get_region(region_id).unwrap();
-    assert_eq!(region.write_bytes.load(Ordering::Relaxed), 0);
+    assert_eq!(region.written_bytes.load(Ordering::Relaxed), 0);
 
     let request = ScanRequest::default();
     let stream = engine.scan_to_stream(region_id, request).await.unwrap();
diff --git a/src/mito2/src/region.rs b/src/mito2/src/region.rs
@@ -134,7 +134,7 @@ pub struct MitoRegion {
     /// which means these WAL entries maybe able to be pruned up to `topic_latest_entry_id`.
     pub(crate) topic_latest_entry_id: AtomicU64,
     /// The total bytes written to the region.
-    pub(crate) write_bytes: Arc<AtomicU64>,
+    pub(crate) written_bytes: Arc<AtomicU64>,
     /// Memtable builder for the region.
     pub(crate) memtable_builder: MemtableBuilderRef,
     /// manifest stats
@@ -453,7 +453,7 @@ impl MitoRegion {
         let manifest_version = self.stats.manifest_version();
 
         let topic_latest_entry_id = self.topic_latest_entry_id.load(Ordering::Relaxed);
-        let write_bytes = self.write_bytes.load(Ordering::Relaxed);
+        let written_bytes = self.written_bytes.load(Ordering::Relaxed);
 
         RegionStatistic {
             num_rows,
@@ -469,7 +469,7 @@ impl MitoRegion {
             },
             data_topic_latest_entry_id: topic_latest_entry_id,
             metadata_topic_latest_entry_id: topic_latest_entry_id,
-            write_bytes,
+            written_bytes,
         }
     }
 
@@ -1217,7 +1217,7 @@ mod tests {
             last_compaction_millis: Default::default(),
             time_provider: Arc::new(StdTimeProvider),
             topic_latest_entry_id: Default::default(),
-            write_bytes: Arc::new(AtomicU64::new(0)),
+            written_bytes: Arc::new(AtomicU64::new(0)),
             memtable_builder: Arc::new(EmptyMemtableBuilder::default()),
             stats: ManifestStats::default(),
         };
diff --git a/src/mito2/src/region/opener.rs b/src/mito2/src/region/opener.rs
@@ -303,7 +303,7 @@ impl RegionOpener {
             time_provider: self.time_provider.clone(),
             topic_latest_entry_id: AtomicU64::new(0),
             memtable_builder,
-            write_bytes: Arc::new(AtomicU64::new(0)),
+            written_bytes: Arc::new(AtomicU64::new(0)),
             stats: self.stats,
         })
     }
@@ -505,7 +505,7 @@ impl RegionOpener {
             last_compaction_millis: AtomicI64::new(now),
             time_provider: self.time_provider.clone(),
             topic_latest_entry_id: AtomicU64::new(topic_latest_entry_id),
-            write_bytes: Arc::new(AtomicU64::new(0)),
+            written_bytes: Arc::new(AtomicU64::new(0)),
             memtable_builder,
             stats: self.stats.clone(),
         };
diff --git a/src/mito2/src/region_write_ctx.rs b/src/mito2/src/region_write_ctx.rs
@@ -107,8 +107,8 @@ pub(crate) struct RegionWriteCtx {
     pub(crate) put_num: usize,
     /// Rows to delete.
     pub(crate) delete_num: usize,
-    /// Write bytes per second.
-    pub(crate) write_bytes: Option<Arc<AtomicU64>>,
+    /// The total bytes written to the region.
+    pub(crate) written_bytes: Option<Arc<AtomicU64>>,
 }
 
 impl RegionWriteCtx {
@@ -117,7 +117,7 @@ impl RegionWriteCtx {
         region_id: RegionId,
         version_control: &VersionControlRef,
         provider: Provider,
-        write_bytes: Option<Arc<AtomicU64>>,
+        written_bytes: Option<Arc<AtomicU64>>,
     ) -> RegionWriteCtx {
         let VersionControlData {
             version,
@@ -140,7 +140,7 @@ impl RegionWriteCtx {
             put_num: 0,
             delete_num: 0,
             bulk_parts: vec![],
-            write_bytes,
+            written_bytes,
         }
     }
 
@@ -219,7 +219,7 @@ impl RegionWriteCtx {
         }
 
         let mutable_memtable = self.version.memtables.mutable.clone();
-        let prev_memory_usage = if self.write_bytes.is_some() {
+        let prev_memory_usage = if self.written_bytes.is_some() {
             Some(mutable_memtable.memory_usage())
         } else {
             None
@@ -257,11 +257,10 @@ impl RegionWriteCtx {
             }
         }
 
-        if let Some(write_bytes) = &self.write_bytes {
+        if let Some(written_bytes) = &self.written_bytes {
             let new_memory_usage = mutable_memtable.memory_usage();
-            let written_bytes =
-                new_memory_usage.saturating_sub(prev_memory_usage.unwrap_or_default());
-            write_bytes.fetch_add(written_bytes as u64, Ordering::Relaxed);
+            let bytes = new_memory_usage.saturating_sub(prev_memory_usage.unwrap_or_default());
+            written_bytes.fetch_add(bytes as u64, Ordering::Relaxed);
         }
         // Updates region sequence and entry id. Since we stores last sequence and entry id in region, we need
         // to decrease `next_sequence` and `next_entry_id` by 1.
@@ -298,7 +297,7 @@ impl RegionWriteCtx {
             .start_timer();
 
         let mutable_memtable = &self.version.memtables.mutable;
-        let prev_memory_usage = if self.write_bytes.is_some() {
+        let prev_memory_usage = if self.written_bytes.is_some() {
             Some(mutable_memtable.memory_usage())
         } else {
             None
@@ -333,11 +332,10 @@ impl RegionWriteCtx {
             }
         }
 
-        if let Some(write_bytes) = &self.write_bytes {
+        if let Some(written_bytes) = &self.written_bytes {
             let new_memory_usage = mutable_memtable.memory_usage();
-            let written_bytes =
-                new_memory_usage.saturating_sub(prev_memory_usage.unwrap_or_default());
-            write_bytes.fetch_add(written_bytes as u64, Ordering::Relaxed);
+            let bytes = new_memory_usage.saturating_sub(prev_memory_usage.unwrap_or_default());
+            written_bytes.fetch_add(bytes as u64, Ordering::Relaxed);
         }
         self.version_control
             .set_sequence_and_entry_id(self.next_sequence - 1, self.next_entry_id - 1);
diff --git a/src/mito2/src/worker/handle_write.rs b/src/mito2/src/worker/handle_write.rs
diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs
diff --git a/tests/cases/standalone/common/information_schema/region_statistics.result b/tests/cases/standalone/common/information_schema/region_statistics.result
diff --git a/tests/cases/standalone/common/information_schema/region_statistics.sql b/tests/cases/standalone/common/information_schema/region_statistics.sql
diff --git a/tests/cases/standalone/common/system/information_schema.result b/tests/cases/standalone/common/system/information_schema.result

Original file line number	Diff line number	Diff line change
`@@ -834,7 +834,7 @@ impl InformationExtension for StandaloneInformationExtension {`
`834`	`834`	`region_manifest: region_stat.manifest.into(),`
`835`	`835`	`data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id,`
`836`	`836`	`metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id,`
`837`		`- write_bytes: 0,`
	`837`	`+ written_bytes: region_stat.written_bytes,`
`838`	`838`	`}`
`839`	`839`	`})`
`840`	`840`	`.collect::<Vec<_>>();`
Original file line number	Diff line number	Diff line change
`@@ -87,7 +87,7 @@ mod tests {`
`87`	`87`	`index_size: 0,`
`88`	`88`	`data_topic_latest_entry_id: 0,`
`89`	`89`	`metadata_topic_latest_entry_id: 0,`
`90`		`- write_bytes: 0,`
	`90`	`+ written_bytes: 0,`
`91`	`91`	`}`
`92`	`92`	`}`
`93`	`93`
Original file line number	Diff line number	Diff line change
`@@ -105,7 +105,7 @@ mod tests {`
`105`	`105`	`},`
`106`	`106`	`data_topic_latest_entry_id: 0,`
`107`	`107`	`metadata_topic_latest_entry_id: 0,`
`108`		`- write_bytes: 0,`
	`108`	`+ written_bytes: 0,`
`109`	`109`	`}`
`110`	`110`	`}`
`111`	`111`	`acc.stat = Some(Stat {`
Original file line number	Diff line number	Diff line change
`@@ -167,7 +167,7 @@ mod test {`
`167`	`167`	`},`
`168`	`168`	`data_topic_latest_entry_id: 0,`
`169`	`169`	`metadata_topic_latest_entry_id: 0,`
`170`		`- write_bytes: 0,`
	`170`	`+ written_bytes: 0,`
`171`	`171`	`}`
`172`	`172`	`}`
`173`	`173`