Avoid explicit CASA read locking by opening tables in 'usernoread' mode

daskms/reads.py

@@ -306,7 +306,7 @@ def __init__(self, table, select_cols, group_cols, index_cols, **kwargs):
 
     def _table_proxy_factory(self):
         return TableProxy(pt.table, self.table_path, ack=False,
-                          readonly=True, lockoptions='user',
+                          readonly=True, lockoptions="usernoread",
                           __executor_key__=executor_key(self.canonical_name))
 
     def _table_schema(self):

daskms/table_proxy.py

@@ -82,29 +82,14 @@ def proxied_method_factory(method, locktype):
     on a concurrent.futures.ThreadPoolExecutor.
     """
 
-    if locktype == NOLOCK:
+    if locktype in (NOLOCK, READLOCK):
         def _impl(table_future, args, kwargs):
             try:
                 return getattr(table_future.result(), method)(*args, **kwargs)
             except Exception:
                 if logging.DEBUG >= log.getEffectiveLevel():
                     log.exception("Exception in %s", method)
                 raise
-
-    elif locktype == READLOCK:
-        def _impl(table_future, args, kwargs):
-            table = table_future.result()
-            table.lock(write=False)
-
-            try:
-                return getattr(table, method)(*args, **kwargs)
-            except Exception:
-                if logging.DEBUG >= log.getEffectiveLevel():
-                    log.exception("Exception in %s", method)
-                raise
-            finally:
-                table.unlock()
-
     elif locktype == WRITELOCK:
         def _impl(table_future, args, kwargs):
             table = table_future.result()
@@ -177,21 +162,19 @@ def taql_factory(query, style='Python', tables=(), readonly=True):
     tables = [t._table_future.result() for t in tables]
 
     if isinstance(readonly, (tuple, list)):
-        it = zip_longest(tables, readonly[:len(tables)],
-                         fillvalue=readonly[-1])
+        it = list(zip_longest(tables, readonly[:len(tables)],
+                              fillvalue=readonly[-1]))
     elif isinstance(readonly, bool):
-        it = zip(tables, (readonly,)*len(tables))
+        it = list(zip(tables, (readonly,)*len(tables)))
     else:
         raise TypeError("readonly must be a bool or list of bools")
 
-    for t, ro in it:
-        t.lock(write=ro is False)
+    # See https://github.com/ska-sa/dask-ms/pull/141
+    if any(ro is False for ro in it):
+        raise ValueError(f"write-locking requested for '{query}'"
+                         f"but locking has been removed!")
 
-    try:
-        return pt.taql(query, style=style, tables=tables)
-    finally:
-        for t in tables:
-            t.unlock()
+    return pt.taql(query, style=style, tables=tables)
 
 
 def _nolock_runner(table_future, fn, args, kwargs):
@@ -203,20 +186,6 @@ def _nolock_runner(table_future, fn, args, kwargs):
         raise
 
 
-def _readlock_runner(table_future, fn, args, kwargs):
-    table = table_future.result()
-    table.lock(write=False)
-
-    try:
-        return fn(table, *args, **kwargs)
-    except Exception:
-        if logging.DEBUG >= log.getEffectiveLevel():
-            log.exception("Exception in %s", fn.__name__)
-        raise
-    finally:
-        table.unlock()
-
-
 def _writelock_runner(table_future, fn, args, kwargs):
     table = table_future.result()
     table.lock(write=True)
@@ -391,12 +360,9 @@ def submit(self, fn, locktype, *args, **kwargs):
         future : :class:`concurrent.futures.Future`
             Future containing the result of :code:`fn(table, *args, **kwargs)`
         """
-        if locktype == NOLOCK:
+        if locktype in (NOLOCK, READLOCK):
             return self._ex.submit(_nolock_runner, self._table_future,
                                    fn, args, kwargs)
-        elif locktype == READLOCK:
-            return self._ex.submit(_readlock_runner, self._table_future,
-                                   fn, args, kwargs)
         elif locktype == WRITELOCK:
             return self._ex.submit(_writelock_runner, self._table_future,
                                    fn, args, kwargs)

daskms/tests/test_ordering.py

@@ -16,7 +16,7 @@
 
 def table_proxy(ms):
     return TableProxy(pt.table, ms, ack=False,
-                      lockoptions='user', readonly=True)
+                      lockoptions='usernoread', readonly=True)
 
 
 @pytest.mark.parametrize("group_cols", [

daskms/writes.py

@@ -230,7 +230,7 @@ def descriptor_builder(table, descriptor):
 
 def _writable_table_proxy(table_name):
     return TableProxy(pt.table, table_name, ack=False,
-                      readonly=False, lockoptions='user',
+                      readonly=False, lockoptions="usernoread",
                       __executor_key__=executor_key(table_name))