This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/paimon-rust.git
The following commit(s) were added to refs/heads/main by this push:
new 1dab4960 feat: rust examples for creating and querying Paimon tables
(#648)
1dab4960 is described below
commit 1dab496083ae44de8bdbd2ac4614b2477450f38a
Author: Ganesh Sivakumar <[email protected]>
AuthorDate: Sat Aug 15 13:47:07 2026 +0530
feat: rust examples for creating and querying Paimon tables (#648)
---
Cargo.toml | 10 +-
.../datafusion/examples/datafusion_query.rs | 85 +++++++++++++
crates/paimon/Cargo.toml | 13 +-
crates/paimon/examples/create_table.rs | 134 +++++++++++++++++++++
4 files changed, 239 insertions(+), 3 deletions(-)
diff --git a/Cargo.toml b/Cargo.toml
index fab0a8db..08fe20d3 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -17,7 +17,15 @@
[workspace]
resolver = "2"
-members = ["crates/paimon", "crates/paimon-rest-server",
"crates/integration_tests", "bindings/c", "bindings/python",
"crates/integrations/datafusion", "benchmarks/tpcds"]
+members = [
+ "crates/paimon",
+ "crates/paimon-rest-server",
+ "crates/integration_tests",
+ "bindings/c",
+ "bindings/python",
+ "crates/integrations/datafusion",
+ "benchmarks/tpcds",
+]
[workspace.package]
version = "0.4.0"
diff --git a/crates/integrations/datafusion/examples/datafusion_query.rs
b/crates/integrations/datafusion/examples/datafusion_query.rs
new file mode 100644
index 00000000..b3873a00
--- /dev/null
+++ b/crates/integrations/datafusion/examples/datafusion_query.rs
@@ -0,0 +1,85 @@
+// 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.
+
+use std::error::Error;
+use std::sync::Arc;
+
+use datafusion::prelude::{col, lit, SessionContext};
+use paimon::catalog::Identifier;
+use paimon::{Catalog, CatalogFactory, CatalogOptions, Options};
+use paimon_datafusion::PaimonTableProvider;
+
+// This example demonstrates how to query a Paimon table
+// using the DataFusion DataFrame API.
+//
+// Before running this example, create the sample table at
+// examples/create_table, then pass the catalog warehouse path:
+// cargo run --package paimon-datafusion --example datafusion_query --
/path/to/warehouse
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+ let warehouse = std::env::args().nth(1).ok_or_else(|| {
+ std::io::Error::new(
+ std::io::ErrorKind::InvalidInput,
+ "usage: cargo run --package paimon-datafusion --example
datafusion_query -- <warehouse-path>",
+ )
+ })?;
+
+ // Open the local Paimon catalog
+ let catalog = create_catalog(warehouse).await?;
+
+ // Load the users table
+ let identifier = Identifier::new("my_db", "users");
+ let table = catalog.get_table(&identifier).await?;
+
+ // DataFusion TableProvider for the Paimon table
+ let provider = PaimonTableProvider::try_new(table)?;
+
+ let ctx = SessionContext::new();
+
+ // Register table
+ ctx.register_table("user_table", Arc::new(provider))?;
+
+ let df = ctx.table("user_table").await?;
+
+ // Filter users with score >= 90 and select a subset of columns
+ let df = df.filter(col("score").gt_eq(lit(90)))?.select(vec![
+ col("name"),
+ col("city"),
+ col("score"),
+ ])?;
+
+ // Expected output:
+ //
+ // +-------+-----------+-------+
+ // | name | city | score |
+ // +-------+-----------+-------+
+ // | Alice | New York | 95 |
+ // | Paul | Bengaluru | 91 |
+ // +-------+-----------+-------+
+
+ // Display the results
+ df.show().await?;
+
+ Ok(())
+}
+
+pub async fn create_catalog(warehouse: String) -> Result<Arc<dyn Catalog>,
Box<dyn Error>> {
+ let mut options = Options::new();
+ options.set(CatalogOptions::WAREHOUSE, warehouse);
+ let catalog = CatalogFactory::create(options).await?;
+ Ok(catalog)
+}
diff --git a/crates/paimon/Cargo.toml b/crates/paimon/Cargo.toml
index d4049eb1..4c0bd8dd 100644
--- a/crates/paimon/Cargo.toml
+++ b/crates/paimon/Cargo.toml
@@ -54,7 +54,10 @@ storage-oss = [
]
storage-s3 = ["dep:opendal-http-transport-reqwest", "dep:opendal-service-s3"]
storage-cos = ["dep:opendal-http-transport-reqwest", "dep:opendal-service-cos"]
-storage-azdls = ["dep:opendal-http-transport-reqwest",
"dep:opendal-service-azdls"]
+storage-azdls = [
+ "dep:opendal-http-transport-reqwest",
+ "dep:opendal-service-azdls",
+]
storage-obs = ["dep:opendal-http-transport-reqwest", "dep:opendal-service-obs"]
storage-gcs = ["dep:opendal-http-transport-reqwest", "dep:opendal-service-gcs"]
storage-hdfs = ["dep:opendal-service-hdfs-native"]
@@ -64,7 +67,13 @@ url = "2.5.2"
async-trait = "0.1.81"
bytes = "1.7.1"
bitflags = "2.6.0"
-tokio = { version = "1.39.2", features = ["fs", "io-util", "macros", "sync",
"time"] }
+tokio = { version = "1.39.2", features = [
+ "fs",
+ "io-util",
+ "macros",
+ "sync",
+ "time",
+] }
chrono = { version = "0.4.38", features = ["serde"] }
serde = { version = "1", features = ["derive", "rc"] }
serde_bytes = "0.11.15"
diff --git a/crates/paimon/examples/create_table.rs
b/crates/paimon/examples/create_table.rs
new file mode 100644
index 00000000..5f9e8724
--- /dev/null
+++ b/crates/paimon/examples/create_table.rs
@@ -0,0 +1,134 @@
+// 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.
+
+use std::collections::HashMap;
+use std::error::Error;
+use std::sync::Arc;
+
+use arrow_array::{Int32Array, RecordBatch, StringArray};
+use arrow_schema::{DataType as ArrowDataType, Field, Schema as ArrowSchema};
+use paimon::catalog::Identifier;
+use paimon::spec::{DataType, IntType, Schema, VarCharType};
+use paimon::{Catalog, CatalogFactory, CatalogOptions, Options};
+
+// This example creates a paimon table and inserts test data
+// Run the example by passing the catalog warehouse path first after `--`:
+// Eg: cargo run --package paimon --example create_table -- /path/to/warehouse
--overwrite
+// Use optional --overwrite flag after the warehouse path to automatically
drop and re-create
+// the table if it already exists.
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+ let mut args = std::env::args().skip(1);
+
+ let warehouse = args.next().ok_or_else(|| {
+ std::io::Error::new(
+ std::io::ErrorKind::InvalidInput,
+ "usage: cargo run --package paimon --example create_table --
<warehouse-path> --overwrite",
+ )
+ })?;
+
+ let overwrite = args.any(|arg| arg == "--overwrite");
+
+ // Open local catalog
+ let catalog = create_catalog(warehouse).await?;
+
+ // Create new database
+ catalog
+ .create_database("my_db", true, HashMap::new())
+ .await?;
+
+ // Define table schema and its data types
+ let schema = Schema::builder()
+ .column("id", DataType::Int(IntType::new()))
+ .column("name", DataType::VarChar(VarCharType::string_type()))
+ .column("city", DataType::VarChar(VarCharType::string_type()))
+ .column("age", DataType::Int(IntType::new()))
+ .column("score", DataType::Int(IntType::new()))
+ .build()?;
+
+ let identifier = Identifier::new("my_db", "users");
+
+ // Check if table exists in catalog
+ let table_exists = match catalog.get_table(&identifier).await {
+ Ok(_) => true,
+ Err(paimon::Error::TableNotExist { .. }) => false,
+ Err(error) => return Err(error.into()),
+ };
+
+ if table_exists {
+ if !overwrite {
+ return Err(format!(
+ "table {} already exists, pass --overwrite to automatically
drop and re-create it",
+ identifier
+ )
+ .into());
+ }
+
+ catalog.drop_table(&identifier, false).await?;
+ }
+ catalog.create_table(&identifier, schema, false).await?;
+
+ let table = catalog.get_table(&identifier).await?;
+
+ let builder = table.new_write_builder();
+ let txn = builder.new_commit();
+
+ let mut writer = builder.new_write()?;
+
+ let arrow_schema = Arc::new(ArrowSchema::new(vec![
+ Field::new("id", ArrowDataType::Int32, false),
+ Field::new("name", ArrowDataType::Utf8, false),
+ Field::new("city", ArrowDataType::Utf8, false),
+ Field::new("age", ArrowDataType::Int32, false),
+ Field::new("score", ArrowDataType::Int32, false),
+ ]));
+
+ // sample data
+ let batch = RecordBatch::try_new(
+ arrow_schema,
+ vec![
+ Arc::new(Int32Array::from(vec![1, 2, 3, 4, 5])),
+ Arc::new(StringArray::from(vec![
+ "Alice", "Bob", "Paul", "Diana", "Ethan",
+ ])),
+ Arc::new(StringArray::from(vec![
+ "New York",
+ "San Francisco",
+ "Bengaluru",
+ "Amsterdam",
+ "Berlin",
+ ])),
+ Arc::new(Int32Array::from(vec![28, 34, 22, 31, 27])),
+ Arc::new(Int32Array::from(vec![95, 82, 91, 88, 76])),
+ ],
+ )?;
+
+ writer.write_arrow_batch(&batch).await?;
+
+ let msg = writer.prepare_commit().await?;
+
+ txn.commit(msg).await?;
+
+ Ok(())
+}
+
+pub async fn create_catalog(warehouse: String) -> Result<Arc<dyn Catalog>,
Box<dyn Error>> {
+ let mut options = Options::new();
+ options.set(CatalogOptions::WAREHOUSE, warehouse);
+ let catalog = CatalogFactory::create(options).await?;
+ Ok(catalog)
+}
