connectorx/
partition.rs

1use std::sync::Arc;
2
3use crate::errors::{ConnectorXOutError, OutResult};
4use crate::source_router::{SourceConn, SourceType};
5#[cfg(feature = "src_bigquery")]
6use crate::sources::bigquery::BigQueryDialect;
7#[cfg(feature = "src_clickhouse")]
8use crate::sources::clickhouse::{ClickHouseSource, ClickHouseSourceError};
9#[cfg(feature = "src_mssql")]
10use crate::sources::mssql::{mssql_config, FloatN, IntN, MsSQLTypeSystem};
11#[cfg(feature = "src_mysql")]
12use crate::sources::mysql::{MySQLSourceError, MySQLTypeSystem};
13#[cfg(feature = "src_oracle")]
14use crate::sources::oracle::{OracleDialect, OracleSource};
15#[cfg(feature = "src_postgres")]
16use crate::sources::postgres::{rewrite_tls_args, PostgresTypeSystem};
17#[cfg(feature = "src_trino")]
18use crate::sources::trino::TrinoDialect;
19#[cfg(feature = "src_sqlite")]
20use crate::sql::get_partition_range_query_sep;
21use crate::sql::{get_partition_range_query, single_col_partition_query, CXQuery};
22use anyhow::anyhow;
23use fehler::{throw, throws};
24#[cfg(feature = "src_bigquery")]
25use gcp_bigquery_client;
26#[cfg(feature = "src_mysql")]
27use r2d2_mysql::mysql::{prelude::Queryable, Opts, Pool, Row};
28#[cfg(feature = "src_sqlite")]
29use rusqlite::{types::Type, Connection};
30#[cfg(feature = "src_postgres")]
31use rust_decimal::{prelude::ToPrimitive, Decimal};
32#[cfg(feature = "src_postgres")]
33use rust_decimal_macros::dec;
34#[cfg(feature = "src_clickhouse")]
35use serde::Deserialize;
36#[cfg(feature = "src_clickhouse")]
37use serde_json::Value as JsonValue;
38#[cfg(feature = "src_clickhouse")]
39use sqlparser::dialect::ClickHouseDialect;
40#[cfg(feature = "src_mssql")]
41use sqlparser::dialect::MsSqlDialect;
42#[cfg(feature = "src_mysql")]
43use sqlparser::dialect::MySqlDialect;
44#[cfg(feature = "src_postgres")]
45use sqlparser::dialect::PostgreSqlDialect;
46#[cfg(feature = "src_sqlite")]
47use sqlparser::dialect::SQLiteDialect;
48#[cfg(feature = "src_mssql")]
49use tiberius::Client;
50#[cfg(any(feature = "src_bigquery", feature = "src_mssql", feature = "src_trino"))]
51use tokio::{net::TcpStream, runtime::Runtime};
52#[cfg(feature = "src_mssql")]
53use tokio_util::compat::TokioAsyncWriteCompatExt;
54use url::Url;
55
56pub struct PartitionQuery {
57    query: String,
58    column: String,
59    min: Option<i64>,
60    max: Option<i64>,
61    num: usize,
62}
63
64impl PartitionQuery {
65    pub fn new(query: &str, column: &str, min: Option<i64>, max: Option<i64>, num: usize) -> Self {
66        Self {
67            query: query.into(),
68            column: column.into(),
69            min,
70            max,
71            num,
72        }
73    }
74}
75
76pub fn partition(part: &PartitionQuery, source_conn: &SourceConn) -> OutResult<Vec<CXQuery>> {
77    let mut queries = vec![];
78    let num = part.num as i64;
79    let (min, max) = match (part.min, part.max) {
80        (None, None) => get_col_range(source_conn, &part.query, &part.column)?,
81        (Some(min), Some(max)) => (min, max),
82        _ => throw!(anyhow!(
83            "partition_query range can not be partially specified",
84        )),
85    };
86
87    let partition_size = (max - min + 1) / num;
88
89    for i in 0..num {
90        let lower = min + i * partition_size;
91        let upper = match i == num - 1 {
92            true => max + 1,
93            false => min + (i + 1) * partition_size,
94        };
95        let partition_query = get_part_query(source_conn, &part.query, &part.column, lower, upper)?;
96        queries.push(partition_query);
97    }
98    Ok(queries)
99}
100
101pub fn get_col_range(source_conn: &SourceConn, query: &str, col: &str) -> OutResult<(i64, i64)> {
102    match source_conn.ty {
103        #[cfg(feature = "src_postgres")]
104        SourceType::Postgres => pg_get_partition_range(&source_conn.conn, query, col),
105        #[cfg(feature = "src_sqlite")]
106        SourceType::SQLite => sqlite_get_partition_range(&source_conn.conn, query, col),
107        #[cfg(feature = "src_mysql")]
108        SourceType::MySQL => mysql_get_partition_range(&source_conn.conn, query, col),
109        #[cfg(feature = "src_mssql")]
110        SourceType::MsSQL => mssql_get_partition_range(&source_conn.conn, query, col),
111        #[cfg(feature = "src_oracle")]
112        SourceType::Oracle => oracle_get_partition_range(&source_conn.conn, query, col),
113        #[cfg(feature = "src_bigquery")]
114        SourceType::BigQuery => bigquery_get_partition_range(&source_conn.conn, query, col),
115        #[cfg(feature = "src_trino")]
116        SourceType::Trino => trino_get_partition_range(&source_conn.conn, query, col),
117        #[cfg(feature = "src_clickhouse")]
118        SourceType::ClickHouse => clickhouse_get_partition_range(&source_conn.conn, query, col),
119        _ => unimplemented!("{:?} not implemented!", source_conn.ty),
120    }
121}
122
123#[throws(ConnectorXOutError)]
124pub fn get_part_query(
125    source_conn: &SourceConn,
126    query: &str,
127    col: &str,
128    lower: i64,
129    upper: i64,
130) -> CXQuery<String> {
131    let query = match source_conn.ty {
132        #[cfg(feature = "src_postgres")]
133        SourceType::Postgres => {
134            single_col_partition_query(query, col, lower, upper, &PostgreSqlDialect {})?
135        }
136        #[cfg(feature = "src_sqlite")]
137        SourceType::SQLite => {
138            single_col_partition_query(query, col, lower, upper, &SQLiteDialect {})?
139        }
140        #[cfg(feature = "src_mysql")]
141        SourceType::MySQL => {
142            single_col_partition_query(query, col, lower, upper, &MySqlDialect {})?
143        }
144        #[cfg(feature = "src_mssql")]
145        SourceType::MsSQL => {
146            single_col_partition_query(query, col, lower, upper, &MsSqlDialect {})?
147        }
148        #[cfg(feature = "src_oracle")]
149        SourceType::Oracle => {
150            single_col_partition_query(query, col, lower, upper, &OracleDialect {})?
151        }
152        #[cfg(feature = "src_bigquery")]
153        SourceType::BigQuery => {
154            single_col_partition_query(query, col, lower, upper, &BigQueryDialect {})?
155        }
156        #[cfg(feature = "src_trino")]
157        SourceType::Trino => {
158            single_col_partition_query(query, col, lower, upper, &TrinoDialect {})?
159        }
160        #[cfg(feature = "src_clickhouse")]
161        SourceType::ClickHouse => {
162            single_col_partition_query(query, col, lower, upper, &ClickHouseDialect {})?
163        }
164        _ => unimplemented!("{:?} not implemented!", source_conn.ty),
165    };
166    CXQuery::Wrapped(query)
167}
168
169#[cfg(feature = "src_postgres")]
170#[throws(ConnectorXOutError)]
171fn pg_get_partition_range(conn: &Url, query: &str, col: &str) -> (i64, i64) {
172    let (config, tls) = rewrite_tls_args(conn)?;
173    let mut client = match tls {
174        None => config.connect(postgres::NoTls)?,
175        Some(tls_conn) => config.connect(tls_conn)?,
176    };
177    let range_query = get_partition_range_query(query, col, &PostgreSqlDialect {})?;
178    let row = client.query_one(range_query.as_str(), &[])?;
179
180    let col_type = PostgresTypeSystem::from(row.columns()[0].type_());
181    let (min_v, max_v) = match col_type {
182        PostgresTypeSystem::Int2(_) => {
183            let min_v: Option<i16> = row.get(0);
184            let max_v: Option<i16> = row.get(1);
185            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
186        }
187        PostgresTypeSystem::Int4(_) => {
188            let min_v: Option<i32> = row.get(0);
189            let max_v: Option<i32> = row.get(1);
190            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
191        }
192        PostgresTypeSystem::Int8(_) => {
193            let min_v: Option<i64> = row.get(0);
194            let max_v: Option<i64> = row.get(1);
195            (min_v.unwrap_or(0), max_v.unwrap_or(0))
196        }
197        PostgresTypeSystem::Float4(_) => {
198            let min_v: Option<f32> = row.get(0);
199            let max_v: Option<f32> = row.get(1);
200            (min_v.unwrap_or(0.0) as i64, max_v.unwrap_or(0.0) as i64)
201        }
202        PostgresTypeSystem::Float8(_) => {
203            let min_v: Option<f64> = row.get(0);
204            let max_v: Option<f64> = row.get(1);
205            (min_v.unwrap_or(0.0) as i64, max_v.unwrap_or(0.0) as i64)
206        }
207        PostgresTypeSystem::Numeric(_) => {
208            let min_v: Option<Decimal> = row.get(0);
209            let max_v: Option<Decimal> = row.get(1);
210            (
211                min_v.unwrap_or(dec!(0.0)).to_i64().unwrap_or(0),
212                max_v.unwrap_or(dec!(0.0)).to_i64().unwrap_or(0),
213            )
214        }
215        _ => throw!(anyhow!(
216            "Partition can only be done on int or float columns"
217        )),
218    };
219
220    (min_v, max_v)
221}
222
223#[cfg(feature = "src_sqlite")]
224#[throws(ConnectorXOutError)]
225fn sqlite_get_partition_range(conn: &Url, query: &str, col: &str) -> (i64, i64) {
226    // remove the first "sqlite://" manually since url.path is not correct for windows and for relative path
227    let conn = Connection::open(&conn.as_str()[9..])?;
228    // SQLite only optimize min max queries when there is only one aggregation
229    // https://www.sqlite.org/optoverview.html#minmax
230    let (min_query, max_query) = get_partition_range_query_sep(query, col, &SQLiteDialect {})?;
231    let mut error = None;
232    let min_v = conn.query_row(min_query.as_str(), [], |row| {
233        // declare type for count query will be None, only need to check the returned value type
234        let col_type = row.get_ref(0)?.data_type();
235        match col_type {
236            Type::Integer => row.get(0),
237            Type::Real => {
238                let v: f64 = row.get(0)?;
239                Ok(v as i64)
240            }
241            Type::Null => Ok(0),
242            _ => {
243                error = Some(anyhow!("Partition can only be done on integer columns"));
244                Ok(0)
245            }
246        }
247    })?;
248    match error {
249        None => {}
250        Some(e) => throw!(e),
251    }
252    let max_v = conn.query_row(max_query.as_str(), [], |row| {
253        let col_type = row.get_ref(0)?.data_type();
254        match col_type {
255            Type::Integer => row.get(0),
256            Type::Real => {
257                let v: f64 = row.get(0)?;
258                Ok(v as i64)
259            }
260            Type::Null => Ok(0),
261            _ => {
262                error = Some(anyhow!("Partition can only be done on integer columns"));
263                Ok(0)
264            }
265        }
266    })?;
267    match error {
268        None => {}
269        Some(e) => throw!(e),
270    }
271
272    (min_v, max_v)
273}
274
275#[cfg(feature = "src_mysql")]
276#[throws(ConnectorXOutError)]
277fn mysql_get_partition_range(conn: &Url, query: &str, col: &str) -> (i64, i64) {
278    let pool = Pool::new(Opts::from_url(conn.as_str()).map_err(MySQLSourceError::MySQLUrlError)?)?;
279    let mut conn = pool.get_conn()?;
280    let range_query = get_partition_range_query(query, col, &MySqlDialect {})?;
281    let row: Row = conn
282        .query_first(range_query)?
283        .ok_or_else(|| anyhow!("mysql range: no row returns"))?;
284
285    let col_type = MySQLTypeSystem::from((
286        &row.columns()[0].column_type(),
287        &row.columns()[0].flags(),
288        row.columns()[0].character_set(),
289    ));
290
291    let (min_v, max_v) = match col_type {
292        MySQLTypeSystem::Tiny(_) => {
293            let min_v: Option<i8> = row
294                .get(0)
295                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
296            let max_v: Option<i8> = row
297                .get(1)
298                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
299            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
300        }
301        MySQLTypeSystem::Short(_) => {
302            let min_v: Option<i16> = row
303                .get(0)
304                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
305            let max_v: Option<i16> = row
306                .get(1)
307                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
308            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
309        }
310        MySQLTypeSystem::Int24(_) => {
311            let min_v: Option<i32> = row
312                .get(0)
313                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
314            let max_v: Option<i32> = row
315                .get(1)
316                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
317            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
318        }
319        MySQLTypeSystem::Long(_) => {
320            let min_v: Option<i64> = row
321                .get(0)
322                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
323            let max_v: Option<i64> = row
324                .get(1)
325                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
326            (min_v.unwrap_or(0), max_v.unwrap_or(0))
327        }
328        MySQLTypeSystem::LongLong(_) => {
329            let min_v: Option<i64> = row
330                .get(0)
331                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
332            let max_v: Option<i64> = row
333                .get(1)
334                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
335            (min_v.unwrap_or(0), max_v.unwrap_or(0))
336        }
337        MySQLTypeSystem::UTiny(_) => {
338            let min_v: Option<u8> = row
339                .get(0)
340                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
341            let max_v: Option<u8> = row
342                .get(1)
343                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
344            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
345        }
346        MySQLTypeSystem::UShort(_) => {
347            let min_v: Option<u16> = row
348                .get(0)
349                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
350            let max_v: Option<u16> = row
351                .get(1)
352                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
353            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
354        }
355        MySQLTypeSystem::UInt24(_) => {
356            let min_v: Option<u32> = row
357                .get(0)
358                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
359            let max_v: Option<u32> = row
360                .get(1)
361                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
362            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
363        }
364        MySQLTypeSystem::ULong(_) => {
365            let min_v: Option<u32> = row
366                .get(0)
367                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
368            let max_v: Option<u32> = row
369                .get(1)
370                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
371            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
372        }
373        MySQLTypeSystem::ULongLong(_) => {
374            let min_v: Option<u64> = row
375                .get(0)
376                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
377            let max_v: Option<u64> = row
378                .get(1)
379                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
380            (min_v.unwrap_or(0) as i64, max_v.unwrap_or(0) as i64)
381        }
382        MySQLTypeSystem::Float(_) => {
383            let min_v: Option<f32> = row
384                .get(0)
385                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
386            let max_v: Option<f32> = row
387                .get(1)
388                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
389            (min_v.unwrap_or(0.0) as i64, max_v.unwrap_or(0.0) as i64)
390        }
391        MySQLTypeSystem::Double(_) => {
392            let min_v: Option<f64> = row
393                .get(0)
394                .ok_or_else(|| anyhow!("mysql range: cannot get min value"))?;
395            let max_v: Option<f64> = row
396                .get(1)
397                .ok_or_else(|| anyhow!("mysql range: cannot get max value"))?;
398            (min_v.unwrap_or(0.0) as i64, max_v.unwrap_or(0.0) as i64)
399        }
400        _ => throw!(anyhow!("Partition can only be done on int columns")),
401    };
402
403    (min_v, max_v)
404}
405
406#[cfg(feature = "src_mssql")]
407#[throws(ConnectorXOutError)]
408fn mssql_get_partition_range(conn: &Url, query: &str, col: &str) -> (i64, i64) {
409    let rt = Runtime::new().expect("Failed to create runtime");
410    let config = mssql_config(conn)?;
411    let tcp = rt.block_on(TcpStream::connect(config.get_addr()))?;
412    tcp.set_nodelay(true)?;
413
414    let mut client = rt.block_on(Client::connect(config, tcp.compat_write()))?;
415
416    let range_query = get_partition_range_query(query, col, &MsSqlDialect {})?;
417    let query_result = rt.block_on(client.query(range_query.as_str(), &[]))?;
418    let row = rt.block_on(query_result.into_row())?.unwrap();
419
420    let col_type = MsSQLTypeSystem::from(&row.columns()[0].column_type());
421    let (min_v, max_v) = match col_type {
422        MsSQLTypeSystem::Tinyint(_) => {
423            let min_v: u8 = row.get(0).unwrap_or(0);
424            let max_v: u8 = row.get(1).unwrap_or(0);
425            (min_v as i64, max_v as i64)
426        }
427        MsSQLTypeSystem::Smallint(_) => {
428            let min_v: i16 = row.get(0).unwrap_or(0);
429            let max_v: i16 = row.get(1).unwrap_or(0);
430            (min_v as i64, max_v as i64)
431        }
432        MsSQLTypeSystem::Int(_) => {
433            let min_v: i32 = row.get(0).unwrap_or(0);
434            let max_v: i32 = row.get(1).unwrap_or(0);
435            (min_v as i64, max_v as i64)
436        }
437        MsSQLTypeSystem::Bigint(_) => {
438            let min_v: i64 = row.get(0).unwrap_or(0);
439            let max_v: i64 = row.get(1).unwrap_or(0);
440            (min_v, max_v)
441        }
442        MsSQLTypeSystem::Intn(_) => {
443            let min_v: IntN = row.get(0).unwrap_or(IntN(0));
444            let max_v: IntN = row.get(1).unwrap_or(IntN(0));
445            (min_v.0, max_v.0)
446        }
447        MsSQLTypeSystem::Float24(_) => {
448            let min_v: f32 = row.get(0).unwrap_or(0.0);
449            let max_v: f32 = row.get(1).unwrap_or(0.0);
450            (min_v as i64, max_v as i64)
451        }
452        MsSQLTypeSystem::Float53(_) => {
453            let min_v: f64 = row.get(0).unwrap_or(0.0);
454            let max_v: f64 = row.get(1).unwrap_or(0.0);
455            (min_v as i64, max_v as i64)
456        }
457        MsSQLTypeSystem::Floatn(_) => {
458            let min_v: FloatN = row.get(0).unwrap_or(FloatN(0.0));
459            let max_v: FloatN = row.get(1).unwrap_or(FloatN(0.0));
460            (min_v.0 as i64, max_v.0 as i64)
461        }
462        _ => throw!(anyhow!(
463            "Partition can only be done on int or float columns"
464        )),
465    };
466
467    (min_v, max_v)
468}
469
470#[cfg(feature = "src_oracle")]
471#[throws(ConnectorXOutError)]
472fn oracle_get_partition_range(conn: &Url, query: &str, col: &str) -> (i64, i64) {
473    let source = OracleSource::new(conn.as_str(), 1)?;
474    let conn = source.get_conn()?;
475    let range_query = get_partition_range_query(query, col, &OracleDialect {})?;
476    let row = conn.query_row(range_query.as_str(), &[])?;
477    let min_v: i64 = row.get(0).unwrap_or(0);
478    let max_v: i64 = row.get(1).unwrap_or(0);
479    (min_v, max_v)
480}
481
482#[cfg(feature = "src_bigquery")]
483#[throws(ConnectorXOutError)] // TODO
484fn bigquery_get_partition_range(conn: &Url, query: &str, col: &str) -> (i64, i64) {
485    let rt = Runtime::new().expect("Failed to create runtime");
486    let url = Url::parse(conn.as_str())?;
487    let sa_key_path = url.path();
488    let client = rt.block_on(gcp_bigquery_client::Client::from_service_account_key_file(
489        sa_key_path,
490    ))?;
491
492    let auth_data = std::fs::read_to_string(sa_key_path)?;
493    let auth_json: serde_json::Value = serde_json::from_str(&auth_data)?;
494    let project_id = auth_json
495        .get("project_id")
496        .ok_or_else(|| anyhow!("Cannot get project_id from auth file"))?
497        .as_str()
498        .ok_or_else(|| anyhow!("Cannot get project_id as string from auth file"))?;
499    let range_query = get_partition_range_query(query, col, &BigQueryDialect {})?;
500
501    let query_result = rt.block_on(client.job().query(
502        project_id,
503        gcp_bigquery_client::model::query_request::QueryRequest::new(range_query.as_str()),
504    ))?;
505    let mut rs = gcp_bigquery_client::model::query_response::ResultSet::new_from_query_response(
506        query_result,
507    );
508    rs.next_row();
509    let min_v = rs.get_i64(0)?.unwrap_or(0);
510    let max_v = rs.get_i64(1)?.unwrap_or(0);
511
512    (min_v, max_v)
513}
514
515#[cfg(feature = "src_trino")]
516#[throws(ConnectorXOutError)]
517fn trino_get_partition_range(conn: &Url, query: &str, col: &str) -> (i64, i64) {
518    use crate::sources::trino::{build_client_from_url, TrinoDialect, TrinoPartitionQueryResult};
519
520    let rt = Runtime::new().expect("Failed to create runtime");
521
522    let client =
523        build_client_from_url(conn).map_err(|e| anyhow!("Failed to build Trino client: {}", e))?;
524
525    let range_query = get_partition_range_query(query, col, &TrinoDialect {})?;
526    let query_result = rt.block_on(client.get_all::<TrinoPartitionQueryResult>(range_query));
527
528    let query_result = match query_result {
529        Ok(query_result) => Ok(query_result.into_vec()),
530        Err(e) => match e {
531            prusto::error::Error::EmptyData => {
532                Ok(vec![TrinoPartitionQueryResult { _col0: 0, _col1: 0 }])
533            }
534            _ => Err(anyhow!("Failed to get query result: {}", e)),
535        },
536    }?;
537
538    let result = query_result
539        .first()
540        .unwrap_or(&TrinoPartitionQueryResult { _col0: 0, _col1: 0 });
541
542    (result._col0, result._col1)
543}
544
545#[cfg(feature = "src_clickhouse")]
546#[throws(ConnectorXOutError)]
547fn clickhouse_get_partition_range(conn: &Url, query: &str, col: &str) -> (i64, i64) {
548    use sqlparser::dialect::ClickHouseDialect;
549
550    let rt = Arc::new(tokio::runtime::Runtime::new().expect("Failed to create runtime"));
551    let clickhouse_source = ClickHouseSource::new(rt.clone(), conn.as_str())
552        .expect("Failed to create ClickHouse client");
553
554    let range_query = get_partition_range_query(query, col, &ClickHouseDialect {})?;
555
556    let response = rt.block_on(async {
557        let mut cursor = clickhouse_source
558            .client
559            .query(range_query.as_str())
560            .fetch_bytes("JSONCompact")
561            .map_err(|e| anyhow!("ClickHouse error: {}", e))?;
562        let bytes = cursor
563            .collect()
564            .await
565            .map_err(|e| anyhow!("ClickHouse error: {}", e))?;
566        Ok::<_, ClickHouseSourceError>(bytes)
567    })?;
568
569    #[derive(Debug, Deserialize)]
570    struct MinMaxResponse {
571        data: Vec<Vec<JsonValue>>,
572    }
573
574    let parsed: MinMaxResponse = serde_json::from_slice(&response)
575        .map_err(|e| anyhow!("Failed to parse min max response: {}", e))?;
576
577    let (min_v, max_v) = if let Some(row) = parsed.data.first() {
578        let min_v = row.get(0).and_then(|v| v.as_i64()).unwrap_or(0);
579        let max_v = row.get(1).and_then(|v| v.as_i64()).unwrap_or(0);
580
581        (min_v, max_v)
582    } else {
583        (0, 0)
584    };
585
586    (min_v, max_v)
587}
connectorx/partition.rs

connectorx/
partition.rs
