Support table-model OBJECT writes and OBJECT result decoding

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 |

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/` | 两种模型及会话池的可运行示例 |

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");

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]

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};