rust/greptimedb
1
0
Fork 0
mirror of https://github.com/GreptimeTeam/greptimedb.git synced 2026-10-26 21:56:17 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: improve Parquet metadata diagnostics

Browse Source 
Signed-off-by: jeremyhi <fengjiachun@gmail.com>
This commit is contained in:
jeremyhi
2026-08-31 12:34:40 +08:00
parent a6a679aa68
commit 5af1aa0104
2 changed files with 68 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/common/datasource/src/file_format/parquet.rs
+62 -39
View File
@@ -76,9 +76,13 @@ pub fn parquet_table_name_from_metadata(metadata: Option<&[KeyValue]>) -> Result
}
.fail();
}
if table_name.is_some_and(|table_name| table_name != value) {
if let Some(table_name) = table_name
&& table_name != value
{
return InvalidParquetTableNameMetadataSnafu {
reason: format!("metadata key '{PARQUET_TABLE_NAME_KEY}' has conflicting values"),
reason: format!(
"metadata key '{PARQUET_TABLE_NAME_KEY}' has conflicting values: {table_name:?} and {value:?}"
),
}
.fail();
}
@@ -97,18 +101,32 @@ pub fn parquet_table_name_from_metadata_batch<'a>(
for metadata in metadata {
match parquet_table_name_from_metadata(metadata)? {
Some(value) => {
if saw_missing || table_name.is_some_and(|table_name| table_name != value) {
if saw_missing {
return InvalidParquetTableNameMetadataSnafu {
reason: "Parquet files have inconsistent table names".to_string(),
reason: format!(
"Parquet table name metadata is missing from some files but other files use {value:?}"
),
}
.fail();
}
if let Some(table_name) = table_name
&& table_name != value
{
return InvalidParquetTableNameMetadataSnafu {
reason: format!(
"Parquet files have conflicting table names: {table_name:?} and {value:?}"
),
}
.fail();
}
table_name = Some(value);
}
None => {
if table_name.is_some() {
if let Some(table_name) = table_name {
return InvalidParquetTableNameMetadataSnafu {
reason: "Parquet files have inconsistent table names".to_string(),
reason: format!(
"Parquet table name metadata is missing from some files but other files use {table_name:?}"
),
}
.fail();
}
@@ -289,7 +307,7 @@ pub async fn stream_to_parquet(
path: &str,
concurrency: usize,
) -> Result<usize> {
stream_to_parquet_with_metadata(stream, schema, store, path, concurrency, Vec::new()).await
stream_to_parquet_with_metadata(stream, schema, store, path, concurrency, None).await
}
/// Output the stream to a parquet file with custom key/value footer metadata.
@@ -301,18 +319,9 @@ pub async fn stream_to_parquet_with_metadata(
store: ObjectStore,
path: &str,
concurrency: usize,
metadata: Vec<(String, String)>,
metadata: Option<(String, String)>,
) -> Result<usize> {
let metadata = if metadata.is_empty() {
None
} else {
Some(
metadata
.into_iter()
.map(|(key, value)| KeyValue::new(key, value))
.collect(),
)
};
let metadata = metadata.map(|(key, value)| vec![KeyValue::new(key, value)]);
let write_props = column_wise_config(
WriterProperties::builder()
.set_compression(Compression::ZSTD(ZstdLevel::default()))
@@ -448,7 +457,7 @@ mod tests {
store.clone(),
&path,
1,
vec![(PARQUET_TABLE_NAME_KEY.to_string(), "my_table".to_string(),)],
Some((PARQUET_TABLE_NAME_KEY.to_string(), "my_table".to_string(),)),
)
.await
.unwrap()
@@ -494,7 +503,11 @@ mod tests {
None::<String>,
)];
assert!(parquet_table_name_from_metadata(Some(&metadata)).is_err());
let err = parquet_table_name_from_metadata(Some(&metadata)).unwrap_err();
assert_eq!(
"Invalid Parquet table name metadata: metadata key 'greptime.table_name' has no value",
err.to_string()
);
}
#[test]
@@ -514,7 +527,11 @@ mod tests {
KeyValue::new(PARQUET_TABLE_NAME_KEY.to_string(), "second".to_string()),
];
assert!(parquet_table_name_from_metadata(Some(&metadata)).is_err());
let err = parquet_table_name_from_metadata(Some(&metadata)).unwrap_err();
assert_eq!(
"Invalid Parquet table name metadata: metadata key 'greptime.table_name' has conflicting values: \"first\" and \"second\"",
err.to_string()
);
}
#[test]
@@ -578,12 +595,14 @@ mod tests {
"second".to_string(),
)];
assert!(
parquet_table_name_from_metadata_batch([
Some(first.as_slice()),
Some(second.as_slice())
])
.is_err()
let err = parquet_table_name_from_metadata_batch([
Some(first.as_slice()),
Some(second.as_slice()),
])
.unwrap_err();
assert_eq!(
"Invalid Parquet table name metadata: Parquet files have conflicting table names: \"first\" and \"second\"",
err.to_string()
);
}
@@ -605,19 +624,23 @@ mod tests {
)];
let missing = [KeyValue::new("other".to_string(), "value".to_string())];
assert!(
parquet_table_name_from_metadata_batch([
Some(present.as_slice()),
Some(missing.as_slice()),
])
.is_err()
let err = parquet_table_name_from_metadata_batch([
Some(present.as_slice()),
Some(missing.as_slice()),
])
.unwrap_err();
assert_eq!(
"Invalid Parquet table name metadata: Parquet table name metadata is missing from some files but other files use \"table\"",
err.to_string()
);
assert!(
parquet_table_name_from_metadata_batch([
Some(missing.as_slice()),
Some(present.as_slice()),
])
.is_err()
let err = parquet_table_name_from_metadata_batch([
Some(missing.as_slice()),
Some(present.as_slice()),
])
.unwrap_err();
assert_eq!(
"Invalid Parquet table name metadata: Parquet table name metadata is missing from some files but other files use \"table\"",
err.to_string()
);
}
src/operator/src/statement/copy_table_to.rs
+6 -6
View File
@@ -51,12 +51,12 @@ const WRITE_BUFFER_THRESHOLD: ReadableSize = ReadableSize::mb(8);
/// Default number of concurrent write, it only works on object store backend(e.g., S3).
const WRITE_CONCURRENCY: usize = 8;
fn parquet_metadata(format: &Format, table_name: Option<&str>) -> Vec<(String, String)> {
fn parquet_metadata(format: &Format, table_name: Option<&str>) -> Option<(String, String)> {
match (format, table_name) {
(Format::Parquet(_), Some(table_name)) => {
vec![(PARQUET_TABLE_NAME_KEY.to_string(), table_name.to_string())]
Some((PARQUET_TABLE_NAME_KEY.to_string(), table_name.to_string()))
}
_ => Vec::new(),
_ => None,
}
}
@@ -223,10 +223,10 @@ mod tests {
let json = Format::Json(JsonFormat::default());
assert_eq!(
vec![(PARQUET_TABLE_NAME_KEY.to_string(), "my_table".to_string())],
Some((PARQUET_TABLE_NAME_KEY.to_string(), "my_table".to_string())),
parquet_metadata(&parquet, Some("my_table"))
);
assert!(parquet_metadata(&parquet, None).is_empty());
assert!(parquet_metadata(&json, Some("my_table")).is_empty());
assert!(parquet_metadata(&parquet, None).is_none());
assert!(parquet_metadata(&json, Some("my_table")).is_none());
}
}