fix: DH-21128: Always call Java Consumer's accept method regardless of whether partition value is provided (#7480)

Two new tests are added to cover the no-partition cases. However, the
ticking table test seems to reveal a dead-lock situation, which could be
a problem of the test data service.

Author: jmao-denver

py/server/deephaven/experimental/table_data_service.py

@@ -401,8 +401,9 @@ def location_cb_proxy(
         ):
             j_tbl_location_key = _JTableLocationKeyImpl(pt_location_key)
             if pt_table is None or pt_table.to_batches() is None:
-                location_cb.apply(
-                    j_tbl_location_key, jpy.array("java.nio.ByteBuffer", [])
+                location_cb.accept(
+                    j_tbl_location_key,
+                    jpy.array("java.nio.ByteBuffer", []),
                 )
             else:
                 if pt_table.num_rows != 1:
@@ -456,8 +457,9 @@ def location_cb_proxy(
         ):
             j_tbl_location_key = _JTableLocationKeyImpl(pt_location_key)
             if pt_table is None:
-                location_cb.apply(
-                    j_tbl_location_key, jpy.array("java.nio.ByteBuffer", [])
+                location_cb.accept(
+                    j_tbl_location_key,
+                    jpy.array("java.nio.ByteBuffer", []),
                 )
             else:
                 if pt_table.num_rows != 1:

py/server/tests/test_table_data_service.py

@@ -77,6 +77,8 @@ def __init__(
         self.sub_new_partition_cancelled: bool = False
         self.sub_new_partition_fail_test: bool = False
         self.sub_partition_size_fail_test: bool = False
+        self.table_location_no_partition_test: bool = False
+        self.sub_table_location_no_partition_test: bool = False
         self.partitions: dict[TableLocationKey, pa.Table] = {}
         self.partitions_size_subscriptions: dict[TableLocationKey, bool] = {}
         self.existing_partitions_called: int = 0
@@ -115,13 +117,16 @@ def table_locations(
             partition_key = TableLocationKeyImpl(f"{ticker}/NYSE")
             self.partitions[partition_key] = pa_table
 
-            expr = (pc.field("Ticker") == f"{ticker}") & (
-                pc.field("Exchange") == "NYSE"
-            )
-            location_cb(
-                partition_key,
-                pa_table.filter(expr).select(["Ticker", "Exchange"]).slice(0, 1),
-            )
+            if self.table_location_no_partition_test:
+                location_cb(partition_key, None)
+            else:
+                expr = (pc.field("Ticker") == f"{ticker}") & (
+                    pc.field("Exchange") == "NYSE"
+                )
+                location_cb(
+                    partition_key,
+                    pa_table.filter(expr).select(["Ticker", "Exchange"]).slice(0, 1),
+                )
             self.existing_partitions_called += 1
 
             # indicate that we've finished notifying existing table locations
@@ -185,10 +190,13 @@ def _th_new_partitions(
             expr = (pc.field("Ticker") == f"{ticker}") & (
                 pc.field("Exchange") == "NYSE"
             )
-            location_cb(
-                partition_key,
-                pa_table.filter(expr).select(["Ticker", "Exchange"]).slice(0, 1),
-            )
+            if self.sub_table_location_no_partition_test:
+                location_cb(partition_key, None)
+            else:
+                location_cb(
+                    partition_key,
+                    pa_table.filter(expr).select(["Ticker", "Exchange"]).slice(0, 1),
+                )
             if self.sub_new_partition_fail_test:
                 failure_cb(Exception("table location subscription failure"))
                 return
@@ -217,10 +225,14 @@ def subscribe_to_table_locations(
             expr = (pc.field("Ticker") == f"{ticker}") & (
                 pc.field("Exchange") == "NYSE"
             )
-            location_cb(
-                partition_key,
-                pa_table.filter(expr).select(["Ticker", "Exchange"]).slice(0, 1),
-            )
+            if self.sub_table_location_no_partition_test:
+                location_cb(partition_key, None)
+            else:
+                location_cb(
+                    partition_key,
+                    pa_table.filter(expr).select(["Ticker", "Exchange"]).slice(0, 1),
+                )
+        success_cb()
 
         exec_ctx = get_exec_ctx()
         th = threading.Thread(
@@ -233,7 +245,6 @@ def _cancellation_callback():
             self.sub_new_partition_cancelled = True
             self.location_cb = None
 
-        success_cb()
         return _cancellation_callback
 
     def _th_partition_size_changes(
@@ -296,6 +307,7 @@ def subscribe_to_table_location_size(
 
         # need to initial size
         size_cb(self.partitions[table_location_key].num_rows)
+        success_cb()
 
         self.partitions_size_subscriptions[table_location_key] = True
         th = threading.Thread(
@@ -307,7 +319,6 @@ def subscribe_to_table_location_size(
         def _cancellation_callback():
             self.partitions_size_subscriptions[table_location_key] = False
 
-        success_cb()
         return _cancellation_callback
 
 
@@ -608,6 +619,35 @@ def test_partition_size_sub_failure(self):
 
         self.assertTrue(table.is_failed)
 
+    def test_make_static_table_no_partition(self):
+        backend = TestBackend(self.gen_pa_table(), pt_schema=self.pa_table.schema)
+        backend.table_location_no_partition_test = True
+        data_service = TableDataService(backend)
+        table = data_service.make_table(TableKeyImpl("test"), refreshing=False)
+        self.assertIsNotNone(table)
+        self.assertFalse(table.columns[0].column_type == ColumnType.PARTITIONING)
+        self.assertFalse(table.columns[1].column_type == ColumnType.PARTITIONING)
+        self.assertEqual(table.columns[2:], self.test_table.columns[2:])
+        self.assertEqual(table.size, 2)
+        self.assertEqual(backend.existing_partitions_called, 1)
+        self.assertEqual(backend.partition_size_called, 1)
+
+    def test_make_live_table_with_no_partition(self):
+        backend = TestBackend(self.gen_pa_table(), pt_schema=self.pa_table.schema)
+        data_service = TableDataService(backend)
+        backend.sub_table_location_no_partition_test = True
+        table = data_service.make_table(TableKeyImpl("test"), refreshing=True)
+        self.assertIsNotNone(table)
+        self.assertFalse(table.columns[0].column_type == ColumnType.PARTITIONING)
+        self.assertFalse(table.columns[1].column_type == ColumnType.PARTITIONING)
+        self.assertEqual(table.columns[2:], self.test_table.columns[2:])
+
+        self.wait_ticking_table_update(table, 20, 5)
+
+        self.assertGreaterEqual(table.size, 20)
+        self.assertEqual(backend.existing_partitions_called, 0)
+        self.assertEqual(backend.partition_size_called, 0)
+
 
 if __name__ == "__main__":
     unittest.main()