From fc5831043f28ca88eff96b75b3da36a027b15833 Mon Sep 17 00:00:00 2001 From: jeremyhi Date: Mon, 12 Oct 2026 10:09:26 +0000 Subject: [PATCH] feat(cli): simplify packed metric exports with a boolean flag (#9531) * feat(cli): select packed metric export with the layout option Signed-off-by: jeremyhi * test(cli): inspect the packed capability error cause Signed-off-by: jeremyhi * test(cli): follow the standard capability error source Signed-off-by: jeremyhi * refactor(cli): clarify the metric export gate helper name Signed-off-by: jeremyhi * feat(cli): use a boolean flag for packed metric exports Signed-off-by: jeremyhi --------- Signed-off-by: jeremyhi --- docs/rfcs/2026-09-11-metric-export-import.md | 6 + src/cli/src/data/export_v2.rs | 5 +- src/cli/src/data/export_v2/command.rs | 137 ++++++++++-------- src/cli/src/data/export_v2/coordinator.rs | 6 +- src/cli/src/data/export_v2/data.rs | 24 ++- .../tests/export_logical_tables.rs | 104 +++++++++---- 6 files changed, 183 insertions(+), 99 deletions(-) diff --git a/docs/rfcs/2026-09-11-metric-export-import.md b/docs/rfcs/2026-09-11-metric-export-import.md index 57212ea53cb..bf66df932b9 100644 --- a/docs/rfcs/2026-09-11-metric-export-import.md +++ b/docs/rfcs/2026-09-11-metric-export-import.md @@ -193,6 +193,12 @@ Packed Metric exports write manifest version 2 with a required data-layout marke exports can keep version 1. A version-2 snapshot indexes all data in its schema chunks, including ordinary tables stored as standalone Parquet objects. +`export-v2 create --pack-metric-data` selects shared physical scans and +packed output. Omitting the flag keeps the default version-1 export path; +schema-only snapshots also retain version 1. The source frontend or standalone +server must enable `experimental_metric_export = true`. Resume derives the layout +from the manifest, so packed snapshots do not need the flag repeated. + Before creating or replacing a snapshot, the CLI checks the source server's explicit packed-export capability. Before target DDL or data writes, the importer checks the manifest version/layout and the target's packed-import capability. diff --git a/src/cli/src/data/export_v2.rs b/src/cli/src/data/export_v2.rs index 9887e525030..8785bcf3e31 100644 --- a/src/cli/src/data/export_v2.rs +++ b/src/cli/src/data/export_v2.rs @@ -38,10 +38,11 @@ //! --end-time 2025-01-31T23:59:59Z //! ``` //! -//! `--experimental-metric-export` enables shared Metric physical scans for Parquet. +//! `--pack-metric-data` enables shared Metric physical scans for Parquet +//! and writes version-2 packed snapshots. //! Enable `experimental_metric_export = true` on the frontend or standalone server //! and use an endpoint whose frontends all support and enable this option. -//! Add `--metric-data-layout packed` to write version-2 packed snapshots. +//! Resume reads the layout from the existing manifest. //! This requires the server packed-export capability; schema-only exports remain version 1. //! The command checks the server capability before modifying the snapshot. //! Each snapshot path belongs to one export task. Resume requires stable source diff --git a/src/cli/src/data/export_v2/command.rs b/src/cli/src/data/export_v2/command.rs index 48779f3c435..fe38c6b7c7e 100644 --- a/src/cli/src/data/export_v2/command.rs +++ b/src/cli/src/data/export_v2/command.rs @@ -292,14 +292,11 @@ pub struct ExportCreateCommand { #[clap(long, value_enum, default_value = "parquet")] format: DataFormat, - /// Use shared Metric physical scans (Parquet only). Resume requires that the previous - /// export and storage writes have ended; an HTTP timeout does not establish this. + /// Use shared Metric physical scans and packed objects (Parquet only). + /// Resume uses the manifest layout and requires that previous export and storage + /// writes have ended; an HTTP timeout does not establish this. #[clap(long)] - experimental_metric_export: bool, - - /// Store independent Metric Parquet streams in shared packed objects. - #[clap(long, value_parser = ["packed"], requires = "experimental_metric_export")] - metric_data_layout: Option, + pack_metric_data: bool, /// Delete existing snapshot and recreate. #[clap(long)] @@ -350,10 +347,7 @@ impl ExportCreateCommand { let time_range = TimeRange::parse(self.start_time.as_deref(), self.end_time.as_deref()) .map_err(BoxedError::new)?; - if self.metric_data_layout.is_some() - && time_range.is_bounded() - && time_range.start == time_range.end - { + if self.pack_metric_data && time_range.is_bounded() && time_range.start == time_range.end { return crate::error::InvalidArgumentsSnafu { msg: "Packed export requires --start-time to be earlier than --end-time", } @@ -413,40 +407,18 @@ impl ExportCreateCommand { ); // The filesystem storage constructor can create the snapshot root. - if self.experimental_metric_export { - if self.format != DataFormat::Parquet { - return crate::data::export_v2::error::MetricExportUnavailableSnafu - .fail() - .map_err(BoxedError::new); - } - let capability = database_client - .sql_response( - "SHOW VARIABLES experimental_metric_export", - DEFAULT_SCHEMA_NAME, - ) + if self.pack_metric_data { + require_metric_export_enabled(&database_client, self.format) .await - .context(DatabaseSnafu) .map_err(BoxedError::new)?; - let expected_schema = OutputSchema::new(vec![ColumnSchema::new( - "EXPERIMENTAL_METRIC_EXPORT".to_string(), - "String".to_string(), - )]); - if !matches!(capability.output(), [GreptimeQueryOutput::Records(records)] - if records.schema() == &expected_schema - && records.rows() == &vec![vec![Value::String("true".to_string())]]) - { - return crate::data::export_v2::error::MetricExportUnavailableSnafu - .fail() - .map_err(BoxedError::new); + if !self.schema_only { + database_client + .require_packed_export() + .await + .context(DatabaseSnafu) + .map_err(BoxedError::new)?; } } - if self.metric_data_layout.is_some() && !self.schema_only { - database_client - .require_packed_export() - .await - .context(DatabaseSnafu) - .map_err(BoxedError::new)?; - } let storage = OpenDalStorage::from_uri(&self.to, &self.storage).map_err(BoxedError::new)?; Ok(Box::new(ExportCreate { @@ -455,8 +427,7 @@ impl ExportCreateCommand { schemas, schema_only: self.schema_only, format: self.format, - experimental_metric_export: self.experimental_metric_export, - packed: self.metric_data_layout.is_some(), + packed: self.pack_metric_data, force: self.force, time_range, chunk_time_window: self.chunk_time_window, @@ -484,7 +455,6 @@ struct ExportConfig { schemas: Option>, schema_only: bool, format: DataFormat, - experimental_metric_export: bool, packed: bool, force: bool, time_range: TimeRange, @@ -538,13 +508,11 @@ impl ExportCreate { validate_resume_config(&manifest, &self.config)?; if manifest.is_packed() { + require_metric_export_enabled(&self.database_client, manifest.format).await?; self.database_client .require_packed_export() .await .context(DatabaseSnafu)?; - if !self.config.experimental_metric_export { - return crate::data::export_v2::error::MetricExportUnavailableSnafu.fail(); - } } info!( @@ -573,7 +541,6 @@ impl ExportCreate { storage_config: &self.config.storage_config, parallelism: self.config.parallelism, chunk_parallelism: self.config.chunk_parallelism, - experimental_metric_export: self.config.experimental_metric_export, resume: true, }, progress.as_ref(), @@ -608,7 +575,7 @@ impl ExportCreate { manifest.version = 2; manifest.data_layout = Some(common_datasource::packed_snapshot::PACKED_LAYOUT.into()); } - if self.config.experimental_metric_export { + if manifest.is_packed() { for chunk in &manifest.chunks { self.storage .prepare_export_chunk(&schema_names, chunk.id, false, manifest.is_packed()) @@ -651,7 +618,6 @@ impl ExportCreate { storage_config: &self.config.storage_config, parallelism: self.config.parallelism, chunk_parallelism: self.config.chunk_parallelism, - experimental_metric_export: self.config.experimental_metric_export, resume: false, }, progress.as_ref(), @@ -761,10 +727,34 @@ impl ExportCreate { } } +async fn require_metric_export_enabled(client: &DatabaseClient, format: DataFormat) -> Result<()> { + if format != DataFormat::Parquet { + return crate::data::export_v2::error::MetricExportUnavailableSnafu.fail(); + } + let capability = client + .sql_response( + "SHOW VARIABLES experimental_metric_export", + DEFAULT_SCHEMA_NAME, + ) + .await + .context(DatabaseSnafu)?; + let expected_schema = OutputSchema::new(vec![ColumnSchema::new( + "EXPERIMENTAL_METRIC_EXPORT".to_string(), + "String".to_string(), + )]); + if !matches!(capability.output(), [GreptimeQueryOutput::Records(records)] + if records.schema() == &expected_schema + && records.rows() == &vec![vec![Value::String("true".to_string())]]) + { + return crate::data::export_v2::error::MetricExportUnavailableSnafu.fail(); + } + Ok(()) +} + fn validate_resume_config(manifest: &Manifest, config: &ExportConfig) -> Result<()> { if config.packed && !manifest.schema_only && !manifest.is_packed() { return ResumeConfigMismatchSnafu { - field: "metric_data_layout", + field: "data_layout", existing: "standalone".to_string(), requested: "packed".to_string(), } @@ -1683,6 +1673,44 @@ mod tests { ); } + #[tokio::test] + async fn test_pack_metric_data_arguments() { + let directory = tempfile::tempdir().unwrap(); + let snapshot = directory.path().join("snapshot"); + let uri = url::Url::from_file_path(&snapshot).unwrap(); + let args = [ + "export-v2-create", + "--addr", + "127.0.0.1:1", + "--to", + uri.as_str(), + ]; + assert!(!ExportCreateCommand::parse_from(args).pack_metric_data); + let packed = + ExportCreateCommand::parse_from(args.into_iter().chain(["--pack-metric-data"])); + assert!(packed.pack_metric_data); + for removed in [ + &["--experimental-metric-export"][..], + &["--metric-data-layout", "packed"][..], + ] { + let error = ExportCreateCommand::try_parse_from( + args.into_iter().chain(removed.iter().copied()), + ) + .unwrap_err(); + assert_eq!(error.kind(), clap::error::ErrorKind::UnknownArgument); + } + for format in ["csv", "json"] { + let command = ExportCreateCommand::parse_from(args.into_iter().chain([ + "--pack-metric-data", + "--format", + format, + ])); + let error = command.build().await.err().unwrap(); + assert!(error.to_string().contains("Metric export requires Parquet")); + assert!(!snapshot.exists()); + } + } + #[test] fn test_chunk_parallelism_parses_valid_value() { let cmd = ExportCreateCommand::parse_from([ @@ -1755,7 +1783,6 @@ mod tests { schemas: None, schema_only: false, format: DataFormat::Parquet, - experimental_metric_export: false, packed: false, force: false, time_range: TimeRange::unbounded(), @@ -1794,7 +1821,6 @@ mod tests { ]), schema_only: false, format: DataFormat::Parquet, - experimental_metric_export: false, packed: false, force: false, time_range: TimeRange::unbounded(), @@ -1812,7 +1838,7 @@ mod tests { validate_resume_config(&manifest, &config) .unwrap_err() .to_string() - .contains("metric_data_layout") + .contains("data_layout") ); manifest.version = 2; manifest.data_layout = Some(common_datasource::packed_snapshot::PACKED_LAYOUT.into()); @@ -1840,7 +1866,6 @@ mod tests { schemas: None, schema_only: false, format: DataFormat::Parquet, - experimental_metric_export: false, packed: false, force: false, time_range, @@ -1875,7 +1900,6 @@ mod tests { schemas: None, schema_only: false, format: DataFormat::Csv, - experimental_metric_export: false, packed: false, force: false, time_range: TimeRange::unbounded(), @@ -1912,7 +1936,6 @@ mod tests { schemas: None, schema_only: false, format: DataFormat::Parquet, - experimental_metric_export: false, packed: false, force: false, time_range: TimeRange::new(Some(start), Some(start)), diff --git a/src/cli/src/data/export_v2/coordinator.rs b/src/cli/src/data/export_v2/coordinator.rs index c6181e51211..e18014af32f 100644 --- a/src/cli/src/data/export_v2/coordinator.rs +++ b/src/cli/src/data/export_v2/coordinator.rs @@ -39,7 +39,6 @@ struct ExportContext<'a> { schemas: Vec, format: DataFormat, parallelism: usize, - experimental_metric_export: bool, packed: bool, resume: bool, } @@ -49,7 +48,6 @@ pub struct ExportDataOptions<'a> { pub storage_config: &'a ObjectStoreConfig, pub parallelism: usize, pub chunk_parallelism: usize, - pub experimental_metric_export: bool, pub resume: bool, } @@ -74,7 +72,6 @@ pub async fn export_data( format: manifest.format, packed: manifest.is_packed(), parallelism: options.parallelism, - experimental_metric_export: options.experimental_metric_export, resume: options.resume, }; @@ -278,7 +275,7 @@ async fn export_chunk( chunk_id: u32, time_range: TimeRange, ) -> Result> { - if context.experimental_metric_export { + if context.packed { context .storage .prepare_export_chunk(&context.schemas, chunk_id, context.resume, context.packed) @@ -290,7 +287,6 @@ async fn export_chunk( format: context.format, time_range, parallelism: context.parallelism, - experimental_metric_export: context.experimental_metric_export, packed: context.packed, }; diff --git a/src/cli/src/data/export_v2/data.rs b/src/cli/src/data/export_v2/data.rs index 53c3899f59e..dac094c0dda 100644 --- a/src/cli/src/data/export_v2/data.rs +++ b/src/cli/src/data/export_v2/data.rs @@ -30,7 +30,6 @@ pub(super) struct CopyOptions { pub(super) format: DataFormat, pub(super) time_range: TimeRange, pub(super) parallelism: usize, - pub(crate) experimental_metric_export: bool, pub(super) packed: bool, } @@ -218,10 +217,8 @@ pub(crate) async fn execute_copy_database_from( fn build_with_options(options: &CopyOptions) -> String { let mut parts = vec![format!("FORMAT='{}'", options.format)]; - if options.experimental_metric_export { - parts.push("experimental_metric_export='true'".to_string()); - } if options.packed { + parts.push("experimental_metric_export='true'".to_string()); parts.push("metric_data_layout='packed'".to_string()); } if let Some(start) = options.time_range.start { @@ -430,6 +427,25 @@ mod tests { use super::*; use crate::common::{PrefixedAzblobConnection, PrefixedGcsConnection, PrefixedOssConnection}; + #[test] + fn test_packed_copy_options() { + let mut options = CopyOptions { + format: DataFormat::Parquet, + time_range: TimeRange::unbounded(), + parallelism: 2, + packed: false, + }; + assert_eq!( + build_with_options(&options), + "FORMAT='parquet', PARALLELISM=2" + ); + options.packed = true; + assert_eq!( + build_with_options(&options), + "FORMAT='parquet', experimental_metric_export='true', metric_data_layout='packed', PARALLELISM=2" + ); + } + #[test] fn test_build_oss_connection_includes_endpoint() { let storage = ObjectStoreConfig { diff --git a/tests-integration/tests/export_logical_tables.rs b/tests-integration/tests/export_logical_tables.rs index 13f9384cb5c..101b2b6f476 100644 --- a/tests-integration/tests/export_logical_tables.rs +++ b/tests-integration/tests/export_logical_tables.rs @@ -1713,9 +1713,7 @@ async fn metric_export_v2_cli_roundtrip(s3: bool, packed: bool, older: bool) { uri.as_str(), "--schemas", "public", - "--experimental-metric-export", - "--metric-data-layout", - "packed", + "--pack-metric-data", "--no-proxy", "--start-time", "1970-01-01T00:00:00Z", @@ -1792,7 +1790,6 @@ async fn metric_export_v2_cli_roundtrip(s3: bool, packed: bool, older: bool) { } else { "public,z_later" }, - "--experimental-metric-export", "--no-proxy", "--start-time", "1970-01-01T00:00:00Z", @@ -1805,7 +1802,7 @@ async fn metric_export_v2_cli_roundtrip(s3: bool, packed: bool, older: bool) { ]; args.extend(storage_args.iter().map(String::as_str)); if packed { - args.extend(["--metric-data-layout", "packed"]); + args.push("--pack-metric-data"); } assert!(run_data_cli(&args).await.is_err()); let before: cli::export_v2::manifest::Manifest = @@ -1838,25 +1835,28 @@ async fn metric_export_v2_cli_roundtrip(s3: bool, packed: bool, older: bool) { .await .unwrap() ); - store - .write("data/public/2/unknown.txt", "keep") - .await - .unwrap(); - let copies = faults.copies.load(std::sync::atomic::Ordering::SeqCst); - assert!(run_data_cli(&args).await.is_err()); - assert_eq!( - faults.copies.load(std::sync::atomic::Ordering::SeqCst), - copies - ); - assert_eq!( + if packed { + args.pop(); store - .read("data/public/2/unknown.txt") + .write("data/public/2/unknown.txt", "keep") .await - .unwrap() - .to_vec(), - b"keep" - ); - store.delete("data/public/2/unknown.txt").await.unwrap(); + .unwrap(); + let copies = faults.copies.load(std::sync::atomic::Ordering::SeqCst); + assert!(run_data_cli(&args).await.is_err()); + assert_eq!( + faults.copies.load(std::sync::atomic::Ordering::SeqCst), + copies + ); + assert_eq!( + store + .read("data/public/2/unknown.txt") + .await + .unwrap() + .to_vec(), + b"keep" + ); + store.delete("data/public/2/unknown.txt").await.unwrap(); + } run_data_cli(&args).await.unwrap(); let after: cli::export_v2::manifest::Manifest = serde_json::from_slice(&store.read("manifest.json").await.unwrap().to_vec()).unwrap(); @@ -2031,9 +2031,7 @@ async fn metric_export_v2_cli_roundtrip(s3: bool, packed: bool, older: bool) { "--schemas", "public", "--schema-only", - "--experimental-metric-export", - "--metric-data-layout", - "packed", + "--pack-metric-data", "--no-proxy", "--progress", "never", @@ -2115,7 +2113,7 @@ async fn metric_export_v2_disabled_and_legacy_cli() { "--to", &uri, "--force", - "--experimental-metric-export", + "--pack-metric-data", "--no-proxy", "--progress", "never", @@ -2156,6 +2154,7 @@ async fn metric_export_v2_disabled_and_legacy_cli() { #[tokio::test] async fn metric_export_v2_refuses_missing_or_malformed_capability_before_force() { use axum::http::StatusCode; + use common_error::ext::ErrorExt; use serde_json::json; let column = json!({"name": "EXPERIMENTAL_METRIC_EXPORT", "data_type": "String"}); let records = |rows, columns| { @@ -2168,6 +2167,7 @@ async fn metric_export_v2_refuses_missing_or_malformed_capability_before_force() let mut responses = [ (StatusCode::BAD_REQUEST, json!([])), (StatusCode::OK, json!([])), + (StatusCode::OK, json!([["false"]])), (StatusCode::OK, json!([[true]])), (StatusCode::OK, json!([["true", "extra"]])), (StatusCode::OK, json!([["true"], ["true"]])), @@ -2205,9 +2205,11 @@ async fn metric_export_v2_refuses_missing_or_malformed_capability_before_force() )]), ), ]); + responses.push((StatusCode::OK, json!([valid.clone()]))); for (status, output) in responses { - let app = - axum::Router::new().route( + let gate_enabled = output == json!([valid.clone()]); + let app = axum::Router::new() + .route( "/v1/sql", axum::routing::post( move |axum::Form(form): axum::Form< @@ -2223,6 +2225,10 @@ async fn metric_export_v2_refuses_missing_or_malformed_capability_before_force() ) }, ), + ) + .route( + "/v1/capabilities", + axum::routing::get(|| async { axum::Json(json!({})) }), ); let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let addr = listener.local_addr().unwrap().to_string(); @@ -2237,14 +2243,17 @@ async fn metric_export_v2_refuses_missing_or_malformed_capability_before_force() &addr, "--to", &uri, - "--experimental-metric-export", + "--pack-metric-data", "--force", "--no-proxy", "--progress", "never", ]; let error = run_data_cli(&args).await.unwrap_err(); - if status == StatusCode::OK { + let cause = error.root_cause().unwrap_or(&error).to_string(); + if gate_enabled { + assert!(cause.contains("metric_packed_export"), "{error:?}"); + } else if status == StatusCode::OK { assert!( error.to_string().contains("Metric export requires"), "{error}" @@ -2255,7 +2264,40 @@ async fn metric_export_v2_refuses_missing_or_malformed_capability_before_force() let manifest = snapshot.join("manifest.json"); std::fs::write(&manifest, b"preserve").unwrap(); assert!(run_data_cli(&args).await.is_err()); - assert_eq!(std::fs::read(manifest).unwrap(), b"preserve"); + assert_eq!(std::fs::read(&manifest).unwrap(), b"preserve"); + + let mut packed = cli::export_v2::manifest::Manifest::new_for_export( + "greptime".into(), + vec!["public".into()], + false, + cli::export_v2::manifest::TimeRange::unbounded(), + cli::export_v2::manifest::DataFormat::Parquet, + None, + ) + .unwrap(); + packed.version = 2; + packed.data_layout = Some(common_datasource::packed_snapshot::PACKED_LAYOUT.into()); + packed.validate_layout().unwrap(); + let original = serde_json::to_vec(&packed).unwrap(); + std::fs::write(&manifest, &original).unwrap(); + let chunk = snapshot.join("data/public/1"); + std::fs::create_dir_all(&chunk).unwrap(); + let part = chunk.join("pack-000000.bin"); + std::fs::write(&part, b"unfinished").unwrap(); + let resume = args + .into_iter() + .filter(|arg| !matches!(*arg, "--force" | "--pack-metric-data")) + .collect::>(); + let resume_error = run_data_cli(&resume).await.unwrap_err(); + assert_eq!( + resume_error + .root_cause() + .unwrap_or(&resume_error) + .to_string(), + cause + ); + assert_eq!(std::fs::read(&manifest).unwrap(), original); + assert_eq!(std::fs::read(&part).unwrap(), b"unfinished"); server.abort(); } }