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 7c7120b0 feat(go): expose snapshot and tag APIs (#828)
7c7120b0 is described below
commit 7c7120b0afffbf1ad0be28e14ce9bc5e13b19560
Author: XiaoHongbo <[email protected]>
AuthorDate: Mon Sep 14 21:46:37 2026 +0800
feat(go): expose snapshot and tag APIs (#828)
---
bindings/c/src/catalog.rs | 133 ++++++++++++++++-
bindings/c/src/error.rs | 3 +
bindings/c/src/result.rs | 12 ++
bindings/c/src/table.rs | 42 +++++-
bindings/c/src/tests.rs | 142 +++++++++++++++++-
bindings/c/src/types.rs | 7 +
bindings/go/catalog.go | 7 +-
bindings/go/catalog_tag.go | 263 ++++++++++++++++++++++++++++++++++
bindings/go/tests/catalog_tag_test.go | 110 ++++++++++++++
bindings/go/types.go | 41 ++++++
docs/src/go-binding.md | 38 +++++
11 files changed, 785 insertions(+), 13 deletions(-)
diff --git a/bindings/c/src/catalog.rs b/bindings/c/src/catalog.rs
index 9c5c683b..49c2fb48 100644
--- a/bindings/c/src/catalog.rs
+++ b/bindings/c/src/catalog.rs
@@ -15,16 +15,16 @@
// specific language governing permissions and limitations
// under the License.
-use std::ffi::c_void;
+use std::ffi::{c_char, c_void};
use std::sync::Arc;
use paimon::catalog::Identifier;
use paimon::{Catalog, CatalogFactory, Options};
-use crate::error::{check_non_null, paimon_error, validate_cstr};
-use crate::result::{paimon_result_catalog_new, paimon_result_get_table};
+use crate::error::{check_non_null, paimon_error, validate_cstr,
PaimonErrorCode};
+use crate::result::{paimon_result_catalog_new, paimon_result_get_table,
paimon_result_get_tag};
use crate::runtime;
-use crate::types::{paimon_catalog, paimon_option, paimon_table};
+use crate::types::{paimon_bytes, paimon_catalog, paimon_identifier,
paimon_option, paimon_table};
/// Create a catalog using CatalogFactory with the given options.
///
@@ -136,3 +136,128 @@ pub unsafe extern "C" fn paimon_catalog_get_table(
},
}
}
+
+/// Create a tag for a snapshot, or the latest snapshot when `snapshot_id` is
null.
+///
+/// # Safety
+/// `catalog` and `identifier` must be valid pointers from previous paimon C
calls.
+/// `tag_name` must be a valid null-terminated C string.
+/// If non-null, `snapshot_id` must point to a valid `i64`.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_catalog_create_tag(
+ catalog: *const paimon_catalog,
+ identifier: *const paimon_identifier,
+ tag_name: *const c_char,
+ snapshot_id: *const i64,
+ ignore_if_exists: bool,
+) -> *mut paimon_error {
+ if let Err(error) = check_non_null(catalog, "catalog") {
+ return error;
+ }
+ if let Err(error) = check_non_null(identifier, "identifier") {
+ return error;
+ }
+ let tag_name = match validate_cstr(tag_name, "tag_name") {
+ Ok(tag_name) => tag_name,
+ Err(error) => return error,
+ };
+
+ let catalog = &*((*catalog).inner as *const Arc<dyn Catalog>);
+ let identifier = &*((*identifier).inner as *const Identifier);
+ let snapshot_id = snapshot_id.as_ref().copied();
+ match runtime().block_on(catalog.create_tag(
+ identifier,
+ &tag_name,
+ snapshot_id,
+ ignore_if_exists,
+ )) {
+ Ok(()) => std::ptr::null_mut(),
+ Err(error) => paimon_error::from_paimon(error),
+ }
+}
+
+/// Get a tag and its snapshot metadata as JSON.
+///
+/// # Safety
+/// `catalog` and `identifier` must be valid pointers from previous paimon C
calls.
+/// `tag_name` must be a valid null-terminated C string.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_catalog_get_tag(
+ catalog: *const paimon_catalog,
+ identifier: *const paimon_identifier,
+ tag_name: *const c_char,
+) -> paimon_result_get_tag {
+ if let Err(error) = check_non_null(catalog, "catalog") {
+ return paimon_result_get_tag {
+ tag: paimon_bytes::empty(),
+ error,
+ };
+ }
+ if let Err(error) = check_non_null(identifier, "identifier") {
+ return paimon_result_get_tag {
+ tag: paimon_bytes::empty(),
+ error,
+ };
+ }
+ let tag_name = match validate_cstr(tag_name, "tag_name") {
+ Ok(tag_name) => tag_name,
+ Err(error) => {
+ return paimon_result_get_tag {
+ tag: paimon_bytes::empty(),
+ error,
+ }
+ }
+ };
+
+ let catalog = &*((*catalog).inner as *const Arc<dyn Catalog>);
+ let identifier = &*((*identifier).inner as *const Identifier);
+ match runtime().block_on(catalog.get_tag(identifier, &tag_name)) {
+ Ok(tag) => match serde_json::to_vec(&tag) {
+ Ok(tag) => paimon_result_get_tag {
+ tag: paimon_bytes::new(tag),
+ error: std::ptr::null_mut(),
+ },
+ Err(error) => paimon_result_get_tag {
+ tag: paimon_bytes::empty(),
+ error: paimon_error::new(
+ PaimonErrorCode::Unexpected,
+ format!("failed to serialize tag metadata: {error}"),
+ ),
+ },
+ },
+ Err(error) => paimon_result_get_tag {
+ tag: paimon_bytes::empty(),
+ error: paimon_error::from_paimon(error),
+ },
+ }
+}
+
+/// Delete a tag.
+///
+/// # Safety
+/// `catalog` and `identifier` must be valid pointers from previous paimon C
calls.
+/// `tag_name` must be a valid null-terminated C string.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_catalog_delete_tag(
+ catalog: *const paimon_catalog,
+ identifier: *const paimon_identifier,
+ tag_name: *const c_char,
+) -> *mut paimon_error {
+ if let Err(error) = check_non_null(catalog, "catalog") {
+ return error;
+ }
+ if let Err(error) = check_non_null(identifier, "identifier") {
+ return error;
+ }
+ let tag_name = match validate_cstr(tag_name, "tag_name") {
+ Ok(tag_name) => tag_name,
+ Err(error) => return error,
+ };
+
+ let catalog = &*((*catalog).inner as *const Arc<dyn Catalog>);
+ let identifier = &*((*identifier).inner as *const Identifier);
+ match runtime().block_on(catalog.delete_tag(identifier, &tag_name, false))
{
+ Ok(()) => std::ptr::null_mut(),
+ Err(error) => paimon_error::from_paimon(error),
+ }
+}
diff --git a/bindings/c/src/error.rs b/bindings/c/src/error.rs
index 7b0a88fc..db82773d 100644
--- a/bindings/c/src/error.rs
+++ b/bindings/c/src/error.rs
@@ -52,9 +52,12 @@ impl paimon_error {
}
paimon::Error::TableNotExist { .. }
| paimon::Error::DatabaseNotExist { .. }
+ | paimon::Error::SnapshotNotExist { .. }
+ | paimon::Error::TagNotExist { .. }
| paimon::Error::ColumnNotExist { .. } =>
PaimonErrorCode::NotFound,
paimon::Error::TableAlreadyExist { .. }
| paimon::Error::DatabaseAlreadyExist { .. }
+ | paimon::Error::TagAlreadyExist { .. }
| paimon::Error::ColumnAlreadyExist { .. } =>
PaimonErrorCode::AlreadyExists,
paimon::Error::ConfigInvalid { .. }
| paimon::Error::DataTypeInvalid { .. }
diff --git a/bindings/c/src/result.rs b/bindings/c/src/result.rs
index 3f3b7f98..60da7c09 100644
--- a/bindings/c/src/result.rs
+++ b/bindings/c/src/result.rs
@@ -72,6 +72,18 @@ pub struct paimon_result_get_table {
pub error: *mut paimon_error,
}
+#[repr(C)]
+pub struct paimon_result_get_tag {
+ pub tag: paimon_bytes,
+ pub error: *mut paimon_error,
+}
+
+#[repr(C)]
+pub struct paimon_result_latest_snapshot {
+ pub snapshot: paimon_bytes,
+ pub error: *mut paimon_error,
+}
+
#[repr(C)]
pub struct paimon_result_new_read {
pub read: *mut paimon_table_read,
diff --git a/bindings/c/src/table.rs b/bindings/c/src/table.rs
index 0b18de70..058c95d8 100644
--- a/bindings/c/src/table.rs
+++ b/bindings/c/src/table.rs
@@ -30,9 +30,9 @@ use paimon::Plan;
use crate::error::{check_non_null, paimon_error, validate_cstr,
PaimonErrorCode};
use crate::file_io::file_io_ref;
use crate::result::{
- paimon_result_get_table, paimon_result_new_read, paimon_result_next_batch,
paimon_result_plan,
- paimon_result_predicate, paimon_result_read_builder,
paimon_result_record_batch_reader,
- paimon_result_table_scan,
+ paimon_result_get_table, paimon_result_latest_snapshot,
paimon_result_new_read,
+ paimon_result_next_batch, paimon_result_plan, paimon_result_predicate,
+ paimon_result_read_builder, paimon_result_record_batch_reader,
paimon_result_table_scan,
};
use crate::runtime;
use crate::types::*;
@@ -336,6 +336,42 @@ pub unsafe extern "C" fn paimon_table_free(table: *mut
paimon_table) {
free_table_wrapper(table, |t| t.inner);
}
+/// Return the latest snapshot as JSON, or JSON null when the table is empty.
+///
+/// # Safety
+/// `table` must be a valid pointer from a previous paimon C call.
+#[no_mangle]
+pub unsafe extern "C" fn paimon_table_latest_snapshot(
+ table: *const paimon_table,
+) -> paimon_result_latest_snapshot {
+ if let Err(error) = check_non_null(table, "table") {
+ return paimon_result_latest_snapshot {
+ snapshot: paimon_bytes::empty(),
+ error,
+ };
+ }
+ let table = &*((*table).inner as *const Table);
+ match runtime().block_on(table.snapshot_manager().get_latest_snapshot()) {
+ Ok(snapshot) => match serde_json::to_vec(&snapshot) {
+ Ok(snapshot) => paimon_result_latest_snapshot {
+ snapshot: paimon_bytes::new(snapshot),
+ error: std::ptr::null_mut(),
+ },
+ Err(error) => paimon_result_latest_snapshot {
+ snapshot: paimon_bytes::empty(),
+ error: paimon_error::new(
+ PaimonErrorCode::Unexpected,
+ format!("failed to serialize snapshot metadata: {error}"),
+ ),
+ },
+ },
+ Err(error) => paimon_result_latest_snapshot {
+ snapshot: paimon_bytes::empty(),
+ error: paimon_error::from_paimon(error),
+ },
+ }
+}
+
/// Time-travel selector option names, in the core's resolution priority order.
const TIME_TRAVEL_SELECTORS: [&str; 5] = [
"scan.timestamp-millis",
diff --git a/bindings/c/src/tests.rs b/bindings/c/src/tests.rs
index dc66432f..570c85a8 100644
--- a/bindings/c/src/tests.rs
+++ b/bindings/c/src/tests.rs
@@ -44,11 +44,14 @@ use paimon::spec::{
BlobDescriptor, CommitKind, DataType, IntType, Schema, TableSchema,
VarCharType,
};
use paimon::table::{SnapshotManager, Table};
+use paimon::{Catalog, FileSystemCatalog, Options};
use crate::blob_reader::*;
use crate::bucket_vector_search_split::*;
+use crate::catalog::*;
use crate::error::*;
use crate::file_io::*;
+use crate::identifier::*;
use crate::table::*;
use crate::types::*;
use crate::vector_read::*;
@@ -64,13 +67,16 @@ fn memory_file_io() -> paimon::io::FileIO {
FileIOBuilder::new("memory").build().unwrap()
}
-fn simple_table_schema() -> TableSchema {
- let schema = Schema::builder()
+fn simple_schema() -> Schema {
+ Schema::builder()
.column("id", DataType::Int(IntType::new()))
.column("name", DataType::VarChar(VarCharType::string_type()))
.build()
- .unwrap();
- TableSchema::new(0, &schema)
+ .unwrap()
+}
+
+fn simple_table_schema() -> TableSchema {
+ TableSchema::new(0, &simple_schema())
}
fn not_null_table_schema() -> TableSchema {
@@ -100,6 +106,11 @@ unsafe fn wrap_table(table: Table) -> *mut paimon_table {
Box::into_raw(Box::new(paimon_table { inner }))
}
+unsafe fn wrap_catalog(catalog: Arc<dyn Catalog>) -> *mut paimon_catalog {
+ let inner = Box::into_raw(Box::new(catalog)) as *mut c_void;
+ Box::into_raw(Box::new(paimon_catalog { inner }))
+}
+
unsafe fn unwrap_table(table: *mut paimon_table) {
let wrapper = Box::from_raw(table);
if !wrapper.inner.is_null() {
@@ -953,6 +964,129 @@ fn
test_table_from_schema_json_rejects_invalid_identifier() {
}
}
+// =========================================================================
+// Catalog tag tests
+// =========================================================================
+
+#[test]
+fn test_catalog_tag_lifecycle() {
+ let temp_dir = tempfile::tempdir().unwrap();
+ let mut options = Options::new();
+ options.set("warehouse", temp_dir.path().to_string_lossy());
+ let catalog = FileSystemCatalog::new(options).unwrap();
+ let identifier = Identifier::new("default", "test");
+ let table = crate::runtime().block_on(async {
+ catalog
+ .create_database("default", false, HashMap::new())
+ .await
+ .unwrap();
+ catalog
+ .create_table(&identifier, simple_schema(), false)
+ .await
+ .unwrap();
+ catalog.get_table(&identifier).await.unwrap()
+ });
+ unsafe {
+ let empty_table = wrap_table(table.clone());
+ let latest = paimon_table_latest_snapshot(empty_table);
+ assert!(latest.error.is_null());
+ let json = std::slice::from_raw_parts(latest.snapshot.data,
latest.snapshot.len);
+ assert!(
+ serde_json::from_slice::<Option<paimon::spec::Snapshot>>(json)
+ .unwrap()
+ .is_none()
+ );
+ paimon_bytes_free(latest.snapshot);
+ paimon_table_free(empty_table);
+ }
+ write_data_rust(&table, &[make_batch(vec![1], vec!["a"])]);
+
+ unsafe {
+ let table = wrap_table(table);
+ let latest_result = paimon_table_latest_snapshot(table);
+ assert!(latest_result.error.is_null());
+ let latest_json =
+ std::slice::from_raw_parts(latest_result.snapshot.data,
latest_result.snapshot.len);
+ let latest: Option<paimon::spec::Snapshot> =
serde_json::from_slice(latest_json).unwrap();
+ assert_eq!(latest.unwrap().id(), 1);
+ paimon_bytes_free(latest_result.snapshot);
+ paimon_table_free(table);
+
+ let catalog = wrap_catalog(Arc::new(catalog));
+ let database = CString::new("default").unwrap();
+ let object = CString::new("test").unwrap();
+ let identifier = paimon_identifier_new(database.as_ptr(),
object.as_ptr());
+ assert!(identifier.error.is_null());
+ let tag_name = CString::new("release-1").unwrap();
+
+ let error = paimon_catalog_create_tag(
+ catalog,
+ identifier.identifier,
+ tag_name.as_ptr(),
+ ptr::null(),
+ false,
+ );
+ assert!(error.is_null());
+
+ let tag_result = paimon_catalog_get_tag(catalog,
identifier.identifier, tag_name.as_ptr());
+ assert!(tag_result.error.is_null());
+ let tag_json = std::slice::from_raw_parts(tag_result.tag.data,
tag_result.tag.len);
+ let tag: paimon::api::GetTagResponse =
serde_json::from_slice(tag_json).unwrap();
+ assert_eq!(tag.tag_name, "release-1");
+ assert_eq!(tag.snapshot.id(), 1);
+
+ let explicit_name = CString::new("release-explicit").unwrap();
+ let snapshot_id = 1;
+ let error = paimon_catalog_create_tag(
+ catalog,
+ identifier.identifier,
+ explicit_name.as_ptr(),
+ &snapshot_id,
+ false,
+ );
+ assert!(error.is_null());
+ let missing_snapshot_id = 2;
+ let missing_snapshot_name = CString::new("missing-snapshot").unwrap();
+ let missing = paimon_catalog_create_tag(
+ catalog,
+ identifier.identifier,
+ missing_snapshot_name.as_ptr(),
+ &missing_snapshot_id,
+ false,
+ );
+ assert_eq!((*missing).code, PaimonErrorCode::NotFound as i32);
+ paimon_error_free(missing);
+
+ let duplicate = paimon_catalog_create_tag(
+ catalog,
+ identifier.identifier,
+ tag_name.as_ptr(),
+ ptr::null(),
+ false,
+ );
+ assert_eq!((*duplicate).code, PaimonErrorCode::AlreadyExists as i32);
+ paimon_error_free(duplicate);
+
+ let error = paimon_catalog_delete_tag(catalog, identifier.identifier,
tag_name.as_ptr());
+ assert!(error.is_null());
+ let missing = paimon_catalog_get_tag(catalog, identifier.identifier,
tag_name.as_ptr());
+ assert_eq!((*missing.error).code, PaimonErrorCode::NotFound as i32);
+ paimon_error_free(missing.error);
+
+ let missing = paimon_catalog_delete_tag(catalog,
identifier.identifier, tag_name.as_ptr());
+ assert_eq!((*missing).code, PaimonErrorCode::NotFound as i32);
+ paimon_error_free(missing);
+
+ let error =
+ paimon_catalog_delete_tag(catalog, identifier.identifier,
explicit_name.as_ptr());
+ assert!(error.is_null());
+
+ paimon_bytes_free(tag_result.tag);
+ paimon_identifier_free(identifier.identifier);
+ paimon_catalog_free(catalog);
+ }
+}
+
// =========================================================================
// Read path tests
// =========================================================================
diff --git a/bindings/c/src/types.rs b/bindings/c/src/types.rs
index ede86664..212a60ce 100644
--- a/bindings/c/src/types.rs
+++ b/bindings/c/src/types.rs
@@ -41,6 +41,13 @@ pub struct paimon_bytes {
}
impl paimon_bytes {
+ pub fn empty() -> Self {
+ Self {
+ data: std::ptr::null_mut(),
+ len: 0,
+ }
+ }
+
pub fn new(v: Vec<u8>) -> Self {
let boxed = v.into_boxed_slice();
let len = boxed.len();
diff --git a/bindings/go/catalog.go b/bindings/go/catalog.go
index 9ebba9db..b813a7ca 100644
--- a/bindings/go/catalog.go
+++ b/bindings/go/catalog.go
@@ -72,8 +72,7 @@ func (c *Catalog) GetTable(id Identifier) (*Table, error) {
if c.inner == nil {
return nil, ErrClosed
}
- createIdFn := ffiIdentifierNew.symbol(c.ctx)
- cID, err := createIdFn(id.database, id.object)
+ cID, err := c.newIdentifier(id)
if err != nil {
return nil, err
}
@@ -88,6 +87,10 @@ func (c *Catalog) GetTable(id Identifier) (*Table, error) {
return &Table{ctx: c.ctx, lib: c.lib, inner: inner}, nil
}
+func (c *Catalog) newIdentifier(id Identifier) (*paimonIdentifier, error) {
+ return ffiIdentifierNew.symbol(c.ctx)(id.database, id.object)
+}
+
var ffiCatalogCreate = newFFI(ffiOpts{
sym: "paimon_catalog_create",
rType: &typeResultCatalogNew,
diff --git a/bindings/go/catalog_tag.go b/bindings/go/catalog_tag.go
new file mode 100644
index 00000000..594c4901
--- /dev/null
+++ b/bindings/go/catalog_tag.go
@@ -0,0 +1,263 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package paimon
+
+import (
+ "context"
+ "encoding/json"
+ "runtime"
+ "unsafe"
+
+ "github.com/jupiterrider/ffi"
+)
+
+// CommitKind identifies the change represented by a snapshot.
+type CommitKind string
+
+const (
+ CommitKindAppend CommitKind = "APPEND"
+ CommitKindCompact CommitKind = "COMPACT"
+ CommitKindOverwrite CommitKind = "OVERWRITE"
+ CommitKindAnalyze CommitKind = "ANALYZE"
+)
+
+// Snapshot describes a Paimon table snapshot.
+type Snapshot struct {
+ ID int64 `json:"id"`
+ CommitKind CommitKind `json:"commitKind"`
+ TimeMillis int64 `json:"timeMillis"`
+ TotalRecordCount *int64 `json:"totalRecordCount,omitempty"`
+ DeltaRecordCount *int64 `json:"deltaRecordCount,omitempty"`
+}
+
+// GetTagResponse contains a tag and its snapshot metadata.
+type GetTagResponse struct {
+ TagName string `json:"tagName"`
+ Snapshot Snapshot `json:"snapshot"`
+ TagCreateTime *int64 `json:"tagCreateTime,omitempty"`
+ TagTimeRetained *string `json:"tagTimeRetained,omitempty"`
+}
+
+// LatestSnapshot returns the latest snapshot, or nil when the table is empty.
+func (t *Table) LatestSnapshot() (*Snapshot, error) {
+ if t.inner == nil {
+ return nil, ErrClosed
+ }
+ return ffiTableLatestSnapshot.symbol(t.ctx)(t.inner)
+}
+
+// CreateTag creates a tag for snapshotID, or the latest snapshot when
snapshotID is nil.
+func (c *Catalog) CreateTag(
+ id Identifier,
+ tagName string,
+ snapshotID *int64,
+ ignoreIfExists bool,
+) error {
+ if c.inner == nil {
+ return ErrClosed
+ }
+ cID, err := c.newIdentifier(id)
+ if err != nil {
+ return err
+ }
+ defer ffiIdentifierFree.symbol(c.ctx)(cID)
+ return ffiCatalogCreateTag.symbol(c.ctx)(
+ c.inner,
+ cID,
+ tagName,
+ snapshotID,
+ ignoreIfExists,
+ )
+}
+
+// GetTag returns a tag and its snapshot metadata.
+func (c *Catalog) GetTag(id Identifier, tagName string) (GetTagResponse,
error) {
+ if c.inner == nil {
+ return GetTagResponse{}, ErrClosed
+ }
+ cID, err := c.newIdentifier(id)
+ if err != nil {
+ return GetTagResponse{}, err
+ }
+ defer ffiIdentifierFree.symbol(c.ctx)(cID)
+ return ffiCatalogGetTag.symbol(c.ctx)(c.inner, cID, tagName)
+}
+
+// DeleteTag deletes a tag.
+func (c *Catalog) DeleteTag(id Identifier, tagName string) error {
+ if c.inner == nil {
+ return ErrClosed
+ }
+ cID, err := c.newIdentifier(id)
+ if err != nil {
+ return err
+ }
+ defer ffiIdentifierFree.symbol(c.ctx)(cID)
+ return ffiCatalogDeleteTag.symbol(c.ctx)(c.inner, cID, tagName)
+}
+
+var ffiCatalogCreateTag = newFFI(ffiOpts{
+ sym: "paimon_catalog_create_tag",
+ rType: &ffi.TypePointer,
+ aTypes: []*ffi.Type{
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ &ffi.TypeUint8,
+ },
+}, func(ctx context.Context, ffiCall ffiCall) func(
+ *paimonCatalog,
+ *paimonIdentifier,
+ string,
+ *int64,
+ bool,
+) error {
+ return func(
+ catalog *paimonCatalog,
+ id *paimonIdentifier,
+ tagName string,
+ snapshotID *int64,
+ ignoreIfExists bool,
+ ) error {
+ tagNamePtr, err := bytePtrFromString(tagName)
+ if err != nil {
+ return err
+ }
+ var snapshotIDPtr unsafe.Pointer
+ if snapshotID != nil {
+ snapshotIDPtr = unsafe.Pointer(snapshotID)
+ }
+ ignore := uint8(0)
+ if ignoreIfExists {
+ ignore = 1
+ }
+ var ffiError *paimonError
+ ffiCall(
+ unsafe.Pointer(&ffiError),
+ unsafe.Pointer(&catalog),
+ unsafe.Pointer(&id),
+ unsafe.Pointer(&tagNamePtr),
+ unsafe.Pointer(&snapshotIDPtr),
+ unsafe.Pointer(&ignore),
+ )
+ runtime.KeepAlive(tagNamePtr)
+ runtime.KeepAlive(snapshotID)
+ return parseError(ctx, ffiError)
+ }
+})
+
+var ffiTableLatestSnapshot = newFFI(ffiOpts{
+ sym: "paimon_table_latest_snapshot",
+ rType: &typeResultLatestSnapshot,
+ aTypes: []*ffi.Type{&ffi.TypePointer},
+}, func(ctx context.Context, ffiCall ffiCall) func(*paimonTable) (*Snapshot,
error) {
+ return func(table *paimonTable) (*Snapshot, error) {
+ var result resultLatestSnapshot
+ ffiCall(unsafe.Pointer(&result), unsafe.Pointer(&table))
+ if result.error != nil {
+ return nil, parseError(ctx, result.error)
+ }
+ defer ffiBytesFree.symbol(ctx)(result.snapshot)
+ var snapshot *Snapshot
+ if err := json.Unmarshal(parseBytes(result.snapshot),
&snapshot); err != nil {
+ return nil, err
+ }
+ return snapshot, nil
+ }
+})
+
+var ffiCatalogGetTag = newFFI(ffiOpts{
+ sym: "paimon_catalog_get_tag",
+ rType: &typeResultGetTag,
+ aTypes: []*ffi.Type{&ffi.TypePointer, &ffi.TypePointer,
&ffi.TypePointer},
+}, func(ctx context.Context, ffiCall ffiCall) func(
+ *paimonCatalog,
+ *paimonIdentifier,
+ string,
+) (GetTagResponse, error) {
+ return func(catalog *paimonCatalog, id *paimonIdentifier, tagName
string) (GetTagResponse, error) {
+ tagNamePtr, err := bytePtrFromString(tagName)
+ if err != nil {
+ return GetTagResponse{}, err
+ }
+ var result resultGetTag
+ ffiCall(
+ unsafe.Pointer(&result),
+ unsafe.Pointer(&catalog),
+ unsafe.Pointer(&id),
+ unsafe.Pointer(&tagNamePtr),
+ )
+ runtime.KeepAlive(tagNamePtr)
+ if result.error != nil {
+ return GetTagResponse{}, parseError(ctx, result.error)
+ }
+ defer ffiBytesFree.symbol(ctx)(result.tag)
+ var tag GetTagResponse
+ if err := json.Unmarshal(parseBytes(result.tag), &tag); err !=
nil {
+ return GetTagResponse{}, err
+ }
+ return tag, nil
+ }
+})
+
+var ffiCatalogDeleteTag = newFFI(ffiOpts{
+ sym: "paimon_catalog_delete_tag",
+ rType: &ffi.TypePointer,
+ aTypes: []*ffi.Type{
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ },
+}, func(ctx context.Context, ffiCall ffiCall) func(
+ *paimonCatalog,
+ *paimonIdentifier,
+ string,
+) error {
+ return func(
+ catalog *paimonCatalog,
+ id *paimonIdentifier,
+ tagName string,
+ ) error {
+ tagNamePtr, err := bytePtrFromString(tagName)
+ if err != nil {
+ return err
+ }
+ var ffiError *paimonError
+ ffiCall(
+ unsafe.Pointer(&ffiError),
+ unsafe.Pointer(&catalog),
+ unsafe.Pointer(&id),
+ unsafe.Pointer(&tagNamePtr),
+ )
+ runtime.KeepAlive(tagNamePtr)
+ return parseError(ctx, ffiError)
+ }
+})
+
+var ffiBytesFree = newFFI(ffiOpts{
+ sym: "paimon_bytes_free",
+ rType: &ffi.TypeVoid,
+ aTypes: []*ffi.Type{&typePaimonBytes},
+}, func(_ context.Context, ffiCall ffiCall) func(paimonBytes) {
+ return func(value paimonBytes) {
+ ffiCall(nil, unsafe.Pointer(&value))
+ }
+})
diff --git a/bindings/go/tests/catalog_tag_test.go
b/bindings/go/tests/catalog_tag_test.go
new file mode 100644
index 00000000..0a84beb7
--- /dev/null
+++ b/bindings/go/tests/catalog_tag_test.go
@@ -0,0 +1,110 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package paimon_test
+
+import (
+ "errors"
+ "path/filepath"
+ "testing"
+
+ paimon "github.com/apache/paimon-rust/bindings/go"
+)
+
+func TestCatalogTagLifecycle(t *testing.T) {
+ source := filepath.Join("testdata", "map_blob_table")
+ warehouse := t.TempDir()
+ if err := copyDirectory(source, filepath.Join(warehouse, "default.db",
"map_blob_table")); err != nil {
+ t.Fatal(err)
+ }
+ catalog, err := paimon.NewCatalog(map[string]string{"warehouse":
warehouse})
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer catalog.Close()
+
+ id := paimon.NewIdentifier("default", "map_blob_table")
+ table, err := catalog.GetTable(id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer table.Close()
+ latest, err := table.LatestSnapshot()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if latest == nil || latest.ID != 1 {
+ t.Fatalf("latest snapshot = %#v", latest)
+ }
+
+ if err := catalog.CreateTag(id, "release-1", nil, false); err != nil {
+ t.Fatal(err)
+ }
+ tag, err := catalog.GetTag(id, "release-1")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if tag.TagName != "release-1" || tag.Snapshot.ID != 1 {
+ t.Fatalf("unexpected tag: %#v", tag)
+ }
+ if tag.Snapshot.CommitKind != paimon.CommitKindAppend {
+ t.Fatalf("commit kind = %q, want %q", tag.Snapshot.CommitKind,
paimon.CommitKindAppend)
+ }
+ readBuilder, err := table.NewReadBuilderWithOptions(map[string]string{
+ "scan.tag-name": "release-1",
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ readBuilder.Close()
+
+ snapshotID := int64(1)
+ if err := catalog.CreateTag(id, "release-explicit", &snapshotID,
false); err != nil {
+ t.Fatal(err)
+ }
+
+ err = catalog.CreateTag(id, "release-1", nil, false)
+ var paimonErr *paimon.Error
+ if !errors.As(err, &paimonErr) || paimonErr.Code() !=
paimon.CodeAlreadyExist {
+ t.Fatalf("duplicate tag error = %v", err)
+ }
+ if err := catalog.CreateTag(id, "release-1", nil, true); err != nil {
+ t.Fatal(err)
+ }
+
+ if err := catalog.DeleteTag(id, "release-1"); err != nil {
+ t.Fatal(err)
+ }
+ _, err = catalog.GetTag(id, "release-1")
+ if !errors.As(err, &paimonErr) || paimonErr.Code() !=
paimon.CodeNotFound {
+ t.Fatalf("missing tag error = %v", err)
+ }
+ err = catalog.DeleteTag(id, "release-1")
+ if !errors.As(err, &paimonErr) || paimonErr.Code() !=
paimon.CodeNotFound {
+ t.Fatalf("delete missing tag error = %v", err)
+ }
+ if err := catalog.DeleteTag(id, "release-explicit"); err != nil {
+ t.Fatal(err)
+ }
+
+ catalog.Close()
+ if err := catalog.CreateTag(id, "closed", nil, false); !errors.Is(err,
paimon.ErrClosed) {
+ t.Fatalf("closed catalog error = %v", err)
+ }
+}
diff --git a/bindings/go/types.go b/bindings/go/types.go
index 6adbac94..5412bcd6 100644
--- a/bindings/go/types.go
+++ b/bindings/go/types.go
@@ -74,6 +74,15 @@ var (
}[0],
}
+ typePaimonBytes = ffi.Type{
+ Type: ffi.Struct,
+ Elements: &[]*ffi.Type{
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ nil,
+ }[0],
+ }
+
typeResultReadBlobs = ffi.Type{
Type: ffi.Struct,
Elements: &[]*ffi.Type{
@@ -105,6 +114,28 @@ var (
}[0],
}
+ // paimon_result_get_tag { tag: paimon_bytes, error: *paimon_error }
+ typeResultGetTag = ffi.Type{
+ Type: ffi.Struct,
+ Elements: &[]*ffi.Type{
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ nil,
+ }[0],
+ }
+
+ // paimon_result_latest_snapshot { snapshot: paimon_bytes, error:
*paimon_error }
+ typeResultLatestSnapshot = ffi.Type{
+ Type: ffi.Struct,
+ Elements: &[]*ffi.Type{
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ &ffi.TypePointer,
+ nil,
+ }[0],
+ }
+
// paimon_result_identifier_new { identifier: paimon_identifier, error:
*paimon_error }
typeResultIdentifierNew = ffi.Type{
Type: ffi.Struct,
@@ -370,6 +401,16 @@ type resultGetTable struct {
error *paimonError
}
+type resultGetTag struct {
+ tag paimonBytes
+ error *paimonError
+}
+
+type resultLatestSnapshot struct {
+ snapshot paimonBytes
+ error *paimonError
+}
+
type resultIdentifierNew struct {
identifier *paimonIdentifier
error *paimonError
diff --git a/docs/src/go-binding.md b/docs/src/go-binding.md
index 40acda2d..025f8e54 100644
--- a/docs/src/go-binding.md
+++ b/docs/src/go-binding.md
@@ -292,6 +292,44 @@ catalog, err := paimon.NewCatalog(map[string]string{
})
```
+## Managing Tags
+
+Catalog tags pin table snapshots. Pass `nil` to tag the latest snapshot, then
+read the tag to record the snapshot ID:
+
+```go
+id := paimon.NewIdentifier("default", "my_table")
+if err := catalog.CreateTag(id, "dataset-42", nil, false); err != nil {
+ log.Fatal(err)
+}
+tag, err := catalog.GetTag(id, "dataset-42")
+if err != nil {
+ log.Fatal(err)
+}
+log.Printf("pinned snapshot %d", tag.Snapshot.ID)
+```
+
+Use `table.LatestSnapshot()` when the current snapshot must be inspected before
+tagging; it returns `nil` for an empty table.
+
+Read the pinned snapshot with the existing read options:
+
+```go
+builder, err := table.NewReadBuilderWithOptions(map[string]string{
+ "scan.tag-name": "dataset-42",
+})
+```
+
+To tag a specific snapshot, pass its ID:
+
+```go
+snapshotID := int64(123)
+err := catalog.CreateTag(id, "dataset-43", &snapshotID, false)
+```
+
+The final argument to `CreateTag` ignores an existing tag. Creating tags with
+a retention duration is not yet supported.
+
## Writing a Table
Use `NewWriteBuilder` for ordinary and fixed-bucket tables. The `arrow.Record`
