Copilot commented on code in PR #600: URL: https://github.com/apache/fluss-rust/pull/600#discussion_r3366925715
########## README.md: ########## @@ -1,125 +1,326 @@ -# Apache Fluss™ Rust (Incubating) +<!-- + 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. +--> + +# Apache Fluss™ Rust Client (Incubating)  +[](https://crates.io/crates/fluss-rs) +[](https://docs.rs/fluss-rs/) +[](https://www.apache.org/licenses/LICENSE-2.0) + +The official Rust **client** library for [Apache Fluss™](https://fluss.apache.org/) (Incubating) — a streaming storage built for real-time analytics, serving as the real-time data layer for Lakehouse architectures. This is a **client SDK**, not the Fluss server itself. + +This repository contains: + +- **`fluss-rs`** — the Rust core client (crates.io: [`fluss-rs`](https://crates.io/crates/fluss-rs)) +- **Language bindings** — Python, C++, and Elixir clients built on top of `fluss-rs` + +--- + +## What is Fluss? + +[Fluss](https://fluss.apache.org/) bridges the gap between streaming data and the data Lakehouse by enabling **low-latency, high-throughput data ingestion and processing** while seamlessly integrating with popular compute engines (Flink, Spark, Trino). -Rust implementation of [Apache Fluss™](https://fluss.apache.org/). +Key concepts: +- **Log table** — append-only table (no primary key). Immutable records, ideal for event streams and audit trails. +- **Primary Key (KV) table** — keyed table supporting upsert, delete, and point/prefixed lookups. +- **Bucket** — parallelism unit within a table (similar to Kafka partitions). +- **Partition** — data organization by column values (e.g., by date or region). -## Why Fluss? -[Fluss](https://fluss.apache.org/) is a streaming storage built for real-time analytics which can serve as the real-time data layer for Lakehouse architectures. -It bridges the gap between streaming data and the data Lakehouse by enabling low-latency, high-throughput data ingestion and processing while seamlessly integrating with popular compute engines. +--- + +## Features + +### Core Client (`fluss-rs`) + +| Category | Capabilities | +| --------------- | ------------------------------------------------------------------------- | +| **Connection** | Bootstrap to Fluss cluster, SASL authentication, graceful shutdown | +| **Admin** | Create/drop/list databases & tables, manage partitions, list offsets | +| **Log Tables** | Append (single-row + Arrow `RecordBatch`), scan with subscribe/poll | +| **KV Tables** | Upsert, delete, point lookup, **prefix lookup**, partitioned KV support | +| **Data Types** | Int, BigInt, String, Float, Double, Boolean, Bytes, Decimal, Date, Time, Timestamp, TimestampLTZ, Char, Binary | +| **Config** | Batch sizing, buffering, retries, compression, timeouts, prefetch, concurrency | +| **Storage** | Memory, Filesystem, S3, OSS (via [OpenDAL](https://opendal.apache.org/)) | +| **WASM** | Compiles for `wasm32` target | + +### Language Bindings + +| Language | Package / Build | Async Runtime | Data Format | +| ---------- | ------------------------ | ------------------------ | --------------------------- | +| **Rust** | [fluss-rs](https://crates.io/crates/fluss-rs) (crates.io) | Tokio | Arrow `RecordBatch` / `GenericRow` | +| **Python** | Build from source (PyO3) | asyncio | PyArrow / Pandas / dict | +| **C++** | CMake / Bazel (FFI) | Synchronous (Tokio internally) | Arrow RecordBatch / GenericRow | +| **Elixir** | [Rustler](https://github.com/rusterlium/rustler) NIFs | Erlang processes | Elixir values | + +--- + +## Project Structure + +``` +fluss-rust/ +├── crates/ +│ ├── fluss/ # Core Rust client (fluss-rs) +│ │ ├── src/client/ # Connection, Admin, Table, Scan, Upsert, Lookup +│ │ ├── src/metadata/ # Schema, TableDescriptor, DataTypes, Partitions +│ │ ├── src/row/ # GenericRow, InternalRow, Arrow integration +│ │ ├── src/rpc/ # gRPC transport layer +│ │ └── src/config.rs # Client configuration +│ ├── examples/ # 5 runnable examples (log, KV, partitioned, prefix lookup) +│ └── fluss-test-cluster/ # Test harness for integration tests +├── bindings/ +│ ├── python/ # Python binding (PyO3) +│ ├── cpp/ # C++ binding (FFI + header) +│ └── elixir/ # Elixir binding (Rustler NIF) +├── website/ # Docusaurus documentation site +├── docs/ # Supplementary documentation +└── scripts/ # Release & version management +``` -## Why Fluss Rust Client -It's an official Rust client for interacting with Fluss. This client provides foundational capabilities for table management and log streaming operations, enabling developers to explore Fluss within Rust ecosystems. +--- -## Quick-Start +## Quick Start -### Step1 Start Fluss cluster -#### Requirements -Fluss runs on all UNIX-like environments, e.g. Linux, Mac OS X. Before you start to setup the system, make sure you have the following software installed on your test machine: +### Prerequisites -Java 17 or higher (Java 8 and Java 11 are not recommended) -If your cluster does not fulfill these software requirements you will need to install/upgrade it. +- [Java 17+](https://adoptium.net/) for running the Fluss cluster +- [Rust](https://www.rust-lang.org/tools/install) (latest stable) +- Linux or macOS (Windows is not currently supported) -Fluss requires the JAVA_HOME environment variable to be set on all nodes and point to the directory of your Java installation. +### 1. Start a Fluss Cluster -#### Fluss Setup -Go to the [downloads](https://fluss.apache.org/downloads/) page and download the latest Fluss release (currently 0.8.0). Make sure to pick the Fluss package matching your Java version. After downloading the latest release, extract it: ```shell +# Download and extract Fluss (0.8.0+) +curl -LO https://dlcdn.apache.org/incubator/fluss/0.8.0/fluss-0.8.0-incubating-bin.tgz tar -xzf fluss-0.8.0-incubating-bin.tgz cd fluss-0.8.0-incubating/ -``` -You can start Fluss local cluster by running the following command: -```shell + +# Start a local cluster ./bin/local-cluster.sh start ``` -After that, the Fluss local cluster is started. -### Run Provided Example -Only supports Linux or macOs. You will need to [install Rust](https://www.rust-lang.org/tools/install) firstly. +### 2. Add `fluss-rs` to Your Project -After that, go the project directory, build it and run the example: -```shell -cargo build --example example-table --release -cd target/release/examples -./example-table +```toml +[dependencies] +fluss = { package = "fluss-rs", version = "0.2" } +tokio = "1" Review Comment: The `tokio = "1"` dependency line is missing the feature flags required by `#[tokio::main]`, so the Quick Start examples won’t compile when copy/pasted into a new project. ########## README.md: ########## @@ -1,125 +1,326 @@ -# Apache Fluss™ Rust (Incubating) +<!-- + 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. +--> + +# Apache Fluss™ Rust Client (Incubating)  +[](https://crates.io/crates/fluss-rs) +[](https://docs.rs/fluss-rs/) +[](https://www.apache.org/licenses/LICENSE-2.0) + +The official Rust **client** library for [Apache Fluss™](https://fluss.apache.org/) (Incubating) — a streaming storage built for real-time analytics, serving as the real-time data layer for Lakehouse architectures. This is a **client SDK**, not the Fluss server itself. + +This repository contains: + +- **`fluss-rs`** — the Rust core client (crates.io: [`fluss-rs`](https://crates.io/crates/fluss-rs)) +- **Language bindings** — Python, C++, and Elixir clients built on top of `fluss-rs` + +--- + +## What is Fluss? + +[Fluss](https://fluss.apache.org/) bridges the gap between streaming data and the data Lakehouse by enabling **low-latency, high-throughput data ingestion and processing** while seamlessly integrating with popular compute engines (Flink, Spark, Trino). -Rust implementation of [Apache Fluss™](https://fluss.apache.org/). +Key concepts: +- **Log table** — append-only table (no primary key). Immutable records, ideal for event streams and audit trails. +- **Primary Key (KV) table** — keyed table supporting upsert, delete, and point/prefixed lookups. +- **Bucket** — parallelism unit within a table (similar to Kafka partitions). +- **Partition** — data organization by column values (e.g., by date or region). -## Why Fluss? -[Fluss](https://fluss.apache.org/) is a streaming storage built for real-time analytics which can serve as the real-time data layer for Lakehouse architectures. -It bridges the gap between streaming data and the data Lakehouse by enabling low-latency, high-throughput data ingestion and processing while seamlessly integrating with popular compute engines. +--- + +## Features + +### Core Client (`fluss-rs`) + +| Category | Capabilities | +| --------------- | ------------------------------------------------------------------------- | +| **Connection** | Bootstrap to Fluss cluster, SASL authentication, graceful shutdown | +| **Admin** | Create/drop/list databases & tables, manage partitions, list offsets | +| **Log Tables** | Append (single-row + Arrow `RecordBatch`), scan with subscribe/poll | +| **KV Tables** | Upsert, delete, point lookup, **prefix lookup**, partitioned KV support | +| **Data Types** | Int, BigInt, String, Float, Double, Boolean, Bytes, Decimal, Date, Time, Timestamp, TimestampLTZ, Char, Binary | +| **Config** | Batch sizing, buffering, retries, compression, timeouts, prefetch, concurrency | +| **Storage** | Memory, Filesystem, S3, OSS (via [OpenDAL](https://opendal.apache.org/)) | +| **WASM** | Compiles for `wasm32` target | + +### Language Bindings + +| Language | Package / Build | Async Runtime | Data Format | +| ---------- | ------------------------ | ------------------------ | --------------------------- | +| **Rust** | [fluss-rs](https://crates.io/crates/fluss-rs) (crates.io) | Tokio | Arrow `RecordBatch` / `GenericRow` | +| **Python** | Build from source (PyO3) | asyncio | PyArrow / Pandas / dict | +| **C++** | CMake / Bazel (FFI) | Synchronous (Tokio internally) | Arrow RecordBatch / GenericRow | +| **Elixir** | [Rustler](https://github.com/rusterlium/rustler) NIFs | Erlang processes | Elixir values | + +--- + +## Project Structure + +``` +fluss-rust/ +├── crates/ +│ ├── fluss/ # Core Rust client (fluss-rs) +│ │ ├── src/client/ # Connection, Admin, Table, Scan, Upsert, Lookup +│ │ ├── src/metadata/ # Schema, TableDescriptor, DataTypes, Partitions +│ │ ├── src/row/ # GenericRow, InternalRow, Arrow integration +│ │ ├── src/rpc/ # gRPC transport layer +│ │ └── src/config.rs # Client configuration +│ ├── examples/ # 5 runnable examples (log, KV, partitioned, prefix lookup) +│ └── fluss-test-cluster/ # Test harness for integration tests +├── bindings/ +│ ├── python/ # Python binding (PyO3) +│ ├── cpp/ # C++ binding (FFI + header) +│ └── elixir/ # Elixir binding (Rustler NIF) +├── website/ # Docusaurus documentation site +├── docs/ # Supplementary documentation +└── scripts/ # Release & version management +``` -## Why Fluss Rust Client -It's an official Rust client for interacting with Fluss. This client provides foundational capabilities for table management and log streaming operations, enabling developers to explore Fluss within Rust ecosystems. +--- -## Quick-Start +## Quick Start -### Step1 Start Fluss cluster -#### Requirements -Fluss runs on all UNIX-like environments, e.g. Linux, Mac OS X. Before you start to setup the system, make sure you have the following software installed on your test machine: +### Prerequisites -Java 17 or higher (Java 8 and Java 11 are not recommended) -If your cluster does not fulfill these software requirements you will need to install/upgrade it. +- [Java 17+](https://adoptium.net/) for running the Fluss cluster +- [Rust](https://www.rust-lang.org/tools/install) (latest stable) +- Linux or macOS (Windows is not currently supported) -Fluss requires the JAVA_HOME environment variable to be set on all nodes and point to the directory of your Java installation. +### 1. Start a Fluss Cluster -#### Fluss Setup -Go to the [downloads](https://fluss.apache.org/downloads/) page and download the latest Fluss release (currently 0.8.0). Make sure to pick the Fluss package matching your Java version. After downloading the latest release, extract it: ```shell +# Download and extract Fluss (0.8.0+) +curl -LO https://dlcdn.apache.org/incubator/fluss/0.8.0/fluss-0.8.0-incubating-bin.tgz tar -xzf fluss-0.8.0-incubating-bin.tgz cd fluss-0.8.0-incubating/ -``` -You can start Fluss local cluster by running the following command: -```shell + +# Start a local cluster ./bin/local-cluster.sh start ``` -After that, the Fluss local cluster is started. -### Run Provided Example -Only supports Linux or macOs. You will need to [install Rust](https://www.rust-lang.org/tools/install) firstly. +### 2. Add `fluss-rs` to Your Project -After that, go the project directory, build it and run the example: -```shell -cargo build --example example-table --release -cd target/release/examples -./example-table +```toml +[dependencies] +fluss = { package = "fluss-rs", version = "0.2" } +tokio = "1" ``` -The example code is as follows: + +### 3. Write Code + +#### Log Table: Append + Scan + ```rust +use fluss::client::{EARLIEST_OFFSET, FlussConnection}; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; +use std::time::Duration; + #[tokio::main] -pub async fn main() -> Result<()> { - // 1: create the table; - let mut args = Args::default(); - args.bootstrap_servers = "127.0.0.1:9123".to_string(); - let conn_config = ConnectionConfig::from_args(args); - let conn = FlussConnection::new(conn_config).await; - - let admin = conn.get_admin(); - - let table_descriptor = TableDescriptor::builder() - .schema( - Schema::builder() - .column("c1", DataTypes::int()) - .column("c2", DataTypes::string()) - .build(), - ) - .build(); - - let table_path = TablePath::new("fluss".to_owned(), "rust_test".to_owned()); - - admin - .create_table(&table_path, &table_descriptor, true) - .await - .unwrap(); - - // 2: get the table - let table_info = admin.get_table_info(&table_path).await.unwrap(); - print!("Get created table:\n {}\n", table_info); - - // let's sleep 2 seconds to wait leader ready - thread::sleep(Duration::from_secs(2)); - - // 3: append log to the table - let table = conn.get_table(&table_path).await; - let append_writer = table.new_append().create_writer(); - let batch = record_batch!(("c1", Int32, [1, 2, 3, 4, 5, 6]), ("c2", Utf8, ["a1", "a2", "a3", "a4", "a5", "a6"])).unwrap(); - append_writer.append(batch)?; - append_writer.flush().await?; - println!("Start to scan log records......"); - // 4: scan the records - let log_scanner = table.new_scan().create_log_scanner(); - log_scanner.subscribe(0, 0).await; +async fn main() -> Result<()> { + let mut config = Config::default(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + // Create a log table + let table_path = TablePath::new("my_db", "events"); Review Comment: The Log Table Quick Start creates `TablePath::new("my_db", ...)` but never creates the database. `create_table` will fail if the database doesn’t already exist. ########## README.md: ########## @@ -1,125 +1,326 @@ -# Apache Fluss™ Rust (Incubating) +<!-- + 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. +--> + +# Apache Fluss™ Rust Client (Incubating)  +[](https://crates.io/crates/fluss-rs) +[](https://docs.rs/fluss-rs/) +[](https://www.apache.org/licenses/LICENSE-2.0) + +The official Rust **client** library for [Apache Fluss™](https://fluss.apache.org/) (Incubating) — a streaming storage built for real-time analytics, serving as the real-time data layer for Lakehouse architectures. This is a **client SDK**, not the Fluss server itself. + +This repository contains: + +- **`fluss-rs`** — the Rust core client (crates.io: [`fluss-rs`](https://crates.io/crates/fluss-rs)) +- **Language bindings** — Python, C++, and Elixir clients built on top of `fluss-rs` + +--- + +## What is Fluss? + +[Fluss](https://fluss.apache.org/) bridges the gap between streaming data and the data Lakehouse by enabling **low-latency, high-throughput data ingestion and processing** while seamlessly integrating with popular compute engines (Flink, Spark, Trino). -Rust implementation of [Apache Fluss™](https://fluss.apache.org/). +Key concepts: +- **Log table** — append-only table (no primary key). Immutable records, ideal for event streams and audit trails. +- **Primary Key (KV) table** — keyed table supporting upsert, delete, and point/prefixed lookups. +- **Bucket** — parallelism unit within a table (similar to Kafka partitions). +- **Partition** — data organization by column values (e.g., by date or region). -## Why Fluss? -[Fluss](https://fluss.apache.org/) is a streaming storage built for real-time analytics which can serve as the real-time data layer for Lakehouse architectures. -It bridges the gap between streaming data and the data Lakehouse by enabling low-latency, high-throughput data ingestion and processing while seamlessly integrating with popular compute engines. +--- + +## Features + +### Core Client (`fluss-rs`) + +| Category | Capabilities | +| --------------- | ------------------------------------------------------------------------- | +| **Connection** | Bootstrap to Fluss cluster, SASL authentication, graceful shutdown | +| **Admin** | Create/drop/list databases & tables, manage partitions, list offsets | +| **Log Tables** | Append (single-row + Arrow `RecordBatch`), scan with subscribe/poll | +| **KV Tables** | Upsert, delete, point lookup, **prefix lookup**, partitioned KV support | +| **Data Types** | Int, BigInt, String, Float, Double, Boolean, Bytes, Decimal, Date, Time, Timestamp, TimestampLTZ, Char, Binary | +| **Config** | Batch sizing, buffering, retries, compression, timeouts, prefetch, concurrency | +| **Storage** | Memory, Filesystem, S3, OSS (via [OpenDAL](https://opendal.apache.org/)) | +| **WASM** | Compiles for `wasm32` target | + +### Language Bindings + +| Language | Package / Build | Async Runtime | Data Format | +| ---------- | ------------------------ | ------------------------ | --------------------------- | +| **Rust** | [fluss-rs](https://crates.io/crates/fluss-rs) (crates.io) | Tokio | Arrow `RecordBatch` / `GenericRow` | +| **Python** | Build from source (PyO3) | asyncio | PyArrow / Pandas / dict | +| **C++** | CMake / Bazel (FFI) | Synchronous (Tokio internally) | Arrow RecordBatch / GenericRow | +| **Elixir** | [Rustler](https://github.com/rusterlium/rustler) NIFs | Erlang processes | Elixir values | + +--- + +## Project Structure + +``` +fluss-rust/ +├── crates/ +│ ├── fluss/ # Core Rust client (fluss-rs) +│ │ ├── src/client/ # Connection, Admin, Table, Scan, Upsert, Lookup +│ │ ├── src/metadata/ # Schema, TableDescriptor, DataTypes, Partitions +│ │ ├── src/row/ # GenericRow, InternalRow, Arrow integration +│ │ ├── src/rpc/ # gRPC transport layer +│ │ └── src/config.rs # Client configuration +│ ├── examples/ # 5 runnable examples (log, KV, partitioned, prefix lookup) +│ └── fluss-test-cluster/ # Test harness for integration tests +├── bindings/ +│ ├── python/ # Python binding (PyO3) +│ ├── cpp/ # C++ binding (FFI + header) +│ └── elixir/ # Elixir binding (Rustler NIF) +├── website/ # Docusaurus documentation site +├── docs/ # Supplementary documentation +└── scripts/ # Release & version management +``` -## Why Fluss Rust Client -It's an official Rust client for interacting with Fluss. This client provides foundational capabilities for table management and log streaming operations, enabling developers to explore Fluss within Rust ecosystems. +--- -## Quick-Start +## Quick Start -### Step1 Start Fluss cluster -#### Requirements -Fluss runs on all UNIX-like environments, e.g. Linux, Mac OS X. Before you start to setup the system, make sure you have the following software installed on your test machine: +### Prerequisites -Java 17 or higher (Java 8 and Java 11 are not recommended) -If your cluster does not fulfill these software requirements you will need to install/upgrade it. +- [Java 17+](https://adoptium.net/) for running the Fluss cluster +- [Rust](https://www.rust-lang.org/tools/install) (latest stable) +- Linux or macOS (Windows is not currently supported) -Fluss requires the JAVA_HOME environment variable to be set on all nodes and point to the directory of your Java installation. +### 1. Start a Fluss Cluster -#### Fluss Setup -Go to the [downloads](https://fluss.apache.org/downloads/) page and download the latest Fluss release (currently 0.8.0). Make sure to pick the Fluss package matching your Java version. After downloading the latest release, extract it: ```shell +# Download and extract Fluss (0.8.0+) +curl -LO https://dlcdn.apache.org/incubator/fluss/0.8.0/fluss-0.8.0-incubating-bin.tgz tar -xzf fluss-0.8.0-incubating-bin.tgz cd fluss-0.8.0-incubating/ -``` -You can start Fluss local cluster by running the following command: -```shell + +# Start a local cluster ./bin/local-cluster.sh start ``` -After that, the Fluss local cluster is started. -### Run Provided Example -Only supports Linux or macOs. You will need to [install Rust](https://www.rust-lang.org/tools/install) firstly. +### 2. Add `fluss-rs` to Your Project -After that, go the project directory, build it and run the example: -```shell -cargo build --example example-table --release -cd target/release/examples -./example-table +```toml +[dependencies] +fluss = { package = "fluss-rs", version = "0.2" } +tokio = "1" ``` -The example code is as follows: + +### 3. Write Code + +#### Log Table: Append + Scan + ```rust +use fluss::client::{EARLIEST_OFFSET, FlussConnection}; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; +use std::time::Duration; + #[tokio::main] -pub async fn main() -> Result<()> { - // 1: create the table; - let mut args = Args::default(); - args.bootstrap_servers = "127.0.0.1:9123".to_string(); - let conn_config = ConnectionConfig::from_args(args); - let conn = FlussConnection::new(conn_config).await; - - let admin = conn.get_admin(); - - let table_descriptor = TableDescriptor::builder() - .schema( - Schema::builder() - .column("c1", DataTypes::int()) - .column("c2", DataTypes::string()) - .build(), - ) - .build(); - - let table_path = TablePath::new("fluss".to_owned(), "rust_test".to_owned()); - - admin - .create_table(&table_path, &table_descriptor, true) - .await - .unwrap(); - - // 2: get the table - let table_info = admin.get_table_info(&table_path).await.unwrap(); - print!("Get created table:\n {}\n", table_info); - - // let's sleep 2 seconds to wait leader ready - thread::sleep(Duration::from_secs(2)); - - // 3: append log to the table - let table = conn.get_table(&table_path).await; - let append_writer = table.new_append().create_writer(); - let batch = record_batch!(("c1", Int32, [1, 2, 3, 4, 5, 6]), ("c2", Utf8, ["a1", "a2", "a3", "a4", "a5", "a6"])).unwrap(); - append_writer.append(batch)?; - append_writer.flush().await?; - println!("Start to scan log records......"); - // 4: scan the records - let log_scanner = table.new_scan().create_log_scanner(); - log_scanner.subscribe(0, 0).await; +async fn main() -> Result<()> { + let mut config = Config::default(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + // Create a log table + let table_path = TablePath::new("my_db", "events"); + let schema = Schema::builder() + .column("ts", DataTypes::bigint()) + .column("message", DataTypes::string()) + .build()?; + let descriptor = TableDescriptor::builder().schema(schema).build()?; + admin.create_table(&table_path, &descriptor, true).await?; + + // Append rows + let table = conn.get_table(&table_path).await?; + let writer = table.new_append()?.create_writer()?; + let mut row = GenericRow::new(2); + row.set_field(0, 1_700_000_000_000i64); + row.set_field(1, "hello fluss"); + writer.append(&row)?; + writer.flush().await?; + // Scan logs + let scanner = table.new_scan()?.create_log_scanner()?; + scanner.subscribe(0, EARLIEST_OFFSET).await?; Review Comment: `FlussTable::new_scan()` returns `TableScan` directly (not a `Result`), so `table.new_scan()?` won’t compile. ########## README.md: ########## @@ -1,125 +1,326 @@ -# Apache Fluss™ Rust (Incubating) +<!-- + 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. +--> + +# Apache Fluss™ Rust Client (Incubating)  +[](https://crates.io/crates/fluss-rs) +[](https://docs.rs/fluss-rs/) +[](https://www.apache.org/licenses/LICENSE-2.0) + +The official Rust **client** library for [Apache Fluss™](https://fluss.apache.org/) (Incubating) — a streaming storage built for real-time analytics, serving as the real-time data layer for Lakehouse architectures. This is a **client SDK**, not the Fluss server itself. + +This repository contains: + +- **`fluss-rs`** — the Rust core client (crates.io: [`fluss-rs`](https://crates.io/crates/fluss-rs)) +- **Language bindings** — Python, C++, and Elixir clients built on top of `fluss-rs` + +--- + +## What is Fluss? + +[Fluss](https://fluss.apache.org/) bridges the gap between streaming data and the data Lakehouse by enabling **low-latency, high-throughput data ingestion and processing** while seamlessly integrating with popular compute engines (Flink, Spark, Trino). -Rust implementation of [Apache Fluss™](https://fluss.apache.org/). +Key concepts: +- **Log table** — append-only table (no primary key). Immutable records, ideal for event streams and audit trails. +- **Primary Key (KV) table** — keyed table supporting upsert, delete, and point/prefixed lookups. +- **Bucket** — parallelism unit within a table (similar to Kafka partitions). +- **Partition** — data organization by column values (e.g., by date or region). -## Why Fluss? -[Fluss](https://fluss.apache.org/) is a streaming storage built for real-time analytics which can serve as the real-time data layer for Lakehouse architectures. -It bridges the gap between streaming data and the data Lakehouse by enabling low-latency, high-throughput data ingestion and processing while seamlessly integrating with popular compute engines. +--- + +## Features + +### Core Client (`fluss-rs`) + +| Category | Capabilities | +| --------------- | ------------------------------------------------------------------------- | +| **Connection** | Bootstrap to Fluss cluster, SASL authentication, graceful shutdown | +| **Admin** | Create/drop/list databases & tables, manage partitions, list offsets | +| **Log Tables** | Append (single-row + Arrow `RecordBatch`), scan with subscribe/poll | +| **KV Tables** | Upsert, delete, point lookup, **prefix lookup**, partitioned KV support | +| **Data Types** | Int, BigInt, String, Float, Double, Boolean, Bytes, Decimal, Date, Time, Timestamp, TimestampLTZ, Char, Binary | +| **Config** | Batch sizing, buffering, retries, compression, timeouts, prefetch, concurrency | +| **Storage** | Memory, Filesystem, S3, OSS (via [OpenDAL](https://opendal.apache.org/)) | +| **WASM** | Compiles for `wasm32` target | + +### Language Bindings + +| Language | Package / Build | Async Runtime | Data Format | +| ---------- | ------------------------ | ------------------------ | --------------------------- | +| **Rust** | [fluss-rs](https://crates.io/crates/fluss-rs) (crates.io) | Tokio | Arrow `RecordBatch` / `GenericRow` | +| **Python** | Build from source (PyO3) | asyncio | PyArrow / Pandas / dict | +| **C++** | CMake / Bazel (FFI) | Synchronous (Tokio internally) | Arrow RecordBatch / GenericRow | +| **Elixir** | [Rustler](https://github.com/rusterlium/rustler) NIFs | Erlang processes | Elixir values | + +--- + +## Project Structure + +``` +fluss-rust/ +├── crates/ +│ ├── fluss/ # Core Rust client (fluss-rs) +│ │ ├── src/client/ # Connection, Admin, Table, Scan, Upsert, Lookup +│ │ ├── src/metadata/ # Schema, TableDescriptor, DataTypes, Partitions +│ │ ├── src/row/ # GenericRow, InternalRow, Arrow integration +│ │ ├── src/rpc/ # gRPC transport layer +│ │ └── src/config.rs # Client configuration +│ ├── examples/ # 5 runnable examples (log, KV, partitioned, prefix lookup) +│ └── fluss-test-cluster/ # Test harness for integration tests +├── bindings/ +│ ├── python/ # Python binding (PyO3) +│ ├── cpp/ # C++ binding (FFI + header) +│ └── elixir/ # Elixir binding (Rustler NIF) +├── website/ # Docusaurus documentation site +├── docs/ # Supplementary documentation +└── scripts/ # Release & version management +``` -## Why Fluss Rust Client -It's an official Rust client for interacting with Fluss. This client provides foundational capabilities for table management and log streaming operations, enabling developers to explore Fluss within Rust ecosystems. +--- -## Quick-Start +## Quick Start -### Step1 Start Fluss cluster -#### Requirements -Fluss runs on all UNIX-like environments, e.g. Linux, Mac OS X. Before you start to setup the system, make sure you have the following software installed on your test machine: +### Prerequisites -Java 17 or higher (Java 8 and Java 11 are not recommended) -If your cluster does not fulfill these software requirements you will need to install/upgrade it. +- [Java 17+](https://adoptium.net/) for running the Fluss cluster +- [Rust](https://www.rust-lang.org/tools/install) (latest stable) +- Linux or macOS (Windows is not currently supported) -Fluss requires the JAVA_HOME environment variable to be set on all nodes and point to the directory of your Java installation. +### 1. Start a Fluss Cluster -#### Fluss Setup -Go to the [downloads](https://fluss.apache.org/downloads/) page and download the latest Fluss release (currently 0.8.0). Make sure to pick the Fluss package matching your Java version. After downloading the latest release, extract it: ```shell +# Download and extract Fluss (0.8.0+) +curl -LO https://dlcdn.apache.org/incubator/fluss/0.8.0/fluss-0.8.0-incubating-bin.tgz tar -xzf fluss-0.8.0-incubating-bin.tgz cd fluss-0.8.0-incubating/ -``` -You can start Fluss local cluster by running the following command: -```shell + +# Start a local cluster ./bin/local-cluster.sh start ``` -After that, the Fluss local cluster is started. -### Run Provided Example -Only supports Linux or macOs. You will need to [install Rust](https://www.rust-lang.org/tools/install) firstly. +### 2. Add `fluss-rs` to Your Project -After that, go the project directory, build it and run the example: -```shell -cargo build --example example-table --release -cd target/release/examples -./example-table +```toml +[dependencies] +fluss = { package = "fluss-rs", version = "0.2" } +tokio = "1" ``` -The example code is as follows: + +### 3. Write Code + +#### Log Table: Append + Scan + ```rust +use fluss::client::{EARLIEST_OFFSET, FlussConnection}; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; +use std::time::Duration; + #[tokio::main] -pub async fn main() -> Result<()> { - // 1: create the table; - let mut args = Args::default(); - args.bootstrap_servers = "127.0.0.1:9123".to_string(); - let conn_config = ConnectionConfig::from_args(args); - let conn = FlussConnection::new(conn_config).await; - - let admin = conn.get_admin(); - - let table_descriptor = TableDescriptor::builder() - .schema( - Schema::builder() - .column("c1", DataTypes::int()) - .column("c2", DataTypes::string()) - .build(), - ) - .build(); - - let table_path = TablePath::new("fluss".to_owned(), "rust_test".to_owned()); - - admin - .create_table(&table_path, &table_descriptor, true) - .await - .unwrap(); - - // 2: get the table - let table_info = admin.get_table_info(&table_path).await.unwrap(); - print!("Get created table:\n {}\n", table_info); - - // let's sleep 2 seconds to wait leader ready - thread::sleep(Duration::from_secs(2)); - - // 3: append log to the table - let table = conn.get_table(&table_path).await; - let append_writer = table.new_append().create_writer(); - let batch = record_batch!(("c1", Int32, [1, 2, 3, 4, 5, 6]), ("c2", Utf8, ["a1", "a2", "a3", "a4", "a5", "a6"])).unwrap(); - append_writer.append(batch)?; - append_writer.flush().await?; - println!("Start to scan log records......"); - // 4: scan the records - let log_scanner = table.new_scan().create_log_scanner(); - log_scanner.subscribe(0, 0).await; +async fn main() -> Result<()> { + let mut config = Config::default(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + // Create a log table + let table_path = TablePath::new("my_db", "events"); + let schema = Schema::builder() + .column("ts", DataTypes::bigint()) + .column("message", DataTypes::string()) + .build()?; + let descriptor = TableDescriptor::builder().schema(schema).build()?; + admin.create_table(&table_path, &descriptor, true).await?; + + // Append rows + let table = conn.get_table(&table_path).await?; + let writer = table.new_append()?.create_writer()?; + let mut row = GenericRow::new(2); + row.set_field(0, 1_700_000_000_000i64); + row.set_field(1, "hello fluss"); + writer.append(&row)?; + writer.flush().await?; + // Scan logs + let scanner = table.new_scan()?.create_log_scanner()?; + scanner.subscribe(0, EARLIEST_OFFSET).await?; loop { - let scan_records = log_scanner.poll(Duration::from_secs(10)).await?; - println!("Start to poll records......"); - for record in scan_records { + let records = scanner.poll(Duration::from_secs(5)).await?; + for record in records { let row = record.row(); - println!( - "{{{}, {}}}@{}", - row.get_int(0), - row.get_string(1), - record.offset() - ); + println!("offset={}, c1={}, c2={}", + record.offset(), row.get_long(0)?, row.get_string(1)?); Review Comment: The log table example prints column labels `c1`/`c2`, but the schema defines `ts` and `message`, which is confusing for readers. ########## README.md: ########## @@ -1,125 +1,326 @@ -# Apache Fluss™ Rust (Incubating) +<!-- + 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. +--> + +# Apache Fluss™ Rust Client (Incubating)  +[](https://crates.io/crates/fluss-rs) +[](https://docs.rs/fluss-rs/) +[](https://www.apache.org/licenses/LICENSE-2.0) + +The official Rust **client** library for [Apache Fluss™](https://fluss.apache.org/) (Incubating) — a streaming storage built for real-time analytics, serving as the real-time data layer for Lakehouse architectures. This is a **client SDK**, not the Fluss server itself. + +This repository contains: + +- **`fluss-rs`** — the Rust core client (crates.io: [`fluss-rs`](https://crates.io/crates/fluss-rs)) +- **Language bindings** — Python, C++, and Elixir clients built on top of `fluss-rs` + +--- + +## What is Fluss? + +[Fluss](https://fluss.apache.org/) bridges the gap between streaming data and the data Lakehouse by enabling **low-latency, high-throughput data ingestion and processing** while seamlessly integrating with popular compute engines (Flink, Spark, Trino). -Rust implementation of [Apache Fluss™](https://fluss.apache.org/). +Key concepts: +- **Log table** — append-only table (no primary key). Immutable records, ideal for event streams and audit trails. +- **Primary Key (KV) table** — keyed table supporting upsert, delete, and point/prefixed lookups. +- **Bucket** — parallelism unit within a table (similar to Kafka partitions). +- **Partition** — data organization by column values (e.g., by date or region). -## Why Fluss? -[Fluss](https://fluss.apache.org/) is a streaming storage built for real-time analytics which can serve as the real-time data layer for Lakehouse architectures. -It bridges the gap between streaming data and the data Lakehouse by enabling low-latency, high-throughput data ingestion and processing while seamlessly integrating with popular compute engines. +--- + +## Features + +### Core Client (`fluss-rs`) + +| Category | Capabilities | +| --------------- | ------------------------------------------------------------------------- | +| **Connection** | Bootstrap to Fluss cluster, SASL authentication, graceful shutdown | +| **Admin** | Create/drop/list databases & tables, manage partitions, list offsets | +| **Log Tables** | Append (single-row + Arrow `RecordBatch`), scan with subscribe/poll | +| **KV Tables** | Upsert, delete, point lookup, **prefix lookup**, partitioned KV support | +| **Data Types** | Int, BigInt, String, Float, Double, Boolean, Bytes, Decimal, Date, Time, Timestamp, TimestampLTZ, Char, Binary | +| **Config** | Batch sizing, buffering, retries, compression, timeouts, prefetch, concurrency | +| **Storage** | Memory, Filesystem, S3, OSS (via [OpenDAL](https://opendal.apache.org/)) | +| **WASM** | Compiles for `wasm32` target | + +### Language Bindings + +| Language | Package / Build | Async Runtime | Data Format | +| ---------- | ------------------------ | ------------------------ | --------------------------- | +| **Rust** | [fluss-rs](https://crates.io/crates/fluss-rs) (crates.io) | Tokio | Arrow `RecordBatch` / `GenericRow` | +| **Python** | Build from source (PyO3) | asyncio | PyArrow / Pandas / dict | +| **C++** | CMake / Bazel (FFI) | Synchronous (Tokio internally) | Arrow RecordBatch / GenericRow | +| **Elixir** | [Rustler](https://github.com/rusterlium/rustler) NIFs | Erlang processes | Elixir values | + +--- + +## Project Structure + +``` +fluss-rust/ +├── crates/ +│ ├── fluss/ # Core Rust client (fluss-rs) +│ │ ├── src/client/ # Connection, Admin, Table, Scan, Upsert, Lookup +│ │ ├── src/metadata/ # Schema, TableDescriptor, DataTypes, Partitions +│ │ ├── src/row/ # GenericRow, InternalRow, Arrow integration +│ │ ├── src/rpc/ # gRPC transport layer +│ │ └── src/config.rs # Client configuration +│ ├── examples/ # 5 runnable examples (log, KV, partitioned, prefix lookup) +│ └── fluss-test-cluster/ # Test harness for integration tests +├── bindings/ +│ ├── python/ # Python binding (PyO3) +│ ├── cpp/ # C++ binding (FFI + header) +│ └── elixir/ # Elixir binding (Rustler NIF) +├── website/ # Docusaurus documentation site +├── docs/ # Supplementary documentation +└── scripts/ # Release & version management +``` -## Why Fluss Rust Client -It's an official Rust client for interacting with Fluss. This client provides foundational capabilities for table management and log streaming operations, enabling developers to explore Fluss within Rust ecosystems. +--- -## Quick-Start +## Quick Start -### Step1 Start Fluss cluster -#### Requirements -Fluss runs on all UNIX-like environments, e.g. Linux, Mac OS X. Before you start to setup the system, make sure you have the following software installed on your test machine: +### Prerequisites -Java 17 or higher (Java 8 and Java 11 are not recommended) -If your cluster does not fulfill these software requirements you will need to install/upgrade it. +- [Java 17+](https://adoptium.net/) for running the Fluss cluster +- [Rust](https://www.rust-lang.org/tools/install) (latest stable) +- Linux or macOS (Windows is not currently supported) -Fluss requires the JAVA_HOME environment variable to be set on all nodes and point to the directory of your Java installation. +### 1. Start a Fluss Cluster -#### Fluss Setup -Go to the [downloads](https://fluss.apache.org/downloads/) page and download the latest Fluss release (currently 0.8.0). Make sure to pick the Fluss package matching your Java version. After downloading the latest release, extract it: ```shell +# Download and extract Fluss (0.8.0+) +curl -LO https://dlcdn.apache.org/incubator/fluss/0.8.0/fluss-0.8.0-incubating-bin.tgz tar -xzf fluss-0.8.0-incubating-bin.tgz cd fluss-0.8.0-incubating/ -``` -You can start Fluss local cluster by running the following command: -```shell + +# Start a local cluster ./bin/local-cluster.sh start ``` -After that, the Fluss local cluster is started. -### Run Provided Example -Only supports Linux or macOs. You will need to [install Rust](https://www.rust-lang.org/tools/install) firstly. +### 2. Add `fluss-rs` to Your Project -After that, go the project directory, build it and run the example: -```shell -cargo build --example example-table --release -cd target/release/examples -./example-table +```toml +[dependencies] +fluss = { package = "fluss-rs", version = "0.2" } +tokio = "1" ``` -The example code is as follows: + +### 3. Write Code + +#### Log Table: Append + Scan + ```rust +use fluss::client::{EARLIEST_OFFSET, FlussConnection}; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; +use std::time::Duration; + #[tokio::main] -pub async fn main() -> Result<()> { - // 1: create the table; - let mut args = Args::default(); - args.bootstrap_servers = "127.0.0.1:9123".to_string(); - let conn_config = ConnectionConfig::from_args(args); - let conn = FlussConnection::new(conn_config).await; - - let admin = conn.get_admin(); - - let table_descriptor = TableDescriptor::builder() - .schema( - Schema::builder() - .column("c1", DataTypes::int()) - .column("c2", DataTypes::string()) - .build(), - ) - .build(); - - let table_path = TablePath::new("fluss".to_owned(), "rust_test".to_owned()); - - admin - .create_table(&table_path, &table_descriptor, true) - .await - .unwrap(); - - // 2: get the table - let table_info = admin.get_table_info(&table_path).await.unwrap(); - print!("Get created table:\n {}\n", table_info); - - // let's sleep 2 seconds to wait leader ready - thread::sleep(Duration::from_secs(2)); - - // 3: append log to the table - let table = conn.get_table(&table_path).await; - let append_writer = table.new_append().create_writer(); - let batch = record_batch!(("c1", Int32, [1, 2, 3, 4, 5, 6]), ("c2", Utf8, ["a1", "a2", "a3", "a4", "a5", "a6"])).unwrap(); - append_writer.append(batch)?; - append_writer.flush().await?; - println!("Start to scan log records......"); - // 4: scan the records - let log_scanner = table.new_scan().create_log_scanner(); - log_scanner.subscribe(0, 0).await; +async fn main() -> Result<()> { + let mut config = Config::default(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + // Create a log table + let table_path = TablePath::new("my_db", "events"); + let schema = Schema::builder() + .column("ts", DataTypes::bigint()) + .column("message", DataTypes::string()) + .build()?; + let descriptor = TableDescriptor::builder().schema(schema).build()?; + admin.create_table(&table_path, &descriptor, true).await?; + + // Append rows + let table = conn.get_table(&table_path).await?; + let writer = table.new_append()?.create_writer()?; + let mut row = GenericRow::new(2); + row.set_field(0, 1_700_000_000_000i64); + row.set_field(1, "hello fluss"); + writer.append(&row)?; + writer.flush().await?; + // Scan logs + let scanner = table.new_scan()?.create_log_scanner()?; + scanner.subscribe(0, EARLIEST_OFFSET).await?; loop { - let scan_records = log_scanner.poll(Duration::from_secs(10)).await?; - println!("Start to poll records......"); - for record in scan_records { + let records = scanner.poll(Duration::from_secs(5)).await?; + for record in records { let row = record.row(); - println!( - "{{{}, {}}}@{}", - row.get_int(0), - row.get_string(1), - record.offset() - ); + println!("offset={}, c1={}, c2={}", + record.offset(), row.get_long(0)?, row.get_string(1)?); } } +} +``` + +#### KV Table: Upsert + Lookup + +```rust +use fluss::client::FlussConnection; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; + +#[tokio::main] +async fn main() -> Result<()> { + let mut config = Config::default(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + // Create a KV table + let table_path = TablePath::new("my_db", "users"); Review Comment: The KV table Quick Start also uses `TablePath::new("my_db", ...)` without ensuring the database exists, so `create_table` can fail at runtime in a fresh cluster. ########## README.md: ########## @@ -1,125 +1,326 @@ -# Apache Fluss™ Rust (Incubating) +<!-- + 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. +--> + +# Apache Fluss™ Rust Client (Incubating)  +[](https://crates.io/crates/fluss-rs) +[](https://docs.rs/fluss-rs/) +[](https://www.apache.org/licenses/LICENSE-2.0) + +The official Rust **client** library for [Apache Fluss™](https://fluss.apache.org/) (Incubating) — a streaming storage built for real-time analytics, serving as the real-time data layer for Lakehouse architectures. This is a **client SDK**, not the Fluss server itself. + +This repository contains: + +- **`fluss-rs`** — the Rust core client (crates.io: [`fluss-rs`](https://crates.io/crates/fluss-rs)) +- **Language bindings** — Python, C++, and Elixir clients built on top of `fluss-rs` + +--- + +## What is Fluss? + +[Fluss](https://fluss.apache.org/) bridges the gap between streaming data and the data Lakehouse by enabling **low-latency, high-throughput data ingestion and processing** while seamlessly integrating with popular compute engines (Flink, Spark, Trino). -Rust implementation of [Apache Fluss™](https://fluss.apache.org/). +Key concepts: +- **Log table** — append-only table (no primary key). Immutable records, ideal for event streams and audit trails. +- **Primary Key (KV) table** — keyed table supporting upsert, delete, and point/prefixed lookups. +- **Bucket** — parallelism unit within a table (similar to Kafka partitions). +- **Partition** — data organization by column values (e.g., by date or region). -## Why Fluss? -[Fluss](https://fluss.apache.org/) is a streaming storage built for real-time analytics which can serve as the real-time data layer for Lakehouse architectures. -It bridges the gap between streaming data and the data Lakehouse by enabling low-latency, high-throughput data ingestion and processing while seamlessly integrating with popular compute engines. +--- + +## Features + +### Core Client (`fluss-rs`) + +| Category | Capabilities | +| --------------- | ------------------------------------------------------------------------- | +| **Connection** | Bootstrap to Fluss cluster, SASL authentication, graceful shutdown | +| **Admin** | Create/drop/list databases & tables, manage partitions, list offsets | +| **Log Tables** | Append (single-row + Arrow `RecordBatch`), scan with subscribe/poll | +| **KV Tables** | Upsert, delete, point lookup, **prefix lookup**, partitioned KV support | +| **Data Types** | Int, BigInt, String, Float, Double, Boolean, Bytes, Decimal, Date, Time, Timestamp, TimestampLTZ, Char, Binary | +| **Config** | Batch sizing, buffering, retries, compression, timeouts, prefetch, concurrency | +| **Storage** | Memory, Filesystem, S3, OSS (via [OpenDAL](https://opendal.apache.org/)) | +| **WASM** | Compiles for `wasm32` target | + +### Language Bindings + +| Language | Package / Build | Async Runtime | Data Format | +| ---------- | ------------------------ | ------------------------ | --------------------------- | +| **Rust** | [fluss-rs](https://crates.io/crates/fluss-rs) (crates.io) | Tokio | Arrow `RecordBatch` / `GenericRow` | +| **Python** | Build from source (PyO3) | asyncio | PyArrow / Pandas / dict | +| **C++** | CMake / Bazel (FFI) | Synchronous (Tokio internally) | Arrow RecordBatch / GenericRow | +| **Elixir** | [Rustler](https://github.com/rusterlium/rustler) NIFs | Erlang processes | Elixir values | + +--- + +## Project Structure + +``` +fluss-rust/ +├── crates/ +│ ├── fluss/ # Core Rust client (fluss-rs) +│ │ ├── src/client/ # Connection, Admin, Table, Scan, Upsert, Lookup +│ │ ├── src/metadata/ # Schema, TableDescriptor, DataTypes, Partitions +│ │ ├── src/row/ # GenericRow, InternalRow, Arrow integration +│ │ ├── src/rpc/ # gRPC transport layer +│ │ └── src/config.rs # Client configuration +│ ├── examples/ # 5 runnable examples (log, KV, partitioned, prefix lookup) +│ └── fluss-test-cluster/ # Test harness for integration tests +├── bindings/ +│ ├── python/ # Python binding (PyO3) +│ ├── cpp/ # C++ binding (FFI + header) +│ └── elixir/ # Elixir binding (Rustler NIF) +├── website/ # Docusaurus documentation site +├── docs/ # Supplementary documentation +└── scripts/ # Release & version management +``` -## Why Fluss Rust Client -It's an official Rust client for interacting with Fluss. This client provides foundational capabilities for table management and log streaming operations, enabling developers to explore Fluss within Rust ecosystems. +--- -## Quick-Start +## Quick Start -### Step1 Start Fluss cluster -#### Requirements -Fluss runs on all UNIX-like environments, e.g. Linux, Mac OS X. Before you start to setup the system, make sure you have the following software installed on your test machine: +### Prerequisites -Java 17 or higher (Java 8 and Java 11 are not recommended) -If your cluster does not fulfill these software requirements you will need to install/upgrade it. +- [Java 17+](https://adoptium.net/) for running the Fluss cluster +- [Rust](https://www.rust-lang.org/tools/install) (latest stable) +- Linux or macOS (Windows is not currently supported) -Fluss requires the JAVA_HOME environment variable to be set on all nodes and point to the directory of your Java installation. +### 1. Start a Fluss Cluster -#### Fluss Setup -Go to the [downloads](https://fluss.apache.org/downloads/) page and download the latest Fluss release (currently 0.8.0). Make sure to pick the Fluss package matching your Java version. After downloading the latest release, extract it: ```shell +# Download and extract Fluss (0.8.0+) +curl -LO https://dlcdn.apache.org/incubator/fluss/0.8.0/fluss-0.8.0-incubating-bin.tgz tar -xzf fluss-0.8.0-incubating-bin.tgz cd fluss-0.8.0-incubating/ -``` -You can start Fluss local cluster by running the following command: -```shell + +# Start a local cluster ./bin/local-cluster.sh start ``` -After that, the Fluss local cluster is started. -### Run Provided Example -Only supports Linux or macOs. You will need to [install Rust](https://www.rust-lang.org/tools/install) firstly. +### 2. Add `fluss-rs` to Your Project -After that, go the project directory, build it and run the example: -```shell -cargo build --example example-table --release -cd target/release/examples -./example-table +```toml +[dependencies] +fluss = { package = "fluss-rs", version = "0.2" } +tokio = "1" ``` -The example code is as follows: + +### 3. Write Code + +#### Log Table: Append + Scan + ```rust +use fluss::client::{EARLIEST_OFFSET, FlussConnection}; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; +use std::time::Duration; + #[tokio::main] -pub async fn main() -> Result<()> { - // 1: create the table; - let mut args = Args::default(); - args.bootstrap_servers = "127.0.0.1:9123".to_string(); - let conn_config = ConnectionConfig::from_args(args); - let conn = FlussConnection::new(conn_config).await; - - let admin = conn.get_admin(); - - let table_descriptor = TableDescriptor::builder() - .schema( - Schema::builder() - .column("c1", DataTypes::int()) - .column("c2", DataTypes::string()) - .build(), - ) - .build(); - - let table_path = TablePath::new("fluss".to_owned(), "rust_test".to_owned()); - - admin - .create_table(&table_path, &table_descriptor, true) - .await - .unwrap(); - - // 2: get the table - let table_info = admin.get_table_info(&table_path).await.unwrap(); - print!("Get created table:\n {}\n", table_info); - - // let's sleep 2 seconds to wait leader ready - thread::sleep(Duration::from_secs(2)); - - // 3: append log to the table - let table = conn.get_table(&table_path).await; - let append_writer = table.new_append().create_writer(); - let batch = record_batch!(("c1", Int32, [1, 2, 3, 4, 5, 6]), ("c2", Utf8, ["a1", "a2", "a3", "a4", "a5", "a6"])).unwrap(); - append_writer.append(batch)?; - append_writer.flush().await?; - println!("Start to scan log records......"); - // 4: scan the records - let log_scanner = table.new_scan().create_log_scanner(); - log_scanner.subscribe(0, 0).await; +async fn main() -> Result<()> { + let mut config = Config::default(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + // Create a log table + let table_path = TablePath::new("my_db", "events"); + let schema = Schema::builder() + .column("ts", DataTypes::bigint()) + .column("message", DataTypes::string()) + .build()?; + let descriptor = TableDescriptor::builder().schema(schema).build()?; + admin.create_table(&table_path, &descriptor, true).await?; + + // Append rows + let table = conn.get_table(&table_path).await?; + let writer = table.new_append()?.create_writer()?; + let mut row = GenericRow::new(2); + row.set_field(0, 1_700_000_000_000i64); + row.set_field(1, "hello fluss"); + writer.append(&row)?; + writer.flush().await?; + // Scan logs + let scanner = table.new_scan()?.create_log_scanner()?; + scanner.subscribe(0, EARLIEST_OFFSET).await?; loop { - let scan_records = log_scanner.poll(Duration::from_secs(10)).await?; - println!("Start to poll records......"); - for record in scan_records { + let records = scanner.poll(Duration::from_secs(5)).await?; + for record in records { let row = record.row(); - println!( - "{{{}, {}}}@{}", - row.get_int(0), - row.get_string(1), - record.offset() - ); + println!("offset={}, c1={}, c2={}", + record.offset(), row.get_long(0)?, row.get_string(1)?); } } +} +``` + +#### KV Table: Upsert + Lookup + +```rust +use fluss::client::FlussConnection; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; + +#[tokio::main] +async fn main() -> Result<()> { + let mut config = Config::default(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + // Create a KV table + let table_path = TablePath::new("my_db", "users"); + let schema = Schema::builder() + .column("id", DataTypes::int()) + .column("name", DataTypes::string()) + .column("score", DataTypes::bigint()) + .primary_key(vec!["id"]) + .build()?; + let descriptor = TableDescriptor::builder().schema(schema).build()?; + admin.create_table(&table_path, &descriptor, true).await?; + + // Upsert rows + let table = conn.get_table(&table_path).await?; + let writer = table.new_upsert()?.create_writer()?; + for (id, name, score) in [(1, "Alice", 95i64), (2, "Bob", 87)] { + let mut row = GenericRow::new(3); + row.set_field(0, id); + row.set_field(1, name); + row.set_field(2, score); + writer.upsert(&row)?; + } + writer.flush().await?; + + // Point lookup by primary key + let lookuper = table.new_lookup()?.create_lookuper()?; + let mut key = GenericRow::new(1); + key.set_field(0, 1i32); + if let Some(row) = lookuper.lookup(&key).await?.get_single_row()? { + println!("id={}, name={}, score={}", + row.get_int(0)?, row.get_string(1)?, row.get_long(2)?); + } + Ok(()) } ``` -You can change it according to your needs, have fun! +#### More Examples + +| Example | Description | +| ---------------------------------------- | ---------------------------------------------- | +| `example-table` | Log table: append + scan with Arrow batch | +| `example-kv-table` | KV table: upsert + point lookup | +| `example-partitioned-kv-table` | KV table with partitions | +| `example-prefix-lookup` | Prefix lookup on bucket keys | +| `example-partitioned-prefix-lookup` | Prefix lookup on partitioned tables | Review Comment: The example names in the "More Examples" table don’t match the actual Cargo example targets in `crates/examples/Cargo.toml` (e.g., the KV examples are `example-upsert-lookup` / `example-partitioned-upsert-lookup`). As written, these commands won’t work. ########## README.md: ########## @@ -1,125 +1,326 @@ -# Apache Fluss™ Rust (Incubating) +<!-- + 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. +--> + +# Apache Fluss™ Rust Client (Incubating)  +[](https://crates.io/crates/fluss-rs) +[](https://docs.rs/fluss-rs/) +[](https://www.apache.org/licenses/LICENSE-2.0) + +The official Rust **client** library for [Apache Fluss™](https://fluss.apache.org/) (Incubating) — a streaming storage built for real-time analytics, serving as the real-time data layer for Lakehouse architectures. This is a **client SDK**, not the Fluss server itself. + +This repository contains: + +- **`fluss-rs`** — the Rust core client (crates.io: [`fluss-rs`](https://crates.io/crates/fluss-rs)) +- **Language bindings** — Python, C++, and Elixir clients built on top of `fluss-rs` + +--- + +## What is Fluss? + +[Fluss](https://fluss.apache.org/) bridges the gap between streaming data and the data Lakehouse by enabling **low-latency, high-throughput data ingestion and processing** while seamlessly integrating with popular compute engines (Flink, Spark, Trino). -Rust implementation of [Apache Fluss™](https://fluss.apache.org/). +Key concepts: +- **Log table** — append-only table (no primary key). Immutable records, ideal for event streams and audit trails. +- **Primary Key (KV) table** — keyed table supporting upsert, delete, and point/prefixed lookups. +- **Bucket** — parallelism unit within a table (similar to Kafka partitions). +- **Partition** — data organization by column values (e.g., by date or region). -## Why Fluss? -[Fluss](https://fluss.apache.org/) is a streaming storage built for real-time analytics which can serve as the real-time data layer for Lakehouse architectures. -It bridges the gap between streaming data and the data Lakehouse by enabling low-latency, high-throughput data ingestion and processing while seamlessly integrating with popular compute engines. +--- + +## Features + +### Core Client (`fluss-rs`) + +| Category | Capabilities | +| --------------- | ------------------------------------------------------------------------- | +| **Connection** | Bootstrap to Fluss cluster, SASL authentication, graceful shutdown | +| **Admin** | Create/drop/list databases & tables, manage partitions, list offsets | +| **Log Tables** | Append (single-row + Arrow `RecordBatch`), scan with subscribe/poll | +| **KV Tables** | Upsert, delete, point lookup, **prefix lookup**, partitioned KV support | +| **Data Types** | Int, BigInt, String, Float, Double, Boolean, Bytes, Decimal, Date, Time, Timestamp, TimestampLTZ, Char, Binary | +| **Config** | Batch sizing, buffering, retries, compression, timeouts, prefetch, concurrency | +| **Storage** | Memory, Filesystem, S3, OSS (via [OpenDAL](https://opendal.apache.org/)) | +| **WASM** | Compiles for `wasm32` target | + +### Language Bindings + +| Language | Package / Build | Async Runtime | Data Format | +| ---------- | ------------------------ | ------------------------ | --------------------------- | +| **Rust** | [fluss-rs](https://crates.io/crates/fluss-rs) (crates.io) | Tokio | Arrow `RecordBatch` / `GenericRow` | +| **Python** | Build from source (PyO3) | asyncio | PyArrow / Pandas / dict | +| **C++** | CMake / Bazel (FFI) | Synchronous (Tokio internally) | Arrow RecordBatch / GenericRow | +| **Elixir** | [Rustler](https://github.com/rusterlium/rustler) NIFs | Erlang processes | Elixir values | + +--- + +## Project Structure + +``` +fluss-rust/ +├── crates/ +│ ├── fluss/ # Core Rust client (fluss-rs) +│ │ ├── src/client/ # Connection, Admin, Table, Scan, Upsert, Lookup +│ │ ├── src/metadata/ # Schema, TableDescriptor, DataTypes, Partitions +│ │ ├── src/row/ # GenericRow, InternalRow, Arrow integration +│ │ ├── src/rpc/ # gRPC transport layer +│ │ └── src/config.rs # Client configuration +│ ├── examples/ # 5 runnable examples (log, KV, partitioned, prefix lookup) +│ └── fluss-test-cluster/ # Test harness for integration tests +├── bindings/ +│ ├── python/ # Python binding (PyO3) +│ ├── cpp/ # C++ binding (FFI + header) +│ └── elixir/ # Elixir binding (Rustler NIF) +├── website/ # Docusaurus documentation site +├── docs/ # Supplementary documentation +└── scripts/ # Release & version management +``` -## Why Fluss Rust Client -It's an official Rust client for interacting with Fluss. This client provides foundational capabilities for table management and log streaming operations, enabling developers to explore Fluss within Rust ecosystems. +--- -## Quick-Start +## Quick Start -### Step1 Start Fluss cluster -#### Requirements -Fluss runs on all UNIX-like environments, e.g. Linux, Mac OS X. Before you start to setup the system, make sure you have the following software installed on your test machine: +### Prerequisites -Java 17 or higher (Java 8 and Java 11 are not recommended) -If your cluster does not fulfill these software requirements you will need to install/upgrade it. +- [Java 17+](https://adoptium.net/) for running the Fluss cluster +- [Rust](https://www.rust-lang.org/tools/install) (latest stable) +- Linux or macOS (Windows is not currently supported) -Fluss requires the JAVA_HOME environment variable to be set on all nodes and point to the directory of your Java installation. +### 1. Start a Fluss Cluster -#### Fluss Setup -Go to the [downloads](https://fluss.apache.org/downloads/) page and download the latest Fluss release (currently 0.8.0). Make sure to pick the Fluss package matching your Java version. After downloading the latest release, extract it: ```shell +# Download and extract Fluss (0.8.0+) +curl -LO https://dlcdn.apache.org/incubator/fluss/0.8.0/fluss-0.8.0-incubating-bin.tgz tar -xzf fluss-0.8.0-incubating-bin.tgz cd fluss-0.8.0-incubating/ -``` -You can start Fluss local cluster by running the following command: -```shell + +# Start a local cluster ./bin/local-cluster.sh start ``` -After that, the Fluss local cluster is started. -### Run Provided Example -Only supports Linux or macOs. You will need to [install Rust](https://www.rust-lang.org/tools/install) firstly. +### 2. Add `fluss-rs` to Your Project -After that, go the project directory, build it and run the example: -```shell -cargo build --example example-table --release -cd target/release/examples -./example-table +```toml +[dependencies] +fluss = { package = "fluss-rs", version = "0.2" } +tokio = "1" ``` -The example code is as follows: + +### 3. Write Code + +#### Log Table: Append + Scan + ```rust +use fluss::client::{EARLIEST_OFFSET, FlussConnection}; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; +use std::time::Duration; + #[tokio::main] -pub async fn main() -> Result<()> { - // 1: create the table; - let mut args = Args::default(); - args.bootstrap_servers = "127.0.0.1:9123".to_string(); - let conn_config = ConnectionConfig::from_args(args); - let conn = FlussConnection::new(conn_config).await; - - let admin = conn.get_admin(); - - let table_descriptor = TableDescriptor::builder() - .schema( - Schema::builder() - .column("c1", DataTypes::int()) - .column("c2", DataTypes::string()) - .build(), - ) - .build(); - - let table_path = TablePath::new("fluss".to_owned(), "rust_test".to_owned()); - - admin - .create_table(&table_path, &table_descriptor, true) - .await - .unwrap(); - - // 2: get the table - let table_info = admin.get_table_info(&table_path).await.unwrap(); - print!("Get created table:\n {}\n", table_info); - - // let's sleep 2 seconds to wait leader ready - thread::sleep(Duration::from_secs(2)); - - // 3: append log to the table - let table = conn.get_table(&table_path).await; - let append_writer = table.new_append().create_writer(); - let batch = record_batch!(("c1", Int32, [1, 2, 3, 4, 5, 6]), ("c2", Utf8, ["a1", "a2", "a3", "a4", "a5", "a6"])).unwrap(); - append_writer.append(batch)?; - append_writer.flush().await?; - println!("Start to scan log records......"); - // 4: scan the records - let log_scanner = table.new_scan().create_log_scanner(); - log_scanner.subscribe(0, 0).await; +async fn main() -> Result<()> { + let mut config = Config::default(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + // Create a log table + let table_path = TablePath::new("my_db", "events"); + let schema = Schema::builder() + .column("ts", DataTypes::bigint()) + .column("message", DataTypes::string()) + .build()?; + let descriptor = TableDescriptor::builder().schema(schema).build()?; + admin.create_table(&table_path, &descriptor, true).await?; + + // Append rows + let table = conn.get_table(&table_path).await?; + let writer = table.new_append()?.create_writer()?; + let mut row = GenericRow::new(2); + row.set_field(0, 1_700_000_000_000i64); + row.set_field(1, "hello fluss"); + writer.append(&row)?; + writer.flush().await?; + // Scan logs + let scanner = table.new_scan()?.create_log_scanner()?; + scanner.subscribe(0, EARLIEST_OFFSET).await?; loop { - let scan_records = log_scanner.poll(Duration::from_secs(10)).await?; - println!("Start to poll records......"); - for record in scan_records { + let records = scanner.poll(Duration::from_secs(5)).await?; + for record in records { let row = record.row(); - println!( - "{{{}, {}}}@{}", - row.get_int(0), - row.get_string(1), - record.offset() - ); + println!("offset={}, c1={}, c2={}", + record.offset(), row.get_long(0)?, row.get_string(1)?); } } +} +``` + +#### KV Table: Upsert + Lookup + +```rust +use fluss::client::FlussConnection; +use fluss::config::Config; +use fluss::error::Result; +use fluss::metadata::{DataTypes, Schema, TableDescriptor, TablePath}; +use fluss::row::{DataGetters, GenericRow}; + +#[tokio::main] +async fn main() -> Result<()> { + let mut config = Config::default(); + config.bootstrap_servers = "127.0.0.1:9123".to_string(); + let conn = FlussConnection::new(config).await?; + let admin = conn.get_admin()?; + + // Create a KV table + let table_path = TablePath::new("my_db", "users"); + let schema = Schema::builder() + .column("id", DataTypes::int()) + .column("name", DataTypes::string()) + .column("score", DataTypes::bigint()) + .primary_key(vec!["id"]) + .build()?; + let descriptor = TableDescriptor::builder().schema(schema).build()?; + admin.create_table(&table_path, &descriptor, true).await?; + + // Upsert rows + let table = conn.get_table(&table_path).await?; + let writer = table.new_upsert()?.create_writer()?; + for (id, name, score) in [(1, "Alice", 95i64), (2, "Bob", 87)] { + let mut row = GenericRow::new(3); + row.set_field(0, id); + row.set_field(1, name); + row.set_field(2, score); + writer.upsert(&row)?; + } + writer.flush().await?; + + // Point lookup by primary key + let lookuper = table.new_lookup()?.create_lookuper()?; + let mut key = GenericRow::new(1); + key.set_field(0, 1i32); + if let Some(row) = lookuper.lookup(&key).await?.get_single_row()? { + println!("id={}, name={}, score={}", + row.get_int(0)?, row.get_string(1)?, row.get_long(2)?); + } + Ok(()) } ``` -You can change it according to your needs, have fun! +#### More Examples + +| Example | Description | +| ---------------------------------------- | ---------------------------------------------- | +| `example-table` | Log table: append + scan with Arrow batch | +| `example-kv-table` | KV table: upsert + point lookup | +| `example-partitioned-kv-table` | KV table with partitions | +| `example-prefix-lookup` | Prefix lookup on bucket keys | +| `example-partitioned-prefix-lookup` | Prefix lookup on partitioned tables | + +Build and run any example: -#### Clear environment -Then, stop your Fluss cluster. Go to your Fluss home, stop it via the following commands: ```shell -./bin/local-cluster.sh stop +cargo build --example example-table --release +./target/release/examples/example-table ``` +--- + +## Configuration + +`Config` supports the following key options (all with sensible defaults): + +| Option | Default | Description | +| ------------------------------------- | ----------------- | --------------------------------------------- | +| `bootstrap_servers` | `127.0.0.1:9123` | Fluss coordinator address | +| `writer.batch.size` | 2 MB | Max batch size before flushing | +| `writer.batch.timeout_ms` | 100 ms | Max time before auto-flush | +| `writer.buffer.memory` | 64 MB | Total buffer memory for pending writes | +| `writer.retries` | `i32::MAX` | Max write retries | +| `scanner.log.fetch.max.bytes` | 16 MB | Max bytes per fetch request | +| `scanner.log.fetch.wait.max.time_ms` | 500 ms | Max wait time for fetch | +| `scanner.remote_log.read.concurrency` | 4 | Concurrency for remote log reads | +| `connect.timeout_ms` | 120 s | Connection timeout | +| `security.sasl.username` / `password` | — | SASL PLAIN authentication | + Review Comment: The configuration option names in this table use dotted keys (e.g., `writer.batch.size`) that don’t correspond to the actual `Config` fields / clap flags (which are snake_case, e.g., `writer_batch_size`). This will mislead users trying to set options. -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
