This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new c7735709 feat(catalog): support tag creation with retention (#1000)
c7735709 is described below
commit c7735709c694459ae73e1a9551db6434c9f59a3d
Author: QuakeWang <[email protected]>
AuthorDate: Fri Oct 2 09:37:12 2026 +0800
feat(catalog): support tag creation with retention (#1000)
---
crates/paimon/src/api/rest_api.rs | 15 +-
crates/paimon/src/catalog/filesystem.rs | 196 +++++++++++++++++++-
crates/paimon/src/catalog/mod.rs | 23 +++
crates/paimon/src/catalog/rest/rest_catalog.rs | 14 +-
crates/paimon/src/table/tag_manager.rs | 242 ++++++++++++++++++++++++-
crates/paimon/tests/rest_api_test.rs | 33 ++++
crates/paimon/tests/rest_catalog_test.rs | 64 +++++++
7 files changed, 568 insertions(+), 19 deletions(-)
diff --git a/crates/paimon/src/api/rest_api.rs
b/crates/paimon/src/api/rest_api.rs
index 7e435ae1..bf48c2ac 100644
--- a/crates/paimon/src/api/rest_api.rs
+++ b/crates/paimon/src/api/rest_api.rs
@@ -963,6 +963,18 @@ impl RESTApi {
identifier: &Identifier,
tag_name: &str,
snapshot_id: Option<i64>,
+ ) -> Result<()> {
+ self.create_tag_with_retention(identifier, tag_name, snapshot_id, None)
+ .await
+ }
+
+ /// Create a tag, forwarding the optional retention string to the server
unchanged.
+ pub async fn create_tag_with_retention(
+ &self,
+ identifier: &Identifier,
+ tag_name: &str,
+ snapshot_id: Option<i64>,
+ time_retained: Option<&str>,
) -> Result<()> {
let database = identifier.database();
let table = identifier.object();
@@ -972,7 +984,8 @@ impl RESTApi {
(tag_name, "tag name"),
])?;
let path = self.resource_paths.tags(database, table);
- let request = CreateTagRequest::new(tag_name.to_string(), snapshot_id);
+ let mut request = CreateTagRequest::new(tag_name.to_string(),
snapshot_id);
+ request.time_retained = time_retained.map(str::to_string);
let _response: serde_json::Value = self.client.post(&path,
&request).await?;
Ok(())
}
diff --git a/crates/paimon/src/catalog/filesystem.rs
b/crates/paimon/src/catalog/filesystem.rs
index 23195c78..4d48de2a 100644
--- a/crates/paimon/src/catalog/filesystem.rs
+++ b/crates/paimon/src/catalog/filesystem.rs
@@ -565,6 +565,18 @@ impl Catalog for FileSystemCatalog {
tag_name: &str,
snapshot_id: Option<i64>,
ignore_if_exists: bool,
+ ) -> Result<()> {
+ self.create_tag_with_retention(identifier, tag_name, snapshot_id,
None, ignore_if_exists)
+ .await
+ }
+
+ async fn create_tag_with_retention(
+ &self,
+ identifier: &Identifier,
+ tag_name: &str,
+ snapshot_id: Option<i64>,
+ time_retained: Option<&str>,
+ ignore_if_exists: bool,
) -> Result<()> {
let table = self.get_table(identifier).await?;
let manager = table.tag_manager();
@@ -579,7 +591,9 @@ impl Catalog for FileSystemCatalog {
}
let snapshot = resolve_tag_snapshot(&table, snapshot_id).await?;
- manager.create(tag_name, &snapshot).await
+ manager
+ .create_with_retention(tag_name, &snapshot, time_retained)
+ .await
}
async fn get_tag(&self, identifier: &Identifier, tag_name: &str) ->
Result<GetTagResponse> {
@@ -655,8 +669,14 @@ async fn resolve_tag_snapshot(table: &Table, snapshot_id:
Option<i64>) -> Result
}
}
-fn format_tag_time_retained(seconds: f64) -> Option<String> {
- let duration = std::time::Duration::try_from_secs_f64(seconds).ok()?;
+fn format_tag_time_retained(seconds: serde_json::Number) -> Option<String> {
+ // Java writes decimal seconds with nanosecond precision. Parse that
decimal directly
+ // before falling back to the existing float behavior for other JSON
number forms.
+ let duration = format!("PT{seconds}S")
+ .parse::<jiff::SignedDuration>()
+ .ok()
+ .and_then(|duration| std::time::Duration::try_from(duration).ok())
+ .or_else(||
std::time::Duration::try_from_secs_f64(seconds.as_f64()?).ok())?;
let total_seconds = duration.as_secs();
let hours = total_seconds / 3600;
let minutes = total_seconds % 3600 / 60;
@@ -1125,6 +1145,156 @@ mod tests {
.unwrap();
}
+ #[tokio::test]
+ async fn test_create_tag_with_retention() {
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("db1", "t");
+ catalog
+ .create_table(&identifier, testing_schema(), false)
+ .await
+ .unwrap();
+ give_the_table_a_snapshot(&catalog, &identifier).await;
+ let catalog: &dyn Catalog = &catalog;
+ for (name, snapshot_id, retention, expected) in [
+ ("explicit", Some(1), "1d", "PT24H"),
+ ("latest", None, "12h", "PT12H"),
+ ("micro", Some(1), "500micro", "PT0.0005S"),
+ ("nano", Some(1), "1ns", "PT0.000000001S"),
+ (
+ "large-nano",
+ Some(1),
+ "9223372036854775807ns",
+ "PT2562047H47M16.854775807S",
+ ),
+ (
+ "max-seconds",
+ Some(1),
+ "9223372036854775807s",
+ "PT2562047788015215H30M7S",
+ ),
+ ("zero", Some(1), "0", "PT0S"),
+ ] {
+ catalog
+ .create_tag_with_retention(&identifier, name, snapshot_id,
Some(retention), false)
+ .await
+ .unwrap();
+ let tag = catalog.get_tag(&identifier, name).await.unwrap();
+ assert_eq!(tag.snapshot.id(), 1);
+ assert!(tag.tag_create_time.is_some());
+ assert_eq!(tag.tag_time_retained.as_deref(), Some(expected));
+ }
+ catalog
+ .create_tag_with_retention(&identifier, "plain", None, None, false)
+ .await
+ .unwrap();
+ let plain = catalog.get_tag(&identifier, "plain").await.unwrap();
+ assert!(plain.tag_create_time.is_none());
+ assert!(plain.tag_time_retained.is_none());
+ }
+
+ #[tokio::test]
+ async fn test_create_tag_with_retention_preserves_existing_tag() {
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("db1", "t");
+ catalog
+ .create_table(&identifier, testing_schema(), false)
+ .await
+ .unwrap();
+ give_the_table_a_snapshot(&catalog, &identifier).await;
+ catalog
+ .create_tag_with_retention(&identifier, "retained", Some(1),
Some("1d"), false)
+ .await
+ .unwrap();
+ let table = catalog.get_table(&identifier).await.unwrap();
+ let input = table
+ .file_io()
+ .new_input(&table.tag_manager().tag_path("retained"))
+ .unwrap();
+ let before = input.read().await.unwrap();
+ catalog
+ .create_tag_with_retention(&identifier, "retained", Some(2),
Some("invalid"), true)
+ .await
+ .unwrap();
+ catalog
+ .create_tag(&identifier, "retained", None, true)
+ .await
+ .unwrap();
+ assert_eq!(input.read().await.unwrap(), before);
+ assert!(matches!(
+ catalog
+ .create_tag_with_retention(&identifier, "retained", Some(2),
Some("invalid"), false)
+ .await,
+ Err(Error::TagAlreadyExist { .. })
+ ));
+ }
+
+ #[tokio::test]
+ async fn test_create_tag_with_retention_errors_and_snapshot_fallback() {
+ let (_temp_dir, catalog) = create_test_catalog();
+ catalog
+ .create_database("db1", false, HashMap::new())
+ .await
+ .unwrap();
+ let identifier = Identifier::new("db1", "t");
+ catalog
+ .create_table(&identifier, testing_schema(), false)
+ .await
+ .unwrap();
+ assert!(matches!(
+ catalog
+ .create_tag_with_retention(&identifier, "empty", None,
Some("1d"), false)
+ .await,
+ Err(Error::DataInvalid { .. })
+ ));
+ give_the_table_a_snapshot(&catalog, &identifier).await;
+ assert!(matches!(
+ catalog
+ .create_tag_with_retention(&identifier, "missing", Some(2),
Some("1d"), false)
+ .await,
+ Err(Error::SnapshotNotExist { snapshot_id: 2 })
+ ));
+ assert!(matches!(
+ catalog
+ .create_tag_with_retention(&identifier, "invalid", Some(1),
Some("-1d"), false)
+ .await,
+ Err(Error::ConfigInvalid { .. })
+ ));
+ assert!(matches!(
+ catalog
+ .create_tag_with_retention(&identifier, "nested/tag", Some(1),
Some("1d"), false)
+ .await,
+ Err(Error::ConfigInvalid { .. })
+ ));
+ let table = catalog.get_table(&identifier).await.unwrap();
+ assert!(table
+ .tag_manager()
+ .list_all_names()
+ .await
+ .unwrap()
+ .is_empty());
+
+ catalog
+ .create_tag(&identifier, "source", Some(1), false)
+ .await
+ .unwrap();
+ table.snapshot_manager().delete_snapshot(1).await.unwrap();
+ catalog
+ .create_tag_with_retention(&identifier, "from-tag", Some(1),
Some("1d"), false)
+ .await
+ .unwrap();
+ let tag = catalog.get_tag(&identifier, "from-tag").await.unwrap();
+ assert_eq!(tag.snapshot.id(), 1);
+ assert_eq!(tag.tag_time_retained.as_deref(), Some("PT24H"));
+ }
+
#[tokio::test]
async fn test_delete_expired_tag_is_unsupported() {
let (_temp_dir, catalog) = create_test_catalog();
@@ -1168,13 +1338,19 @@ mod tests {
#[test]
fn test_format_tag_time_retained() {
- assert_eq!(format_tag_time_retained(0.0).as_deref(), Some("PT0S"));
- assert_eq!(format_tag_time_retained(90.0).as_deref(), Some("PT1M30S"));
- assert_eq!(
- format_tag_time_retained(259_200.0).as_deref(),
- Some("PT72H")
- );
- assert_eq!(format_tag_time_retained(1.5).as_deref(), Some("PT1.5S"));
+ for (seconds, expected) in [
+ ("0.0", Some("PT0S")),
+ ("90.0", Some("PT1M30S")),
+ ("259200.0", Some("PT72H")),
+ ("1.5", Some("PT1.5S")),
+ ("1e-6", Some("PT0.000001S")),
+ ("9223372036.854775807", Some("PT2562047H47M16.854775807S")),
+ ("-1.0", None),
+ ("1e100", None),
+ ] {
+ let number = serde_json::from_str(seconds).unwrap();
+ assert_eq!(format_tag_time_retained(number).as_deref(), expected);
+ }
}
#[test]
diff --git a/crates/paimon/src/catalog/mod.rs b/crates/paimon/src/catalog/mod.rs
index b610680c..3ffdedd7 100644
--- a/crates/paimon/src/catalog/mod.rs
+++ b/crates/paimon/src/catalog/mod.rs
@@ -510,6 +510,29 @@ pub trait Catalog: Send + Sync {
})
}
+ /// Create a tag with an optional retention, such as `"1d"`, `"12h"`, or
`"500micro"`.
+ ///
+ /// `None` preserves [`Self::create_tag`] behavior. An explicit retention
records the
+ /// creation time and TTL; this method does not run tag expiration.
+ /// Catalogs that do not support retention return [`Error::Unsupported`].
+ async fn create_tag_with_retention(
+ &self,
+ identifier: &Identifier,
+ tag_name: &str,
+ snapshot_id: Option<i64>,
+ time_retained: Option<&str>,
+ ignore_if_exists: bool,
+ ) -> Result<()> {
+ if time_retained.is_none() {
+ return self
+ .create_tag(identifier, tag_name, snapshot_id,
ignore_if_exists)
+ .await;
+ }
+ Err(Error::Unsupported {
+ message: "tag retention is not supported by this
catalog".to_string(),
+ })
+ }
+
/// Return a tag and its snapshot metadata.
async fn get_tag(&self, _identifier: &Identifier, _tag_name: &str) ->
Result<GetTagResponse> {
Err(Error::Unsupported {
diff --git a/crates/paimon/src/catalog/rest/rest_catalog.rs
b/crates/paimon/src/catalog/rest/rest_catalog.rs
index de42003d..c7a5cfaa 100644
--- a/crates/paimon/src/catalog/rest/rest_catalog.rs
+++ b/crates/paimon/src/catalog/rest/rest_catalog.rs
@@ -409,10 +409,22 @@ impl Catalog for RESTCatalog {
tag_name: &str,
snapshot_id: Option<i64>,
ignore_if_exists: bool,
+ ) -> Result<()> {
+ self.create_tag_with_retention(identifier, tag_name, snapshot_id,
None, ignore_if_exists)
+ .await
+ }
+
+ async fn create_tag_with_retention(
+ &self,
+ identifier: &Identifier,
+ tag_name: &str,
+ snapshot_id: Option<i64>,
+ time_retained: Option<&str>,
+ ignore_if_exists: bool,
) -> Result<()> {
let result = self
.api
- .create_tag(identifier, tag_name, snapshot_id)
+ .create_tag_with_retention(identifier, tag_name, snapshot_id,
time_retained)
.await
.map_err(|error| map_rest_error_for_tag(error, identifier,
tag_name, snapshot_id));
ignore_error_if(result, |error| {
diff --git a/crates/paimon/src/table/tag_manager.rs
b/crates/paimon/src/table/tag_manager.rs
index 668dc9b3..b3a1cb2a 100644
--- a/crates/paimon/src/table/tag_manager.rs
+++ b/crates/paimon/src/table/tag_manager.rs
@@ -15,15 +15,17 @@
// specific language governing permissions and limitations
// under the License.
-//! Tag manager for reading tag metadata using FileIO.
+//! Tag manager for reading and writing tag metadata using FileIO.
//!
//! Reference:
[org.apache.paimon.utils.TagManager](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java)
//! and
[pypaimon.tag.tag_manager.TagManager](https://github.com/apache/paimon/blob/master/paimon-python/pypaimon/tag/tag_manager.py).
use crate::io::FileIO;
use crate::spec::Snapshot;
+use chrono::{Datelike, Timelike};
use futures::future::try_join_all;
use opendal::raw::get_basename;
+use std::time::Duration;
const TAG_DIR: &str = "tag";
const TAG_PREFIX: &str = "tag-";
@@ -31,7 +33,8 @@ const TAG_PREFIX: &str = "tag-";
/// Manager for tag files using unified FileIO.
///
/// Tags are named snapshots stored as JSON files at
`{table_path}/tag/tag-{name}`.
-/// The tag file format is identical to a Snapshot JSON file.
+/// Tags without retention use Snapshot JSON. Retained tags also include
Java-compatible
+/// creation time and retention metadata.
///
/// Reference:
[org.apache.paimon.utils.TagManager](https://github.com/apache/paimon/blob/master/paimon-core/src/main/java/org/apache/paimon/utils/TagManager.java)
#[derive(Debug, Clone)]
@@ -110,7 +113,7 @@ impl TagManager {
(
snapshot,
create_time.map(|value|
value.and_utc().timestamp_millis()),
- time_retained,
+ time_retained.and_then(|value| value.as_f64()),
)
},
))
@@ -119,7 +122,13 @@ impl TagManager {
pub(crate) async fn get_with_raw_metadata(
&self,
tag_name: &str,
- ) -> crate::Result<Option<(Snapshot, Option<chrono::NaiveDateTime>,
Option<f64>)>> {
+ ) -> crate::Result<
+ Option<(
+ Snapshot,
+ Option<chrono::NaiveDateTime>,
+ Option<serde_json::Number>,
+ )>,
+ > {
validate_tag_name(tag_name)?;
let path = self.tag_path(tag_name);
let input = self.file_io.new_input(&path)?;
@@ -147,7 +156,8 @@ impl TagManager {
.and_then(parse_tag_create_time);
let time_retained = value
.get(FIELD_TAG_TIME_RETAINED)
- .and_then(serde_json::Value::as_f64);
+ .and_then(serde_json::Value::as_number)
+ .cloned();
Ok(Some((snapshot, create_time, time_retained)))
}
@@ -199,12 +209,54 @@ impl TagManager {
/// Create a tag by writing the snapshot JSON to the tag path.
pub async fn create(&self, tag_name: &str, snapshot: &Snapshot) ->
crate::Result<()> {
+ self.create_with_retention(tag_name, snapshot, None).await
+ }
+
+ /// Create a tag with an optional retention in Java's
`TimeUtils.parseDuration` format.
+ /// A bare integer denotes milliseconds; units from days through
nanoseconds are supported.
+ pub async fn create_with_retention(
+ &self,
+ tag_name: &str,
+ snapshot: &Snapshot,
+ time_retained: Option<&str>,
+ ) -> crate::Result<()> {
validate_tag_name(tag_name)?;
let path = self.tag_path(tag_name);
- let json = serde_json::to_string(snapshot).map_err(|e|
crate::Error::DataInvalid {
+ let serialize_error = |e| crate::Error::DataInvalid {
message: format!("failed to serialize snapshot for tag
'{tag_name}': {e}"),
source: Some(Box::new(e)),
- })?;
+ };
+ let json = if let Some(value) = time_retained {
+ let duration =
+ parse_tag_time_retained(value).ok_or_else(||
crate::Error::ConfigInvalid {
+ message: format!("Invalid tag retention: {value:?}"),
+ })?;
+ let now = chrono::Local::now();
+ let tag = RetainedTag {
+ snapshot,
+ tag_create_time: [
+ i64::from(now.year()),
+ i64::from(now.month()),
+ i64::from(now.day()),
+ i64::from(now.hour()),
+ i64::from(now.minute()),
+ i64::from(now.second()),
+ i64::from(now.nanosecond()),
+ ],
+ // Decimal seconds preserve nanoseconds even when the duration
is too large
+ // for an f64 to represent its fractional part.
+ tag_time_retained: serde_json::from_str(&format!(
+ "{}.{:09}",
+ duration.as_secs(),
+ duration.subsec_nanos()
+ ))
+ .map_err(&serialize_error)?,
+ };
+ serde_json::to_string(&tag)
+ } else {
+ serde_json::to_string(snapshot)
+ }
+ .map_err(serialize_error)?;
self.file_io.mkdirs(&self.tag_directory()).await?;
let output = self.file_io.new_output(&path)?;
output.write(bytes::Bytes::from(json)).await
@@ -238,6 +290,40 @@ impl TagManager {
const FIELD_TAG_CREATE_TIME: &str = "tagCreateTime";
const FIELD_TAG_TIME_RETAINED: &str = "tagTimeRetained";
+#[derive(serde::Serialize)]
+#[serde(rename_all = "camelCase")]
+struct RetainedTag<'a> {
+ #[serde(flatten)]
+ snapshot: &'a Snapshot,
+ tag_create_time: [i64; 7],
+ tag_time_retained: serde_json::Number,
+}
+
+/// Match Java's unsigned integer syntax and signed 64-bit value/seconds
limits.
+fn parse_tag_time_retained(value: &str) -> Option<Duration> {
+ let value = value.trim();
+ let pos = value
+ .find(|c: char| !c.is_ascii_digit())
+ .unwrap_or(value.len());
+ let (number, unit) = value.split_at(pos);
+ let number = number.parse::<i64>().ok()? as u64;
+ let duration = match unit.trim().to_ascii_lowercase().as_str() {
+ "d" | "day" | "days" =>
Duration::from_secs(number.checked_mul(86_400)?),
+ "h" | "hour" | "hours" =>
Duration::from_secs(number.checked_mul(3_600)?),
+ "m" | "min" | "minute" | "minutes" =>
Duration::from_secs(number.checked_mul(60)?),
+ "s" | "sec" | "secs" | "second" | "seconds" =>
Duration::from_secs(number),
+ "" | "ms" | "milli" | "millis" | "millisecond" | "milliseconds" => {
+ Duration::from_millis(number)
+ }
+ "\u{b5}s" | "micro" | "micros" | "microsecond" | "microseconds" => {
+ Duration::from_micros(number)
+ }
+ "ns" | "nano" | "nanos" | "nanosecond" | "nanoseconds" =>
Duration::from_nanos(number),
+ _ => return None,
+ };
+ (duration.as_secs() <= i64::MAX as u64).then_some(duration)
+}
+
fn validate_tag_name(tag_name: &str) -> crate::Result<()> {
let invalid = tag_name.trim().is_empty()
|| tag_name.trim_end() != tag_name
@@ -316,6 +402,148 @@ mod tests {
output.write(Bytes::from(json)).await.unwrap();
}
+ #[tokio::test]
+ async fn test_create_tag_without_retention_writes_plain_snapshot() {
+ let file_io = test_file_io();
+ let tm = TagManager::new(file_io.clone(),
"memory:/test_tag_create_plain".to_string());
+ let snapshot = test_snapshot(1);
+ for name in ["legacy", "none"] {
+ if name == "legacy" {
+ tm.create(name, &snapshot).await.unwrap();
+ } else {
+ tm.create_with_retention(name, &snapshot, None)
+ .await
+ .unwrap();
+ }
+ let bytes = file_io
+ .new_input(&tm.tag_path(name))
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ assert_eq!(bytes.as_ref(), serde_json::to_vec(&snapshot).unwrap());
+ }
+ }
+
+ #[tokio::test]
+ async fn test_create_tag_with_retention_writes_java_metadata() {
+ let file_io = test_file_io();
+ let tm = TagManager::new(
+ file_io.clone(),
+ "memory:/test_tag_create_retained".to_string(),
+ );
+ let snapshot = test_snapshot(1);
+ for (retention, expected) in [
+ ("1d", "86400.000000000"),
+ ("12h", "43200.000000000"),
+ ("30m", "1800.000000000"),
+ ("500micro", "0.000500000"),
+ ("1ns", "0.000000001"),
+ ("0", "0.000000000"),
+ ("9223372036854775807ns", "9223372036.854775807"),
+ ("9223372036854775807", "9223372036854775.807000000"),
+ ("9223372036854775807s", "9223372036854775807.000000000"),
+ ] {
+ let before = chrono::Local::now().naive_local();
+ tm.create_with_retention("retained", &snapshot, Some(retention))
+ .await
+ .unwrap();
+ let after = chrono::Local::now().naive_local();
+ let bytes = file_io
+ .new_input(&tm.tag_path("retained"))
+ .unwrap()
+ .read()
+ .await
+ .unwrap();
+ let value: serde_json::Value =
serde_json::from_slice(&bytes).unwrap();
+ assert_eq!(
+ value[FIELD_TAG_TIME_RETAINED]
+ .as_number()
+ .unwrap()
+ .to_string(),
+ expected
+ );
+ assert_eq!(value[FIELD_TAG_CREATE_TIME].as_array().unwrap().len(),
7);
+ let created =
parse_tag_create_time(&value[FIELD_TAG_CREATE_TIME]).unwrap();
+ assert!(before <= created && created <= after);
+ assert_eq!(serde_json::from_value::<Snapshot>(value).unwrap(),
snapshot);
+ assert_eq!(tm.get("retained").await.unwrap(),
Some(snapshot.clone()));
+ }
+ }
+
+ #[test]
+ fn test_parse_tag_retention_matches_java_units() {
+ for (units, duration) in [
+ (vec!["d", "day", "days"], Duration::from_secs(86_400)),
+ (vec!["h", "hour", "hours"], Duration::from_secs(3_600)),
+ (
+ vec!["m", "min", "minute", "minutes"],
+ Duration::from_secs(60),
+ ),
+ (
+ vec!["s", "sec", "secs", "second", "seconds"],
+ Duration::from_secs(1),
+ ),
+ (
+ vec!["", "ms", "milli", "millis", "millisecond",
"milliseconds"],
+ Duration::from_millis(1),
+ ),
+ (
+ vec!["\u{b5}s", "micro", "micros", "microsecond",
"microseconds"],
+ Duration::from_micros(1),
+ ),
+ (
+ vec!["ns", "nano", "nanos", "nanosecond", "nanoseconds"],
+ Duration::from_nanos(1),
+ ),
+ ] {
+ for unit in units {
+ assert_eq!(
+ parse_tag_time_retained(&format!(" 1 {} ",
unit.to_ascii_uppercase())),
+ Some(duration)
+ );
+ }
+ }
+ assert_eq!(
+ parse_tag_time_retained("106751991167300d"),
+ Some(Duration::from_secs(106_751_991_167_300 * 86_400))
+ );
+ }
+
+ #[tokio::test]
+ async fn test_create_tag_with_invalid_retention_does_not_write() {
+ let file_io = test_file_io();
+ let tm = TagManager::new(
+ file_io.clone(),
+ "memory:/test_tag_create_invalid".to_string(),
+ );
+ let snapshot = test_snapshot(1);
+ for value in [
+ "",
+ " ",
+ "-1d",
+ "+1d",
+ "1.5h",
+ "PT24H",
+ "1fortnight",
+ "1us",
+ "9223372036854775808ns",
+ "106751991167301d",
+ "9223372036854775807h",
+ ] {
+ assert!(
+ matches!(
+ tm.create_with_retention("invalid", &snapshot, Some(value))
+ .await,
+ Err(crate::Error::ConfigInvalid { .. })
+ ),
+ "{value:?}"
+ );
+ assert!(!tm.tag_exists("invalid").await.unwrap());
+ }
+ assert!(!file_io.exists(&tm.tag_directory()).await.unwrap());
+ }
+
#[tokio::test]
async fn test_list_all_names_missing_dir_returns_empty() {
let file_io = test_file_io();
diff --git a/crates/paimon/tests/rest_api_test.rs
b/crates/paimon/tests/rest_api_test.rs
index 9ca3d15a..1649b5a2 100644
--- a/crates/paimon/tests/rest_api_test.rs
+++ b/crates/paimon/tests/rest_api_test.rs
@@ -83,6 +83,39 @@ async fn setup_partition_api() -> (TestContext, Identifier) {
(ctx, identifier)
}
+// ==================== Tag Tests ====================
+#[tokio::test]
+async fn test_create_tag_with_retention() {
+ let ctx = setup_test_server(vec!["default"]).await;
+ ctx.server.add_table("default", "managed_table");
+ let identifier = Identifier::new("default", "managed_table");
+ for (name, snapshot_id, retention) in [
+ ("explicit", Some(1), Some("1d")),
+ ("latest", None, Some(" 12 HOURS ")),
+ ("nano", Some(1), Some("1ns")),
+ ("plain", None, None),
+ ] {
+ ctx.api
+ .create_tag_with_retention(&identifier, name, snapshot_id,
retention)
+ .await
+ .unwrap();
+ let response = ctx.api.get_tag(&identifier, name).await.unwrap();
+ assert_eq!(response.snapshot.id(), 1);
+ assert_eq!(response.tag_time_retained.as_deref(), retention);
+ }
+ ctx.api
+ .create_tag(&identifier, "legacy", Some(1))
+ .await
+ .unwrap();
+ assert!(ctx
+ .api
+ .get_tag(&identifier, "legacy")
+ .await
+ .unwrap()
+ .tag_time_retained
+ .is_none());
+}
+
// ==================== Database Tests ====================
#[tokio::test]
async fn test_list_databases() {
diff --git a/crates/paimon/tests/rest_catalog_test.rs
b/crates/paimon/tests/rest_catalog_test.rs
index 004d435e..c7ddad54 100644
--- a/crates/paimon/tests/rest_catalog_test.rs
+++ b/crates/paimon/tests/rest_catalog_test.rs
@@ -1856,6 +1856,70 @@ async fn test_catalog_alter_table() {
.unwrap();
}
+#[tokio::test]
+async fn test_catalog_create_tag_with_retention() {
+ let ctx = setup_catalog(vec!["default"]).await;
+ ctx.server.add_table("default", "managed_table");
+ let identifier = Identifier::new("default", "managed_table");
+ let catalog: &dyn Catalog = &ctx.catalog;
+ for (name, snapshot_id, retention) in [
+ ("explicit", Some(1), Some("1d")),
+ ("latest", None, Some(" 12 HOURS ")),
+ ("micro", Some(1), Some("500micro")),
+ ("plain", None, None),
+ ] {
+ catalog
+ .create_tag_with_retention(&identifier, name, snapshot_id,
retention, false)
+ .await
+ .unwrap();
+ let tag = catalog.get_tag(&identifier, name).await.unwrap();
+ assert_eq!(tag.snapshot.id(), 1);
+ assert_eq!(tag.tag_time_retained.as_deref(), retention);
+ }
+
+ assert!(matches!(
+ catalog
+ .create_tag_with_retention(&identifier, "explicit", Some(1),
Some("12h"), false)
+ .await,
+ Err(paimon::Error::TagAlreadyExist { .. })
+ ));
+ catalog
+ .create_tag_with_retention(&identifier, "explicit", Some(1),
Some("12h"), true)
+ .await
+ .unwrap();
+ catalog
+ .create_tag(&identifier, "explicit", Some(1), true)
+ .await
+ .unwrap();
+ assert_eq!(
+ catalog
+ .get_tag(&identifier, "explicit")
+ .await
+ .unwrap()
+ .tag_time_retained
+ .as_deref(),
+ Some("1d")
+ );
+ assert!(matches!(
+ catalog
+ .create_tag_with_retention(&identifier, "missing", Some(2),
Some("1d"), false)
+ .await,
+ Err(paimon::Error::SnapshotNotExist { snapshot_id: 2 })
+ ));
+ assert!(matches!(
+ catalog
+ .create_tag_with_retention(
+ &Identifier::new("default", "missing"),
+ "tag",
+ None,
+ Some("1d"),
+ false
+ )
+ .await,
+ Err(paimon::Error::TableNotExist { .. })
+ ));
+}
+
#[tokio::test]
async fn test_catalog_tag_lifecycle() {
let ctx = setup_catalog(vec!["default"]).await;
