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 <fengjiachun@gmail.com>

* test(cli): inspect the packed capability error cause

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* test(cli): follow the standard capability error source

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* refactor(cli): clarify the metric export gate helper name

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

* feat(cli): use a boolean flag for packed metric exports

Signed-off-by: jeremyhi <fengjiachun@gmail.com>

---------

Signed-off-by: jeremyhi <fengjiachun@gmail.com>
This commit is contained in:
jeremyhi
2026-10-12 10:09:26 +00:00
committed by GitHub
parent 2df33380a7
commit fc5831043f
6 changed files with 183 additions and 99 deletions
@@ -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.
+3 -2
View File
@@ -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
+80 -57
View File
@@ -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<String>,
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<Vec<String>>,
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)),
+1 -5
View File
@@ -39,7 +39,6 @@ struct ExportContext<'a> {
schemas: Vec<String>,
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<Vec<String>> {
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,
};
+20 -4
View File
@@ -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 {
@@ -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::<Vec<_>>();
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();
}
}