| # 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. |
| |
| """Type stubs for Fluss Python bindings.""" |
| |
| from enum import IntEnum |
| from types import TracebackType |
| from typing import Dict, Iterator, List, Optional, Tuple, Union, overload |
| |
| import pandas as pd |
| import pyarrow as pa |
| |
| class ChangeType(IntEnum): |
| """Represents the type of change for a record in a log.""" |
| |
| AppendOnly = 0 |
| """Append-only operation""" |
| Insert = 1 |
| """Insert operation""" |
| UpdateBefore = 2 |
| """Update operation containing the previous content of the updated row""" |
| UpdateAfter = 3 |
| """Update operation containing the new content of the updated row""" |
| Delete = 4 |
| """Delete operation""" |
| |
| def short_string(self) -> str: |
| """Returns a short string representation (+A, +I, -U, +U, -D).""" |
| ... |
| |
| class ScanRecord: |
| """Represents a single scan record with metadata. |
| |
| The bucket is the key in ScanRecords, not on the individual record |
| (matches Rust/Java). |
| """ |
| |
| @property |
| def offset(self) -> int: |
| """The position of this record in the log.""" |
| ... |
| @property |
| def timestamp(self) -> int: |
| """The timestamp of this record.""" |
| ... |
| @property |
| def change_type(self) -> ChangeType: |
| """The type of change (insert, update, delete, etc.).""" |
| ... |
| @property |
| def row(self) -> Dict[str, object]: |
| """The row data as a dictionary mapping column names to values.""" |
| ... |
| def __str__(self) -> str: ... |
| def __repr__(self) -> str: ... |
| |
| class RecordBatch: |
| """Represents a batch of records with metadata.""" |
| |
| @property |
| def batch(self) -> pa.RecordBatch: |
| """The Arrow RecordBatch containing the data.""" |
| ... |
| @property |
| def bucket(self) -> TableBucket: |
| """The bucket this batch belongs to.""" |
| ... |
| @property |
| def base_offset(self) -> int: |
| """The offset of the first record in this batch.""" |
| ... |
| @property |
| def last_offset(self) -> int: |
| """The offset of the last record in this batch.""" |
| ... |
| def __str__(self) -> str: ... |
| def __repr__(self) -> str: ... |
| |
| class ScanRecords: |
| """A collection of scan records grouped by bucket. |
| |
| Returned by ``LogScanner.poll()``. Supports flat iteration |
| (``for rec in records``) and per-bucket access (``records.records(bucket)``). |
| """ |
| |
| def buckets(self) -> List[TableBucket]: |
| """List of distinct buckets that have records.""" |
| ... |
| def records(self, bucket: TableBucket) -> List[ScanRecord]: |
| """Get records for a specific bucket. Returns empty list if bucket not present.""" |
| ... |
| def count(self) -> int: |
| """Total number of records across all buckets.""" |
| ... |
| def is_empty(self) -> bool: |
| """Whether the result set is empty.""" |
| ... |
| def keys(self) -> List[TableBucket]: |
| """Mapping protocol: alias for ``buckets()``.""" |
| ... |
| def values(self) -> Iterator[List[ScanRecord]]: |
| """Mapping protocol: lazy iterator over record lists, one per bucket.""" |
| ... |
| def items(self) -> Iterator[Tuple[TableBucket, List[ScanRecord]]]: |
| """Mapping protocol: lazy iterator over ``(bucket, records)`` pairs.""" |
| ... |
| def __len__(self) -> int: ... |
| @overload |
| def __getitem__(self, index: int) -> ScanRecord: ... |
| @overload |
| def __getitem__(self, index: slice) -> List[ScanRecord]: ... |
| @overload |
| def __getitem__(self, bucket: TableBucket) -> List[ScanRecord]: ... |
| def __getitem__(self, key: Union[int, slice, TableBucket]) -> Union[ScanRecord, List[ScanRecord]]: ... |
| def __contains__(self, bucket: TableBucket) -> bool: ... |
| def __iter__(self) -> Iterator[ScanRecord]: ... |
| def __str__(self) -> str: ... |
| def __repr__(self) -> str: ... |
| |
| class Config: |
| def __init__(self, properties: Optional[Dict[str, str]] = None) -> None: ... |
| @property |
| def bootstrap_servers(self) -> str: ... |
| @bootstrap_servers.setter |
| def bootstrap_servers(self, server: str) -> None: ... |
| @property |
| def writer_request_max_size(self) -> int: ... |
| @writer_request_max_size.setter |
| def writer_request_max_size(self, size: int) -> None: ... |
| @property |
| def writer_acks(self) -> str: ... |
| @writer_acks.setter |
| def writer_acks(self, acks: str) -> None: ... |
| @property |
| def writer_retries(self) -> int: ... |
| @writer_retries.setter |
| def writer_retries(self, retries: int) -> None: ... |
| @property |
| def writer_batch_size(self) -> int: ... |
| @writer_batch_size.setter |
| def writer_batch_size(self, size: int) -> None: ... |
| @property |
| def writer_bucket_no_key_assigner(self) -> str: ... |
| @writer_bucket_no_key_assigner.setter |
| def writer_bucket_no_key_assigner(self, value: str) -> None: ... |
| @property |
| def scanner_remote_log_prefetch_num(self) -> int: ... |
| @scanner_remote_log_prefetch_num.setter |
| def scanner_remote_log_prefetch_num(self, num: int) -> None: ... |
| @property |
| def remote_file_download_thread_num(self) -> int: ... |
| @remote_file_download_thread_num.setter |
| def remote_file_download_thread_num(self, num: int) -> None: ... |
| @property |
| def scanner_remote_log_read_concurrency(self) -> int: ... |
| @scanner_remote_log_read_concurrency.setter |
| def scanner_remote_log_read_concurrency(self, num: int) -> None: ... |
| @property |
| def scanner_log_max_poll_records(self) -> int: ... |
| @scanner_log_max_poll_records.setter |
| def scanner_log_max_poll_records(self, num: int) -> None: ... |
| @property |
| def scanner_log_fetch_max_bytes(self) -> int: ... |
| @scanner_log_fetch_max_bytes.setter |
| def scanner_log_fetch_max_bytes(self, bytes: int) -> None: ... |
| @property |
| def scanner_log_fetch_min_bytes(self) -> int: ... |
| @scanner_log_fetch_min_bytes.setter |
| def scanner_log_fetch_min_bytes(self, bytes: int) -> None: ... |
| @property |
| def scanner_log_fetch_wait_max_time_ms(self) -> int: ... |
| @scanner_log_fetch_wait_max_time_ms.setter |
| def scanner_log_fetch_wait_max_time_ms(self, ms: int) -> None: ... |
| @property |
| def scanner_log_fetch_max_bytes_for_bucket(self) -> int: ... |
| @scanner_log_fetch_max_bytes_for_bucket.setter |
| def scanner_log_fetch_max_bytes_for_bucket(self, bytes: int) -> None: ... |
| @property |
| def writer_batch_timeout_ms(self) -> int: ... |
| @writer_batch_timeout_ms.setter |
| def writer_batch_timeout_ms(self, timeout: int) -> None: ... |
| @property |
| def writer_enable_idempotence(self) -> bool: ... |
| @writer_enable_idempotence.setter |
| def writer_enable_idempotence(self, enabled: bool) -> None: ... |
| @property |
| def writer_max_inflight_requests_per_bucket(self) -> int: ... |
| @writer_max_inflight_requests_per_bucket.setter |
| def writer_max_inflight_requests_per_bucket(self, num: int) -> None: ... |
| @property |
| def writer_buffer_memory_size(self) -> int: ... |
| @writer_buffer_memory_size.setter |
| def writer_buffer_memory_size(self, size: int) -> None: ... |
| @property |
| def writer_buffer_wait_timeout_ms(self) -> int: ... |
| @writer_buffer_wait_timeout_ms.setter |
| def writer_buffer_wait_timeout_ms(self, timeout: int) -> None: ... |
| @property |
| def connect_timeout_ms(self) -> int: ... |
| @connect_timeout_ms.setter |
| def connect_timeout_ms(self, timeout: int) -> None: ... |
| @property |
| def security_protocol(self) -> str: ... |
| @security_protocol.setter |
| def security_protocol(self, protocol: str) -> None: ... |
| @property |
| def security_sasl_mechanism(self) -> str: ... |
| @security_sasl_mechanism.setter |
| def security_sasl_mechanism(self, mechanism: str) -> None: ... |
| @property |
| def security_sasl_username(self) -> str: ... |
| @security_sasl_username.setter |
| def security_sasl_username(self, username: str) -> None: ... |
| @property |
| def security_sasl_password(self) -> str: ... |
| @security_sasl_password.setter |
| def security_sasl_password(self, password: str) -> None: ... |
| |
| class FlussConnection: |
| @staticmethod |
| async def create(config: Config) -> FlussConnection: ... |
| async def get_admin(self) -> FlussAdmin: ... |
| async def get_table(self, table_path: TablePath) -> FlussTable: ... |
| def close(self) -> None: ... |
| def __enter__(self) -> FlussConnection: ... |
| def __exit__( |
| self, |
| exc_type: Optional[type], |
| exc_value: Optional[BaseException], |
| traceback: Optional[TracebackType], |
| ) -> bool: ... |
| def __repr__(self) -> str: ... |
| |
| class ServerNode: |
| """Information about a server node in the Fluss cluster.""" |
| |
| @property |
| def id(self) -> int: |
| """The server node ID.""" |
| ... |
| @property |
| def host(self) -> str: |
| """The hostname of the server.""" |
| ... |
| @property |
| def port(self) -> int: |
| """The port number of the server.""" |
| ... |
| @property |
| def server_type(self) -> str: |
| """The type of server ('CoordinatorServer' or 'TabletServer').""" |
| ... |
| @property |
| def uid(self) -> str: |
| """The unique identifier of the server (e.g. 'cs-0', 'ts-1').""" |
| ... |
| def __repr__(self) -> str: ... |
| |
| class FlussAdmin: |
| async def create_database( |
| self, |
| database_name: str, |
| database_descriptor: Optional["DatabaseDescriptor"] = None, |
| ignore_if_exists: bool = False, |
| ) -> None: |
| """Create a database.""" |
| ... |
| async def drop_database( |
| self, |
| database_name: str, |
| ignore_if_not_exists: bool = False, |
| cascade: bool = True, |
| ) -> None: |
| """Drop a database.""" |
| ... |
| async def list_databases(self) -> List[str]: |
| """List all databases.""" |
| ... |
| async def database_exists(self, database_name: str) -> bool: |
| """Check if a database exists.""" |
| ... |
| async def get_database_info(self, database_name: str) -> "DatabaseInfo": |
| """Get database information.""" |
| ... |
| async def list_tables(self, database_name: str) -> List[str]: |
| """List all tables in a database.""" |
| ... |
| async def table_exists(self, table_path: TablePath) -> bool: |
| """Check if a table exists.""" |
| ... |
| async def drop_partition( |
| self, |
| table_path: TablePath, |
| partition_spec: Dict[str, str], |
| ignore_if_not_exists: bool = False, |
| ) -> None: |
| """Drop a partition from a partitioned table.""" |
| ... |
| async def create_table( |
| self, |
| table_path: TablePath, |
| table_descriptor: TableDescriptor, |
| ignore_if_exists: Optional[bool] = False, |
| ) -> None: ... |
| async def get_table_info(self, table_path: TablePath) -> TableInfo: ... |
| async def get_latest_lake_snapshot(self, table_path: TablePath) -> LakeSnapshot: ... |
| async def drop_table( |
| self, |
| table_path: TablePath, |
| ignore_if_not_exists: bool = False, |
| ) -> None: ... |
| async def list_offsets( |
| self, |
| table_path: TablePath, |
| bucket_ids: List[int], |
| offset_spec: "OffsetSpec", |
| ) -> Dict[int, int]: |
| """List offsets for the specified buckets. |
| |
| Args: |
| table_path: Path to the table |
| bucket_ids: List of bucket IDs to query |
| offset_spec: Offset specification (OffsetSpec.earliest(), OffsetSpec.latest(), |
| or OffsetSpec.timestamp(ts)) |
| |
| Returns: |
| Dict mapping bucket_id -> offset |
| """ |
| ... |
| async def list_partition_offsets( |
| self, |
| table_path: TablePath, |
| partition_name: str, |
| bucket_ids: List[int], |
| offset_spec: "OffsetSpec", |
| ) -> Dict[int, int]: |
| """List offsets for buckets in a specific partition. |
| |
| Args: |
| table_path: Path to the table |
| partition_name: Partition value (e.g., "US" not "region=US") |
| bucket_ids: List of bucket IDs to query |
| offset_spec: Offset specification (OffsetSpec.earliest(), OffsetSpec.latest(), |
| or OffsetSpec.timestamp(ts)) |
| |
| Returns: |
| Dict mapping bucket_id -> offset |
| """ |
| ... |
| async def create_partition( |
| self, |
| table_path: TablePath, |
| partition_spec: Dict[str, str], |
| ignore_if_exists: bool = False, |
| ) -> None: |
| """Create a partition for a partitioned table. |
| |
| Args: |
| table_path: Path to the table |
| partition_spec: Dict mapping partition column name to value (e.g., {"region": "US"}) |
| ignore_if_exists: If True, don't raise error if partition already exists |
| """ |
| ... |
| async def list_partition_infos( |
| self, |
| table_path: TablePath, |
| partition_spec: Optional[Dict[str, str]] = None, |
| ) -> List["PartitionInfo"]: |
| """List partitions for a partitioned table. |
| |
| Args: |
| table_path: Path to the table |
| partition_spec: Optional partial partition spec to filter results. |
| Dict mapping partition column name to value (e.g., {"region": "US"}). |
| If None, returns all partitions. |
| |
| Returns: |
| List of PartitionInfo objects |
| """ |
| ... |
| async def get_server_nodes(self) -> List[ServerNode]: |
| """Get all alive server nodes in the cluster. |
| |
| Returns: |
| List of ServerNode objects (coordinator and tablet servers) |
| """ |
| ... |
| def __repr__(self) -> str: ... |
| |
| |
| class DatabaseDescriptor: |
| """Descriptor for a Fluss database (comment and custom properties).""" |
| |
| def __init__( |
| self, |
| comment: Optional[str] = None, |
| custom_properties: Optional[Dict[str, str]] = None, |
| ) -> None: ... |
| @property |
| def comment(self) -> Optional[str]: ... |
| def get_custom_properties(self) -> Dict[str, str]: ... |
| def __repr__(self) -> str: ... |
| |
| |
| class DatabaseInfo: |
| """Information about a Fluss database.""" |
| |
| @property |
| def database_name(self) -> str: ... |
| def get_database_descriptor(self) -> DatabaseDescriptor: ... |
| @property |
| def created_time(self) -> int: ... |
| @property |
| def modified_time(self) -> int: ... |
| def __repr__(self) -> str: ... |
| |
| class TableScan: |
| """Builder for creating log scanners with flexible configuration. |
| |
| Use this builder to configure projection before creating a log scanner. |
| Obtain a TableScan instance via `FlussTable.new_scan()`. |
| |
| Example: |
| ```python |
| # Record-based scanning with projection |
| scanner = await table.new_scan() \\ |
| .project([0, 1, 2]) \\ |
| .create_log_scanner() |
| |
| # Batch-based scanning with column names |
| scanner = await table.new_scan() \\ |
| .project_by_name(["id", "name"]) \\ |
| .create_record_batch_log_scanner() |
| ``` |
| """ |
| |
| def project(self, indices: List[int]) -> "TableScan": |
| """Project to specific columns by their indices. |
| |
| Args: |
| indices: List of column indices (0-based) to include in the scan. |
| |
| Returns: |
| Self for method chaining. |
| """ |
| ... |
| def project_by_name(self, names: List[str]) -> "TableScan": |
| """Project to specific columns by their names. |
| |
| Args: |
| names: List of column names to include in the scan. |
| |
| Returns: |
| Self for method chaining. |
| """ |
| ... |
| async def create_log_scanner(self) -> LogScanner: |
| """Create a record-based log scanner. |
| |
| Use this scanner with `poll()` to get individual records with metadata |
| (offset, timestamp, change_type). |
| |
| Returns: |
| LogScanner for record-by-record scanning with `poll()` |
| """ |
| ... |
| async def create_record_batch_log_scanner(self) -> LogScanner: |
| """Create a batch-based log scanner. |
| |
| Use this scanner with `poll_arrow()` to get Arrow Tables, or with |
| `poll_record_batch()` to get individual batches with metadata. |
| |
| Returns: |
| LogScanner for batch-based scanning with `poll_arrow()` or `poll_record_batch()` |
| """ |
| ... |
| def __repr__(self) -> str: ... |
| |
| class FlussTable: |
| def new_scan(self) -> TableScan: |
| """Create a new table scan builder for configuring and creating log scanners. |
| |
| Use this method to create scanners with the builder pattern: |
| |
| Example: |
| ```python |
| # Record-based scanning |
| scanner = await table.new_scan() \\ |
| .project([0, 1]) \\ |
| .create_log_scanner() |
| |
| # Batch-based scanning |
| scanner = await table.new_scan() \\ |
| .project_by_name(["id", "name"]) \\ |
| .create_record_batch_log_scanner() |
| ``` |
| |
| Returns: |
| TableScan builder for configuring the scanner. |
| """ |
| ... |
| def new_append(self) -> TableAppend: ... |
| def new_upsert(self) -> TableUpsert: ... |
| def new_lookup(self) -> TableLookup: ... |
| def get_table_info(self) -> TableInfo: ... |
| def get_table_path(self) -> TablePath: ... |
| def has_primary_key(self) -> bool: ... |
| def __repr__(self) -> str: ... |
| |
| class TableAppend: |
| """Builder for creating an AppendWriter. |
| |
| Obtain via `FlussTable.new_append()`, then call `create_writer()`. |
| |
| Example: |
| writer = table.new_append().create_writer() |
| """ |
| |
| def create_writer(self) -> AppendWriter: ... |
| def __repr__(self) -> str: ... |
| |
| class TableUpsert: |
| """Builder for creating an UpsertWriter, with optional partial update. |
| |
| Obtain via `FlussTable.new_upsert()`, then optionally call |
| `partial_update_by_name()` or `partial_update_by_index()`, |
| then call `create_writer()`. |
| |
| Example: |
| # Full row upsert |
| writer = table.new_upsert().create_writer() |
| |
| # Partial update by column names |
| writer = table.new_upsert().partial_update_by_name(["col1", "col2"]).create_writer() |
| |
| # Partial update by column indices |
| writer = table.new_upsert().partial_update_by_index([0, 1]).create_writer() |
| """ |
| |
| def partial_update_by_name(self, columns: List[str]) -> "TableUpsert": ... |
| def partial_update_by_index(self, column_indices: List[int]) -> "TableUpsert": ... |
| def create_writer(self) -> UpsertWriter: ... |
| def __repr__(self) -> str: ... |
| |
| class TableLookup: |
| """Builder for creating a Lookuper. |
| |
| Obtain via `FlussTable.new_lookup()`, then call `create_lookuper()`. |
| |
| Example: |
| lookuper = table.new_lookup().create_lookuper() |
| """ |
| |
| def create_lookuper(self) -> Lookuper: ... |
| def __repr__(self) -> str: ... |
| |
| class AppendWriter: |
| def append(self, row: dict | list | tuple) -> WriteResultHandle: |
| """Append a single row to the table. |
| |
| Args: |
| row: Dictionary mapping field names to values, or |
| list/tuple of values in schema order |
| |
| Returns: |
| WriteResultHandle: Ignore for fire-and-forget, or await handle.wait() for acknowledgement. |
| |
| Supported Types: |
| - Boolean, TinyInt, SmallInt, Int, BigInt (integers) |
| - Float, Double (floating point) |
| - String, Char (text) |
| - Bytes, Binary (binary data) |
| - Date, Time, Timestamp, TimestampLTZ (temporal) |
| - Decimal (arbitrary precision) |
| - Null values |
| |
| Example: |
| writer.append({'id': 1, 'name': 'Alice', 'score': 95.5}) |
| writer.append([1, 'Alice', 95.5]) |
| |
| Note: |
| For high-throughput bulk loading, prefer write_arrow_batch(). |
| Use flush() to ensure all queued records are sent and acknowledged. |
| """ |
| ... |
| def write_arrow(self, table: pa.Table) -> None: ... |
| def write_arrow_batch(self, batch: pa.RecordBatch) -> WriteResultHandle: ... |
| def write_pandas(self, df: pd.DataFrame) -> None: ... |
| async def flush(self) -> None: ... |
| def __repr__(self) -> str: ... |
| |
| class UpsertWriter: |
| """Writer for upserting and deleting data in a Fluss primary key table.""" |
| |
| def upsert(self, row: dict | list | tuple) -> WriteResultHandle: |
| """Upsert a row into the table. |
| |
| If a row with the same primary key exists, it will be updated. |
| Otherwise, a new row will be inserted. |
| |
| Args: |
| row: Dictionary mapping field names to values, or |
| list/tuple of values in schema order |
| |
| Returns: |
| WriteResultHandle: Ignore for fire-and-forget, or await handle.wait() for ack. |
| """ |
| ... |
| def delete(self, pk: dict | list | tuple) -> WriteResultHandle: |
| """Delete a row from the table by primary key. |
| |
| Args: |
| pk: Dictionary with PK column names as keys, or |
| list/tuple of PK values in PK column order |
| |
| Returns: |
| WriteResultHandle: Ignore for fire-and-forget, or await handle.wait() for ack. |
| """ |
| ... |
| async def flush(self) -> None: |
| """Flush all pending upsert/delete operations to the server.""" |
| ... |
| def __repr__(self) -> str: ... |
| |
| |
| class WriteResultHandle: |
| """Handle for a pending write (append/upsert/delete). Ignore for fire-and-forget, or await handle.wait() for ack.""" |
| |
| async def wait(self) -> None: |
| """Wait for server acknowledgment of this write.""" |
| ... |
| def __repr__(self) -> str: ... |
| |
| |
| class Lookuper: |
| """Lookuper for performing primary key lookups on a Fluss table.""" |
| |
| async def lookup(self, pk: dict | list | tuple) -> Optional[Dict[str, object]]: |
| """Lookup a row by its primary key. |
| |
| Args: |
| pk: Dictionary with PK column names as keys, or |
| list/tuple of PK values in PK column order |
| |
| Returns: |
| A dict containing the row data if found, None otherwise. |
| """ |
| ... |
| def __repr__(self) -> str: ... |
| |
| class LogScanner: |
| """Scanner for reading log data from a Fluss table. |
| |
| This scanner supports two modes: |
| - Record-based scanning via `poll()` - returns individual records with metadata |
| - Batch-based scanning via `poll_arrow()` / `poll_record_batch()` - returns Arrow batches |
| |
| Create scanners using the builder pattern: |
| # Record-based scanning |
| scanner = await table.new_scan().create_log_scanner() |
| |
| # Batch-based scanning |
| scanner = await table.new_scan().create_record_batch_log_scanner() |
| |
| # With projection |
| scanner = await table.new_scan().project([0, 1]).create_log_scanner() |
| """ |
| |
| def subscribe(self, bucket_id: int, start_offset: int) -> None: |
| """Subscribe to a single bucket at a specific offset (non-partitioned tables). |
| |
| Args: |
| bucket_id: The bucket ID to subscribe to |
| start_offset: The offset to start reading from (use EARLIEST_OFFSET for beginning) |
| """ |
| ... |
| def subscribe_buckets(self, bucket_offsets: Dict[int, int]) -> None: |
| """Subscribe to multiple buckets at specified offsets (non-partitioned tables). |
| |
| Args: |
| bucket_offsets: Dict mapping bucket_id -> start_offset |
| """ |
| ... |
| def subscribe_partition( |
| self, partition_id: int, bucket_id: int, start_offset: int |
| ) -> None: |
| """Subscribe to a bucket within a specific partition (partitioned tables only). |
| |
| Args: |
| partition_id: The partition ID (from PartitionInfo.partition_id) |
| bucket_id: The bucket ID within the partition |
| start_offset: The offset to start reading from (use EARLIEST_OFFSET for beginning) |
| """ |
| ... |
| def subscribe_partition_buckets( |
| self, partition_bucket_offsets: Dict[Tuple[int, int], int] |
| ) -> None: |
| """Subscribe to multiple partition+bucket combinations at once (partitioned tables only). |
| |
| Args: |
| partition_bucket_offsets: Dict mapping (partition_id, bucket_id) tuples to start_offsets. |
| Example: {(partition_id_1, 0): EARLIEST_OFFSET, (partition_id_2, 1): 100} |
| """ |
| ... |
| def unsubscribe(self, bucket_id: int) -> None: |
| """Unsubscribe from a specific bucket (non-partitioned tables only). |
| |
| Args: |
| bucket_id: The bucket ID to unsubscribe from |
| """ |
| ... |
| def unsubscribe_partition(self, partition_id: int, bucket_id: int) -> None: |
| """Unsubscribe from a specific partition bucket (partitioned tables only). |
| |
| Args: |
| partition_id: The partition ID to unsubscribe from |
| bucket_id: The bucket ID within the partition |
| """ |
| ... |
| def poll(self, timeout_ms: int) -> ScanRecords: |
| """Poll for individual records with metadata. |
| |
| Requires a record-based scanner (created with new_scan().create_log_scanner()). |
| |
| Args: |
| timeout_ms: Timeout in milliseconds to wait for records. |
| |
| Returns: |
| ScanRecords grouped by bucket. Supports flat iteration |
| (``for rec in records``) and per-bucket access |
| (``records.buckets()``, ``records.records(bucket)``). |
| |
| Note: |
| Returns an empty ScanRecords if no records are available or timeout expires. |
| """ |
| ... |
| def poll_record_batch(self, timeout_ms: int) -> List[RecordBatch]: |
| """Poll for batches with metadata. |
| |
| Requires a batch-based scanner (created with new_scan().create_record_batch_log_scanner()). |
| |
| Args: |
| timeout_ms: Timeout in milliseconds to wait for batches. |
| |
| Returns: |
| List of RecordBatch objects, each containing the Arrow batch along with |
| bucket, base_offset, and last_offset metadata. |
| |
| Note: |
| Returns an empty list if no batches are available or timeout expires. |
| """ |
| ... |
| def poll_arrow(self, timeout_ms: int) -> pa.Table: |
| """Poll for records as an Arrow Table. |
| |
| Requires a batch-based scanner (created with new_scan().create_record_batch_log_scanner()). |
| |
| Args: |
| timeout_ms: Timeout in milliseconds to wait for records. |
| |
| Returns: |
| PyArrow Table containing the polled records (batches merged). |
| |
| Note: |
| Returns an empty table (with correct schema) if no records are available |
| or timeout expires. |
| """ |
| ... |
| def to_pandas(self) -> pd.DataFrame: |
| """Convert all data to Pandas DataFrame. |
| |
| Requires a batch-based scanner (created with new_scan().create_record_batch_log_scanner()). |
| Reads from currently subscribed buckets until reaching their latest offsets. |
| |
| You must call subscribe(), subscribe_buckets(), or subscribe_partition() first. |
| """ |
| ... |
| def to_arrow(self) -> pa.Table: |
| """Convert all data to Arrow Table. |
| |
| Requires a batch-based scanner (created with new_scan().create_record_batch_log_scanner()). |
| Reads from currently subscribed buckets until reaching their latest offsets. |
| |
| You must call subscribe(), subscribe_buckets(), or subscribe_partition() first. |
| """ |
| ... |
| def __repr__(self) -> str: ... |
| |
| class Schema: |
| def __init__( |
| self, schema: pa.Schema, primary_keys: Optional[List[str]] = None |
| ) -> None: ... |
| def get_column_names(self) -> List[str]: ... |
| def get_column_types(self) -> List[str]: ... |
| def get_columns(self) -> List[Tuple[str, str]]: ... |
| def get_primary_keys(self) -> List[str]: ... |
| def __str__(self) -> str: ... |
| |
| class TableDescriptor: |
| def __init__( |
| self, |
| schema: Schema, |
| *, |
| partition_keys: Optional[List[str]] = None, |
| bucket_count: Optional[int] = None, |
| bucket_keys: Optional[List[str]] = None, |
| comment: Optional[str] = None, |
| log_format: Optional[str] = None, |
| kv_format: Optional[str] = None, |
| properties: Optional[Dict[str, str]] = None, |
| custom_properties: Optional[Dict[str, str]] = None, |
| ) -> None: ... |
| def get_schema(self) -> Schema: ... |
| |
| class TablePath: |
| def __init__(self, database: str, table: str) -> None: ... |
| @property |
| def database_name(self) -> str: ... |
| @property |
| def table_name(self) -> str: ... |
| def table_path_str(self) -> str: ... |
| def __str__(self) -> str: ... |
| def __repr__(self) -> str: ... |
| def __hash__(self) -> int: ... |
| def __eq__(self, other: object) -> bool: ... |
| |
| class TableInfo: |
| @property |
| def table_id(self) -> int: ... |
| @property |
| def schema_id(self) -> int: ... |
| @property |
| def created_time(self) -> int: ... |
| @property |
| def modified_time(self) -> int: ... |
| @property |
| def table_path(self) -> TablePath: ... |
| @property |
| def num_buckets(self) -> int: ... |
| @property |
| def comment(self) -> Optional[str]: ... |
| def get_primary_keys(self) -> List[str]: ... |
| def get_bucket_keys(self) -> List[str]: ... |
| def get_partition_keys(self) -> List[str]: ... |
| def has_primary_key(self) -> bool: ... |
| def is_partitioned(self) -> bool: ... |
| def get_properties(self) -> Dict[str, str]: ... |
| def get_custom_properties(self) -> Dict[str, str]: ... |
| def get_schema(self) -> Schema: ... |
| def get_column_names(self) -> List[str]: ... |
| def get_column_count(self) -> int: ... |
| |
| class FlussError(Exception): |
| message: str |
| error_code: int |
| def __init__(self, message: str, error_code: int = -2) -> None: ... |
| def __str__(self) -> str: ... |
| @property |
| def is_retriable(self) -> bool: ... |
| |
| class LakeSnapshot: |
| def __init__(self, snapshot_id: int) -> None: ... |
| @property |
| def snapshot_id(self) -> int: ... |
| @property |
| def table_buckets_offset(self) -> Dict[TableBucket, int]: ... |
| def get_bucket_offset(self, bucket: TableBucket) -> Optional[int]: ... |
| def get_table_buckets(self) -> List[TableBucket]: ... |
| def __str__(self) -> str: ... |
| def __repr__(self) -> str: ... |
| |
| class TableBucket: |
| def __init__(self, table_id: int, bucket: int) -> None: ... |
| @staticmethod |
| def with_partition( |
| table_id: int, partition_id: int, bucket: int |
| ) -> TableBucket: ... |
| @property |
| def table_id(self) -> int: ... |
| @property |
| def bucket_id(self) -> int: ... |
| @property |
| def partition_id(self) -> Optional[int]: ... |
| def __hash__(self) -> int: ... |
| def __eq__(self, other: object) -> bool: ... |
| def __str__(self) -> str: ... |
| def __repr__(self) -> str: ... |
| |
| class PartitionInfo: |
| """Information about a partition.""" |
| |
| @property |
| def partition_id(self) -> int: |
| """Get the partition ID (globally unique in the cluster).""" |
| ... |
| @property |
| def partition_name(self) -> str: |
| """Get the partition name.""" |
| ... |
| def __repr__(self) -> str: ... |
| |
| class ErrorCode: |
| """Named constants for Fluss API error codes. |
| |
| Server API errors have error_code > 0 or == -1. |
| Client-side errors have error_code == CLIENT_ERROR (-2). |
| These constants are convenience names — new server error codes work |
| automatically since error_code is a raw int, not a closed enum. |
| """ |
| |
| CLIENT_ERROR: int |
| NONE: int |
| UNKNOWN_SERVER_ERROR: int |
| NETWORK_EXCEPTION: int |
| UNSUPPORTED_VERSION: int |
| CORRUPT_MESSAGE: int |
| DATABASE_NOT_EXIST: int |
| DATABASE_NOT_EMPTY: int |
| DATABASE_ALREADY_EXIST: int |
| TABLE_NOT_EXIST: int |
| TABLE_ALREADY_EXIST: int |
| SCHEMA_NOT_EXIST: int |
| LOG_STORAGE_EXCEPTION: int |
| KV_STORAGE_EXCEPTION: int |
| NOT_LEADER_OR_FOLLOWER: int |
| RECORD_TOO_LARGE_EXCEPTION: int |
| CORRUPT_RECORD_EXCEPTION: int |
| INVALID_TABLE_EXCEPTION: int |
| INVALID_DATABASE_EXCEPTION: int |
| INVALID_REPLICATION_FACTOR: int |
| INVALID_REQUIRED_ACKS: int |
| LOG_OFFSET_OUT_OF_RANGE_EXCEPTION: int |
| NON_PRIMARY_KEY_TABLE_EXCEPTION: int |
| UNKNOWN_TABLE_OR_BUCKET_EXCEPTION: int |
| INVALID_UPDATE_VERSION_EXCEPTION: int |
| INVALID_COORDINATOR_EXCEPTION: int |
| FENCED_LEADER_EPOCH_EXCEPTION: int |
| REQUEST_TIME_OUT: int |
| STORAGE_EXCEPTION: int |
| OPERATION_NOT_ATTEMPTED_EXCEPTION: int |
| NOT_ENOUGH_REPLICAS_AFTER_APPEND_EXCEPTION: int |
| NOT_ENOUGH_REPLICAS_EXCEPTION: int |
| SECURITY_TOKEN_EXCEPTION: int |
| OUT_OF_ORDER_SEQUENCE_EXCEPTION: int |
| DUPLICATE_SEQUENCE_EXCEPTION: int |
| UNKNOWN_WRITER_ID_EXCEPTION: int |
| INVALID_COLUMN_PROJECTION: int |
| INVALID_TARGET_COLUMN: int |
| PARTITION_NOT_EXISTS: int |
| TABLE_NOT_PARTITIONED_EXCEPTION: int |
| INVALID_TIMESTAMP_EXCEPTION: int |
| INVALID_CONFIG_EXCEPTION: int |
| LAKE_STORAGE_NOT_CONFIGURED_EXCEPTION: int |
| KV_SNAPSHOT_NOT_EXIST: int |
| PARTITION_ALREADY_EXISTS: int |
| PARTITION_SPEC_INVALID_EXCEPTION: int |
| LEADER_NOT_AVAILABLE_EXCEPTION: int |
| PARTITION_MAX_NUM_EXCEPTION: int |
| AUTHENTICATE_EXCEPTION: int |
| SECURITY_DISABLED_EXCEPTION: int |
| AUTHORIZATION_EXCEPTION: int |
| BUCKET_MAX_NUM_EXCEPTION: int |
| FENCED_TIERING_EPOCH_EXCEPTION: int |
| RETRIABLE_AUTHENTICATE_EXCEPTION: int |
| INVALID_SERVER_RACK_INFO_EXCEPTION: int |
| LAKE_SNAPSHOT_NOT_EXIST: int |
| LAKE_TABLE_ALREADY_EXIST: int |
| INELIGIBLE_REPLICA_EXCEPTION: int |
| INVALID_ALTER_TABLE_EXCEPTION: int |
| DELETION_DISABLED_EXCEPTION: int |
| |
| class OffsetSpec: |
| """Offset specification for list_offsets(), matching Java's OffsetSpec. |
| |
| Use factory methods to create instances: |
| OffsetSpec.earliest() |
| OffsetSpec.latest() |
| OffsetSpec.timestamp(ts) |
| """ |
| |
| @staticmethod |
| def earliest() -> "OffsetSpec": |
| """Create an OffsetSpec for the earliest available offset.""" |
| ... |
| @staticmethod |
| def latest() -> "OffsetSpec": |
| """Create an OffsetSpec for the latest available offset.""" |
| ... |
| @staticmethod |
| def timestamp(ts: int) -> "OffsetSpec": |
| """Create an OffsetSpec for the offset at or after the given timestamp.""" |
| ... |
| def __repr__(self) -> str: ... |
| |
| # Constant for earliest offset (-2) |
| EARLIEST_OFFSET: int |
| |
| __version__: str |
