Fix arrow inserts (#28)

* Add insert benchmarking methods

Benchmarking methods added for:

* stream inserts using tuples of time, value data
* stream inserts using pyarrow tables of timestamps, value columns

* streamset inserts using a dict map of streamset stream uuids, and lists of tuples of time, value data
* streamset inserts using a dict map of streamset stream uuids, and pyarrow tables of timestamps, values.

* Include nullable false in pyarrow schema inserts

* This was the only difference in the schemas between go and python.
* also using a bytesIO stream to act as the sink for the ipc bytes.

Author: justinGilmer

btrdb/stream.py

@@ -10,6 +10,8 @@
 """
 Module for Stream and related classes
 """
+import io
+
 ##########################################################################
 ## Imports
 ##########################################################################
@@ -512,8 +514,8 @@ def arrow_insert(self, data: pa.Table, merge: str = "never") -> int:
         logger.debug(f"tmp_table schema: {tmp_table.schema}")
         new_schema = pa.schema(
             [
-                (pa.field("time", pa.timestamp(unit="ns", tz="UTC"))),
-                (pa.field("value", pa.float64())),
+                (pa.field("time", pa.timestamp(unit="ns", tz="UTC"), nullable=False)),
+                (pa.field("value", pa.float64(), nullable=False)),
             ]
         )
         tmp_table = tmp_table.cast(new_schema)
@@ -1930,10 +1932,12 @@ def _materialize_stream_as_table(arrow_bytes):
 
 
 def _table_slice_to_feather_bytes(table_slice: pa.Table) -> bytes:
-    sink = pa.BufferOutputStream()
+    # sink = pa.BufferOutputStream()
+    sink = io.BytesIO()
     with pa.ipc.new_stream(sink=sink, schema=table_slice.schema) as writer:
-        writer.write_table(table_slice)
-    return sink.readall()
+        writer.write(table_slice)
+    buf = sink.getvalue()
+    return buf
 
 
 def _coalesce_table_deque(tables: deque):