feat: rust examples for creating and querying Paimon tables

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"
@@ -41,6 +49,7 @@ arrow-string = "58.0"
 datafusion = "54.0.0"
 datafusion-ffi = "54.0.0"
 paimon = { version = "0.4.0", path = "crates/paimon" }
+paimon-datafusion = { path = "crates/integrations/datafusion" }
 parquet = "58.0"
 constant_time_eq = ">=0.4.0, <0.5.0"
 tokio = "1.39.2"

crates/paimon/Cargo.toml

@@ -50,7 +50,10 @@ storage-fs = ["dep:opendal-service-fs"]
 storage-oss = ["dep:opendal-http-transport-reqwest", "dep:opendal-service-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"]
@@ -60,7 +63,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"] }
 serde_bytes = "0.11.15"
@@ -127,3 +136,5 @@ unicode-segmentation = "=1.13.2"
 axum = { version = "0.7", features = ["macros", "tokio", "http1", "http2"] }
 rand = "0.8.5"
 tempfile = "3"
+paimon-datafusion = { workspace = true }
+datafusion = { workspace = true }

crates/paimon/examples/create_table.rs

@@ -0,0 +1,104 @@
+// 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
+// set the catalog path and run example using:
+// cargo run --package paimon --example create_table
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    // Open local catalog
+    let catalog = create_catelog().await?;
+
+    // Create new database
+    catalog
+        .create_database("my_db", false, 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");
+
+    // create table
+    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_catelog() -> Result<Arc<dyn Catalog>, Box<dyn Error>> {
+    let mut options = Options::new();
+    options.set(CatalogOptions::WAREHOUSE, "/path-to/testdata");
+    let catalog = CatalogFactory::create(options).await?;
+    Ok(catalog)
+}

crates/paimon/examples/datafusion_query.rs

@@ -0,0 +1,77 @@
+// 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
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error>> {
+    // Open the local Paimon catalog
+    let catalog = create_catelog().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_catelog() -> Result<Arc<dyn Catalog>, Box<dyn Error>> {
+    let mut options = Options::new();
+    options.set(CatalogOptions::WAREHOUSE, "/path-to/testdata");
+    let catalog = CatalogFactory::create(options).await?;
+    Ok(catalog)
+}