dbkit/remote/
mod.rs

1//! Remote analytical sources.
2//!
3//! A [`RemoteSource`] is an external analytical database (e.g. ClickHouse) that
4//! you **query over the network** — the data already lives there, so unlike an
5//! embedded [`ReadEngine`](crate::ReadEngine) there is no `load_table` or sync.
6//! The contract is the same Arrow `RecordBatch`, so typed deserialization works
7//! identically via [`RemoteSourceExt::query_as`].
8
9use crate::DbkitError;
10use crate::analytical::RecordBatch;
11use crate::value::DbValue;
12use async_trait::async_trait;
13
14#[cfg(feature = "clickhouse")]
15pub mod clickhouse;
16
17/// An external analytical database queried over the network, returning Arrow.
18///
19/// Object-safe (only `query_arrow`), so `Box<dyn RemoteSource>` works. Typed
20/// reads are provided by the [`RemoteSourceExt`] extension trait.
21#[async_trait]
22pub trait RemoteSource: Send + Sync {
23    /// Run an analytical query against the remote store, returning Arrow batches.
24    ///
25    /// Parameter support is source-dependent; sources that don't support bind
26    /// parameters return an error if any are supplied.
27    async fn query_arrow(
28        &self,
29        sql: &str,
30        params: &[DbValue],
31    ) -> Result<Vec<RecordBatch>, DbkitError>;
32}
33
34/// Typed querying for any [`RemoteSource`].
35///
36/// Kept separate from `RemoteSource` so the base trait stays object-safe — this
37/// generic method is provided by a blanket impl and is callable on concrete
38/// sources and on `&dyn RemoteSource` alike.
39#[cfg(feature = "_read-typed")]
40#[async_trait]
41pub trait RemoteSourceExt: RemoteSource {
42    /// Run a query and deserialize each row into `T` (via `serde_arrow`).
43    async fn query_as<T>(&self, sql: &str, params: &[DbValue]) -> Result<Vec<T>, DbkitError>
44    where
45        T: serde::de::DeserializeOwned,
46    {
47        let batches = self.query_arrow(sql, params).await?;
48        crate::analytical::deserialize_batches(&batches)
49    }
50}
51
52#[cfg(feature = "_read-typed")]
53impl<S: RemoteSource + ?Sized> RemoteSourceExt for S {}
54
55/// A remote analytical store you can bulk-load Arrow data into.
56///
57/// The write-side mirror of [`RemoteSource`]: warehouse loads are bulk,
58/// columnar, and append-oriented, so the input is Arrow batches (not row-by-row
59/// transactional writes). Object-safe; typed writes come from [`RemoteSinkExt`].
60#[async_trait]
61pub trait RemoteSink: Send + Sync {
62    /// Append Arrow batches to `table` in the remote store. The table is
63    /// expected to already exist with a compatible schema. Empty input is a
64    /// no-op.
65    async fn write_arrow(
66        &self,
67        table: &str,
68        batches: Vec<RecordBatch>,
69    ) -> Result<(), DbkitError>;
70}
71
72/// Typed writing for any [`RemoteSink`].
73#[cfg(feature = "_read-typed")]
74#[async_trait]
75pub trait RemoteSinkExt: RemoteSink {
76    /// Serialize `rows` to Arrow (via `serde_arrow`) and append them to `table`.
77    async fn write_rows<T>(&self, table: &str, rows: &[T]) -> Result<(), DbkitError>
78    where
79        T: serde::Serialize + Sync,
80    {
81        if rows.is_empty() {
82            return Ok(());
83        }
84        let batch = crate::analytical::serialize_rows(rows)?;
85        self.write_arrow(table, vec![batch]).await
86    }
87}
88
89#[cfg(feature = "_read-typed")]
90impl<S: RemoteSink + ?Sized> RemoteSinkExt for S {}
dbkit/remote/mod.rs

dbkit/remote/
mod.rs
