Add blink_table option to input table factory method in pydeeephaven (deephaven#5317)

* Add blink_table to input table factory method

* Update py/client/pydeephaven/session.py

Co-authored-by: Chip Kent <5250374+chipkent@users.noreply.github.com>

* Respond to review comments

---------

Co-authored-by: Chip Kent <5250374+chipkent@users.noreply.github.com>

Author: jmao-denver

py/client/pydeephaven/_table_ops.py

@@ -606,17 +606,24 @@ def make_grpc_request_for_batch(self, result_id, source_id) -> Any:
 
 
 class CreateInputTableOp(TableOp):
-    def __init__(self, schema: pa.schema, init_table: Any, key_cols: List[str] = None):
+    def __init__(self, schema: pa.schema, init_table: Any, key_cols: List[str] = None, blink: bool = False):
+        if blink and key_cols:
+            raise ValueError("key columns are not supported for blink input tables.")
+
         self.schema = schema
         self.init_table = init_table
         self.key_cols = key_cols
+        self.blink = blink
 
     @classmethod
     def get_stub_func(cls, table_service_stub: table_pb2_grpc.TableServiceStub) -> Any:
         return table_service_stub.CreateInputTable
 
     def make_grpc_request(self, result_id, source_id) -> Any:
-        if self.key_cols:
+        if self.blink:
+            blink_ = table_pb2.CreateInputTableRequest.InputTableKind.Blink()
+            input_table_kind = table_pb2.CreateInputTableRequest.InputTableKind(blink=blink_)
+        elif self.key_cols:
             key_backed = table_pb2.CreateInputTableRequest.InputTableKind.InMemoryKeyBacked(
                 key_columns=self.key_cols)
             input_table_kind = table_pb2.CreateInputTableRequest.InputTableKind(in_memory_key_backed=key_backed)

py/client/pydeephaven/session.py

@@ -514,14 +514,16 @@ def query(self, table: Table) -> Query:
         return Query(self, table)
 
     def input_table(self, schema: pa.Schema = None, init_table: Table = None,
-                    key_cols: Union[str, List[str]] = None) -> InputTable:
-        """Creates an InputTable from either Arrow schema or initial table. When key columns are
-        provided, the InputTable will be keyed, otherwise it will be append-only.
+                    key_cols: Union[str, List[str]] = None, blink_table: bool = False) -> InputTable:
+        """Creates an InputTable from either Arrow schema or initial table.  When blink_table is True, the InputTable
+        will be a blink table. When blink_table is False (default), the InputTable will be
+        keyed if key columns are provided, otherwise it will be append-only.
 
         Args:
             schema (pa.Schema): the schema for the InputTable
             init_table (Table): the initial table
             key_cols (Union[str, Sequence[str]): the name(s) of the key column(s)
+            blink_table (bool): whether the InputTable should be a blink table, default is False
 
         Returns:
             an InputTable
@@ -534,7 +536,10 @@ def input_table(self, schema: pa.Schema = None, init_table: Table = None,
         elif schema and init_table:
             raise ValueError("both arrow schema and init table are provided.")
 
-        table_op = CreateInputTableOp(schema=schema, init_table=init_table, key_cols=to_list(key_cols))
+        if blink_table and key_cols:
+            raise ValueError("key columns are not supported for blink input tables.")
+
+        table_op = CreateInputTableOp(schema=schema, init_table=init_table, key_cols=to_list(key_cols), blink=blink_table)
         input_table = self.table_service.grpc_table_op(None, table_op, table_class=InputTable)
         input_table.key_cols = key_cols
         return input_table

py/client/tests/test_session.py

@@ -1,7 +1,6 @@
 #
 # Copyright (c) 2016-2024 Deephaven Data Labs and Patent Pending
 #
-
 import unittest
 from time import sleep
 
@@ -54,26 +53,21 @@ def test_time_table(self):
             t = session.time_table(period=100000)
             self.assertFalse(t.is_static)
             session.bind_table("t", t)
-            session.run_script("""
+
+            console_script = ("""
 from deephaven import empty_table
 try:
     del t1
 except NameError:
     pass
 t1 = empty_table(0) if t.is_blink else None
 """)
+            session.run_script(console_script)
             self.assertNotIn("t1", session.tables)
 
             t = session.time_table(period=100000, blink_table=True)
             session.bind_table("t", t)
-            session.run_script("""
-from deephaven import empty_table
-try:
-    del t1
-except NameError:
-    pass
-t1 = empty_table(0) if t.is_blink else None
-""")
+            session.run_script(console_script)
             self.assertIn("t1", session.tables)
 
     def test_merge_tables(self):
@@ -273,6 +267,78 @@ def test_auto_close(self):
         session = None
         self.assertIsNone(session)
 
+    def test_blink_input_table(self):
+        pa_types = [
+            pa.bool_(),
+            pa.int8(),
+            pa.int16(),
+            pa.int32(),
+            pa.int64(),
+            pa.timestamp('ns', tz='UTC'),
+            pa.float32(),
+            pa.float64(),
+            pa.string(),
+        ]
+        pa_data = [
+            pa.array([True, False]),
+            pa.array([2 ** 7 - 1, -2 ** 7 + 1]),
+            pa.array([2 ** 15 - 1, -2 ** 15 + 1]),
+            pa.array([2 ** 31 - 1, -2 ** 31 + 1]),
+            pa.array([2 ** 63 - 1, -2 ** 63 + 1]),
+            pa.array([pd.Timestamp('2017-01-01T12:01:01', tz='UTC'),
+                      pd.Timestamp('2017-01-01T11:01:01', tz='Europe/Paris')]),
+            pa.array([1.1, 2.2], pa.float32()),
+            pa.array([1.1, 2.2], pa.float64()),
+            pa.array(["foo", "bar"]),
+        ]
+        fields = [pa.field(f"f{i}", ty) for i, ty in enumerate(pa_types)]
+        schema = pa.schema(fields)
+        pa_table = pa.table(pa_data, schema=schema)
+        with Session() as session:
+            dh_table = session.import_table(pa_table)
+
+            with self.subTest("Create blink Input Table"):
+                with self.assertRaises(ValueError):
+                    session.input_table(schema=schema, key_cols="f1", blink_table=True)
+                blink_input_table = session.input_table(schema=schema, blink_table=True)
+                pa_table = blink_input_table.to_arrow()
+                self.assertEqual(schema, pa_table.schema)
+                session.bind_table("t", blink_input_table)
+                console_script = ("""
+from deephaven import empty_table
+try:
+    del t1
+except NameError:
+    pass
+t1 = empty_table(0) if t.is_blink else None
+        """)
+                session.run_script(console_script)
+                self.assertIn("t1", session.tables)
+
+                with self.assertRaises(ValueError):
+                    session.input_table(schema=schema, init_table=blink_input_table, blink_table=True)
+                with self.assertRaises(ValueError):
+                    session.input_table(key_cols="f0", blink_table=True)
+
+            with self.subTest("blink InputTable ops"):
+                session.bind_table("dh_table", dh_table)
+                console_script = ("""
+from deephaven import empty_table
+try:
+    del t1
+except NameError:
+    pass
+t.add(dh_table)
+t.await_update()
+t1 = empty_table(0) if t.size == 2 else None
+        """)
+                session.run_script(console_script)
+                self.assertIn("t1", session.tables)
+
+                with self.assertRaises(PermissionError):
+                    blink_input_table.delete(dh_table.select(["f1"]))
+
+
 
 if __name__ == '__main__':
     unittest.main()