This is an automated email from the ASF dual-hosted git repository. CritasWang pushed a commit to branch main in repository https://gitbox.apache.org/repos/asf/iotdb-client-rust.git
commit 2a830b0bc706ab426d74418b5d41cbc57774efd7 Author: CritasWang <[email protected]> AuthorDate: Mon Jul 13 12:36:29 2026 +0800 V4: insert_tablets batch (tree only, per-tablet redirect harvest) + pool lazy idle eviction (max_idle_time 60s, sweep on activity, min_size floor) --- README.md | 2 + README_ZH.md | 2 + src/client/pool.rs | 247 +++++++++++++++++++++++++++++++++++++--- src/client/session.rs | 307 +++++++++++++++++++++++++++++++++++++++++++++++++- 4 files changed, 539 insertions(+), 19 deletions(-) diff --git a/README.md b/README.md index e35aa09..746f805 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,8 @@ fn main() -> Result<()> { tablet.add_row(1_720_000_000_000, vec![Some(Value::Double(21.5))])?; tablet.add_row(1_720_000_001_000, vec![None])?; // null cell session.insert_tablet(&tablet)?; + // Multiple tablets in one RPC: insert_tablets(&[t1, t2], false) + // (tree model only; insert_aligned_tablets for aligned devices). // Or write a single row via insertRecord (row-oriented; aligned variants // and multi-row insert_records / insert_records_of_one_device also exist). diff --git a/README_ZH.md b/README_ZH.md index f1f9db3..d743d31 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -60,6 +60,8 @@ fn main() -> Result<()> { tablet.add_row(1_720_000_000_000, vec![Some(Value::Double(21.5))])?; tablet.add_row(1_720_000_001_000, vec![None])?; // null 单元格 session.insert_tablet(&tablet)?; + // 一次 RPC 批量写入多个 tablet：insert_tablets(&[t1, t2], false) + //（仅树模型；aligned 设备用 insert_aligned_tablets）。 // 也可以通过 insertRecord 写入单行（行式编码；另有 aligned 变体 // 以及多行的 insert_records / insert_records_of_one_device）。 diff --git a/src/client/pool.rs b/src/client/pool.rs index 887aff5..64b3564 100644 --- a/src/client/pool.rs +++ b/src/client/pool.rs @@ -45,6 +45,16 @@ pub struct SessionPoolConfig { /// How long [`SessionPool::acquire`] waits for an idle session once the /// pool is at `max_size`. Default 60 s (Node.js `waitTimeout`). pub acquire_timeout: Duration, + /// Idle sessions unused for longer than this are closed by the lazy + /// sweep (never shrinking the pool below `min_size`). Default 60 s + /// (Node.js `maxIdleTime`). + pub max_idle_time: Duration, + /// Minimum time between idle sweeps. There is **no** timer thread — + /// the sweep runs lazily on pool activity (acquire/release) once this + /// much time has passed since the previous sweep, so a completely idle + /// pool keeps its sessions until the next acquire. Default 30 s + /// (Node.js sweep interval). + pub idle_sweep_interval: Duration, } impl Default for SessionPoolConfig { @@ -54,6 +64,8 @@ impl Default for SessionPoolConfig { max_size: 8, min_size: 0, acquire_timeout: Duration::from_secs(60), + max_idle_time: Duration::from_secs(60), + idle_sweep_interval: Duration::from_secs(30), } } } @@ -66,11 +78,29 @@ impl SessionPoolConfig { } } +/// An idle session together with the moment it went idle, so the sweep +/// can measure how long it has been unused. +struct IdleEntry { + session: Session, + since: Instant, +} + +impl IdleEntry { + fn new(session: Session) -> Self { + Self { + session, + since: Instant::now(), + } + } +} + /// Idle sessions plus the pool lifecycle flag, guarded by one mutex so -/// [`Condvar`] waiters observe both consistently. +/// [`Condvar`] waiters observe both consistently. `last_sweep` gates the +/// lazy idle sweep (see [`SessionPoolConfig::idle_sweep_interval`]). struct PoolState { - idle: VecDeque<Session>, + idle: VecDeque<IdleEntry>, closed: bool, + last_sweep: Instant, } /// A pool of open tree-model [`Session`]s. @@ -78,7 +108,10 @@ struct PoolState { /// Sessions are created lazily up to `max_size` (after the eager /// `min_size`); [`SessionPool::acquire`] blocks up to `acquire_timeout` /// when the pool is exhausted. Dead sessions are discarded on acquire and -/// on release. The pool tracks the most recent `USE <db>` seen on any +/// on release; sessions idle longer than `max_idle_time` are closed by a +/// lazy sweep that runs on pool activity (no timer thread — see +/// [`SessionPoolConfig::idle_sweep_interval`]), never shrinking the pool +/// below `min_size`. The pool tracks the most recent `USE <db>` seen on any /// released session and replays it on acquire so every handed-out session /// is in the pool's current database (spec §6.2). pub struct SessionPool { @@ -107,6 +140,7 @@ impl SessionPool { state: Mutex::new(PoolState { idle: VecDeque::new(), closed: false, + last_sweep: Instant::now(), }), available: Condvar::new(), live: AtomicUsize::new(0), @@ -115,7 +149,7 @@ impl SessionPool { for _ in 0..pool.config.min_size { let session = pool.open_session()?; let mut state = pool.state.lock().expect("pool lock poisoned"); - state.idle.push_back(session); + state.idle.push_back(IdleEntry::new(session)); pool.live.fetch_add(1, Ordering::Relaxed); } Ok(pool) @@ -126,6 +160,45 @@ impl SessionPool { self.live.load(Ordering::Relaxed) } + /// Sessions sitting idle in the pool right now. + pub fn idle_count(&self) -> usize { + self.state.lock().expect("pool lock poisoned").idle.len() + } + + /// Lazy idle sweep, run at acquire/release time (there is no timer + /// thread — a completely inactive pool is only swept on its next use). + /// No-op until `idle_sweep_interval` has elapsed since the last sweep; + /// then removes sessions idle longer than `max_idle_time`, oldest + /// first, never shrinking the pool (idle + handed out) below + /// `min_size`. Returns the expired sessions — the caller closes them + /// **after** dropping the state lock, so a slow `closeSession` RPC + /// cannot stall other pool users. + #[must_use] + fn sweep_idle(&self, state: &mut PoolState) -> Vec<Session> { + let now = Instant::now(); + if now.duration_since(state.last_sweep) < self.config.idle_sweep_interval { + return Vec::new(); + } + state.last_sweep = now; + let mut expired = Vec::new(); + let mut i = 0; + while i < state.idle.len() { + if self.live.load(Ordering::Relaxed) - expired.len() <= self.config.min_size { + break; + } + if now.duration_since(state.idle[i].since) > self.config.max_idle_time { + expired.push(state.idle.remove(i).expect("index in bounds").session); + } else { + i += 1; + } + } + if !expired.is_empty() { + self.live.fetch_sub(expired.len(), Ordering::Relaxed); + log::debug!("idle sweep evicted {} session(s)", expired.len()); + } + expired + } + /// Acquire a session, blocking up to `acquire_timeout` when the pool is /// at capacity with nothing idle. Dead idle sessions are discarded and /// the acquire retried; a fresh session is opened while under @@ -137,11 +210,20 @@ impl SessionPool { if state.closed { return Err(Error::Client("session pool is closed".into())); } + let expired = self.sweep_idle(&mut state); + if !expired.is_empty() { + drop(state); + for mut session in expired { + let _ = session.close(); + } + state = self.state.lock().expect("pool lock poisoned"); + continue; // re-check closed/idle after re-locking + } // Idle session available → validate liveness, evict the dead. - while let Some(session) = state.idle.pop_front() { - if session.is_open() { + while let Some(entry) = state.idle.pop_front() { + if entry.session.is_open() { drop(state); - return self.hand_out(session); + return self.hand_out(entry.session); } self.live.fetch_sub(1, Ordering::Relaxed); } @@ -197,16 +279,15 @@ impl SessionPool { let hint = state .idle .iter_mut() - .find_map(|s| s.redirect_hint(device_id)); + .find_map(|e| e.session.redirect_hint(device_id)); if let Some(endpoint) = hint { - let matching = state - .idle - .iter() - .position(|s| s.is_open() && s.current_endpoint() == Some(&endpoint)); + let matching = state.idle.iter().position(|e| { + e.session.is_open() && e.session.current_endpoint() == Some(&endpoint) + }); if let Some(pos) = matching { - let session = state.idle.remove(pos).expect("index in bounds"); + let entry = state.idle.remove(pos).expect("index in bounds"); drop(state); - return self.hand_out(session); + return self.hand_out(entry.session); } } } @@ -230,8 +311,8 @@ impl SessionPool { std::mem::take(&mut state.idle) }; self.live.fetch_sub(drained.len(), Ordering::Relaxed); - for mut session in drained { - let _ = session.close(); + for mut entry in drained { + let _ = entry.session.close(); } self.available.notify_all(); } @@ -288,8 +369,12 @@ impl SessionPool { let mut session = session; let _ = session.close(); } else { - state.idle.push_back(session); + state.idle.push_back(IdleEntry::new(session)); + let expired = self.sweep_idle(&mut state); drop(state); + for mut session in expired { + let _ = session.close(); + } } self.available.notify_one(); } @@ -299,7 +384,7 @@ impl SessionPool { #[cfg(test)] fn inject_idle(&self, session: Session) { let mut state = self.state.lock().expect("pool lock poisoned"); - state.idle.push_back(session); + state.idle.push_back(IdleEntry::new(session)); self.live.fetch_add(1, Ordering::Relaxed); } } @@ -378,6 +463,10 @@ impl TableSessionPool { self.pool.live_count() } + pub fn idle_count(&self) -> usize { + self.pool.idle_count() + } + pub fn close(&self) { self.pool.close() } @@ -412,6 +501,8 @@ mod tests { assert_eq!(cfg.max_size, 8); assert_eq!(cfg.min_size, 0); assert_eq!(cfg.acquire_timeout, Duration::from_secs(60)); + assert_eq!(cfg.max_idle_time, Duration::from_secs(60)); + assert_eq!(cfg.idle_sweep_interval, Duration::from_secs(30)); } #[test] @@ -591,6 +682,126 @@ mod tests { assert_eq!(guard.current_endpoint(), Some(&ep_a)); } + /// Backdate an idle entry and the sweep clock so eviction tests are + /// deterministic without long sleeps. + fn backdate(pool: &SessionPool, entry_ages: Duration, sweep_age: Duration) { + let mut state = pool.state.lock().unwrap(); + state.last_sweep = Instant::now() - sweep_age; + for entry in &mut state.idle { + entry.since = Instant::now() - entry_ages; + } + } + + /// Sessions idle past `max_idle_time` are closed by the sweep on the + /// next acquire; the freed capacity is then used to (try to) grow. + #[test] + fn idle_sessions_are_evicted_on_acquire() { + let ep = fake_listener(); + let cfg = SessionPoolConfig { + max_size: 4, + acquire_timeout: Duration::from_millis(50), + max_idle_time: Duration::from_millis(5), + idle_sweep_interval: Duration::from_millis(1), + session: dead_endpoint_config(), + ..Default::default() + }; + let pool = SessionPool::new(cfg).unwrap(); + pool.inject_idle(injected_session(&ep)); + pool.inject_idle(injected_session(&ep)); + assert_eq!(pool.idle_count(), 2); + + // Exceed max_idle_time, then acquire: the sweep evicts both idle + // sessions and the acquire grows — which fails against the dead + // endpoint (a connect error, proving the idle queue really was + // emptied rather than handed out). + std::thread::sleep(Duration::from_millis(10)); + match pool.acquire() { + Err(Error::Thrift(_)) => {} + other => panic!("expected thrift connect error, got {other:?}"), + } + assert_eq!(pool.idle_count(), 0); + assert_eq!(pool.live_count(), 0); + } + + /// The sweep never shrinks the pool (idle + handed out) below + /// `min_size`, evicting oldest-first. + #[test] + fn idle_sweep_respects_min_size_floor() { + let ep = fake_listener(); + let cfg = SessionPoolConfig { + max_size: 4, + max_idle_time: Duration::from_millis(5), + idle_sweep_interval: Duration::from_millis(1), + session: dead_endpoint_config(), + ..Default::default() + }; + let mut pool = SessionPool::new(cfg).unwrap(); + pool.config.min_size = 2; // set post-new: eager open would need a server + for _ in 0..3 { + pool.inject_idle(injected_session(&ep)); + } + backdate(&pool, Duration::from_millis(10), Duration::from_millis(10)); + + let mut state = pool.state.lock().unwrap(); + let expired = pool.sweep_idle(&mut state); + // All 3 exceed max_idle_time, but only 1 may go: 3 live - 1 = floor. + assert_eq!(expired.len(), 1); + assert_eq!(state.idle.len(), 2); + drop(state); + assert_eq!(pool.live_count(), 2); + } + + /// The sweep is a no-op until `idle_sweep_interval` has passed since the + /// previous sweep — even when idle sessions have already expired. + #[test] + fn idle_sweep_is_gated_by_interval() { + let ep = fake_listener(); + let cfg = SessionPoolConfig { + max_size: 4, + max_idle_time: Duration::from_millis(1), + idle_sweep_interval: Duration::from_secs(3600), + session: dead_endpoint_config(), + ..Default::default() + }; + let pool = SessionPool::new(cfg).unwrap(); + pool.inject_idle(injected_session(&ep)); + // Entry long expired, but the last sweep (pool creation) is recent + // relative to the huge interval → gated. + backdate(&pool, Duration::from_secs(10), Duration::ZERO); + let mut state = pool.state.lock().unwrap(); + assert!(pool.sweep_idle(&mut state).is_empty()); + assert_eq!(state.idle.len(), 1); + + // Once the interval has elapsed the same entry is evicted… + state.last_sweep = Instant::now() - Duration::from_secs(3601); + assert_eq!(pool.sweep_idle(&mut state).len(), 1); + // …and last_sweep was refreshed, so an immediate re-sweep is gated. + assert!(pool.sweep_idle(&mut state).is_empty()); + } + + /// Releasing a session also drives the sweep: the stale idle session is + /// evicted while the just-released (fresh) one stays. + #[test] + fn release_triggers_sweep_but_keeps_fresh_session() { + let ep = fake_listener(); + let cfg = SessionPoolConfig { + max_size: 4, + max_idle_time: Duration::from_millis(5), + idle_sweep_interval: Duration::from_millis(1), + session: dead_endpoint_config(), + ..Default::default() + }; + let pool = SessionPool::new(cfg).unwrap(); + pool.inject_idle(injected_session(&ep)); + backdate(&pool, Duration::from_millis(10), Duration::from_millis(10)); + + // Simulate returning a handed-out session (live already counted). + pool.live.fetch_add(1, Ordering::Relaxed); + pool.release(injected_session(&ep)); + assert_eq!(pool.idle_count(), 1, "stale evicted, fresh kept"); + assert_eq!(pool.live_count(), 1); + } + /// Live-server tests: acquire/release round-trip, reuse, and blocking /// hand-off. Skipped when no IoTDB server is reachable. #[test] diff --git a/src/client/session.rs b/src/client/session.rs index 96199df..f98a308 100644 --- a/src/client/session.rs +++ b/src/client/session.rs @@ -30,7 +30,7 @@ use crate::error::{Error, Result}; use crate::protocol::client::{ TIClientRPCServiceSyncClient, TSCloseOperationReq, TSCloseSessionReq, TSExecuteStatementReq, TSFetchResultsReq, TSInsertRecordReq, TSInsertRecordsOfOneDeviceReq, TSInsertRecordsReq, - TSInsertTabletReq, TSOpenSessionReq, TSProtocolVersion, + TSInsertTabletReq, TSInsertTabletsReq, TSOpenSessionReq, TSProtocolVersion, }; use crate::protocol::common::TSStatus; @@ -515,6 +515,34 @@ impl Session { self.check_insert_status(&[prefix_path], &status) } + /// Insert a batch of [`Tablet`]s in one `insertTablets` RPC (spec §3.6). + /// Each tablet is serialized exactly like [`Session::insert_tablet`] — + /// column-major values with trailing null bitmaps, i64-BE timestamps, + /// rows sorted by timestamp. `is_aligned` applies to the **whole batch** + /// (the RPC carries a single flag); the per-tablet aligned flag is + /// ignored, matching the Java client's `insertTablets` / + /// `insertAlignedTablets` split. + /// + /// Tree model only: the batch request has no `writeToTable` / + /// `columnCategories` fields, so table-model tablets are rejected — + /// send those one at a time via [`Session::insert_tablet`]. + pub fn insert_tablets(&mut self, tablets: &[Tablet], is_aligned: bool) -> Result<()> { + let req = build_insert_tablets_req(self.session_id, tablets, is_aligned)?; + let status = self.with_retry(|session| { + // Clone per attempt: a reconnect refreshes the session id. + let mut req = req.clone(); + req.session_id = session.session_id; + Ok(session.connection_mut()?.client_mut().insert_tablets(req)?) + })?; + let devices: Vec<&str> = req.prefix_paths.iter().map(String::as_str).collect(); + self.check_insert_status(&devices, &status) + } + + /// [`Session::insert_tablets`] against aligned devices. + pub fn insert_aligned_tablets(&mut self, tablets: &[Tablet]) -> Result<()> { + self.insert_tablets(tablets, true) + } + /// Insert one row for one device via `insertRecord`. `values[i]` pairs /// with `measurements[i]`. The value buffer is row-oriented (per-cell /// type marker + big-endian payload), unlike the tablet's column-major @@ -774,6 +802,58 @@ impl Session { } } +/// Assembles a `TSInsertTabletsReq` (spec §3.6): one entry per tablet in +/// each parallel list, every buffer serialized exactly like the +/// single-tablet path. Kept as a free function so request assembly is +/// testable without a connection. Rejects empty batches and table-model +/// tablets (the batch RPC has no table-model fields). +fn build_insert_tablets_req( + session_id: i64, + tablets: &[Tablet], + is_aligned: bool, +) -> Result<TSInsertTabletsReq> { + if tablets.is_empty() { + return Err(Error::Client( + "insert_tablets called with no tablets".into(), + )); + } + let n = tablets.len(); + let mut prefix_paths = Vec::with_capacity(n); + let mut measurements_list = Vec::with_capacity(n); + let mut values_list = Vec::with_capacity(n); + let mut timestamps_list = Vec::with_capacity(n); + let mut types_list = Vec::with_capacity(n); + let mut size_list = Vec::with_capacity(n); + for tablet in tablets { + if tablet.is_table_model() { + return Err(Error::Client(format!( + "insert_tablets is tree-model only; tablet for table {:?} \ + must go through insert_tablet", + tablet.target() + ))); + } + // Serialization sorts in place; clone so the caller's tablet order + // is untouched (the clone is cheap relative to the RPC). + let mut tablet = tablet.clone(); + values_list.push(tablet.serialize_values()); + timestamps_list.push(tablet.serialize_timestamps()); + prefix_paths.push(tablet.target().to_string()); + measurements_list.push(tablet.measurements().to_vec()); + types_list.push(tablet.types().iter().map(|t| t.code()).collect()); + size_list.push(tablet.row_count() as i32); + } + Ok(TSInsertTabletsReq::new( + session_id, + prefix_paths, + measurements_list, + values_list, + timestamps_list, + types_list, + size_list, + is_aligned, + )) +} + /// Stably sorts one-device record rows by timestamp — reordering the /// measurement lists in step and serializing each row's value buffer in the /// sorted order (Java `genTSInsertRecordsOfOneDeviceReq`; the server @@ -1041,6 +1121,95 @@ mod tests { assert_eq!(ms, [vec!["first".to_string()], vec!["second".into()]]); } + /// insert_tablets request assembly: parallel lists pair 1:1 with the + /// input tablets, and every buffer matches what the single-tablet path + /// (`serialize_values`/`serialize_timestamps`) produces for the same + /// tablet — including the timestamp sort and null bitmaps. + #[test] + fn insert_tablets_request_assembly() { + use crate::data::{tablet::Tablet, TSDataType, Value}; + + let mut t1 = Tablet::new( + "root.sg.d1", + vec!["i".into(), "s".into()], + vec![TSDataType::Int32, TSDataType::Text], + ) + .unwrap(); + // Unsorted rows + a null: serialization must sort and bitmap. + t1.add_row(2, vec![Some(Value::Int32(20)), None]).unwrap(); + t1.add_row(1, vec![None, Some(Value::Text("a".into()))]) + .unwrap(); + let mut t2 = Tablet::new("root.sg.d2", vec!["b".into()], vec![TSDataType::Boolean]) // + .unwrap(); + t2.add_row(5, vec![Some(Value::Boolean(true))]).unwrap(); + + let req = build_insert_tablets_req(7, &[t1.clone(), t2.clone()], false).unwrap(); + assert_eq!(req.session_id, 7); + assert_eq!(req.prefix_paths, ["root.sg.d1", "root.sg.d2"]); + assert_eq!( + req.measurements_list, + [vec!["i".to_string(), "s".into()], vec!["b".into()]] + ); + assert_eq!(req.types_list, [vec![1, 5], vec![0]]); // Int32+Text, Boolean + assert_eq!(req.size_list, [2, 1]); + assert_eq!(req.is_aligned, Some(false)); + // Byte-identical to the single-tablet serialization helpers. + assert_eq!( + req.values_list, + [t1.serialize_values(), t2.serialize_values()] + ); + assert_eq!( + req.timestamps_list, + [t1.serialize_timestamps(), t2.serialize_timestamps()] + ); + // Known bytes for t1 after sorting (ts 1 first): int col null@row0, + // text col null@row1. + assert_eq!( + req.values_list[0], + [ + 0x80, 0x00, 0x00, 0x00, // i row0: null placeholder (i32::MIN) + 0x00, 0x00, 0x00, 0x14, // i row1: 20 + 0x00, 0x00, 0x00, 0x01, b'a', // s row0: "a" + 0x00, 0x00, 0x00, 0x00, // s row1: null placeholder (empty) + 0x01, 0x01, // i bitmap: flag + row 0 null (LSB-first) + 0x01, 0x02, // s bitmap: flag + row 1 null + ] + ); + assert_eq!( + req.timestamps_list[0], + [0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 2] + ); + + // The batch flag is a single RPC-level field. + let req = build_insert_tablets_req(7, &[t2], true).unwrap(); + assert_eq!(req.is_aligned, Some(true)); + } + + #[test] + fn insert_tablets_rejects_empty_and_table_model() { + use crate::data::{tablet::Tablet, ColumnCategory, TSDataType}; + + let err = build_insert_tablets_req(1, &[], false).unwrap_err(); + assert!(matches!(err, Error::Client(m) if m.contains("no tablets"))); + + let tree = Tablet::new("root.sg.d1", vec!["s".into()], vec![TSDataType::Int32]).unwrap(); + let table = Tablet::new_table( + "sensors", + vec!["f1".into()], + vec![TSDataType::Double], + vec![ColumnCategory::Field], + ) + .unwrap(); + // A table-model tablet anywhere in the batch fails the whole call. + let err = build_insert_tablets_req(1, &[tree, table], false).unwrap_err(); + assert!(matches!(err, Error::Client(m) if m.contains("tree-model only"))); + + // The errors fire before any connection use. + let mut session = Session::new(SessionConfig::default()); + let err = session.insert_tablets(&[], false).unwrap_err(); + assert!(matches!(err, Error::Client(m) if m.contains("no tablets"))); + } + #[test] fn insert_records_length_mismatch_is_client_error() { use crate::data::Value; @@ -1659,4 +1828,140 @@ mod tests { .expect("cleanup"); session.close().expect("close session"); } + + /// Live roundtrip for `insertTablets`: one batch of three tablets across + /// two devices (with nulls and unsorted rows), plus an aligned batch on + /// an aligned device. Every written cell is read back and asserted. + /// Skipped when no IoTDB instance is reachable on localhost:6667. + #[test] + fn live_insert_tablets_readback() { + use crate::data::{tablet::Tablet, TSDataType, Value}; + use std::net::TcpStream; + if TcpStream::connect_timeout( + &"127.0.0.1:6667".parse().unwrap(), + Duration::from_millis(300), + ) + .is_err() + { + eprintln!("skipping live_insert_tablets_readback: no IoTDB server on 127.0.0.1:6667"); + return; + } + + const DB: &str = "root.rusttest_tablets"; + let _guard = LIVE_DB_LOCK.lock().unwrap_or_else(|p| p.into_inner()); + + let read_rows = |session: &mut Session, sql: &str| -> Vec<(i64, Vec<Value>)> { + let mut rows = Vec::new(); + let mut dataset = session.execute_query(sql).expect("query"); + while let Some(row) = dataset.next_row().expect("next_row") { + rows.push((row.timestamp.expect("timestamp"), row.values.clone())); + } + rows + }; + + let mut session = Session::new(SessionConfig::default()); + session.open().expect("open session"); + let _ = session.execute_non_query(&format!("DELETE DATABASE {DB}")); + session + .execute_non_query(&format!("CREATE DATABASE {DB}")) + .expect("create database"); + + // Three tablets, two devices, one aligned=false batch. t1 and t3 + // both target d1 over disjoint time ranges; t1 has a null and + // unsorted rows (the client must sort per tablet). + let mut t1 = Tablet::new( + format!("{DB}.d1"), + vec!["i".into(), "s".into()], + vec![TSDataType::Int32, TSDataType::Text], + ) + .expect("tablet t1"); + t1.add_row(2, vec![Some(Value::Int32(20)), None]) + .expect("add_row"); + t1.add_row(1, vec![None, Some(Value::Text("one".into()))]) + .expect("add_row"); + let mut t2 = Tablet::new( + format!("{DB}.d2"), + vec!["d".into()], + vec![TSDataType::Double], + ) + .expect("tablet t2"); + t2.add_row(10, vec![Some(Value::Double(1.5))]) + .expect("add_row"); + t2.add_row(11, vec![Some(Value::Double(-2.5))]) + .expect("add_row"); + let mut t3 = Tablet::new( + format!("{DB}.d1"), + vec!["i".into(), "s".into()], + vec![TSDataType::Int32, TSDataType::Text], + ) + .expect("tablet t3"); + t3.add_row( + 3, + vec![Some(Value::Int32(30)), Some(Value::Text("three".into()))], + ) + .expect("add_row"); + session + .insert_tablets(&[t1, t2, t3], false) + .expect("insert_tablets"); + + let rows = read_rows( + &mut session, + &format!("SELECT i, s FROM {DB}.d1 ORDER BY time"), + ); + assert_eq!( + rows, + [ + (1, vec![Value::Null, Value::Text("one".into())]), + (2, vec![Value::Int32(20), Value::Null]), + (3, vec![Value::Int32(30), Value::Text("three".into())]), + ] + ); + let rows = read_rows( + &mut session, + &format!("SELECT d FROM {DB}.d2 ORDER BY time"), + ); + assert_eq!( + rows, + [ + (10, vec![Value::Double(1.5)]), + (11, vec![Value::Double(-2.5)]), + ] + ); + + // Aligned batch on an aligned device. + session + .execute_non_query(&format!( + "CREATE ALIGNED TIMESERIES {DB}.a1(s1 INT32, s2 DOUBLE)" + )) + .expect("create aligned timeseries"); + let mut a1 = Tablet::new_aligned( + format!("{DB}.a1"), + vec!["s1".into(), "s2".into()], + vec![TSDataType::Int32, TSDataType::Double], + ) + .expect("aligned tablet"); + a1.add_row(40, vec![Some(Value::Int32(400)), None]) + .expect("add_row"); + a1.add_row(41, vec![Some(Value::Int32(410)), Some(Value::Double(4.1))]) + .expect("add_row"); + session + .insert_aligned_tablets(&[a1]) + .expect("insert_aligned_tablets"); + let rows = read_rows( + &mut session, + &format!("SELECT s1, s2 FROM {DB}.a1 ORDER BY time"), + ); + assert_eq!( + rows, + [ + (40, vec![Value::Int32(400), Value::Null]), + (41, vec![Value::Int32(410), Value::Double(4.1)]), + ] + ); + + session + .execute_non_query(&format!("DELETE DATABASE {DB}")) + .expect("cleanup"); + session.close().expect("close session"); + } }
