connectorx/
lib.rs

1#![allow(clippy::upper_case_acronyms)]
2#![warn(missing_docs)]
3// The Trino and BigQuery client errors (232 and 176 bytes) are held by value in
4// the source error enums and propagate into `ConnectorXOutError`. Boxing them
5// would change public error types; the error path is cold, so the larger
6// `Result` is not worth an API break.
7#![allow(clippy::result_large_err)]
8
9//! # ConnectorX
10//!
11//! ConnectorX enables you to load data from databases into dataframes in the fastest and most memory efficient way by leveraging
12//! zero-copy and partition-based parallelism.
13//!
14//! Currently, ConnectorX consists of a Rust core library and a python library. This is the documentation for the Rust crate.
15//! For the documentation of the Python library, please refer to our [Github Readme](https://github.com/sfu-db/connector-x).
16//!
17//! # Design
18//!
19//! A data loading problem consists of three sub-problems:
20//! 1. How to connect to the data source and read data.
21//! 2. How to connect to the data destination and write data.
22//! 3. How to map the types between the source and destination.
23//!
24//! Additionally, since ConnectorX will partition a query into partitions and execute them in parallel, we also have
25//! 4. How to partition the query and run them in parallel.
26//!
27//! ConnectorX approaches these problems by defining abstractions on sources, destinations, and mapping rules.
28//! For the partition-based parallelism, ConnectorX will partition the query as well as the source and the destination
29//! together and put them into threads.
30//! Each thread will own exactly 1 query, 1 partitioned source, and 1 partitioned destination.
31//!
32//! The following graph depicts the internal mechanism when ConnectorX is downloading the data.
33//!
34//! ```text
35//!                     +------------------------------------------------------------+
36//!                     |                           Thread 1                         |
37//!                     |                                                            |
38//!     +---+           | +-----------------+   +-------------+  +-----------------+ |          +---+
39//!     |   +-----------+>| Partitioned Src +-->| Type Mapper +->| Partitioned Dst +-+--------->|   |
40//!     |   |           | +-----------------+   +-------------+  +-----------------+ |          |   |
41//!     | D |           |                                                            |          | D |
42//!     | a |           +------------------------------------------------------------+          | a |
43//!     | t |                                          .                                        | t |
44//!     | a |                                          .                                        | a |
45//!     | b |                                          .                                        | f |
46//!     | a |           +------------------------------------------------------------+          | r |
47//!     | s |           |                           Thread n                         |          | a |
48//!     | e |           |                                                            |          | m |
49//!     |   |           | +-----------------+   +-------------+  +-----------------+ |          | e |
50//!     |   +-----------+>| Partitioned Src +-->| Type Mapper +->| Partitioned Dst +-+--------->|   |
51//!     +---+           | +-----------------+   +-------------+  +-----------------+ |          +---+
52//!                     |                                                            |
53//!                     +------------------------------------------------------------+
54//!
55//! ```
56//! ## How does ConnectorX download the data?
57//!
58//! Upon receiving the query, e.g. SELECT * FROM lineitem, ConnectorX will first issue a LIMIT 0 query SELECT * FROM lineitem LIMIT 0 to get the schema of the result set.
59//!
60//! Then, if partition_on is specified, ConnectorX will issue `SELECT MIN($partition_on), MAX($partition_on) FROM (SELECT * FROM lineitem)` to know the range of the partition column.
61//! After that, the original query is split into partitions based on the min/max information, e.g. `SELECT * FROM (SELECT * FROM lineitem) WHERE $partition_on > 0 AND $partition_on < 10000`.
62//! ConnectorX will then run a count query to get the partition size (e.g. `SELECT COUNT(*) FROM (SELECT * FROM lineitem) WHERE $partition_on > 0 AND $partition_on < 10000`).
63//! If the partition is not specified, the count query will be `SELECT COUNT(*) FROM (SELECT * FROM lineitem)`.
64//!
65//! Finally, ConnectorX will use the schema info as well as the count info to allocate memory and download data by executing the queries normally.
66//! Once the downloading begins, there will be one thread for each partition so that the data are downloaded in parallel at the partition level.
67//! The thread will issue the query of the corresponding partition to the database and then write the returned data to the destination row-wise or column-wise (depends on the database) in a streaming fashion.
68//! This mechanism implies that having an index on the partition column is recommended to make full use of the parallel downloading power provided by ConnectorX.
69//!
70//! # Extending ConnectorX
71//! ## Adding a new source
72//!
73//! To add a new data source, you need to implement [`sources::Source`], [`sources::SourcePartition`], [`sources::PartitionParser`], and [`sources::Produce`] for the source.
74//! In detail, [`sources::Source`] describes how to connect to the database from a connection string, as well as how to do partitioning on the source to produce a list of [`sources::SourcePartition`].
75//! [`sources::SourcePartition`] describes how to get the row count for the specific partition so that the destination can preallocate the memory.
76//! Finally, [`sources::PartitionParser`] and [`sources::Produce`] abstracts away the detail about how does each partition parse different types.
77//!
78//! ## Adding a new destination
79//!
80//! To add a new data destination, you need to implement [`destinations::Destination`], [`destinations::DestinationPartition`], and [`destinations::Consume`]. Similar to the sources,
81//! [`destinations::Destination`] describes how to allocate the memory of the data destination, as well as how to do partitioning on the destination to produce a list of [`destinations::DestinationPartition`].
82//! [`destinations::DestinationPartition`] and [`destinations::Consume`] abstract away the detail about how does each partition writes different types.
83//!
84//! ## Adding a new transport (type mapping)
85//!
86//! After having a source and a destination that describes how to read and write the data,
87//! ConnectorX also needs to know how to convert the values with different types from the source to the destination.
88//! For example, Postgres can produce a `uuid` type but there's no uuid in Arrow. It is the transport's duty to convert
89//! the `uuid` into an Arrow compatible type, e.g. string. You can use the [`impl_transport!`] macro to define a transport.
90//!
91//! ## Putting things together
92//!
93//! Say, you decide to load data from SQL Server to Arrow. In ConnectorX we already provided the source for SQL Server as [`sources::sqlite::SQLiteSource`], and the
94//! Arrow destination [`destinations::arrow::ArrowDestination`], as well as the transport [`transports::SQLiteArrowTransport`].
95//! Given the source, destination and transport already implemented, you can use [`dispatcher::Dispatcher`] to load the data:
96//!
97//! ```no_run
98//! use connectorx::prelude::*;
99//!
100//! let mut destination = ArrowDestination::new();
101//! let source = SQLiteSource::new("/path/to/db", 10).expect("cannot create the source");
102//! let queries = &["SELECT * FROM db WHERE id < 100", "SELECT * FROM db WHERE id >= 100"];
103//! let dispatcher = Dispatcher::<SQLiteSource, ArrowDestination, SQLiteArrowTransport>::new(source, &mut destination, queries, None);
104//! dispatcher.run().expect("run failed");
105//!
106//! let data = destination.arrow();
107//! ```
108//!
109//! Or simply you can directly use the [`get_arrow::get_arrow`] in which we wrapped the above procedures:
110//!
111//! ```no_run
112//! use connectorx::prelude::*;
113//! use std::convert::TryFrom;
114//!
115//! let mut source_conn = SourceConn::try_from("postgresql://username:password@host:port/db?cxprotocol=binary").expect("parse conn str failed");
116//! let queries = &[CXQuery::from("SELECT * FROM table WHERE id < 100"), CXQuery::from("SELECT * FROM table WHERE id >= 100")];
117//! let destination = get_arrow(&source_conn, None, queries, None).expect("run failed");
118//!
119//! let data = destination.arrow();
120//! ```
121//!
122//! NOTE: the pool size parameter `nconn` used in initializing the source should be larger than or equal to the number of partitioned queries input later.
123//!
124//! ## Need more examples?
125//! You can use the existing implementation as the example.
126//! [MySQL source](https://github.com/sfu-db/connector-x/tree/main/connectorx/src/sources/mysql),
127//! [Arrow destination](https://github.com/sfu-db/connector-x/tree/main/connectorx/src/destinations/arrow),
128//! [MySQL to Arrow transport](https://github.com/sfu-db/connector-x/blob/main/connectorx/src/transports/mysql_arrow.rs).
129//!
130//! # Sources protocols & Destinations that is implemented in the Rust core.
131//!
132//! ## Sources
133//! - [x] Postgres
134//! - [x] Mysql
135//! - [x] Sqlite
136//! - [x] SQL Server
137//! - [x] Oracle
138//! - [x] BigQuery
139//!
140//! ## Destinations
141//! - [x] Arrow
142//! - [x] Polars
143//!
144//! # Feature gates
145//! By default, ConnectorX does not enable any sources / destinations to keep the dependencies minimal.
146//! Instead, we provide following features for you to opt-in: `src_sqlite`, `src_postgres`, `src_mysql`, `src_mssql`, `src_oracle`, `dst_arrow`, `dst_polars`.
147//! For example, if you'd like to load data from Postgres to Arrow, you can enable `src_postgres` and `dst_arrow` in `Cargo.toml`.
148//! This will enable [`sources::postgres`], [`destinations::arrow`] and [`transports::PostgresArrowTransport`].
149
150pub mod typesystem;
151#[macro_use]
152mod macros;
153#[cfg(feature = "dst_arrow")]
154pub mod arrow_batch_iter;
155pub mod constants;
156pub mod data_order;
157pub mod destinations;
158mod dispatcher;
159pub mod errors;
160#[cfg(feature = "fed_exec")]
161pub mod fed_dispatcher;
162#[cfg(feature = "federation")]
163pub mod fed_rewriter;
164#[cfg(feature = "dst_arrow")]
165pub mod get_arrow;
166pub mod partition;
167pub mod source_router;
168pub mod sources;
169#[doc(hidden)]
170pub mod sql;
171pub mod transports;
172#[doc(hidden)]
173pub mod utils;
174
175pub mod prelude {
176    #[cfg(feature = "dst_arrow")]
177    pub use crate::arrow_batch_iter::{set_global_num_thread, RecordBatchIterator};
178    pub use crate::data_order::{coordinate, DataOrder};
179    #[cfg(feature = "dst_arrow")]
180    pub use crate::destinations::arrow::{ArrowDestination, ArrowPartitionWriter, ArrowTypeSystem};
181    #[cfg(feature = "dst_arrow")]
182    pub use crate::destinations::arrowstream::{
183        ArrowDestination as ArrowStreamDestination,
184        ArrowPartitionWriter as ArrowStreamPartitionWriter,
185        ArrowTypeSystem as ArrowStreamTypeSystem,
186    };
187    pub use crate::destinations::{Consume, Destination, DestinationPartition};
188    pub use crate::dispatcher::Dispatcher;
189    pub use crate::errors::{ConnectorXError, ConnectorXOutError};
190    #[cfg(feature = "federation")]
191    pub use crate::fed_rewriter::{rewrite_sql, FederatedDataSourceInfo, Plan};
192    #[cfg(feature = "dst_arrow")]
193    pub use crate::get_arrow::{get_arrow, new_record_batch_iter};
194    pub use crate::source_router::*;
195    #[cfg(feature = "src_bigquery")]
196    pub use crate::sources::bigquery::BigQuerySource;
197    #[cfg(feature = "src_clickhouse")]
198    pub use crate::sources::clickhouse::ClickHouseSource;
199    #[cfg(feature = "src_csv")]
200    pub use crate::sources::csv::CSVSource;
201    #[cfg(feature = "src_dummy")]
202    pub use crate::sources::dummy::DummySource;
203    #[cfg(feature = "src_mssql_common")]
204    pub use crate::sources::mssql::MsSQLSource;
205    #[cfg(feature = "src_mysql")]
206    pub use crate::sources::mysql::MySQLSource;
207    #[cfg(feature = "src_oracle")]
208    pub use crate::sources::oracle::OracleSource;
209    #[cfg(feature = "src_postgres")]
210    pub use crate::sources::postgres::PostgresSource;
211    #[cfg(feature = "src_sqlite")]
212    pub use crate::sources::sqlite::SQLiteSource;
213    #[cfg(feature = "src_trino")]
214    pub use crate::sources::trino::TrinoSource;
215    pub use crate::sources::{PartitionParser, Produce, Source, SourcePartition};
216    pub use crate::sql::CXQuery;
217    pub use crate::transports::*;
218    pub use crate::typesystem::{
219        ParameterizedFunc, ParameterizedOn, Realize, Transport, TypeAssoc, TypeConversion,
220        TypeSystem,
221    };
222}
connectorx/lib.rs

connectorx/
lib.rs
