This is an automated email from the ASF dual-hosted git repository. CritasWang pushed a commit to branch feat/tablet-object-write in repository https://gitbox.apache.org/repos/asf/iotdb-client-rust.git
commit 1e1345d7cad542193214d29ab2b0e2f1908477e4 Author: CritasWang <[email protected]> AuthorDate: Wed Aug 26 10:17:50 2026 +0800 Support table-model OBJECT writes and OBJECT result decoding Mirrors apache/iotdb-client-go#175, apache/iotdb-client-nodejs#25 and apache/iotdb-client-csharp#64. - TSDataType::Object = 12 (official TSFile code) and from_code. - Value::Object(Vec<u8>) carrying the already-framed OBJECT segment (1 byte isEOF + 8 byte big-endian offset + content), plus object_bytes_to_string for the "(Object) 1.00 KB" display format. - Tablet::build_object_value / set_object_value_at with column/row/offset validation; set_object_value_at overwrites null cells, which clears the derived null bitmap bit. serialize_values writes OBJECT columns with the same 4-byte length-prefixed binary layout as BLOB. - TsBlock BinaryArray decoding accepts OBJECT; SessionDataSet formats select file OBJECT metadata as Value::String summary while select READ_OBJECT(file) keeps returning raw Value::Blob. - Golden-byte unit tests for whole/segmented/null writes, bitmap clearing, invalid arguments, TsBlock decode and dataset display semantics; a live-server OBJECT round-trip that probes support and skips like Go's e2e. - Document OBJECT in the READMEs and demo it in the table_session example. No Thrift IDL or generated protocol changes. --- README.md | 26 ++++++- README_ZH.md | 25 ++++++- examples/table_session.rs | 32 ++++++++ src/client/dataset.rs | 78 ++++++++++++++++---- src/client/session.rs | 62 ++++++++++++++++ src/client/table_session.rs | 117 +++++++++++++++++++++++++++++ src/data/mod.rs | 14 +++- src/data/record.rs | 8 +- src/data/tablet.rs | 174 +++++++++++++++++++++++++++++++++++++++++++- src/data/tsblock.rs | 32 ++++++++ src/data/value.rs | 73 +++++++++++++++++++ src/lib.rs | 2 +- 12 files changed, 617 insertions(+), 26 deletions(-) diff --git a/README.md b/README.md index 93e607b..f4a45fd 100644 --- a/README.md +++ b/README.md @@ -172,6 +172,30 @@ cargo run --example table_session cargo run --example session_pool ``` +## OBJECT columns (table model) + +Table-model OBJECT columns (IoTDB 2.0.8+) are written with +`Tablet::set_object_value_at`. Every call wraps one segment in a 9-byte header — +`[1 byte isEOF][8 byte big-endian offset]` — followed by the segment content, so a +large object can be streamed without holding it in memory (ascending offsets, +`is_eof = true` on the last segment; a whole object is one segment at offset 0). + +```rust +let mut tablet = Tablet::new_table( + "objects", + vec!["region".into(), "file".into()], + vec![TSDataType::String, TSDataType::Object], + vec![ColumnCategory::Tag, ColumnCategory::Field], +)?; +tablet.add_row(1_720_000_000_000, vec![Some(Value::String("east".into())), None])?; +tablet.set_object_value_at(true, 0, &object_bytes, 1, 0)?; +session.insert(&tablet)?; +``` + +On the read side `SELECT file` returns the server's OBJECT metadata rendered as +`Value::String("(Object) 1.00 KB")` (see `object_bytes_to_string`), while +`SELECT READ_OBJECT(file)` keeps returning the raw bytes as `Value::Blob`. + ## TLS & RPC compression **RPC compression** (IoTDB's term for the Thrift *compact protocol*) is a plain config flag: @@ -294,7 +318,7 @@ Throughput scales with points per RPC: wider tablets (100 sensors = 100k points | --- | --- | | `src/client/` | `Session`, `TableSession`, `SessionPool`, `TableSessionPool`, `SessionDataSet` | | `src/connection/` | Low-level Thrift transport (framed transport + binary protocol) | -| `src/data/` | `Tablet`, `Value`, `TSDataType` (official TSFile codes 0–11), TsBlock decoding, bitmaps | +| `src/data/` | `Tablet`, `Value`, `TSDataType` (official TSFile codes 0–12), TsBlock decoding, bitmaps | | `src/protocol/` | Generated Thrift stubs (do not edit) | | `thrift/` | Thrift IDL sources, synced from the IoTDB repo | | `examples/` | Runnable examples for both models and the pools | diff --git a/README_ZH.md b/README_ZH.md index 3da4c98..97fd9fa 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -172,6 +172,29 @@ cargo run --example table_session cargo run --example session_pool ``` +## OBJECT 列（表模型） + +表模型 OBJECT 列（IoTDB 2.0.8+）通过 `Tablet::set_object_value_at` 写入。每次调用 +将一段内容包上 9 字节头——`[1 字节 isEOF][8 字节大端 offset]`——再接上段内容，因此大对象 +可以分段写入而无需整体驻留内存（offset 递增，最后一段 `is_eof = true`；整对象即 offset 0 的 +单段写入）。 + +```rust +let mut tablet = Tablet::new_table( + "objects", + vec!["region".into(), "file".into()], + vec![TSDataType::String, TSDataType::Object], + vec![ColumnCategory::Tag, ColumnCategory::Field], +)?; +tablet.add_row(1_720_000_000_000, vec![Some(Value::String("east".into())), None])?; +tablet.set_object_value_at(true, 0, &object_bytes, 1, 0)?; +session.insert(&tablet)?; +``` + +读取侧 `SELECT file` 会把服务端返回的 OBJECT 元数据渲染为 +`Value::String("(Object) 1.00 KB")`（见 `object_bytes_to_string`）； +`SELECT READ_OBJECT(file)` 仍以 `Value::Blob` 返回原始字节。 + ## TLS 与 RPC 压缩 **RPC 压缩**（IoTDB 术语，实为 Thrift *compact 协议*）只是一个配置开关： @@ -293,7 +316,7 @@ cargo run --release --example benchmark -- --mode table \ | --- | --- | | `src/client/` | `Session`、`TableSession`、`SessionPool`、`TableSessionPool`、`SessionDataSet` | | `src/connection/` | 底层 Thrift 传输（帧传输 + 二进制协议） | -| `src/data/` | `Tablet`、`Value`、`TSDataType`（官方 TSFile 编码 0–11）、TsBlock 解码、位图 | +| `src/data/` | `Tablet`、`Value`、`TSDataType`（官方 TSFile 编码 0–12）、TsBlock 解码、位图 | | `src/protocol/` | 生成的 Thrift 桩代码（勿编辑） | | `thrift/` | Thrift IDL 源文件，从 IoTDB 仓库同步 | | `examples/` | 两种模型及会话池的可运行示例 | diff --git a/examples/table_session.rs b/examples/table_session.rs index 01ef860..4329c3d 100644 --- a/examples/table_session.rs +++ b/examples/table_session.rs @@ -105,6 +105,38 @@ fn main() -> Result<()> { } } // dataset drop closes the query and releases the session borrow + // --- OBJECT column demo (needs IoTDB 2.0.8+; skipped on older servers) - + // OBJECT writes use Tablet::set_object_value_at with a 9-byte segment + // header (isEOF + big-endian offset). SELECT file renders the size + // summary; SELECT READ_OBJECT(file) returns the raw BLOB. + if let Err(e) = session.execute_non_query( + "CREATE TABLE IF NOT EXISTS objects (\ + region STRING TAG, \ + file OBJECT FIELD)", + ) { + eprintln!("skipping OBJECT demo (server does not support OBJECT): {e}"); + } else { + let object_bytes: Vec<u8> = (0..1024u32).map(|i| (i % 251) as u8).collect(); + let mut tablet = Tablet::new_table( + "objects", + vec!["region".into(), "file".into()], + vec![TSDataType::String, TSDataType::Object], + vec![ColumnCategory::Tag, ColumnCategory::Field], + )?; + tablet.add_row(base_ts, vec![Some(Value::String("east".into())), None])?; + tablet.set_object_value_at(true, 0, &object_bytes, 1, 0)?; + session.insert(&tablet)?; + println!("inserted an OBJECT row into `objects`"); + + { + let mut dataset = session.execute_query("SELECT file FROM objects")?; + while let Some(row) = dataset.next_row()? { + println!("{:?}", row.values); // e.g. [String("(Object) 1.00 KB")] + } + } + session.execute_non_query("DROP TABLE objects")?; + } + // --- Cleanup ---------------------------------------------------------- session.execute_non_query(&format!("DROP DATABASE {DB}"))?; println!("database dropped"); diff --git a/src/client/dataset.rs b/src/client/dataset.rs index 6c18ff5..fda9d9a 100644 --- a/src/client/dataset.rs +++ b/src/client/dataset.rs @@ -23,7 +23,7 @@ use std::collections::VecDeque; use crate::client::session::{QueryHandle, Session}; use crate::data::tsblock::TsBlock; -use crate::data::value::Value; +use crate::data::value::{object_bytes_to_string, Value}; use crate::error::{Error, Result}; /// One result row: the timestamp (`None` when the server set @@ -176,7 +176,7 @@ impl<'a> SessionDataSet<'a> { values.push(Self::apply_logical_type( column[i].clone(), self.data_type_list.get(ordinal).map(String::as_str), - )); + )?); } let timestamp = (!self.ignore_time_stamp).then(|| block.timestamps[i]); Ok(Row { timestamp, values }) @@ -185,14 +185,20 @@ impl<'a> SessionDataSet<'a> { /// Re-tag a decoded value with the column's logical type from the /// response's `dataTypeList`. TsBlock headers carry the *physical* type /// (DATE arrives as INT32, TIMESTAMP as INT64, STRING as TEXT), so the - /// block decoder alone cannot distinguish them. - fn apply_logical_type(value: Value, logical: Option<&str>) -> Value { + /// block decoder alone cannot distinguish them. OBJECT metadata (8-byte + /// BE size + internal path) is formatted here into the + /// `(Object) 1.00 KB` display string; `READ_OBJECT(file)` results + /// keep logical type BLOB and stay raw `Value::Blob`. + fn apply_logical_type(value: Value, logical: Option<&str>) -> Result<Value> { match (logical, value) { - (Some("DATE"), Value::Int32(v)) => Value::Date(v), - (Some("TIMESTAMP"), Value::Int64(v)) => Value::Timestamp(v), - (Some("STRING"), Value::Text(s)) => Value::String(s), - (Some("BLOB"), Value::Text(s)) => Value::Blob(s.into_bytes()), - (_, v) => v, + (Some("DATE"), Value::Int32(v)) => Ok(Value::Date(v)), + (Some("TIMESTAMP"), Value::Int64(v)) => Ok(Value::Timestamp(v)), + (Some("STRING"), Value::Text(s)) => Ok(Value::String(s)), + (Some("BLOB"), Value::Text(s)) => Ok(Value::Blob(s.into_bytes())), + (Some("OBJECT"), Value::Object(bytes)) => { + Ok(Value::String(object_bytes_to_string(&bytes)?)) + } + (_, v) => Ok(v), } } @@ -260,28 +266,68 @@ mod tests { use Value::*; // Physical → logical re-tags. assert_eq!( - SessionDataSet::apply_logical_type(Int32(20260713), Some("DATE")), + SessionDataSet::apply_logical_type(Int32(20260713), Some("DATE")).unwrap(), Date(20260713) ); assert_eq!( - SessionDataSet::apply_logical_type(Int64(99), Some("TIMESTAMP")), + SessionDataSet::apply_logical_type(Int64(99), Some("TIMESTAMP")).unwrap(), Timestamp(99) ); assert_eq!( - SessionDataSet::apply_logical_type(Text("s".into()), Some("STRING")), + SessionDataSet::apply_logical_type(Text("s".into()), Some("STRING")).unwrap(), String("s".into()) ); assert_eq!( - SessionDataSet::apply_logical_type(Text("b".into()), Some("BLOB")), + SessionDataSet::apply_logical_type(Text("b".into()), Some("BLOB")).unwrap(), Blob(b"b".to_vec()) ); // Pass-throughs: matching physical types and nulls stay untouched. assert_eq!( - SessionDataSet::apply_logical_type(Int32(5), Some("INT32")), + SessionDataSet::apply_logical_type(Int32(5), Some("INT32")).unwrap(), + Int32(5) + ); + assert_eq!( + SessionDataSet::apply_logical_type(Null, Some("DATE")).unwrap(), + Null + ); + assert_eq!( + SessionDataSet::apply_logical_type(Int32(5), None).unwrap(), Int32(5) ); - assert_eq!(SessionDataSet::apply_logical_type(Null, Some("DATE")), Null); - assert_eq!(SessionDataSet::apply_logical_type(Int32(5), None), Int32(5)); + } + + /// One-object-column TsBlock with the server's 8-byte BE size + path + /// payload for the `select file` shape. + fn object_block(ts: i64, payload: &[u8]) -> Vec<u8> { + let mut b = header(&[TSDataType::Object], 1, &[ENCODING_BINARY_ARRAY]); + b.extend_from_slice(&time_column(&[ts])); + b.push(0); // mayHaveNull + b.extend_from_slice(&(payload.len() as i32).to_be_bytes()); + b.extend_from_slice(payload); + b + } + + #[test] + fn object_column_renders_size_summary() { + let mut payload = 1024u64.to_be_bytes().to_vec(); + payload.extend_from_slice(b"internal/path/1.bin"); + + let mut session = offline_session(); + let mut h = handle(vec![object_block(1, &payload)], false); + h.columns = vec!["file".into()]; + h.data_type_list = vec!["OBJECT".into()]; + let mut ds = SessionDataSet::new(&mut session, h); + let row = ds.next_row().unwrap().unwrap(); + assert_eq!(row.values, vec![Value::String("(Object) 1.00 KB".into())]); + assert!(ds.next_row().unwrap().is_none()); + } + + #[test] + fn short_object_metadata_is_decode_error() { + assert!(matches!( + SessionDataSet::apply_logical_type(Value::Object(vec![0; 7]), Some("OBJECT")), + Err(Error::Decode(_)) + )); } #[test] diff --git a/src/client/session.rs b/src/client/session.rs index 91480c4..bee04c6 100644 --- a/src/client/session.rs +++ b/src/client/session.rs @@ -1325,6 +1325,68 @@ mod tests { assert_eq!(req.is_aligned, Some(true)); } + /// insertTablet request assembly for OBJECT columns: the types list + /// carries code 12, the values buffer uses the length-prefixed framed + /// segment, and the table-model fields (writeToTable + column + /// categories) are set exactly as `Session::insert_tablet` sends them. + #[test] + fn insert_tablet_request_carries_object_type_12() { + use crate::data::{ColumnCategory, TSDataType, Tablet, Value}; + + let mut tablet = Tablet::new_table( + "object_table", + vec!["region_id".into(), "file".into()], + vec![TSDataType::String, TSDataType::Object], + vec![ColumnCategory::Tag, ColumnCategory::Field], + ) + .unwrap(); + tablet + .add_row( + 1_608_268_702_780, + vec![Some(Value::String("r1".into())), None], + ) + .unwrap(); + tablet + .set_object_value_at(true, 0, &[0x01, 0x02, 0x03], 1, 0) + .unwrap(); + + let req = TSInsertTabletReq::new( + 1, + tablet.table_name().to_string(), + tablet.measurements().to_vec(), + tablet.serialize_values(), + tablet.serialize_timestamps(), + tablet.types().iter().map(|t| t.code()).collect(), + tablet.row_count() as i32, + tablet.is_aligned(), + Some(true), + Some( + tablet + .column_categories() + .unwrap() + .iter() + .map(|c| c.code()) + .collect(), + ), + None, + None, + None, + ); + + assert_eq!(req.prefix_path, "object_table"); + assert_eq!(req.types, vec![11, 12]); + assert_eq!(req.write_to_table, Some(true)); + assert_eq!(req.column_categories, Some(vec![0, 1])); + assert_eq!(req.size, 1); + assert_eq!(req.timestamps, 1_608_268_702_780i64.to_be_bytes()); + // STRING 'r1': i32 len 2 + 'r1'; OBJECT segment: i32 len 12 + framed + // payload; two no-null bitmap flags. + let mut expected: Vec<u8> = vec![0, 0, 0, 2, b'r', b'1', 0, 0, 0, 12]; + expected.extend_from_slice(&[1, 0, 0, 0, 0, 0, 0, 0, 0, 0x01, 0x02, 0x03]); + expected.extend_from_slice(&[0, 0]); + assert_eq!(req.values, expected); + } + #[test] fn insert_tablets_rejects_empty_and_table_model() { use crate::data::{tablet::Tablet, ColumnCategory, TSDataType}; diff --git a/src/client/table_session.rs b/src/client/table_session.rs index b708e26..1f8d66d 100644 --- a/src/client/table_session.rs +++ b/src/client/table_session.rs @@ -296,6 +296,123 @@ mod tests { } } + /// Live-server OBJECT write/read round-trip (Go PR 175's + /// `Test_InsertObjectTablet` analogue). Skipped when no server is + /// reachable, and when the server predates OBJECT support — the probe + /// CREATE TABLE fails there and the test skips exactly like the Go e2e. + #[test] + fn live_object_write_roundtrip() { + use crate::data::{ColumnCategory, 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_object_write_roundtrip: no IoTDB server on 127.0.0.1:6667"); + return; + } + + const DB: &str = "rust_client_object_test"; + let mut session = TableSession::builder().build().expect("open"); + let _ = session.execute_non_query(&format!("DROP DATABASE IF EXISTS {DB}")); + session + .execute_non_query(&format!("CREATE DATABASE {DB}")) + .expect("create object test db"); + session + .execute_non_query(&format!("USE {DB}")) + .expect("use object test db"); + + if let Err(e) = session.execute_non_query( + "CREATE TABLE object_table (region_id STRING TAG, file OBJECT FIELD)", + ) { + eprintln!("skipping live_object_write_roundtrip: server does not support OBJECT ({e})"); + let _ = session.execute_non_query(&format!("DROP DATABASE IF EXISTS {DB}")); + return; + } + + let object_bytes: Vec<u8> = (0..1024u32).map(|i| (i % 251) as u8).collect(); + let new_tablet = |region: &str| { + let mut tablet = Tablet::new_table( + "object_table", + vec!["region_id".into(), "file".into()], + vec![TSDataType::String, TSDataType::Object], + vec![ColumnCategory::Tag, ColumnCategory::Field], + ) + .unwrap(); + tablet + .add_row(0, vec![Some(Value::String(region.into())), None]) + .unwrap(); + tablet + }; + + // Whole object at time 1. + let mut tablet = new_tablet("1"); + tablet.timestamps_mut()[0] = 1; + tablet + .set_object_value_at(true, 0, &object_bytes, 1, 0) + .unwrap(); + session.insert(&tablet).expect("insert whole object"); + + // Segmented object at time 2 (512 + 512): one segment per insert, + // the server assembles them into one cell (Go PR 175 does the same). + for (offset, is_eof) in [(0, false), (512, true)] { + let mut tablet = new_tablet("2"); + tablet.timestamps_mut()[0] = 2; + let start = offset as usize; + tablet + .set_object_value_at(is_eof, offset, &object_bytes[start..start + 512], 1, 0) + .unwrap(); + session.insert(&tablet).expect("insert object segment"); + } + + // Null object at time 3. + let mut tablet = new_tablet("3"); + tablet.timestamps_mut()[0] = 3; + session.insert(&tablet).expect("insert null object row"); + + // count(*) + { + let mut dataset = session + .execute_query("select count(*) from object_table") + .unwrap(); + let row = dataset.next_row().unwrap().unwrap(); + assert_eq!(row.values[0], Value::Int64(3)); + } + + // READ_OBJECT(file) stays a raw BLOB, whole and segmented. + for time in [1, 2] { + let mut dataset = session + .execute_query(&format!( + "select READ_OBJECT(file) from object_table where time = {time}" + )) + .unwrap(); + let row = dataset.next_row().unwrap().unwrap(); + match &row.values[0] { + Value::Blob(bytes) => assert_eq!(bytes, &object_bytes), + other => panic!("expected BLOB for READ_OBJECT, got {other:?}"), + } + } + + // select file renders the size summary; the null row stays null. + for (time, expected) in [(1, Some("(Object) 1.00 KB")), (3, None)] { + let mut dataset = session + .execute_query(&format!( + "select file from object_table where time = {time}" + )) + .unwrap(); + let row = dataset.next_row().unwrap().unwrap(); + match expected { + Some(summary) => assert_eq!(row.values[0], Value::String(summary.into())), + None => assert_eq!(row.values[0], Value::Null), + } + } + + let _ = session.execute_non_query(&format!("DROP DATABASE IF EXISTS {DB}")); + session.close().expect("close"); + } + /// Live-server test; skipped when no IoTDB instance is reachable. #[test] fn live_table_session_roundtrip() { diff --git a/src/data/mod.rs b/src/data/mod.rs index c6f9218..182fcca 100644 --- a/src/data/mod.rs +++ b/src/data/mod.rs @@ -16,7 +16,7 @@ // under the License. //! Data structures: TSDataType codes, Value, Tablet, TsBlock, bitmap helpers. -//! Data-type codes must match the official TSFile spec (0–11), identical +//! Data-type codes must match the official TSFile spec (0–12), identical //! across all IoTDB client SDKs. pub mod bitmap; @@ -27,7 +27,7 @@ pub mod value; pub use tablet::Tablet; pub use tsblock::TsBlock; -pub use value::Value; +pub use value::{object_bytes_to_string, Value}; /// Official TSFile data type codes. #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -45,6 +45,10 @@ pub enum TSDataType { Date = 9, Blob = 10, String = 11, + /// Table-model OBJECT (IoTDB 2.0.8+). Cells carry the already-framed + /// segment bytes: `[1 byte isEOF][8 byte big-endian offset][content]` + /// (see `Tablet::build_object_value` / `Tablet::set_object_value_at`). + Object = 12, } impl TSDataType { @@ -69,6 +73,7 @@ impl TSDataType { 9 => TSDataType::Date, 10 => TSDataType::Blob, 11 => TSDataType::String, + 12 => TSDataType::Object, _ => return None, }) } @@ -113,15 +118,16 @@ mod tests { assert_eq!(TSDataType::Date.code(), 9); assert_eq!(TSDataType::Blob.code(), 10); assert_eq!(TSDataType::String.code(), 11); + assert_eq!(TSDataType::Object.code(), 12); } #[test] fn data_type_from_code_round_trips() { - for code in 0u8..=11 { + for code in 0u8..=12 { let ty = TSDataType::from_code(code).expect("valid code"); assert_eq!(ty.code(), i32::from(code)); } - assert_eq!(TSDataType::from_code(12), None); + assert_eq!(TSDataType::from_code(13), None); assert_eq!(TSDataType::from_code(255), None); } diff --git a/src/data/record.rs b/src/data/record.rs index 760c033..4f7d863 100644 --- a/src/data/record.rs +++ b/src/data/record.rs @@ -50,7 +50,10 @@ pub fn serialize_record_values(values: &[Value]) -> Vec<u8> { Value::Float(v) => buf.extend_from_slice(&v.to_be_bytes()), Value::Double(v) => buf.extend_from_slice(&v.to_be_bytes()), Value::Text(s) | Value::String(s) => write_binary(&mut buf, s.as_bytes()), - Value::Blob(b) => write_binary(&mut buf, b), + // OBJECT uses the same length-prefixed binary layout; + // the server rejects tree-model OBJECTs, but the SDK does + // not add an extra client-side restriction (plan §1). + Value::Blob(b) | Value::Object(b) => write_binary(&mut buf, b), Value::Null => unreachable!("handled above"), } } @@ -76,6 +79,7 @@ mod tests { Value::Date(20260713), Value::Blob(vec![0xDE, 0xAD]), Value::String("é".into()), + Value::Object(vec![1, 0, 0, 0, 0, 0, 0, 0, 0, 0xDE, 0xAD]), ]; let mut expected: Vec<u8> = Vec::new(); @@ -95,6 +99,8 @@ mod tests { expected.extend_from_slice(&20260713i32.to_be_bytes()); // yyyyMMdd expected.extend_from_slice(&[0x0A, 0, 0, 0, 2, 0xDE, 0xAD]); expected.extend_from_slice(&[0x0B, 0, 0, 0, 2, 0xC3, 0xA9]); // "é" UTF-8 + // OBJECT: type marker 12 + length-prefixed framed segment. + expected.extend_from_slice(&[0x0C, 0, 0, 0, 11, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0xDE, 0xAD]); assert_eq!(serialize_record_values(&values), expected); } diff --git a/src/data/tablet.rs b/src/data/tablet.rs index 177ece4..81771f5 100644 --- a/src/data/tablet.rs +++ b/src/data/tablet.rs @@ -185,6 +185,65 @@ impl Tablet { Ok(()) } + /// Builds the wire representation of one OBJECT segment: a 1-byte isEOF + /// flag, an 8-byte big-endian offset, then the raw segment content. + /// This matches Java's + /// `Tablet.addValue(rowIndex, columnIndex, isEOF, offset, content)`. + /// + /// Whole objects are written as a single segment with + /// `is_eof = true` and `offset = 0`. + pub fn build_object_value(is_eof: bool, offset: i64, content: &[u8]) -> Result<Vec<u8>> { + if offset < 0 { + return Err(Error::Client(format!( + "OBJECT segment offset must be non-negative, got {offset}" + ))); + } + let mut value = Vec::with_capacity(9 + content.len()); + value.push(u8::from(is_eof)); + value.extend_from_slice(&offset.to_be_bytes()); + value.extend_from_slice(content); + Ok(value) + } + + /// Writes one segment of an OBJECT column value at an existing row — + /// the Rust counterpart of Go `Tablet.SetObjectValueAt`. + /// + /// An OBJECT value can be written in multiple segments so a large + /// object does not need to be fully loaded into memory. Segments must + /// be written with ascending offsets and the last segment must set + /// `is_eof` to `true`. Overwriting a cell sets it back to + /// `Some(..)`, which clears the null bit Rust derives from + /// `Option` at serialization time (the Go PR's `unmarkNullValueAt`). + pub fn set_object_value_at( + &mut self, + is_eof: bool, + offset: i64, + content: &[u8], + column_index: usize, + row_index: usize, + ) -> Result<()> { + let column_type = self.types.get(column_index).ok_or_else(|| { + Error::Client(format!( + "column index {column_index} out of range ({} columns)", + self.types.len() + )) + })?; + if *column_type != TSDataType::Object { + return Err(Error::Client(format!( + "column {column_index} must be of type OBJECT, got {column_type:?}" + ))); + } + if row_index >= self.row_count() { + return Err(Error::Client(format!( + "row index {row_index} out of range ({} rows)", + self.row_count() + ))); + } + let framed = Self::build_object_value(is_eof, offset, content)?; + self.values[column_index][row_index] = Some(Value::Object(framed)); + Ok(()) + } + /// Stably sorts rows by timestamp, reordering all value columns in step. pub fn sort_by_timestamp(&mut self) { let n = self.timestamps.len(); @@ -271,7 +330,12 @@ fn write_cell(buf: &mut Vec<u8>, ty: TSDataType, cell: Option<&Value>) { write_binary(buf, s.as_bytes()); } (TSDataType::Blob, Some(Value::Blob(b))) => write_binary(buf, b), - (TSDataType::Text | TSDataType::String | TSDataType::Blob, None) => write_binary(buf, &[]), + // OBJECT cells carry the framed segment (isEOF + offset + content); + // the wire encoding is the same length-prefixed binary layout as BLOB. + (TSDataType::Object, Some(Value::Object(b))) => write_binary(buf, b), + (TSDataType::Text | TSDataType::String | TSDataType::Blob | TSDataType::Object, None) => { + write_binary(buf, &[]) + } // Null sentinel 10000101 = 1000-01-01 (yyyyMMdd), per C#/Java. (TSDataType::Date, Some(Value::Date(v))) => buf.extend_from_slice(&v.to_be_bytes()), (TSDataType::Date, None) => buf.extend_from_slice(&10000101i32.to_be_bytes()), @@ -336,6 +400,7 @@ mod tests { TSDataType::Date, TSDataType::Blob, TSDataType::String, + TSDataType::Object, ]; let mut t = tree_tablet(types); t.add_row( @@ -351,6 +416,9 @@ mod tests { Some(Value::Date(20260710)), Some(Value::Blob(vec![0xDE, 0xAD])), Some(Value::String("é".into())), + Some(Value::Object( + Tablet::build_object_value(true, 0, &[0xDE, 0xAD]).unwrap(), + )), ], ) .unwrap(); @@ -366,7 +434,9 @@ mod tests { expected.extend_from_slice(&20260710i32.to_be_bytes()); expected.extend_from_slice(&[0, 0, 0, 2, 0xDE, 0xAD]); expected.extend_from_slice(&[0, 0, 0, 2, 0xC3, 0xA9]); // "é" UTF-8 - expected.extend_from_slice(&[0; 10]); // 10 columns, no nulls + // OBJECT: length-prefixed framed segment (isEOF=1, offset=0, content). + expected.extend_from_slice(&[0, 0, 0, 11, 1, 0, 0, 0, 0, 0, 0, 0, 0, 0xDE, 0xAD]); + expected.extend_from_slice(&[0; 11]); // 11 columns, no nulls assert_eq!(t.serialize_values(), expected); } @@ -383,6 +453,7 @@ mod tests { TSDataType::Date, TSDataType::Blob, TSDataType::String, + TSDataType::Object, ]; let n = types.len(); let mut t = tree_tablet(types); @@ -399,6 +470,7 @@ mod tests { expected.extend_from_slice(&10000101i32.to_be_bytes()); // 1000-01-01 expected.extend_from_slice(&[0, 0, 0, 0]); // empty blob expected.extend_from_slice(&[0, 0, 0, 0]); // empty string + expected.extend_from_slice(&[0, 0, 0, 0]); // empty object for _ in 0..n { expected.extend_from_slice(&[0x01, 0x01]); // flag + bitmap (row 0 null) } @@ -519,6 +591,104 @@ mod tests { ); } + fn object_tablet() -> Tablet { + Tablet::new_table( + "object_table", + vec!["region_id".into(), "file".into()], + vec![TSDataType::String, TSDataType::Object], + vec![ColumnCategory::Tag, ColumnCategory::Field], + ) + .unwrap() + } + + #[test] + fn new_table_accepts_object_column() { + let t = object_tablet(); + assert!(t.is_table_model()); + assert_eq!(t.types(), [TSDataType::String, TSDataType::Object]); + } + + #[test] + fn build_object_value_frames_segments() { + assert_eq!( + Tablet::build_object_value(true, 0, &[0x11, 0x22]).unwrap(), + [1, 0, 0, 0, 0, 0, 0, 0, 0, 0x11, 0x22] + ); + assert_eq!( + Tablet::build_object_value(false, 512, &[0x33]).unwrap(), + [0, 0, 0, 0, 0, 0, 0, 2, 0, 0x33] + ); + assert!(matches!( + Tablet::build_object_value(true, -1, &[]), + Err(Error::Client(m)) if m.contains("non-negative") + )); + } + + #[test] + fn set_object_value_at_writes_whole_and_segmented_rows() { + let mut t = object_tablet(); + t.add_row(1, vec![Some(Value::String("r1".into())), None]) + .unwrap(); + t.add_row(2, vec![Some(Value::String("r2".into())), None]) + .unwrap(); + t.set_object_value_at(true, 0, &[0x11, 0x22], 1, 0).unwrap(); + t.set_object_value_at(false, 512, &[0x33], 1, 1).unwrap(); + + let expected: Vec<u8> = vec![ + 0, 0, 0, 2, b'r', b'1', // tag col row 0: "r1" + 0, 0, 0, 2, b'r', b'2', // tag col row 1: "r2" + 0, 0, 0, 11, // object row 0: framed segment length 11 + 1, 0, 0, 0, 0, 0, 0, 0, 0, 0x11, 0x22, // isEOF=1, offset=0 + 0, 0, 0, 10, // object row 1: framed segment length 10 + 0, 0, 0, 0, 0, 0, 0, 2, 0, 0x33, // isEOF=0, offset=512 + 0, 0, // trailing bitmap flags: no nulls + ]; + assert_eq!(t.serialize_values(), expected); + } + + #[test] + fn set_object_value_at_clears_previously_null_cell() { + let mut t = object_tablet(); + t.add_row(1, vec![Some(Value::String("r1".into())), None]) + .unwrap(); + // Null OBJECT cell: empty placeholder + [tag flag 0][object flag 1][bitmap 0x01]. + let with_null = t.serialize_values(); + assert_eq!(&with_null[6..], [0, 0, 0, 0, 0, 1, 1]); + + t.set_object_value_at(true, 0, &[0x44], 1, 0).unwrap(); + let without_null = t.serialize_values(); + assert_eq!(without_null.len(), 6 + 4 + 10 + 2); + // Both columns now report "no nulls". + assert_eq!(&without_null[without_null.len() - 2..], [0, 0]); + } + + #[test] + fn set_object_value_at_rejects_invalid_inputs() { + let mut t = object_tablet(); + t.add_row(1, vec![Some(Value::String("r1".into())), None]) + .unwrap(); + // Non-OBJECT column. + assert!(matches!( + t.set_object_value_at(true, 0, &[0x01], 0, 0), + Err(Error::Client(m)) if m.contains("must be of type OBJECT") + )); + // Row out of range. + assert!(matches!( + t.set_object_value_at(true, 0, &[0x01], 1, 1), + Err(Error::Client(m)) if m.contains("row index") + )); + // Column out of range. + assert!(matches!( + t.set_object_value_at(true, 0, &[0x01], 2, 0), + Err(Error::Client(m)) if m.contains("column index") + )); + // Negative offset. + assert!(matches!( + t.set_object_value_at(true, -1, &[0x01], 1, 0), + Err(Error::Client(m)) if m.contains("non-negative") + )); + } + #[test] fn constructor_and_add_row_validation() { assert!(Tablet::new("d", vec!["s1".into()], vec![]).is_err()); diff --git a/src/data/tsblock.rs b/src/data/tsblock.rs index 40cd26a..05c5e41 100644 --- a/src/data/tsblock.rs +++ b/src/data/tsblock.rs @@ -222,6 +222,10 @@ fn decode_column( Err(e) => Value::Blob(e.into_bytes()), }, TSDataType::String => Value::String(decode_utf8(bytes)?), + // OBJECT metadata (8-byte BE size + internal path) stays + // raw here; row assembly formats it via + // `object_bytes_to_string` when the logical type is OBJECT. + TSDataType::Object => Value::Object(bytes), _ => { return Err(Error::Decode(format!( "BinaryArray encoding with incompatible type {ty:?}" @@ -465,6 +469,34 @@ mod tests { ); } + #[test] + fn binary_array_object_metadata_and_nulls() { + // Server OBJECT cell: 8-byte BE file size + internal path. + let mut payload = 1024u64.to_be_bytes().to_vec(); + payload.extend_from_slice(b"internal/path/1.bin"); + + let mut b = header(&[TSDataType::Object], 2, &[ENCODING_BINARY_ARRAY]); + b.extend_from_slice(&time_column(&[1, 2])); + b.push(1); // mayHaveNull + b.push(0b0100_0000); // MSB-first: row 1 null, consumes no payload bytes + b.extend_from_slice(&(payload.len() as i32).to_be_bytes()); + b.extend_from_slice(&payload); + + let block = TsBlock::decode(&b).unwrap(); + assert_eq!(block.column_types, vec![TSDataType::Object]); + assert_eq!(block.columns[0], vec![Value::Object(payload), Value::Null]); + } + + #[test] + fn object_binary_array_truncated_payload_errors() { + let mut b = header(&[TSDataType::Object], 1, &[ENCODING_BINARY_ARRAY]); + b.extend_from_slice(&time_column(&[1])); + b.push(0); // mayHaveNull + b.extend_from_slice(&16i32.to_be_bytes()); // claims 16 payload bytes + b.extend_from_slice(&[0, 0, 0]); // delivers 3 + assert!(matches!(TsBlock::decode(&b), Err(Error::Decode(_)))); + } + #[test] fn rle_replicates_single_value() { let mut b = header(&[TSDataType::Int64], 4, &[ENCODING_RLE]); diff --git a/src/data/value.rs b/src/data/value.rs index 947b073..3f661f5 100644 --- a/src/data/value.rs +++ b/src/data/value.rs @@ -18,6 +18,7 @@ //! A dynamically-typed IoTDB cell value. use super::TSDataType; +use crate::error::{Error, Result}; /// One cell of an IoTDB row: a typed scalar or `Null`. /// @@ -37,6 +38,10 @@ pub enum Value { Date(i32), Blob(Vec<u8>), String(String), + /// Table-model OBJECT cell. The bytes are the **already-framed OBJECT + /// segment**: `[1 byte isEOF][8 byte big-endian offset][content]` + /// (see `Tablet::build_object_value` / `Tablet::set_object_value_at`). + Object(Vec<u8>), Null, } @@ -54,6 +59,7 @@ impl Value { Value::Date(_) => TSDataType::Date, Value::Blob(_) => TSDataType::Blob, Value::String(_) => TSDataType::String, + Value::Object(_) => TSDataType::Object, Value::Null => return None, }) } @@ -69,6 +75,38 @@ impl Value { } } +/// Formats the wire representation of a stored OBJECT value for display. +/// +/// The server stores OBJECT cells as an 8-byte big-endian file size followed +/// by the internal object path. Mirroring the Go client's +/// `objectBytesToString` and the Node/C# helpers, this renders the size as +/// `(Object) 1023 B` / `(Object) 1.00 KB` / `(Object) 1.00 MB` / +/// `(Object) 1.00 GB`. +pub fn object_bytes_to_string(bytes: &[u8]) -> Result<String> { + if bytes.len() < 8 { + return Err(Error::Decode(format!( + "invalid OBJECT value: expected at least 8 bytes, got {}", + bytes.len() + ))); + } + let mut size_bytes = [0u8; 8]; + size_bytes.copy_from_slice(&bytes[..8]); + let size = u64::from_be_bytes(size_bytes); + const KILOBYTE: f64 = 1024.0; + const MEGABYTE: f64 = KILOBYTE * 1024.0; + const GIGABYTE: f64 = MEGABYTE * 1024.0; + let size = size as f64; + if size < KILOBYTE { + Ok(format!("(Object) {size:.0} B")) + } else if size < MEGABYTE { + Ok(format!("(Object) {:.2} KB", size / KILOBYTE)) + } else if size < GIGABYTE { + Ok(format!("(Object) {:.2} MB", size / MEGABYTE)) + } else { + Ok(format!("(Object) {:.2} GB", size / GIGABYTE)) + } +} + #[cfg(test)] mod tests { use super::*; @@ -85,6 +123,7 @@ mod tests { assert_eq!(Value::Date(20260710).type_code(), Some(9)); assert_eq!(Value::Blob(vec![0]).type_code(), Some(10)); assert_eq!(Value::String("s".into()).type_code(), Some(11)); + assert_eq!(Value::Object(vec![]).type_code(), Some(12)); assert_eq!(Value::Null.type_code(), None); } @@ -94,4 +133,38 @@ mod tests { assert!(Value::Null.is_null()); assert!(!Value::Int32(7).is_null()); } + + fn object_size_bytes(size: u64) -> Vec<u8> { + let mut value = size.to_be_bytes().to_vec(); + value.extend_from_slice(b"internal/path/1.bin"); + value + } + + #[test] + fn object_bytes_to_string_formats_all_units() { + assert_eq!( + object_bytes_to_string(&object_size_bytes(1023)).unwrap(), + "(Object) 1023 B" + ); + assert_eq!( + object_bytes_to_string(&object_size_bytes(1024)).unwrap(), + "(Object) 1.00 KB" + ); + assert_eq!( + object_bytes_to_string(&object_size_bytes(1024 * 1024)).unwrap(), + "(Object) 1.00 MB" + ); + assert_eq!( + object_bytes_to_string(&object_size_bytes(1024 * 1024 * 1024)).unwrap(), + "(Object) 1.00 GB" + ); + } + + #[test] + fn object_bytes_to_string_rejects_short_input() { + assert!(matches!( + object_bytes_to_string(&[0; 7]), + Err(Error::Decode(m)) if m.contains("at least 8 bytes") + )); + } } diff --git a/src/lib.rs b/src/lib.rs index bd26ccd..9c073d2 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -38,5 +38,5 @@ pub use client::table_session::{TableSession, TableSessionBuilder}; #[cfg(feature = "tls")] pub use connection::TlsOptions; pub use connection::{ConnectionOptions, Endpoint, RpcProtocol}; -pub use data::{ColumnCategory, TSDataType, Tablet, TsBlock, Value}; +pub use data::{object_bytes_to_string, ColumnCategory, TSDataType, Tablet, TsBlock, Value}; pub use error::{Error, Result};
