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 ff95db93 [core] Reuse supplied REST data token during native table
initialization (#1076)
ff95db93 is described below
commit ff95db9399a60828d74e032a64b6053ef039cc83
Author: XiaoHongbo <[email protected]>
AuthorDate: Thu Oct 8 20:49:05 2026 +0800
[core] Reuse supplied REST data token during native table initialization
(#1076)
---
.../python/python/pypaimon_rust/datafusion.pyi | 10 ++-
bindings/python/src/table.rs | 33 ++++++++-
.../paimon/src/catalog/rest/rest_token_file_io.rs | 67 +++++++++++++++++-
crates/paimon/src/table/rest_env.rs | 79 ++++++++++++++++++++--
4 files changed, 177 insertions(+), 12 deletions(-)
diff --git a/bindings/python/python/pypaimon_rust/datafusion.pyi
b/bindings/python/python/pypaimon_rust/datafusion.pyi
index ff4453ca..65f7090f 100644
--- a/bindings/python/python/pypaimon_rust/datafusion.pyi
+++ b/bindings/python/python/pypaimon_rust/datafusion.pyi
@@ -199,8 +199,16 @@ class Table:
@staticmethod
def from_rest_response(
response_json: str, *, database: str, table: str, rest_options:
Dict[str, str],
+ initial_data_token: Optional[Tuple[Dict[str, str], int]] = None,
) -> "Table":
- """Reuse matching REST metadata and merged catalog options."""
+ """Reuse REST metadata and optionally an already fetched table data
token."""
+ ...
+ @staticmethod
+ def from_rest_response_with_token(
+ response_json: str, *, database: str, table: str, rest_options:
Dict[str, str],
+ data_token: Dict[str, str], expires_at_millis: int,
+ ) -> "Table":
+ """Reuse REST metadata and a matching table token; Rust refreshes it
on expiry."""
...
def copy_with_resolved_schema(self, schema_json: str, *, branch:
Optional[str] = None) -> "Table":
diff --git a/bindings/python/src/table.rs b/bindings/python/src/table.rs
index 7dbee4d3..2421a0ab 100644
--- a/bindings/python/src/table.rs
+++ b/bindings/python/src/table.rs
@@ -18,7 +18,7 @@
use std::collections::HashMap;
use std::sync::Arc;
-use paimon::catalog::Identifier;
+use paimon::catalog::{Identifier, RESTToken};
use paimon::io::FileIO;
use paimon::spec::TableSchema;
use paimon::Options;
@@ -90,13 +90,14 @@ impl PyTable {
/// Reuse the matching REST table response and merged catalog options.
/// Skips config/get-table requests, preserving REST snapshots and token
refresh.
#[staticmethod]
- #[pyo3(signature = (response_json, *, database, table, rest_options))]
+ #[pyo3(signature = (response_json, *, database, table, rest_options,
initial_data_token=None))]
fn from_rest_response(
py: Python<'_>,
response_json: &str,
database: &str,
table: &str,
rest_options: HashMap<String, String>,
+ initial_data_token: Option<(HashMap<String, String>, i64)>,
) -> PyResult<Self> {
let response: paimon::api::GetTableResponse =
serde_json::from_str(response_json).map_err(|err| {
@@ -104,18 +105,44 @@ impl PyTable {
})?;
let identifier = Identifier::new(database, table);
let rest_options = complete_storage_options(rest_options);
+ let initial_token =
+ initial_data_token.map(|(token, expires_at)| RESTToken::new(token,
expires_at));
let table = py
.detach(|| {
- runtime().block_on(paimon::table::Table::from_rest_response(
+
runtime().block_on(paimon::table::Table::from_rest_response_with_token(
identifier,
response,
Options::from_map(rest_options),
+ initial_token,
))
})
.map_err(to_py_err)?;
Ok(Self::new(Arc::new(table)))
}
+ /// Reuse an already obtained table token during REST table construction.
+ /// Rust keeps its normal refresh path when the token expires.
+ #[staticmethod]
+ #[pyo3(signature = (response_json, *, database, table, rest_options,
data_token, expires_at_millis))]
+ fn from_rest_response_with_token(
+ py: Python<'_>,
+ response_json: &str,
+ database: &str,
+ table: &str,
+ rest_options: HashMap<String, String>,
+ data_token: HashMap<String, String>,
+ expires_at_millis: i64,
+ ) -> PyResult<Self> {
+ Self::from_rest_response(
+ py,
+ response_json,
+ database,
+ table,
+ rest_options,
+ Some((data_token, expires_at_millis)),
+ )
+ }
+
/// Replace the complete schema while retaining FileIO, REST credentials
and branch.
/// The caller has already resolved fields and options; no schema is
reloaded.
#[pyo3(signature = (schema_json, *, branch=None))]
diff --git a/crates/paimon/src/catalog/rest/rest_token_file_io.rs
b/crates/paimon/src/catalog/rest/rest_token_file_io.rs
index 446a56ea..6b115621 100644
--- a/crates/paimon/src/catalog/rest/rest_token_file_io.rs
+++ b/crates/paimon/src/catalog/rest/rest_token_file_io.rs
@@ -52,6 +52,7 @@ pub struct RESTTokenFileIO {
path: String,
catalog_options: Options,
api: Arc<RESTApi>,
+ initial_token: Mutex<Option<RESTToken>>,
state: RwLock<Option<TokenState>>,
refresh_lock: Mutex<()>,
local_cache: Option<Arc<LocalCache>>,
@@ -81,6 +82,7 @@ impl RESTTokenFileIO {
path,
catalog_options,
api,
+ initial_token: Mutex::new(None),
state: RwLock::new(None),
refresh_lock: Mutex::new(()),
local_cache,
@@ -88,6 +90,13 @@ impl RESTTokenFileIO {
}
}
+ /// Reuse a token already obtained by a caller for this table. Normal REST
+ /// refresh remains available when that token approaches expiration.
+ pub(crate) fn with_initial_token(mut self, token: RESTToken) -> Self {
+ self.initial_token = Mutex::new(Some(token));
+ self
+ }
+
pub(crate) async fn build_file_io(self: &Arc<Self>) -> Result<FileIO> {
let file_io = self.current_file_io().await?;
Ok(file_io.with_provider(self.clone()))
@@ -103,7 +112,14 @@ impl RESTTokenFileIO {
return Ok(file_io);
}
- let token = self.refresh_token().await?;
+ let initial_token = self.initial_token.lock().await.take();
+ let token = match initial_token.as_ref() {
+ Some(token) if !Self::is_token_expired(token) => RESTToken::new(
+ self.merge_token_with_catalog_options(token.token.clone()),
+ token.expire_at_millis,
+ ),
+ _ => self.refresh_token().await?,
+ };
let file_io = self.build_static_file_io(&token)?;
*self.state.write().await = Some(TokenState {
token,
@@ -317,6 +333,55 @@ mod tests {
server.abort();
}
+ #[tokio::test]
+ async fn test_initial_token_skips_first_request_and_still_refreshes() {
+ let table_directory = tempfile::tempdir().unwrap();
+ let file_path = table_directory.path().join("data");
+ let (options, api, requests, server) = token_api().await;
+ let file_io_cache = file_io_cache(&options);
+ let now = std::time::SystemTime::now()
+ .duration_since(std::time::UNIX_EPOCH)
+ .unwrap()
+ .as_millis() as i64;
+ let token_file_io = Arc::new(
+ RESTTokenFileIO::new(
+ Identifier::new("database", "table"),
+ table_directory.path().to_string_lossy().into_owned(),
+ options,
+ api,
+ None,
+ file_io_cache,
+ )
+ .with_initial_token(RESTToken::new(
+ HashMap::new(),
+ now + TOKEN_EXPIRATION_SAFE_TIME_MILLIS * 2,
+ )),
+ );
+
+ let file_io = token_file_io.build_file_io().await.unwrap();
+ assert_eq!(requests.load(Ordering::SeqCst), 0);
+ assert!(!file_io
+ .exists(file_path.to_string_lossy().as_ref())
+ .await
+ .unwrap());
+ assert_eq!(requests.load(Ordering::SeqCst), 0);
+
+ token_file_io
+ .state
+ .write()
+ .await
+ .as_mut()
+ .unwrap()
+ .token
+ .expire_at_millis = 0;
+ assert!(!file_io
+ .exists(file_path.to_string_lossy().as_ref())
+ .await
+ .unwrap());
+ assert_eq!(requests.load(Ordering::SeqCst), 1);
+ server.abort();
+ }
+
#[tokio::test]
async fn test_blob_reader_reuses_refreshing_file_io() {
let table_directory = tempfile::tempdir().unwrap();
diff --git a/crates/paimon/src/table/rest_env.rs
b/crates/paimon/src/table/rest_env.rs
index 1b3375bb..fe811404 100644
--- a/crates/paimon/src/table/rest_env.rs
+++ b/crates/paimon/src/table/rest_env.rs
@@ -20,7 +20,7 @@
use crate::api::rest_api::RESTApi;
use crate::api::rest_error::RestError;
use crate::api::GetTableResponse;
-use crate::catalog::{Identifier, RESTTokenFileIO};
+use crate::catalog::{Identifier, RESTToken, RESTTokenFileIO};
use crate::common::{CatalogOptions, Options};
use crate::error::Error;
use crate::io::cache::{create_local_cache_with_namespace, LocalCache};
@@ -38,6 +38,17 @@ impl Table {
identifier: Identifier,
response: crate::api::GetTableResponse,
rest_options: Options,
+ ) -> Result<Self> {
+ Self::from_rest_response_with_token(identifier, response,
rest_options, None).await
+ }
+
+ /// As `from_rest_response`, but reuse a matching table data token already
+ /// fetched by the caller. Rust will refresh it from REST after expiry.
+ pub async fn from_rest_response_with_token(
+ identifier: Identifier,
+ response: crate::api::GetTableResponse,
+ rest_options: Options,
+ initial_token: Option<RESTToken>,
) -> Result<Self> {
identifier.validate()?;
// Reject a stale or misrouted response before initializing auth or
local-cache resources.
@@ -55,7 +66,7 @@ impl Table {
.unwrap_or(false);
let local_cache = create_local_cache_with_namespace(&rest_options,
api.options())?;
let file_io_cache =
FileIOCacheContext::from_props(api.options().to_map())?;
- RESTEnv::build_table(
+ RESTEnv::build_table_with_token(
&identifier,
response,
api,
@@ -63,6 +74,7 @@ impl Table {
data_token_enabled,
local_cache,
file_io_cache,
+ initial_token,
)
.await
}
@@ -245,6 +257,31 @@ impl RESTEnv {
data_token_enabled: bool,
local_cache: Option<Arc<LocalCache>>,
file_io_cache: FileIOCacheContext,
+ ) -> Result<Table> {
+ Self::build_table_with_token(
+ identifier,
+ response,
+ api,
+ options,
+ data_token_enabled,
+ local_cache,
+ file_io_cache,
+ None,
+ )
+ .await
+ }
+
+ // Keep the REST construction inputs explicit, including the optional
initial token.
+ #[allow(clippy::too_many_arguments)]
+ async fn build_table_with_token(
+ identifier: &Identifier,
+ response: crate::api::GetTableResponse,
+ api: Arc<RESTApi>,
+ options: Options,
+ data_token_enabled: bool,
+ local_cache: Option<Arc<LocalCache>>,
+ file_io_cache: FileIOCacheContext,
+ initial_token: Option<RESTToken>,
) -> Result<Table> {
let identifier = response_identifier(identifier, &response)?;
let schema = response.schema.ok_or_else(|| Error::DataInvalid {
@@ -303,7 +340,7 @@ impl RESTEnv {
source: None,
})?;
- let file_io = Self::build_file_io(
+ let file_io = Self::build_file_io_with_token(
&identifier,
&table_path,
api.clone(),
@@ -311,6 +348,7 @@ impl RESTEnv {
data_token_enabled && !is_external,
local_cache.clone(),
file_io_cache.clone(),
+ initial_token,
)
.await?;
@@ -406,18 +444,45 @@ impl RESTEnv {
use_data_token: bool,
local_cache: Option<Arc<LocalCache>>,
file_io_cache: FileIOCacheContext,
+ ) -> Result<FileIO> {
+ Self::build_file_io_with_token(
+ identifier,
+ path,
+ api,
+ options,
+ use_data_token,
+ local_cache,
+ file_io_cache,
+ None,
+ )
+ .await
+ }
+
+ // Match the existing FileIO builder while passing through the optional
initial token.
+ #[allow(clippy::too_many_arguments)]
+ async fn build_file_io_with_token(
+ identifier: &Identifier,
+ path: &str,
+ api: Arc<RESTApi>,
+ options: &Options,
+ use_data_token: bool,
+ local_cache: Option<Arc<LocalCache>>,
+ file_io_cache: FileIOCacheContext,
+ initial_token: Option<RESTToken>,
) -> Result<FileIO> {
if use_data_token {
- return Arc::new(RESTTokenFileIO::new(
+ let mut token_file_io = RESTTokenFileIO::new(
identifier.clone(),
path.to_string(),
options.clone(),
api,
local_cache,
file_io_cache,
- ))
- .build_file_io()
- .await;
+ );
+ if let Some(token) = initial_token {
+ token_file_io = token_file_io.with_initial_token(token);
+ }
+ return Arc::new(token_file_io).build_file_io().await;
}
let mut builder =
FileIO::from_path(path)?.with_props(options.to_map());
