From 4e5b60b869c87be7677e2cd88f1a10cfbeb6bcb0 Mon Sep 17 00:00:00 2001 From: Weny Xu Date: Mon, 9 Nov 2026 04:25:42 +0000 Subject: [PATCH] refactor: rename flushed_parquet_bytes to ingested_parquet_bytes (#10258) Signed-off-by: WenyXu --- src/common/meta/src/datanode.rs | 49 ++++++++++++++----- src/meta-srv/src/gc/handler.rs | 2 +- src/meta-srv/src/gc/mock.rs | 2 +- .../handler/collect_leader_region_handler.rs | 2 +- src/meta-srv/src/handler/failure_handler.rs | 2 +- .../src/handler/persist_stats_handler.rs | 2 +- .../src/handler/region_lease_handler.rs | 2 +- src/meta-srv/src/selector/weight_compute.rs | 6 +-- src/metric-engine/src/utils.rs | 12 ++--- src/mito2/src/engine/flush_test.rs | 16 +++--- .../engine/object_store_wal_recovery_test.rs | 2 +- src/mito2/src/engine/staging_test.rs | 10 ++-- src/mito2/src/region.rs | 8 +-- src/mito2/src/worker/handle_flush.rs | 6 +-- src/standalone/src/information_extension.rs | 2 +- src/store-api/src/region_engine.rs | 33 ++++++++++--- 16 files changed, 99 insertions(+), 57 deletions(-) diff --git a/src/common/meta/src/datanode.rs b/src/common/meta/src/datanode.rs index 899063cf84a..ba4250238d7 100644 --- a/src/common/meta/src/datanode.rs +++ b/src/common/meta/src/datanode.rs @@ -102,11 +102,13 @@ pub struct RegionStat { pub region_manifest: RegionManifestInfo, /// The total bytes written of the region since region opened. pub written_bytes: u64, - /// Total Parquet bytes committed by flushes since the region opened. + /// Total ingested Parquet bytes since the region opened. + /// Currently counts only committed flush output. /// /// Excludes index files, compaction and imported SSTs. Resets on region reopen. - #[serde(default)] - pub flushed_parquet_bytes: u64, + /// Accepts the legacy `flushed_parquet_bytes` key; upgrade metasrv before datanodes. + #[serde(default, alias = "flushed_parquet_bytes")] + pub ingested_parquet_bytes: u64, /// The total query CPU time of the region since region opened. /// /// Unit: nanoseconds. @@ -338,7 +340,7 @@ impl From<&api::v1::meta::RegionStat> for RegionStat { index_size: region_stat.index_size, region_manifest: region_stat.manifest.into(), written_bytes: region_stat.written_bytes, - flushed_parquet_bytes: region_stat.flushed_parquet_bytes, + ingested_parquet_bytes: region_stat.ingested_parquet_bytes, query_cpu_time: region_stat.query_cpu_time, query_scanned_bytes: region_stat.query_scanned_bytes, data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id, @@ -603,7 +605,7 @@ mod tests { file_removed_cnt: 0, }, written_bytes: 0, - flushed_parquet_bytes: 4096, + ingested_parquet_bytes: 4096, query_cpu_time: 10, query_scanned_bytes: 20, data_topic_latest_entry_id: 0, @@ -616,14 +618,14 @@ mod tests { let stat_val = DatanodeStatValue { stats: vec![stat] }; let mut value = serde_json::to_value(stat_val).unwrap(); let region_stat = value[0]["region_stats"][0].as_object_mut().unwrap(); - region_stat.remove("flushed_parquet_bytes"); + region_stat.remove("ingested_parquet_bytes"); region_stat.remove("query_cpu_time"); region_stat.remove("query_scanned_bytes"); let stat_val: DatanodeStatValue = serde_json::from_value(value).unwrap(); let region_stat = &stat_val.stats[0].region_stats[0]; - assert_eq!(region_stat.flushed_parquet_bytes, 0); + assert_eq!(region_stat.ingested_parquet_bytes, 0); assert_eq!(region_stat.query_cpu_time, 0); assert_eq!(region_stat.query_scanned_bytes, 0); } @@ -713,10 +715,10 @@ mod tests { } #[test] - fn test_flushed_parquet_bytes_from_heartbeat() { + fn test_ingested_parquet_bytes_from_heartbeat() { for bytes in [0, 4096, u64::MAX] { let statistic = RegionStatistic { - flushed_parquet_bytes: bytes, + ingested_parquet_bytes: bytes, ..Default::default() }; let mut region = api::v1::meta::RegionStat::default(); @@ -735,12 +737,35 @@ mod tests { }; let stat = Stat::try_from(&request).unwrap(); let stat = &stat.region_stats[0]; - assert_eq!(stat.flushed_parquet_bytes, bytes); + assert_eq!(stat.ingested_parquet_bytes, bytes); let encoded = serde_json::to_vec(stat).unwrap(); + let wire: serde_json::Value = serde_json::from_slice(&encoded).unwrap(); + assert_eq!(wire["ingested_parquet_bytes"], bytes); + assert!(wire.get("flushed_parquet_bytes").is_none()); + let mut old_wire = wire.clone(); + old_wire + .as_object_mut() + .unwrap() + .remove("ingested_parquet_bytes"); + old_wire["flushed_parquet_bytes"] = bytes.into(); + assert_eq!( + &serde_json::from_value::(old_wire).unwrap(), + stat + ); let decoded: RegionStat = serde_json::from_slice(&encoded).unwrap(); assert_eq!(&decoded, stat); let mut legacy = serde_json::to_value(&statistic).unwrap(); + legacy + .as_object_mut() + .unwrap() + .remove("ingested_parquet_bytes"); + legacy["flushed_parquet_bytes"] = bytes.into(); + region.extensions.insert( + REGION_STATISTIC_KEY.to_string(), + serde_json::to_vec(&legacy).unwrap(), + ); + assert_eq!(RegionStat::from(®ion).ingested_parquet_bytes, bytes); legacy .as_object_mut() .unwrap() @@ -749,10 +774,10 @@ mod tests { REGION_STATISTIC_KEY.to_string(), serde_json::to_vec(&legacy).unwrap(), ); - assert_eq!(RegionStat::from(®ion).flushed_parquet_bytes, 0); + assert_eq!(RegionStat::from(®ion).ingested_parquet_bytes, 0); } assert_eq!( - RegionStat::from(&api::v1::meta::RegionStat::default()).flushed_parquet_bytes, + RegionStat::from(&api::v1::meta::RegionStat::default()).ingested_parquet_bytes, 0 ); } diff --git a/src/meta-srv/src/gc/handler.rs b/src/meta-srv/src/gc/handler.rs index 1e490ebab58..ffcfa387c21 100644 --- a/src/meta-srv/src/gc/handler.rs +++ b/src/meta-srv/src/gc/handler.rs @@ -495,7 +495,7 @@ fn dropped_region_stat(region_id: RegionId) -> RegionStat { file_removed_cnt: 0, }, written_bytes: 0, - flushed_parquet_bytes: 0, + ingested_parquet_bytes: 0, query_cpu_time: 0, query_scanned_bytes: 0, data_topic_latest_entry_id: 0, diff --git a/src/meta-srv/src/gc/mock.rs b/src/meta-srv/src/gc/mock.rs index c164c538917..04e70e8d3a7 100644 --- a/src/meta-srv/src/gc/mock.rs +++ b/src/meta-srv/src/gc/mock.rs @@ -453,7 +453,7 @@ fn mock_region_stat( min_timestamp: None, max_timestamp: None, written_bytes: 0, - flushed_parquet_bytes: 0, + ingested_parquet_bytes: 0, query_cpu_time: 0, query_scanned_bytes: 0, } diff --git a/src/meta-srv/src/handler/collect_leader_region_handler.rs b/src/meta-srv/src/handler/collect_leader_region_handler.rs index 0b886a489bf..3ec3d8c69ca 100644 --- a/src/meta-srv/src/handler/collect_leader_region_handler.rs +++ b/src/meta-srv/src/handler/collect_leader_region_handler.rs @@ -91,7 +91,7 @@ mod tests { min_timestamp: None, max_timestamp: None, written_bytes: 0, - flushed_parquet_bytes: 0, + ingested_parquet_bytes: 0, query_cpu_time: 0, query_scanned_bytes: 0, } diff --git a/src/meta-srv/src/handler/failure_handler.rs b/src/meta-srv/src/handler/failure_handler.rs index 243f09caa23..70aa6279a65 100644 --- a/src/meta-srv/src/handler/failure_handler.rs +++ b/src/meta-srv/src/handler/failure_handler.rs @@ -100,7 +100,7 @@ mod tests { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, - flushed_parquet_bytes: 0, + ingested_parquet_bytes: 0, query_cpu_time: 0, query_scanned_bytes: 0, min_timestamp: None, diff --git a/src/meta-srv/src/handler/persist_stats_handler.rs b/src/meta-srv/src/handler/persist_stats_handler.rs index b96af08c12f..f3d7c43452d 100644 --- a/src/meta-srv/src/handler/persist_stats_handler.rs +++ b/src/meta-srv/src/handler/persist_stats_handler.rs @@ -298,7 +298,7 @@ mod tests { file_removed_cnt: 0, }, written_bytes, - flushed_parquet_bytes: 0, + ingested_parquet_bytes: 0, query_cpu_time: 0, query_scanned_bytes: 0, data_topic_latest_entry_id: 200, diff --git a/src/meta-srv/src/handler/region_lease_handler.rs b/src/meta-srv/src/handler/region_lease_handler.rs index e4338720216..a859bf18327 100644 --- a/src/meta-srv/src/handler/region_lease_handler.rs +++ b/src/meta-srv/src/handler/region_lease_handler.rs @@ -180,7 +180,7 @@ mod test { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, - flushed_parquet_bytes: 0, + ingested_parquet_bytes: 0, query_cpu_time: 0, query_scanned_bytes: 0, min_timestamp: None, diff --git a/src/meta-srv/src/selector/weight_compute.rs b/src/meta-srv/src/selector/weight_compute.rs index d39fda3b8e5..eda2a0357ad 100644 --- a/src/meta-srv/src/selector/weight_compute.rs +++ b/src/meta-srv/src/selector/weight_compute.rs @@ -200,7 +200,7 @@ mod tests { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, - flushed_parquet_bytes: 0, + ingested_parquet_bytes: 0, query_cpu_time: 0, query_scanned_bytes: 0, min_timestamp: None, @@ -235,7 +235,7 @@ mod tests { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, - flushed_parquet_bytes: 0, + ingested_parquet_bytes: 0, query_cpu_time: 0, query_scanned_bytes: 0, min_timestamp: None, @@ -270,7 +270,7 @@ mod tests { data_topic_latest_entry_id: 0, metadata_topic_latest_entry_id: 0, written_bytes: 0, - flushed_parquet_bytes: 0, + ingested_parquet_bytes: 0, query_cpu_time: 0, query_scanned_bytes: 0, min_timestamp: None, diff --git a/src/metric-engine/src/utils.rs b/src/metric-engine/src/utils.rs index 64834a3f091..43ec4a13586 100644 --- a/src/metric-engine/src/utils.rs +++ b/src/metric-engine/src/utils.rs @@ -80,8 +80,8 @@ fn merge_region_statistic( metadata_manifest_version: metadata_stat.manifest.data_manifest_version(), }, written_bytes: metadata_stat.written_bytes + data_stat.written_bytes, - flushed_parquet_bytes: metadata_stat.flushed_parquet_bytes - + data_stat.flushed_parquet_bytes, + ingested_parquet_bytes: metadata_stat.ingested_parquet_bytes + + data_stat.ingested_parquet_bytes, query_cpu_time: data_stat.query_cpu_time, query_scanned_bytes: data_stat.query_scanned_bytes, data_topic_latest_entry_id: data_stat.data_topic_latest_entry_id, @@ -153,17 +153,17 @@ mod tests { } #[test] - fn merge_region_statistic_sums_flushed_parquet_bytes() { + fn merge_region_statistic_sums_ingested_parquet_bytes() { let metadata_stat = RegionStatistic { - flushed_parquet_bytes: 123, + ingested_parquet_bytes: 123, ..Default::default() }; let data_stat = RegionStatistic { - flushed_parquet_bytes: 456, + ingested_parquet_bytes: 456, ..Default::default() }; assert_eq!( - merge_region_statistic(&metadata_stat, &data_stat).flushed_parquet_bytes, + merge_region_statistic(&metadata_stat, &data_stat).ingested_parquet_bytes, 579 ); } diff --git a/src/mito2/src/engine/flush_test.rs b/src/mito2/src/engine/flush_test.rs index 40b4cce0dbc..399b5857700 100644 --- a/src/mito2/src/engine/flush_test.rs +++ b/src/mito2/src/engine/flush_test.rs @@ -352,7 +352,7 @@ async fn test_manual_flush_with_format(flat_format: bool) { } #[tokio::test] -async fn test_flushed_parquet_bytes() { +async fn test_ingested_parquet_bytes() { for flat_format in [false, true] { let mut env = TestEnv::with_prefix("flushed-parquet-bytes").await; let engine = env @@ -374,7 +374,7 @@ async fn test_flushed_parquet_bytes() { .await .unwrap(); let statistic = || engine.get_region_statistic(region_id).unwrap(); - assert_eq!(statistic().flushed_parquet_bytes, 0); + assert_eq!(statistic().ingested_parquet_bytes, 0); let mut previous_bytes = 0; for start in [0, 10] { @@ -387,7 +387,7 @@ async fn test_flushed_parquet_bytes() { }, ) .await; - assert_eq!(statistic().flushed_parquet_bytes, previous_bytes); + assert_eq!(statistic().ingested_parquet_bytes, previous_bytes); flush_region(&engine, region_id, None).await; let manifest = engine .get_region(region_id) @@ -397,11 +397,11 @@ async fn test_flushed_parquet_bytes() { .await; let parquet_bytes: u64 = manifest.files.values().map(|file| file.file_size).sum(); assert!(parquet_bytes > previous_bytes); - assert_eq!(statistic().flushed_parquet_bytes, parquet_bytes); + assert_eq!(statistic().ingested_parquet_bytes, parquet_bytes); previous_bytes = parquet_bytes; } flush_region(&engine, region_id, None).await; - assert_eq!(statistic().flushed_parquet_bytes, previous_bytes); + assert_eq!(statistic().ingested_parquet_bytes, previous_bytes); // A generic edit can add the same file metadata without producing a flush. let file = engine @@ -430,7 +430,7 @@ async fn test_flushed_parquet_bytes() { ) .await .unwrap(); - assert_eq!(statistic().flushed_parquet_bytes, previous_bytes); + assert_eq!(statistic().ingested_parquet_bytes, previous_bytes); engine .handle_request( @@ -446,11 +446,11 @@ async fn test_flushed_parquet_bytes() { ) .await .unwrap(); - assert_eq!(statistic().flushed_parquet_bytes, previous_bytes); + assert_eq!(statistic().ingested_parquet_bytes, previous_bytes); assert_eq!(statistic().sst_num, 1); reopen_region(&engine, region_id, table_dir, true, HashMap::new()).await; assert!(statistic().sst_size > 0); - assert_eq!(statistic().flushed_parquet_bytes, 0); + assert_eq!(statistic().ingested_parquet_bytes, 0); } } diff --git a/src/mito2/src/engine/object_store_wal_recovery_test.rs b/src/mito2/src/engine/object_store_wal_recovery_test.rs index 0c028e067db..9313f24fe37 100644 --- a/src/mito2/src/engine/object_store_wal_recovery_test.rs +++ b/src/mito2/src/engine/object_store_wal_recovery_test.rs @@ -1866,7 +1866,7 @@ async fn test_flush_does_not_publish_a_frontier_for_a_lost_enqueued_backlog() { engine .get_region_statistic(REGION_A) .unwrap() - .flushed_parquet_bytes, + .ingested_parquet_bytes, 0 ); assert!(store.stop().await.is_err()); diff --git a/src/mito2/src/engine/staging_test.rs b/src/mito2/src/engine/staging_test.rs index 08c272be6ab..5594e5aa7bb 100644 --- a/src/mito2/src/engine/staging_test.rs +++ b/src/mito2/src/engine/staging_test.rs @@ -818,7 +818,7 @@ async fn test_staging_exit_success_with_manifests_with_format(flat_format: bool) .map(|file| file.file_size) .sum(); assert_eq!( - region.region_statistic().flushed_parquet_bytes, + region.region_statistic().ingested_parquet_bytes, parquet_bytes ); assert_eq!( @@ -883,8 +883,8 @@ async fn test_staging_exit_success_with_manifests_with_format(flat_format: bool) assert_eq!(sst_entries.iter().filter(|e| e.visible).count(), 1); assert_eq!(sst_entries.iter().filter(|e| !e.visible).count(), 2); - let flushed_parquet_bytes = region.region_statistic().flushed_parquet_bytes; - assert!(flushed_parquet_bytes > 0); + let ingested_parquet_bytes = region.region_statistic().ingested_parquet_bytes; + assert!(ingested_parquet_bytes > 0); // Streaming inspection must retain staging entries and their visibility. { @@ -1004,8 +1004,8 @@ async fn test_staging_exit_success_with_manifests_with_format(flat_format: bool) ); assert_eq!( - region.region_statistic().flushed_parquet_bytes, - flushed_parquet_bytes + region.region_statistic().ingested_parquet_bytes, + ingested_parquet_bytes ); // Verify staging manifests have been cleared diff --git a/src/mito2/src/region.rs b/src/mito2/src/region.rs index 2e9f1945bca..ed24421d023 100644 --- a/src/mito2/src/region.rs +++ b/src/mito2/src/region.rs @@ -203,7 +203,7 @@ pub(crate) struct RegionStats { /// The total bytes written to the region. pub(crate) written_bytes: Arc, /// Parquet bytes committed by flushes since this region instance was opened. - pub(crate) flushed_parquet_bytes: AtomicU64, + pub(crate) ingested_parquet_bytes: AtomicU64, /// The total query CPU time of the region in nanoseconds. pub(crate) query_cpu_time: Arc, /// The total scanned bytes of the region. @@ -214,7 +214,7 @@ impl RegionStats { pub(crate) fn new() -> Self { Self { written_bytes: Arc::new(AtomicU64::new(0)), - flushed_parquet_bytes: AtomicU64::new(0), + ingested_parquet_bytes: AtomicU64::new(0), query_cpu_time: Arc::new(AtomicU64::new(0)), query_scanned_bytes: Arc::new(AtomicU64::new(0)), } @@ -790,9 +790,9 @@ impl MitoRegion { data_topic_latest_entry_id: topic_latest_entry_id, metadata_topic_latest_entry_id: topic_latest_entry_id, written_bytes, - flushed_parquet_bytes: self + ingested_parquet_bytes: self .region_stats - .flushed_parquet_bytes + .ingested_parquet_bytes .load(Ordering::Relaxed), query_cpu_time, query_scanned_bytes, diff --git a/src/mito2/src/worker/handle_flush.rs b/src/mito2/src/worker/handle_flush.rs index 690d946fad7..09c0d66344e 100644 --- a/src/mito2/src/worker/handle_flush.rs +++ b/src/mito2/src/worker/handle_flush.rs @@ -420,7 +420,7 @@ impl RegionWorkerLoop { // The flush has committed its manifest, including in staging mode. Count only // its Parquet output, before WAL cleanup (which can fail after the commit). - let flushed_parquet_bytes = request + let ingested_parquet_bytes = request .edit .files_to_add .iter() @@ -428,8 +428,8 @@ impl RegionWorkerLoop { .sum(); region .region_stats - .flushed_parquet_bytes - .fetch_add(flushed_parquet_bytes, Ordering::Relaxed); + .ingested_parquet_bytes + .fetch_add(ingested_parquet_bytes, Ordering::Relaxed); region.update_flush_millis(); // Update topic latest entry id as soon as possible after flush to make sure the prunable entry id is updated timely, diff --git a/src/standalone/src/information_extension.rs b/src/standalone/src/information_extension.rs index 44fc266c5be..083fabadeb8 100644 --- a/src/standalone/src/information_extension.rs +++ b/src/standalone/src/information_extension.rs @@ -138,7 +138,7 @@ impl InformationExtension for StandaloneInformationExtension { data_topic_latest_entry_id: region_stat.data_topic_latest_entry_id, metadata_topic_latest_entry_id: region_stat.metadata_topic_latest_entry_id, written_bytes: region_stat.written_bytes, - flushed_parquet_bytes: region_stat.flushed_parquet_bytes, + ingested_parquet_bytes: region_stat.ingested_parquet_bytes, query_cpu_time: region_stat.query_cpu_time, query_scanned_bytes: region_stat.query_scanned_bytes, min_timestamp: region_stat.min_timestamp, diff --git a/src/store-api/src/region_engine.rs b/src/store-api/src/region_engine.rs index dc9124d36f8..48a802d0c98 100644 --- a/src/store-api/src/region_engine.rs +++ b/src/store-api/src/region_engine.rs @@ -603,12 +603,14 @@ pub struct RegionStatistic { #[serde(default)] /// The total bytes written of the region since region opened. pub written_bytes: u64, - /// Total Parquet bytes committed by flushes since the region opened. + /// Total ingested Parquet bytes since the region opened. + /// Currently counts only committed flush output. /// /// Includes staging flushes, but excludes index files, compaction and imported SSTs. /// Resets when the region instance is recreated; manifest replay does not increment it. - #[serde(default)] - pub flushed_parquet_bytes: u64, + /// Accepts the legacy `flushed_parquet_bytes` key; upgrade metasrv before datanodes. + #[serde(default, alias = "flushed_parquet_bytes")] + pub ingested_parquet_bytes: u64, /// The total query CPU time of the region since region opened. /// /// Unit: nanoseconds. @@ -804,24 +806,39 @@ mod tests { "written_bytes": 11 }"#; let mut statistic = RegionStatistic::deserialize_from_slice(legacy_bytes).unwrap(); - assert_eq!(statistic.flushed_parquet_bytes, 0); - statistic.flushed_parquet_bytes = 1234; + assert_eq!(statistic.ingested_parquet_bytes, 0); + statistic.ingested_parquet_bytes = 1234; let encoded = statistic.serialize_to_vec().unwrap(); let decoded = RegionStatistic::deserialize_from_slice(&encoded).unwrap(); - assert_eq!(decoded.flushed_parquet_bytes, 1234); + assert_eq!(decoded.ingested_parquet_bytes, 1234); let legacy: serde_json::Value = serde_json::from_slice(legacy_bytes).unwrap(); let mut current: serde_json::Value = serde_json::from_slice(&encoded).unwrap(); + assert_eq!(current["ingested_parquet_bytes"], 1234); + assert!(current.get("flushed_parquet_bytes").is_none()); + let mut old_wire = current.clone(); + let bytes = old_wire + .as_object_mut() + .unwrap() + .remove("ingested_parquet_bytes") + .unwrap(); + old_wire["flushed_parquet_bytes"] = bytes; + assert_eq!( + serde_json::from_value::(old_wire) + .unwrap() + .ingested_parquet_bytes, + 1234 + ); for (key, value) in legacy.as_object().unwrap() { assert_eq!(¤t[key], value); } current .as_object_mut() .unwrap() - .remove("flushed_parquet_bytes"); + .remove("ingested_parquet_bytes"); assert_eq!( serde_json::from_value::(current) .unwrap() - .flushed_parquet_bytes, + .ingested_parquet_bytes, 0 ); }