Improve table writer signature flexibility

Author: pbower

rust/Cargo.toml

@@ -158,9 +158,9 @@ mmap = ["dep:libc"]
 # Adds parquet IO
 parquet = []
 
-# CSV reader/writer/encoder/decoder. Pulls in itoa+ryu+memchr for 
+# CSV reader/writer/encoder/decoder. Pulls in itoa+ryu+memchr for
 # fast formatter and quote scanner, atoi+fast-float2 for SIMD-accelerated
-# integer and float parsing on the decode path. 
+# integer and float parsing on the decode path.
 csv = ["dep:ryu", "dep:memchr", "dep:fast-float2"]
 
 # TCP transport

rust/examples/arrow/table_reader.rs

@@ -128,7 +128,7 @@ async fn write_stream(tables: &[Table], path: &Path) -> Result<(), Box<dyn std::
     let schema: Vec<Field> = tables[0].schema().iter().map(|f| (**f).clone()).collect();
     let mut writer = TableStreamWriter::<Vec64<u8>>::new(schema, IPCMessageProtocol::Stream, None);
     for table in tables {
-        writer.write(&table.clone().into())?;
+        writer.write(table.clone())?;
     }
     writer.finish()?;
 

rust/examples/arrow/table_stream_reader.rs

@@ -101,7 +101,7 @@ async fn write_stream(
     let schema: Vec<Field> = tables[0].schema().iter().map(|f| (**f).clone()).collect();
     let mut writer = TableStreamWriter::<Vec64<u8>>::new(schema, protocol, None);
     for table in tables {
-        writer.write(&table.clone().into())?;
+        writer.write(table.clone())?;
     }
     writer.finish()?;
 

rust/examples/arrow/table_stream_writer.rs

@@ -33,7 +33,7 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut writer = TableStreamWriter::<Vec64<u8>>::new(schema, IPCMessageProtocol::Stream, None);
 
     for table in &tables {
-        writer.write(&table.clone().into())?;
+        writer.write(table.clone())?;
     }
     writer.finish()?;
 

rust/src/lib.rs

@@ -130,10 +130,6 @@ pub mod models {
 
         /// TLV sink for simple type-length-value framing
         pub mod tlv_sink;
-
-        /// Live LBuffer-backed table sink for decoded records
-        #[cfg(all(feature = "lbuffer", feature = "json"))]
-        pub mod live_table_sink;
     }
 
     /// Encoders for Arrow IPC, TLV, CSV, and optionally Parquet

rust/src/models/encoders/ipc/table_stream.rs

@@ -294,7 +294,7 @@ mod tests {
             4,
         );
 
-        writer.write(&tbl.clone().into()).unwrap();
+        writer.write(tbl.clone()).unwrap();
         writer.finish().unwrap();
 
         let mut file = StdFile::create(&path).unwrap();
@@ -343,7 +343,7 @@ mod tests {
             4,
         );
 
-        writer.write(&tbl.clone().into()).unwrap();
+        writer.write(tbl.clone()).unwrap();
         writer.finish().unwrap();
 
         let mut file = StdFile::create(&path).unwrap();
@@ -392,7 +392,7 @@ mod tests {
             4,
         );
 
-        writer.write(&tbl.into()).unwrap();
+        writer.write(tbl).unwrap();
         writer.finish().unwrap();
 
         let mut buf = Vec::new();
@@ -437,7 +437,7 @@ mod tests {
             4,
         );
 
-        writer.write(&tbl.into()).unwrap();
+        writer.write(tbl).unwrap();
         writer.finish().unwrap();
 
         let mut buf = Vec::new();
@@ -482,7 +482,7 @@ mod tests {
             4,
         );
 
-        writer.write(&tbl.into()).unwrap();
+        writer.write(tbl).unwrap();
         writer.finish().unwrap();
 
         let mut buf = Vec::new();
@@ -527,7 +527,7 @@ mod tests {
             4,
         );
 
-        writer.write(&tbl.clone().into()).unwrap();
+        writer.write(tbl.clone()).unwrap();
         writer.finish().unwrap();
 
         // Write to temp file
@@ -591,7 +591,7 @@ mod tests {
             4,
         );
 
-        writer.write(&tbl.clone().into()).unwrap();
+        writer.write(tbl.clone()).unwrap();
         writer.finish().unwrap();
 
         let mut file = StdFile::create(&path).unwrap();
@@ -654,7 +654,7 @@ mod tests {
             4,
         );
 
-        writer.write(&tbl.clone().into()).unwrap();
+        writer.write(tbl.clone()).unwrap();
         writer.finish().unwrap();
 
         // Write to temp file

rust/src/models/readers/ipc/table.rs

@@ -273,8 +273,8 @@ mod tests {
         let mut writer =
             TableStreamWriter::<Vec64<u8>>::new(schema.clone(), IPCMessageProtocol::Stream, None);
         register_dictionaries_for_table(&mut writer, &table);
-        writer.write(&table.clone().into()).unwrap();
-        writer.write(&table.clone().into()).unwrap();
+        writer.write(table.clone()).unwrap();
+        writer.write(table.clone()).unwrap();
         writer.finish().unwrap();
         let frames = writer.drain_all_frames();
 
@@ -305,9 +305,9 @@ mod tests {
             TableStreamWriter::<Vec64<u8>>::new(schema.clone(), IPCMessageProtocol::Stream, None);
         register_dictionaries_for_table(&mut writer, &table);
         // three batches
-        writer.write(&table.clone().into()).unwrap();
-        writer.write(&table.clone().into()).unwrap();
-        writer.write(&table.clone().into()).unwrap();
+        writer.write(table.clone()).unwrap();
+        writer.write(table.clone()).unwrap();
+        writer.write(table.clone()).unwrap();
         writer.finish().unwrap();
         let frames = writer.drain_all_frames();
 
@@ -332,8 +332,8 @@ mod tests {
         let mut writer =
             TableStreamWriter::<Vec64<u8>>::new(schema.clone(), IPCMessageProtocol::Stream, None);
         register_dictionaries_for_table(&mut writer, &table);
-        writer.write(&table.clone().into()).unwrap();
-        writer.write(&table.clone().into()).unwrap();
+        writer.write(table.clone()).unwrap();
+        writer.write(table.clone()).unwrap();
         writer.finish().unwrap();
         let frames = writer.drain_all_frames();
 
@@ -363,8 +363,8 @@ mod tests {
         let mut writer =
             TableStreamWriter::<Vec64<u8>>::new(schema.clone(), IPCMessageProtocol::Stream, None);
         register_dictionaries_for_table(&mut writer, &table);
-        writer.write(&table.clone().into()).unwrap();
-        writer.write(&table.clone().into()).unwrap();
+        writer.write(table.clone()).unwrap();
+        writer.write(table.clone()).unwrap();
         writer.finish().unwrap();
         let frames = writer.drain_all_frames();
 
@@ -392,7 +392,7 @@ mod tests {
         let mut writer =
             TableStreamWriter::<Vec64<u8>>::new(schema.clone(), IPCMessageProtocol::Stream, None);
         register_dictionaries_for_table(&mut writer, &table);
-        writer.write(&table.clone().into()).unwrap();
+        writer.write(table.clone()).unwrap();
         writer.finish().unwrap();
         let frames = writer.drain_all_frames();
 
@@ -435,8 +435,8 @@ mod tests {
         let mut writer =
             TableStreamWriter::<Vec64<u8>>::new(schema.clone(), IPCMessageProtocol::Stream, None);
         register_dictionaries_for_table(&mut writer, &table);
-        writer.write(&table.clone().into()).unwrap();
-        writer.write(&table.clone().into()).unwrap();
+        writer.write(table.clone()).unwrap();
+        writer.write(table.clone()).unwrap();
         writer.finish().unwrap();
         let frames = writer.drain_all_frames();
 
@@ -465,7 +465,7 @@ mod tests {
         let mut writer =
             TableStreamWriter::<Vec64<u8>>::new(schema.clone(), IPCMessageProtocol::Stream, None);
         register_dictionaries_for_table(&mut writer, &table);
-        writer.write(&table.clone().into()).unwrap();
+        writer.write(table.clone()).unwrap();
         writer.finish().unwrap();
         let frames = writer.drain_all_frames();