You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by cr...@apache.org on 2026/07/20 03:08:19 UTC
(iotdb-client-rust) 10/20: V4: insert_tablets batch (tree only, per-tablet redirect harvest) + pool lazy idle eviction (max_idle_time 60s, sweep on activity, min_size floor)
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 <cr...@outlook.com>
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");
+ }
}