You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@fluss.apache.org by "fresh-borzoni (via GitHub)" <gi...@apache.org> on 2026/01/11 18:28:19 UTC
[PR] [ISSUE-148] Support drop_table and list_offsets methods in python bin… [fluss-rust]
fresh-borzoni opened a new pull request, #150:
URL: https://github.com/apache/fluss-rust/pull/150
Adds two missing admin operations to achieve feature parity with C++ bindings:
1. Admin.drop_table(table_path, ignore_if_not_exists=False)
- Drop a table from the cluster
- Optional flag to ignore if table doesn't exist
2. Admin.list_offsets(table_path, bucket_ids, offset_type, timestamp=None)
- List offsets for specified buckets
- offset_type: 0=Earliest, 1=Latest, 2=Timestamp
- Returns dict[int, int] mapping bucket_id to offset
Implementation follows existing patterns from create_table() and matches
C++ bindings logic. Both core APIs already existed in admin.rs.
Closes #148
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-148] Support drop_table and list_offsets methods in python bin… [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#issuecomment-3777808664
@luoyuxia PTAL 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-148] Support drop_table and list_offsets methods in python bin… [fluss-rust]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#discussion_r2694731234
##########
bindings/python/src/admin.rs:
##########
@@ -96,6 +96,104 @@ impl FlussAdmin {
})
}
+ /// Drop a table
+ #[pyo3(signature = (table_path, ignore_if_not_exists=false))]
+ pub fn drop_table<'py>(
+ &self,
+ py: Python<'py>,
+ table_path: &TablePath,
+ ignore_if_not_exists: bool,
+ ) -> PyResult<Bound<'py, PyAny>> {
+ let core_table_path = table_path.to_core();
+ let admin = self.__admin.clone();
+
+ future_into_py(py, async move {
+ admin
+ .drop_table(&core_table_path, ignore_if_not_exists)
+ .await
+ .map_err(|e| FlussError::new_err(format!("Failed to drop table: {e}")))?;
+
+ Python::attach(|py| Ok(py.None()))
+ })
+ }
+
+ /// List offsets for buckets.
+ ///
+ /// Args:
+ /// table_path: Path to the table
+ /// bucket_ids: List of bucket IDs to query
+ /// offset_type: Type of offset to retrieve:
+ /// - "earliest" or OffsetType.EARLIEST: Start of the log
+ /// - "latest" or OffsetType.LATEST: End of the log
+ /// - "timestamp" or OffsetType.TIMESTAMP: Offset at given timestamp (requires timestamp arg)
+ /// timestamp: Required when offset_type is "timestamp", ignored otherwise
+ ///
+ /// Returns:
+ /// dict[int, int]: Mapping of bucket_id -> offset
+ ///
+ /// Example:
+ /// >>> offsets = await admin.list_offsets(table_path, [0, 1], "latest")
+ /// >>> print(offsets) # {0: 100, 1: 150}
+ #[pyo3(signature = (table_path, bucket_ids, offset_type, timestamp=None))]
+ pub fn list_offsets<'py>(
+ &self,
+ py: Python<'py>,
+ table_path: &TablePath,
+ bucket_ids: Vec<i32>,
+ offset_type: &str,
+ timestamp: Option<i64>,
+ ) -> PyResult<Bound<'py, PyAny>> {
+ use fcore::rpc::message::OffsetSpec;
+
+ // Validate bucket IDs
+ for &bucket_id in &bucket_ids {
+ if bucket_id < 0 {
+ return Err(FlussError::new_err(format!(
+ "Invalid bucket_id: {}. Bucket IDs must be non-negative",
+ bucket_id
+ )));
+ }
+ }
+
+ let core_table_path = table_path.to_core();
Review Comment:
Inconsistent pattern: `to_core()` is called without `.clone()`, while other methods in this file call `.clone()` after `to_core()`. For consistency with the pattern in `create_table()` (line 41) and `get_table()` (line 61), this should be `table_path.to_core().clone()`.
```suggestion
let core_table_path = table_path.to_core().clone();
```
##########
bindings/python/src/admin.rs:
##########
@@ -96,6 +96,104 @@ impl FlussAdmin {
})
}
+ /// Drop a table
+ #[pyo3(signature = (table_path, ignore_if_not_exists=false))]
+ pub fn drop_table<'py>(
+ &self,
+ py: Python<'py>,
+ table_path: &TablePath,
+ ignore_if_not_exists: bool,
+ ) -> PyResult<Bound<'py, PyAny>> {
+ let core_table_path = table_path.to_core();
Review Comment:
Inconsistent pattern: `to_core()` is called without `.clone()`, while other methods in this file call `.clone()` after `to_core()`. For consistency with the pattern in `create_table()` (line 41) and `get_table()` (line 61), this should be `table_path.to_core().clone()` or the variable should be cloned after assignment.
```suggestion
let core_table_path = table_path.to_core().clone();
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support drop_table, partitions and offsets methods in python bindings [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#issuecomment-3856832023
@fresh-borzoni Could you please rebase it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-148] Support drop_table and list_offsets methods in python bin… [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#issuecomment-3848716140
I'll add partition support, since I see we added this to cpp bindings
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support drop_table, partitions and offsets methods in python bindings [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia commented on PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#issuecomment-3853891761
@fresh-borzoni Sorry for miss it. I'll review this weekend.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-148] Support drop_table and list_offsets methods in python bin… [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#issuecomment-3757875228
resolved conflict
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-148] Support drop_table and list_offsets methods in python bin… [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#discussion_r2694801910
##########
bindings/python/src/admin.rs:
##########
@@ -38,7 +38,7 @@ impl FlussAdmin {
) -> PyResult<Bound<'py, PyAny>> {
let ignore = ignore_if_exists.unwrap_or(false);
- let core_table_path = table_path.to_core().clone();
+ let core_table_path = table_path.to_core();
let core_descriptor = table_descriptor.to_core().clone();
Review Comment:
here it's needed, it's reference
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support drop_table, partitions and offsets methods in python bindings [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#issuecomment-3856927448
@luoyuxia rebased and renamed `subscribe_batch`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support drop_table, partitions and offsets methods in python bindings [fluss-rust]
Posted by "luoyuxia (via GitHub)" <gi...@apache.org>.
luoyuxia merged PR #150:
URL: https://github.com/apache/fluss-rust/pull/150
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-148] Support drop_table and list_offsets methods in python bin… [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#issuecomment-3853158701
@luoyuxia I've reworked this PR to support partitioned tables and have the same methods as CPP bindings have atm.
PTAL 🙏
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support drop_table, partitions and offsets methods in python bindings [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#issuecomment-3853857178
I think #246 should be merged first and then I'll rebase this one one more time.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Support drop_table, partitions and offsets methods in python bindings [fluss-rust]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#discussion_r2771571891
##########
bindings/python/src/table.rs:
##########
@@ -1914,9 +1900,215 @@ impl LogScanner {
table_info,
projected_schema,
projected_row_type,
- start_timestamp: None,
- end_timestamp: None,
+ partition_name_cache: std::sync::RwLock::new(None),
+ }
+ }
+
+ /// Get partition_id -> partition_name mapping, using cache if available
+ fn get_partition_name_map(
+ &self,
+ py: Python,
+ table_path: &fcore::metadata::TablePath,
+ ) -> PyResult<HashMap<i64, String>> {
+ // Check cache first (read lock)
+ {
+ let cache = self.partition_name_cache.read().unwrap();
+ if let Some(map) = cache.as_ref() {
+ return Ok(map.clone());
+ }
+ }
+
+ // Fetch partition infos (releases GIL during async call)
+ let partition_infos: Vec<fcore::metadata::PartitionInfo> = py
+ .detach(|| {
+ TOKIO_RUNTIME.block_on(async { self.admin.list_partition_infos(table_path).await })
+ })
+ .map_err(|e| FlussError::new_err(format!("Failed to list partition infos: {e}")))?;
+
+ // Build and cache the mapping
+ let map: HashMap<i64, String> = partition_infos
+ .into_iter()
+ .map(|info| (info.get_partition_id(), info.get_partition_name()))
+ .collect();
+
+ // Store in cache (write lock)
+ {
+ let mut cache = self.partition_name_cache.write().unwrap();
Review Comment:
The `unwrap()` calls on `RwLock::read()` and `RwLock::write()` will panic if the lock is poisoned (which happens when a thread panics while holding the lock). While poison is rare in Python bindings due to controlled execution, it's better to handle this gracefully. Consider using `expect()` with a descriptive message or mapping to a PyErr using `map_err()` instead of `unwrap()`.
##########
bindings/python/src/table.rs:
##########
@@ -1914,9 +1900,215 @@ impl LogScanner {
table_info,
projected_schema,
projected_row_type,
- start_timestamp: None,
- end_timestamp: None,
+ partition_name_cache: std::sync::RwLock::new(None),
+ }
+ }
+
+ /// Get partition_id -> partition_name mapping, using cache if available
+ fn get_partition_name_map(
+ &self,
+ py: Python,
+ table_path: &fcore::metadata::TablePath,
+ ) -> PyResult<HashMap<i64, String>> {
+ // Check cache first (read lock)
+ {
+ let cache = self.partition_name_cache.read().unwrap();
+ if let Some(map) = cache.as_ref() {
+ return Ok(map.clone());
+ }
+ }
+
+ // Fetch partition infos (releases GIL during async call)
+ let partition_infos: Vec<fcore::metadata::PartitionInfo> = py
+ .detach(|| {
+ TOKIO_RUNTIME.block_on(async { self.admin.list_partition_infos(table_path).await })
+ })
+ .map_err(|e| FlussError::new_err(format!("Failed to list partition infos: {e}")))?;
+
+ // Build and cache the mapping
+ let map: HashMap<i64, String> = partition_infos
+ .into_iter()
+ .map(|info| (info.get_partition_id(), info.get_partition_name()))
+ .collect();
+
+ // Store in cache (write lock)
+ {
+ let mut cache = self.partition_name_cache.write().unwrap();
+ *cache = Some(map.clone());
+ }
+
+ Ok(map)
Review Comment:
The partition name cache is never invalidated. If partitions are added or deleted during the lifetime of a LogScanner, the cache will become stale and may cause incorrect behavior or errors. Consider either:
1. Removing the cache if LogScanner instances are short-lived
2. Adding a cache invalidation mechanism if partitions can be dynamically created/dropped
3. Adding a TTL or max-age for cached entries
Additionally, there's a potential race condition: if multiple threads call this method concurrently before the cache is populated, they may all fetch partition infos simultaneously. Consider using a synchronization primitive like `once_cell::sync::OnceCell` or check-before-write pattern with double-checked locking.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [ISSUE-148] Support drop_table and list_offsets methods in python bin… [fluss-rust]
Posted by "fresh-borzoni (via GitHub)" <gi...@apache.org>.
fresh-borzoni commented on code in PR #150:
URL: https://github.com/apache/fluss-rust/pull/150#discussion_r2694777692
##########
bindings/python/src/admin.rs:
##########
@@ -96,6 +96,104 @@ impl FlussAdmin {
})
}
+ /// Drop a table
+ #[pyo3(signature = (table_path, ignore_if_not_exists=false))]
+ pub fn drop_table<'py>(
+ &self,
+ py: Python<'py>,
+ table_path: &TablePath,
+ ignore_if_not_exists: bool,
+ ) -> PyResult<Bound<'py, PyAny>> {
+ let core_table_path = table_path.to_core();
Review Comment:
well, the other way around.
to_core already returns owned copy. I'll cleanup other occurences, np
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@fluss.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org