refactor: rename flushed_parquet_bytes to ingested_parquet_bytes (#10258)

Signed-off-by: WenyXu <wenymedia@gmail.com>

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::<RegionStat>(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(&region).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(&region).flushed_parquet_bytes, 0);
+            assert_eq!(RegionStat::from(&region).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
         );
     }

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,

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,
     }

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,
         }

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,

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,

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,

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,

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
         );
     }

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);
     }
 }
 

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());

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

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<AtomicU64>,
     /// 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<AtomicU64>,
     /// 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,

src/mito2/src/worker/handle_flush.rs

@@ -420,7 +420,7 @@ impl<S: LogStore> RegionWorkerLoop<S> {
 
         // 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<S: LogStore> RegionWorkerLoop<S> {
             .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,

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,

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::<RegionStatistic>(old_wire)
+                .unwrap()
+                .ingested_parquet_bytes,
+            1234
+        );
         for (key, value) in legacy.as_object().unwrap() {
             assert_eq!(&current[key], value);
         }
         current
             .as_object_mut()
             .unwrap()
-            .remove("flushed_parquet_bytes");
+            .remove("ingested_parquet_bytes");
         assert_eq!(
             serde_json::from_value::<RegionStatistic>(current)
                 .unwrap()
-                .flushed_parquet_bytes,
+                .ingested_parquet_bytes,
             0
         );
     }