Skip to content
Open
7 changes: 5 additions & 2 deletions .github/workflows/ci.yml
Original file line number Diff line number Diff line change
Expand Up @@ -44,8 +44,11 @@ jobs:
- name: Is the code formatted?
run: cargo fmt --all -- --check -q

# -D warnings fails the build on any new warning. missing_docs is kept as a
# warning (lib.rs enables it to track the documentation backlog) until that
# backlog is cleared; --force-warn stops -D warnings from promoting it.
- name: Clippy linting
run: cargo clippy --features all
run: cargo clippy --features all -- -D warnings --force-warn missing_docs

- name: Check each feature gate
run: just test-feature-gate
Expand Down Expand Up @@ -209,7 +212,7 @@ jobs:
key: ${{ runner.os }}-cargo-python-${{ matrix.python-version }}-${{ secrets.CACHE_VERSION }}-${{ hashFiles('**/Cargo.lock') }}

- name: Clippy linting
run: cargo clippy --features all --all
run: cargo clippy --features all --all -- -D warnings --force-warn missing_docs

- name: Cache venv
uses: actions/cache@v6
Expand Down
6 changes: 2 additions & 4 deletions connectorx-cpp/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -251,12 +251,10 @@ pub unsafe extern "C" fn connectorx_scan(conn: *const c_char, query: *const c_ch
result.push(cx_rb);
}

let res = CXResult {
CXResult {
data: CXSlice::<_>::new_from_vec(result),
header: CXSlice::<_>::new_from_vec(names),
};

res
}
}

#[repr(C)]
Expand Down
13 changes: 6 additions & 7 deletions connectorx-cpp/src/plan.rs
Original file line number Diff line number Diff line change
@@ -1,22 +1,21 @@
use crate::CXFederatedPlan;
use connectorx::fed_rewriter::Plan;
use libc::c_char;
use std::convert::Into;
use std::ffi::CString;

impl Into<CXFederatedPlan> for Plan {
fn into(self) -> CXFederatedPlan {
impl From<Plan> for CXFederatedPlan {
fn from(plan: Plan) -> Self {
CXFederatedPlan {
db_name: CString::new(self.db_name.as_str())
db_name: CString::new(plan.db_name.as_str())
.expect("new CString error")
.into_raw() as *const c_char,
db_alias: CString::new(self.db_alias.as_str())
db_alias: CString::new(plan.db_alias.as_str())
.expect("new CString error")
.into_raw() as *const c_char,
sql: CString::new(self.sql.as_str())
sql: CString::new(plan.sql.as_str())
.expect("new CString error")
.into_raw() as *const c_char,
cardinality: self.cardinality,
cardinality: plan.cardinality,
}
}
}
2 changes: 1 addition & 1 deletion connectorx/src/constants.rs
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,7 @@ pub const RECORD_BATCH_SIZE: usize = 64 * KILO;
))]
pub const DB_BUFFER_SIZE: usize = 32;

#[cfg(any(feature = "src_oracle"))]
#[cfg(feature = "src_oracle")]
pub const ORACLE_ARRAY_SIZE: u32 = KILO as u32;

#[cfg(all(not(debug_assertions), feature = "federation"))]
Expand Down
27 changes: 13 additions & 14 deletions connectorx/src/dispatcher.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
///! This module provides [`dispatcher::Dispatcher`], the core struct in ConnectorX
///! that drives the data loading from a source to a destination.
//! This module provides [`dispatcher::Dispatcher`], the core struct in ConnectorX
//! that drives the data loading from a source to a destination.
use crate::{
data_order::{coordinate, DataOrder},
destinations::{Destination, DestinationPartition},
Expand All @@ -13,6 +13,16 @@ use log::debug;
use rayon::prelude::*;
use std::marker::PhantomData;

/// What [`Dispatcher::prepare`] returns: the negotiated data order, the source and
/// destination partitions, and the source and destination schemas.
pub type Prepared<'w, S, D> = (
DataOrder,
Vec<<S as Source>::Partition>,
Vec<<D as Destination>::Partition<'w>>,
Vec<<S as Source>::TypeSystem>,
Vec<<D as Destination>::TypeSystem>,
);

/// A dispatcher takes a `S: Source`, a `D: Destination`, a `TP: Transport` and a vector of `queries` as input to
/// load data from `S` to `D` using the queries.
pub struct Dispatcher<'a, S, D, TP> {
Expand Down Expand Up @@ -47,18 +57,7 @@ where
self.src.set_pre_execution_queries(pre_execution_queries);
}

pub fn prepare(
mut self,
) -> Result<
(
DataOrder,
Vec<S::Partition>,
Vec<D::Partition<'w>>,
Vec<S::TypeSystem>,
Vec<D::TypeSystem>,
),
TP::Error,
> {
pub fn prepare(mut self) -> Result<Prepared<'w, S, D>, TP::Error> {
debug!("Prepare");
let dorder = coordinate(S::DATA_ORDERS, D::DATA_ORDERS)?;
self.src.set_data_order(dorder)?;
Expand Down
3 changes: 1 addition & 2 deletions connectorx/src/fed_rewriter.rs
Original file line number Diff line number Diff line change
Expand Up @@ -88,8 +88,7 @@ fn create_sources(
let db_manual = jvm.create_instance("java.util.HashMap", InvocationArg::empty())?;

for (db_name, db_info) in db_map.iter() {
if db_info.manual_info.is_some() {
let manual_info = db_info.manual_info.as_ref().unwrap();
if let Some(manual_info) = &db_info.manual_info {
let schema_info = jvm.create_instance("java.util.HashMap", InvocationArg::empty())?;
for (name, columns) in manual_info {
let arr_instance = jvm.java_list("java.lang.String", columns.to_vec())?;
Expand Down
5 changes: 5 additions & 0 deletions connectorx/src/lib.rs
Original file line number Diff line number Diff line change
@@ -1,5 +1,10 @@
#![allow(clippy::upper_case_acronyms)]
#![warn(missing_docs)]
// The Trino and BigQuery client errors (232 and 176 bytes) are held by value in
// the source error enums and propagate into `ConnectorXOutError`. Boxing them
// would change public error types; the error path is cold, so the larger
// `Result` is not worth an API break.
#![allow(clippy::result_large_err)]

//! # ConnectorX
//!
Expand Down
24 changes: 11 additions & 13 deletions connectorx/src/sources/mssql/tiberius_impl.rs
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ pub fn mssql_config(url: &Url) -> Config {
}
config.port(url.port().unwrap_or(1433));
// remove the leading "/"
config.database(decode(&url.path()[1..])?.to_owned());
config.database(decode(&url.path()[1..])?.into_owned());
// Using SQL Server authentication.
#[allow(unused)]
let params: HashMap<String, String> = url.query_pairs().into_owned().collect();
Expand All @@ -75,37 +75,35 @@ pub fn mssql_config(url: &Url) -> Config {
_ => {
debug!("mssql auth through sqlserver authentication");
config.authentication(AuthMethod::sql_server(
decode(url.username())?.to_owned(),
decode(url.password().unwrap_or(""))?.to_owned(),
decode(url.username())?.into_owned(),
decode(url.password().unwrap_or(""))?.into_owned(),
));
}
};
#[cfg(all(not(windows), not(feature = "integrated-auth-gssapi")))]
config.authentication(AuthMethod::sql_server(
decode(url.username())?.to_owned(),
decode(url.password().unwrap_or(""))?.to_owned(),
decode(url.username())?.into_owned(),
decode(url.password().unwrap_or(""))?.into_owned(),
));

match params.get("trust_server_certificate") {
Some(v) if v.to_lowercase() == "true" => config.trust_cert(),
_ => {}
};

match params.get("trust_server_certificate_ca") {
Some(v) => config.trust_cert_ca(v),
_ => {}
};
if let Some(v) = params.get("trust_server_certificate_ca") {
config.trust_cert_ca(v);
}

match params.get("encrypt") {
Some(v) if v.to_lowercase() == "true" => config.encryption(EncryptionLevel::Required),
Some(v) if v.to_lowercase() == "false" => config.encryption(EncryptionLevel::Off),
_ => config.encryption(EncryptionLevel::NotSupported),
};

match params.get("appname") {
Some(appname) => config.application_name(decode(appname)?.to_owned()),
_ => {}
};
if let Some(appname) = params.get("appname") {
config.application_name(decode(appname)?.into_owned());
}

config
}
Expand Down
1 change: 0 additions & 1 deletion connectorx/src/sources/trino/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -562,7 +562,6 @@ macro_rules! impl_produce_text {
.parse()
.map_err(|_| anyhow!("Trino cannot convert complex value at ({}, {}): {:?}", ridx, cidx, value))?)
}
_ => throw!(anyhow!("Trino unknown value at position: ({}, {}): {:?}", ridx, cidx, value))
}
}
}
Expand Down
Loading