Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
29 changes: 28 additions & 1 deletion Cargo.toml
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
[package]
name = "chdb-rust"
version = "1.3.1"
version = "1.4.0"
edition = "2021"
authors = ["Auxten"]
description = "chDB FFI bindings for Rust(Experimental)"
Expand All @@ -11,9 +11,13 @@ readme = "README.md"
keywords = ["clickhouse", "chdb", "database", "embedded", "analytics"]

[dependencies]
arrow = { version = "59", optional = true, features = ["ffi"] }
thiserror = "1"

[features]
default = ["arrow"]
# Apache Arrow bulk insert / stream registration (RecordBatch, FFI wrappers).
arrow = ["dep:arrow"]
# By default, we link dynamically.
# Users can opt-in to static linking with --features static
static = []
Expand All @@ -25,9 +29,32 @@ flate2 = "1.0"
tar = "0.4"

[dev-dependencies]
arrow = { version = "59", features = ["ffi"] }
tempdir = "0.3.7"
tempfile = "3"

[[example]]
name = "08_arrow_insert"
required-features = ["arrow"]

[[test]]
name = "arrow_insert"
required-features = ["arrow"]

[[test]]
name = "arrow_insert_bench"
required-features = ["arrow"]

[[test]]
name = "arrow_stream"
required-features = ["arrow"]

[[test]]
name = "arrow_stream_roundtrip"
required-features = ["arrow"]

[package.metadata.docs.rs]
# docs.rs cannot download libchdb or link native libraries.
# Setting DOCS_RS env var tells build.rs to skip native build steps.
features = ["arrow"]
rustc-args = ["--cfg", "docsrs"]
21 changes: 21 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,27 @@ cargo test -- --test-threads=1
- **Detailed documentation**: See [docs/examples.md](docs/examples.md) for comprehensive examples and explanations
- **Test examples**: See [tests/](tests/) directory for additional usage examples

## Arrow Bulk Insert

Apache Arrow bulk insert is enabled by default via the `arrow` feature (Arrow 59). Import Arrow types through `chdb_rust::arrow` so your `RecordBatch` types match the crate:

```rust
use chdb_rust::arrow::array::{Int64Array, RecordBatch};
use chdb_rust::arrow::datatypes::{DataType, Field, Schema};
```

SQL-only users can disable it to avoid building Arrow:

```toml
chdb-rust = { version = "1.4", default-features = false }
```

See [docs/examples.md](docs/examples.md#fast-bulk-inserts-arrow) for usage, or run:

```bash
cargo run --example 08_arrow_insert
```

## Contributing

We welcome contributions! Here's how you can help:
Expand Down
45 changes: 43 additions & 2 deletions build.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4,6 +4,8 @@ use std::path::{Path, PathBuf};
use std::process::Command;

fn main() {
println!("cargo::rustc-check-cfg=cfg(direct_arrow_insert)");

if env::var("DOCS_RS").is_ok() {
return;
}
Expand All @@ -13,7 +15,7 @@ fn main() {
let libchdb_info = find_libchdb_or_download(&out_path);
match libchdb_info {
Ok((lib_dir, header_path)) => {
setup_link_paths(&lib_dir);
setup_link_paths(&lib_dir, &header_path);
generate_bindings(&header_path, &out_path);
}
Err(e) => {
Expand Down Expand Up @@ -136,7 +138,30 @@ fn get_platform_string() -> Result<String, &'static str> {
}
}

fn setup_link_paths(lib_dir: &Path) {
fn lib_exports_symbol(lib_dir: &Path, symbol: &str) -> bool {
let lib_path = lib_dir.join("libchdb.so");
if !lib_path.exists() {
return false;
}
let output = Command::new("nm")
.args(["-D", lib_path.to_str().unwrap_or_default()])
.output();
match output {
Ok(o) if o.status.success() => {
let stdout = String::from_utf8_lossy(&o.stdout);
stdout.contains(symbol)
}
_ => false,
}
}

fn header_declares_direct_insert(header_path: &Path) -> bool {
fs::read_to_string(header_path)
.map(|s| s.contains("chdb_insert_arrow_array"))
.unwrap_or(false)
}

fn setup_link_paths(lib_dir: &Path, header_path: &Path) {
println!("cargo:rustc-link-search=native={}", lib_dir.display());
println!("cargo:rustc-link-search=native=./");
println!("cargo:rustc-link-search=native=/usr/local/lib");
Expand All @@ -162,11 +187,27 @@ fn setup_link_paths(lib_dir: &Path) {
println!("cargo:rustc-link-lib=chdb");
}

if header_declares_direct_insert(header_path)
&& lib_exports_symbol(lib_dir, "chdb_insert_arrow_array")
{
println!("cargo:rustc-cfg=direct_arrow_insert");
}

println!("cargo:rerun-if-changed=wrapper.h");
println!("cargo:rerun-if-changed=build.rs");
println!("cargo:rerun-if-changed={}", header_path.display());
if lib_dir.join("libchdb.so").exists() {
println!(
"cargo:rerun-if-changed={}",
lib_dir.join("libchdb.so").display()
);
}
}

fn generate_bindings(header_path: &Path, out_dir: &Path) {
let header_path = header_path
.canonicalize()
.unwrap_or_else(|_| header_path.to_path_buf());
let wrapper_content = format!("#include \"{}\"", header_path.display());
let temp_wrapper = out_dir.join("temp_wrapper.h");
if fs::read_to_string(&temp_wrapper)
Expand Down
72 changes: 72 additions & 0 deletions docs/examples.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ This document provides simple and easy-to-follow examples for using chdb-rust, a
5. [Output Formats](#output-formats)
6. [Reading from Files](#reading-from-files)
7. [Error Handling](#error-handling)
8. [Fast Bulk Inserts (Arrow)](#fast-bulk-inserts-arrow)

## Basic Setup

Expand Down Expand Up @@ -322,6 +323,77 @@ fn main() -> Result<(), chdb_rust::error::Error> {
}
```

## Fast Bulk Inserts (Arrow)

For high throughput and low latency, register in-memory Arrow data and insert with `INSERT ... SELECT` from the `ArrowStream` table function. This avoids temp-file Arrow IPC and SQL `VALUES` parsing.

### Ranked ingest paths

| Rank | Method | Notes |
|------|--------|-------|
| 1 | `register_arrow_stream` + `INSERT INTO dest SELECT * FROM ArrowStream('n')` | Multi-batch; no disk hop |
| 2 | `register_arrow_array` + same SQL | Single `RecordBatch` |
| 3 | `file(path, 'Native')` | Pre-encoded Native files |
| 4 | `file(path, 'RowBinary')` | Dense binary; requires `structure` in SQL |
| 5 | `file(path, 'ArrowStream')` | Arrow IPC on disk |
| 6 | `file(path, 'Parquet')` | Cold bulk loads |
| 7 | `INSERT ... VALUES` | Fine for demos; worst at scale |

`OutputFormat` on `query()` / `execute()` affects **result** encoding only, not ingest speed.

### SQL syntax

Registration exposes a **table function**, not a MergeTree table:

```sql
-- Correct
INSERT INTO dest SELECT * FROM ArrowStream('my_batch');

-- Wrong (UNKNOWN_TABLE)
SELECT * FROM my_batch;
```

In Rust, build the expression with `arrow_stream_table_sql("my_batch")`.

### FFI lifetime rules

- Pass raw `FFI_ArrowArrayStream` / `FFI_ArrowSchema` / `FFI_ArrowArray` pointers via [`ArrowStream::from_raw`](https://docs.rs/chdb-rust/latest/chdb_rust/arrow_stream/struct.ArrowStream.html) (not the opaque `chdb_arrow_stream_` typedef in `chdb.h`).
- Keep backing `RecordBatch` data and FFI structs alive until `unregister_arrow_table` or the connection closes.
- `chdb_arrow_scan` does not take ownership of the stream; do not free it before unregister.

### High-level helpers

```rust
use std::sync::Arc;
use chdb_rust::arrow::array::{Int64Array, RecordBatch};
use chdb_rust::arrow::datatypes::{DataType, Field, Schema};
use chdb_rust::arrow_insert::insert_record_batch;
use chdb_rust::session::SessionBuilder;

let session = SessionBuilder::new()
.with_data_path("/tmp/chdb-ingest")
.with_auto_cleanup(true)
.build()?;

session.execute(
"CREATE TABLE IF NOT EXISTS metrics (id Int64) ENGINE=MergeTree ORDER BY id",
None,
)?;

let batch = RecordBatch::try_new(
Arc::new(Schema::new(vec![Field::new("id", DataType::Int64, false)])),
vec![Arc::new(Int64Array::from(vec![1i64, 2, 3]))],
)?;

session.insert_record_batch("metrics", "flush_1", &batch)?;
```

`insert_record_batches` accepts multiple batches via an Arrow `RecordBatchReader`.

`dest_table` must be a bare ClickHouse table identifier (e.g. `metrics`); dotted names (`db.table`) and reserved words are not quoted automatically.

See `examples/08_arrow_insert.rs` for a runnable program.

## Additional Resources

- For more information about chDB, visit: https://github.com/chdb-io/chdb
Expand Down
62 changes: 62 additions & 0 deletions examples/08_arrow_insert.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,62 @@
//! Fast bulk insert: RecordBatch → ArrowStream registration → MergeTree.
//!
//! Run: cargo run --example 08_arrow_insert

use std::sync::Arc;

use chdb_rust::arrow::array::{Float64Array, Int64Array, RecordBatch};
use chdb_rust::arrow::datatypes::{DataType, Field, Schema};

use chdb_rust::arg::Arg;
use chdb_rust::format::OutputFormat;
use chdb_rust::session::SessionBuilder;
use chdb_rust::InsertOptions;

fn main() -> Result<(), chdb_rust::error::Error> {
let data_dir = std::env::temp_dir().join("chdb-rust-arrow-insert-example");
let session = SessionBuilder::new()
.with_data_path(&data_dir)
.with_auto_cleanup(true)
.build()?;

session.execute(
"CREATE TABLE IF NOT EXISTS metrics (id Int64, value Float64) \
ENGINE = MergeTree ORDER BY id",
None,
)?;

let schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Int64, false),
Field::new("value", DataType::Float64, false),
]));

let batch = RecordBatch::try_new(
schema,
vec![
Arc::new(Int64Array::from(vec![1i64, 2, 3, 4, 5])),
Arc::new(Float64Array::from(vec![10.0, 20.0, 30.0, 40.0, 50.0])),
],
)
.map_err(|e| chdb_rust::error::Error::QueryError(e.to_string()))?;

session.insert_record_batch(
"metrics",
"example_batch",
batch,
InsertOptions::default_bulk(),
)?;

let count = session.execute(
"SELECT count() AS c FROM metrics",
Some(&[Arg::OutputFormat(OutputFormat::TabSeparated)]),
)?;
println!("Row count after Arrow insert: {}", count.data_utf8_lossy());

let sum = session.execute(
"SELECT sum(value) FROM metrics",
Some(&[Arg::OutputFormat(OutputFormat::TabSeparated)]),
)?;
println!("Sum(value): {}", sum.data_utf8_lossy());

Ok(())
}
2 changes: 2 additions & 0 deletions examples/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@ cargo run --example 04_output_formats
cargo run --example 05_reading_from_files
cargo run --example 06_error_handling
cargo run --example 07_analytics
cargo run --example 08_arrow_insert
```

## Example Files
Expand All @@ -25,6 +26,7 @@ cargo run --example 07_analytics
5. **05_reading_from_files.rs** - Querying data from CSV and JSON files
6. **06_error_handling.rs** - Proper error handling patterns
7. **07_analytics.rs** - Complete analytics example with event tracking and aggregation
8. **08_arrow_insert.rs** - Fast bulk insert via Arrow C Data Interface (`insert_record_batch`)

## Prerequisites

Expand Down
Loading
Loading