Repository: cogent3/cogent3
Pull request #1036 (2022-04-21 03:07:48)

ENH: new parallel execution option


Patch:
diff --git a/src/cogent3/app/composable.py b/src/cogent3/app/composable.py
index b9fa0cb79..5d7f2dfe5 100644
--- a/src/cogent3/app/composable.py
+++ b/src/cogent3/app/composable.py
@@ -10,13 +10,18 @@ import traceback
 from copy import deepcopy
 from functools import wraps
 
-import scitrack
+from scitrack import CachingLogger
 
 from cogent3 import make_aligned_seqs, make_unaligned_seqs
 from cogent3.core.alignment import SequenceCollection
+from cogent3.util import parallel as PAR
 from cogent3.util import progress_display as UI
 from cogent3.util.io import open_
-from cogent3.util.misc import extend_docstring_from, get_object_provenance
+from cogent3.util.misc import (
+    extend_docstring_from,
+    get_object_provenance,
+    in_jupyter,
+)
 
 from .data_store import (
     IGNORE,
@@ -24,7 +29,6 @@ from .data_store import (
     RAISE,
     SKIP,
     DataStoreMember,
-    SingleReadDataStore,
     WritableDirectoryDataStore,
     get_data_source,
 )
@@ -39,6 +43,8 @@ __maintainer__ = "Gavin Huttley"
 __email__ = "Gavin.Huttley@anu.edu.au"
 __status__ = "Alpha"
 
+from ..util.warning import discontinued
+
 
 def _make_logfile_name(process):
     text = str(process)
@@ -224,12 +230,10 @@ class Composable(ComposableType):
                 v = v.name
             except AttributeError:
                 pass
-            try:
-                get_ipython()
+
+            if in_jupyter():
                 if p == "kwargs" and v == {"store_history": True, "silent": False}:
                     continue
-            except NameError:
-                pass
             formatted.append(f"{p}={v!r}")
         self._formatted += formatted
 
@@ -360,11 +364,11 @@ class Composable(ComposableType):
         self,
         dstore,
         parallel=False,
-        mininterval=2,
         par_kw=None,
-        logger=True,
+        logger=None,
         cleanup=False,
         ui=None,
+        **kwargs,
     ):
         """invokes self composable function on the provided data store
 
@@ -381,15 +385,12 @@ class Composable(ComposableType):
         par_kw
             dict of values for configuring parallel execution.
         logger
-            Argument ignored if not an io.writer. A scitrack logger, a logfile
-            name or True. If True, a scitrack logger is created with a name that
-            defaults to the composable function names and the process ID,
-            e.g. load_unaligned-progressive_align-write_seqs-pid6962.log.
-            If string, that name is used as the logfile name. Otherwise the
-            logger is used as is.
+            Argument ignored if not an io.writer. If a scitrack logger not provided,
+            one is created with a name that defaults to the composable function names
+            and the process ID.
         cleanup : bool
-            after copying of log files into the data store, they are deleted
-            from their original location
+            after copying of log files into the data store, it is deleted
+            from the original location
 
         Returns
         -------
@@ -400,6 +401,9 @@ class Composable(ComposableType):
         If run in parallel, this instance serves as the master object and
         aggregates results.
         """
+        if "mininterval" in kwargs:
+            discontinued("argument", "mininterval", "2022.10", stack_level=1)
+
         if isinstance(dstore, str):
             dstore = [dstore]
 
@@ -407,32 +411,23 @@ class Composable(ComposableType):
         if not dstore:
             raise ValueError("dstore is empty")
 
-        start = time.time()
-        loggable = hasattr(self, "data_store")
-        if (
-            not loggable
-            or type(logger) != scitrack.CachingLogger
-            and type(logger) != str
-            and logger != True
-        ):
-            LOGGER = None
-        elif type(logger) == scitrack.CachingLogger:
-            LOGGER = logger
-            log_file_path = LOGGER.log_file_path
-        elif type(logger) == str:
-            LOGGER = scitrack.CachingLogger()
-            log_file_path = logger
-        else:
-            LOGGER = scitrack.CachingLogger()
-            log_file_path = None
+        am_writer = hasattr(self, "data_store")
+        if am_writer:
+            start = time.time()
+            if logger is None:
+                logger = CachingLogger()
+            elif not isinstance(logger, CachingLogger):
+                raise TypeError(
+                    f"logger must be scitrack.CachingLogger, not {type(logger)}"
+                )
 
-        if LOGGER:
-            if not log_file_path and not LOGGER.log_file_path:
+            if not logger.log_file_path:
                 src = pathlib.Path(self.data_store.source)
-                log_file_path = src.parent / _make_logfile_name(self)
-                LOGGER.log_file_path = str(log_file_path)
-            LOGGER.log_message(str(self), label="composable function")
-            LOGGER.log_versions(["cogent3"])
+                logger.log_file_path = str(src.parent / _make_logfile_name(self))
+
+            log_file_path = str(logger.log_file_path)
+            logger.log_message(str(self), label="composable function")
+            logger.log_versions(["cogent3"])
 
         results = []
         process = self.input or self
@@ -445,62 +440,80 @@ class Composable(ComposableType):
             self.input = None
 
         # with a tinydb dstore, this also excludes data that failed to complete
-        todo = [m for m in dstore if not self.job_done(m)]
-
-        for i, result in enumerate(
-            ui.imap(
-                process, todo, parallel=parallel, par_kw=par_kw, mininterval=mininterval
+        # todo update to consider different database backends
+        # we want a dict mapping input member names to their md5 sums so these can
+        # be logged
+        inputs = {}
+        for m in dstore:
+            input_id = pathlib.Path(
+                m if isinstance(m, DataStoreMember) else get_data_source(m)
+            )
+            suffixes = input_id.suffixes
+            input_id = input_id.name.replace("".join(suffixes), "")
+            inputs[input_id] = m
+
+        if len(inputs) < len(dstore):
+            diff = len(dstore) - len(inputs)
+            raise ValueError(
+                f"could not construct unique identifiers for {diff} records, "
+                "avoid using '.' as a delimiter in names."
             )
-        ):
-            outcome = result if process is self else self(result)
+
+        if parallel:
+            par_kw = par_kw or {}
+            to_do = PAR.as_completed(process, inputs.values(), **par_kw)
+        else:
+            to_do = map(process, inputs.values())
+
+        for result in ui.series(to_do, count=len(inputs)):
+            if process is not self and am_writer:
+                outcome = self.write(data=result)
+                assert isinstance(outcome, DataStoreMember)
+                # if result is NotCompleted, it will be written as such if
+                # supported by data store backend and the outcome is just the
+                # associated db identifier for tracking steps below we need to
+                # know it's NotCompleted
+                if result:
+                    input_id = outcome.name
+                else:
+                    input_id = get_data_source(result)
+                    outcome = result
+                input_id = pathlib.Path(pathlib.Path(input_id))
+                suffixes = input_id.suffixes
+                input_id = input_id.name.replace("".join(suffixes), "")
+            else:
+                outcome = result
+
             results.append(outcome)
-            if LOGGER:
-                member = todo[i]
-                # ensure member is a DataStoreMember instance
-                if not isinstance(member, DataStoreMember):
-                    member = SingleReadDataStore(member)[0]
-
-                mem_id = self.data_store.make_relative_identifier(member.name)
-                # src points the origins of the data, which can differ from
-                # the db identifier but share in common the name prefix.
-                # todo this is very fragile and needs to be thoroughly refactored
-                src = self.data_store.make_relative_identifier(result)
-                assert (
-                    src.split(".")[0] == mem_id.split(".")[0]
-                ), f"mismatched input data and result identifiers: {src} != {mem_id}"
-
-                LOGGER.log_message(member, label="input")
-                if member.md5:
-                    LOGGER.log_message(member.md5, label="input md5sum")
+
+            if am_writer:
+                # now need to search for the source member
+                m = inputs[input_id]
+                input_md5 = getattr(m, "md5", None)
+                logger.log_message(input_id, label="input")
+                if input_md5:
+                    logger.log_message(input_md5, label="input md5sum")
 
                 if isinstance(outcome, NotCompleted):
-                    try:
-                        # tinydb supports storage
-                        self.data_store.write_incomplete(mem_id, outcome)
-                    except AttributeError:
-                        pass
-                    LOGGER.log_message(
+                    # log error/fail details
+                    logger.log_message(
                         f"{outcome.origin} : {outcome.message}", label=outcome.type
                     )
                     continue
+                elif not outcome:
+                    # other cases where outcome is Falsy (e.g. None)
+                    logger.log_message(f"unexpected value {outcome!r}", label="FAIL")
+                    continue
 
-                if (
-                    isinstance(outcome, DataStoreMember)
-                    and outcome.parent is self.data_store
-                ):
-                    member = outcome
-                else:
-                    member = self.data_store.get_member(mem_id)
+                logger.log_message(outcome, label="output")
+                logger.log_message(outcome.md5, label="output md5sum")
 
-                LOGGER.log_message(member, label="output")
-                LOGGER.log_message(member.md5, label="output md5sum")
+        if am_writer:
+            finish = time.time()
+            taken = finish - start
 
-        finish = time.time()
-        taken = finish - start
-        if LOGGER:
-            LOGGER.log_message(f"{taken}", label="TIME TAKEN")
-            LOGGER.shutdown()
-            log_file_path = str(log_file_path)
+            logger.log_message(f"{taken}", label="TIME TAKEN")
+            logger.shutdown()
             self.data_store.add_file(log_file_path, cleanup=cleanup, keep_suffix=True)
             self.data_store.close()
 
diff --git a/src/cogent3/app/data_store.py b/src/cogent3/app/data_store.py
index 7dd2f123d..516b76f0f 100644
--- a/src/cogent3/app/data_store.py
+++ b/src/cogent3/app/data_store.py
@@ -615,6 +615,9 @@ class WritableDirectoryDataStore(ReadOnlyDirectoryDataStore, WritableDataStoreBa
 
     @extend_docstring_from(WritableDataStoreBase.write)
     def write(self, identifier, data):
+        if not data:
+            return data
+
         super().write(identifier, data)
         id_suffix = identifier.split(".")[-1]
         if id_suffix not in (self.suffix, "log"):
@@ -976,6 +979,11 @@ class WritableTinyDbDataStore(ReadOnlyTinyDbDataStore, WritableDataStoreBase):
     def write(self, identifier, data):
         # writing into a tinydb has its own logic for conversion to json
         #  so we don't validate data is a string for this case
+        from cogent3.app.composable import NotCompleted
+
+        if isinstance(data, NotCompleted):
+            return self.write_incomplete(identifier, data)
+
         super().write(identifier, "")
         id_suffix = identifier.split(".")[-1]
         if id_suffix not in (self.suffix, "log"):
diff --git a/src/cogent3/app/io.py b/src/cogent3/app/io.py
index ca2f3f7ba..158d6c67e 100644
--- a/src/cogent3/app/io.py
+++ b/src/cogent3/app/io.py
@@ -428,8 +428,13 @@ class write_tabular(_checkpointable, ComposableTabular):
         self._format = format
 
     def write(self, data, identifier=None):
+        if not data:
+            msg = f"{self.__class__.__name__!r} does not support writing {data!r}"
+            raise NotImplementedError(msg)
+
         if identifier is None:
             identifier = self._make_output_identifier(data)
+
         output = data.to_string(format=self._format)
         self.data_store.write(identifier, output)
         return identifier
@@ -490,8 +495,13 @@ class write_seqs(_checkpointable):
         self._load_checkpoint = loader
 
     def write(self, data, identifier=None):
+        if not data:
+            msg = f"{self.__class__.__name__!r} does not support writing {data!r}"
+            raise NotImplementedError(msg)
+
         if identifier is None:
             identifier = self._make_output_identifier(data)
+
         data.info.stored = self.data_store.write(identifier, data.to_fasta())
         return identifier
 
@@ -557,8 +567,18 @@ class write_json(_checkpointable):
         self._load_checkpoint = self
 
     def write(self, data, identifier=None):
+        from cogent3.app.composable import NotCompleted
+
+        if isinstance(data, NotCompleted):
+            return self.data_store.write_incomplete(identifier, data)
+
+        if not data:
+            msg = f"{self.__class__.__name__!r} does not support writing {data!r}"
+            raise NotImplementedError(msg)
+
         if identifier is None:
             identifier = self._make_output_identifier(data)
+
         out = make_record_for_json(os.path.basename(identifier), data, True)
         out = json.dumps(out)
         stored = self.data_store.write(identifier, out)
@@ -652,22 +672,27 @@ class write_db(_checkpointable):
         -------
         identifier
         """
-        # todo skip next line if identifier provided
-        data_source = get_data_source(data)  # todo -- this is being ignored!
-        if (data_source and identifier is not None) and str(data_source) != str(
-            identifier
-        ):
-            raise ValueError(f"identifier {identifier} != data source {data_source}")
+        from cogent3.app.composable import NotCompleted
 
-        if identifier is None:
-            identifier = self._make_output_identifier(data)
+        identifier = identifier or get_data_source(data)
+        identifier = self._make_output_identifier(identifier)
+
+        if isinstance(data, NotCompleted):
+            return self.data_store.write_incomplete(identifier, data)
+
+        if not data:
+            msg = f"{self.__class__.__name__!r} does not support writing {data!r}"
+            raise NotImplementedError(msg)
+
+        identifier = identifier or get_data_source(data)
+        identifier = self._make_output_identifier(identifier)
         # todo revisit this when we establish immutability behaviour of database
         try:
             out = data.to_json()
         except AttributeError:
             out = json.dumps(data)
         stored = self.data_store.write(identifier, out)
-        # todo is anything actually using this stored attriubte? if not, delete this
+        # todo is anything actually using this stored attribute? if not, delete this
         #  code and all other cases
         if hasattr(data, "info"):
             data.info["stored"] = stored
@@ -676,4 +701,4 @@ class write_db(_checkpointable):
                 data.stored = stored
             except AttributeError:
                 pass
-        return identifier
+        return stored
diff --git a/src/cogent3/util/misc.py b/src/cogent3/util/misc.py
index 9266b2614..b83388af7 100644
--- a/src/cogent3/util/misc.py
+++ b/src/cogent3/util/misc.py
@@ -1101,3 +1101,8 @@ def get_setting_from_environ(environ_var, params_types):
             )
 
     return result
+
+
+def in_jupyter() -> bool:
+    """whether code is being executed within a jupyter notebook"""
+    return callable(globals().get("get_ipython"))
diff --git a/src/cogent3/util/parallel.py b/src/cogent3/util/parallel.py
index f18c8bd8c..a5edba896 100644
--- a/src/cogent3/util/parallel.py
+++ b/src/cogent3/util/parallel.py
@@ -86,7 +86,7 @@ class PicklableAndCallable:
         return self.func(*args, **kw)
 
 
-def set_default_chunksize(s, max_workers):
+def get_default_chunksize(s, max_workers):
     chunksize, remainder = divmod(len(s), max_workers * 4)
     if remainder:
         chunksize += 1
@@ -110,12 +110,13 @@ def imap(f, s, max_workers=None, use_mpi=False, if_serial="raise", chunksize=Non
         values are 'raise', 'ignore', 'warn'. Defaults to 'raise'.
     chunksize : int or None
         Size of data chunks executed by worker processes. Defaults to None
-        where stable chunksize is determined by set_default_chunksize()
+        where stable chunksize is determined by get_default_chunksize()
 
     Returns
     -------
-    imap is a generator yielding result of f(s[i]), map returns the result
-    series
+    imap and as_completed are generators yielding result of f(s[i]), map returns the result
+    series. imap and map return results in the same order as s, as_completed returns results
+    in the order completed (which can differ from the order in s).
 
     Notes
     -----
@@ -135,15 +136,16 @@ def imap(f, s, max_workers=None, use_mpi=False, if_serial="raise", chunksize=Non
         if not USING_MPI:
             raise RuntimeError("Cannot use MPI")
 
-        err_msg = (
-            "Execution in serial. For parallel MPI execution, use:\n"
-            " $ mpiexec -n <number CPUs> python3 -m mpi4py.futures <executable script>"
-        )
+        if COMM.Get_attr(MPI.UNIVERSE_SIZE) == 1:
+            err_msg = (
+                "Execution in serial. For parallel MPI execution, use:\n"
+                " $ mpiexec -n <number CPUs> python3 -m mpi4py.futures <executable script>"
+            )
 
-        if COMM.Get_attr(MPI.UNIVERSE_SIZE) == 1 and if_serial == "raise":
-            raise RuntimeError(err_msg)
-        elif COMM.Get_attr(MPI.UNIVERSE_SIZE) == 1 and if_serial == "warn":
-            warnings.warn(err_msg, UserWarning)
+            if if_serial == "raise":
+                raise RuntimeError(err_msg)
+            elif if_serial == "warn":
+                warnings.warn(err_msg, UserWarning)
 
         max_workers = max_workers or 1
 
@@ -154,26 +156,85 @@ def imap(f, s, max_workers=None, use_mpi=False, if_serial="raise", chunksize=Non
 
         max_workers = min(max_workers, COMM.Get_attr(MPI.UNIVERSE_SIZE) - 1)
         if not chunksize:
-            chunksize = set_default_chunksize(s, max_workers)
+            chunksize = get_default_chunksize(s, max_workers)
 
         with MPIfutures.MPIPoolExecutor(max_workers=max_workers) as executor:
-            for result in executor.map(f, s, chunksize=chunksize):
-                yield result
+            yield from executor.map(f, s, chunksize=chunksize)
     else:
         if not max_workers:
             max_workers = multiprocessing.cpu_count() - 1
         assert max_workers < multiprocessing.cpu_count()
 
         if not chunksize:
-            chunksize = set_default_chunksize(s, max_workers)
+            chunksize = get_default_chunksize(s, max_workers)
 
         f = PicklableAndCallable(f)
 
         with concurrentfutures.ProcessPoolExecutor(max_workers) as executor:
-            for result in executor.map(f, s, chunksize=chunksize):
-                yield result
+            yield from executor.map(f, s, chunksize=chunksize)
 
 
 @extend_docstring_from(imap)
 def map(f, s, max_workers=None, use_mpi=False, if_serial="raise", chunksize=None):
     return list(imap(f, s, max_workers, use_mpi, if_serial, chunksize))
+
+
+def _as_completed_mpi(f, s, max_workers, if_serial, chunksize=None):
+    """MPI version of as_completed"""
+    if not USING_MPI:
+        raise RuntimeError("Cannot use MPI")
+
+    if COMM.Get_attr(MPI.UNIVERSE_SIZE) == 1:
+        err_msg = (
+            "Execution in serial. For parallel MPI execution, use:\n"
+            " $ mpiexec -n <number CPUs> python3 -m mpi4py.futures <executable script>"
+        )
+
+        if if_serial == "raise":
+            raise RuntimeError(err_msg)
+        elif if_serial == "warn":
+            warnings.warn(err_msg, UserWarning)
+
+    max_workers = max_workers or 1
+
+    f = PicklableAndCallable(f)
+
+    if max_workers > COMM.Get_attr(MPI.UNIVERSE_SIZE):
+        warnings.warn("max_workers too large, reducing to UNIVERSE_SIZE-1", UserWarning)
+
+    max_workers = min(max_workers, COMM.Get_attr(MPI.UNIVERSE_SIZE) - 1)
+    if not chunksize:
+        chunksize = get_default_chunksize(s, max_workers)
+
+    with MPIfutures.MPIPoolExecutor(
+        max_workers=max_workers, chunksize=chunksize
+    ) as executor:
+        to_do = [executor.submit(f, e) for e in s]
+        for result in concurrentfutures.as_completed(to_do):
+            yield result.result()
+
+
+def _as_completed_mproc(f, s, max_workers):
+    """multiprocess version of as_completed"""
+    if not max_workers:
+        max_workers = multiprocessing.cpu_count() - 1
+    assert max_workers < multiprocessing.cpu_count()
+
+    f = PicklableAndCallable(f)
+
+    with concurrentfutures.ProcessPoolExecutor(max_workers=max_workers) as executor:
+        to_do = [executor.submit(f, e) for e in s]
+        for result in concurrentfutures.as_completed(to_do):
+            yield result.result()
+
+
+@extend_docstring_from(imap, pre=True)
+def as_completed(
+    f, s, max_workers=None, use_mpi=False, if_serial="raise", chunksize=None
+):
+    if_serial = if_serial.lower()
+    assert if_serial in ("ignore", "raise", "warn"), f"invalid choice '{if_serial}'"
+    if use_mpi:
+        yield from _as_completed_mpi(f, s, max_workers, if_serial, chunksize)
+    else:
+        yield from _as_completed_mproc(f, s, max_workers)
diff --git a/src/cogent3/util/progress_display.py b/src/cogent3/util/progress_display.py
index e38808f21..1fc9e0d24 100644
--- a/src/cogent3/util/progress_display.py
+++ b/src/cogent3/util/progress_display.py
@@ -18,6 +18,8 @@ __maintainer__ = "Gavin Huttley"
 __email__ = "gavin.huttley@anu.edu.au"
 __status__ = "Alpha"
 
+from cogent3.util.misc import in_jupyter
+
 
 class LogFileOutput:
     """A fake progress bar for when progress bars are impossible"""
@@ -124,7 +126,7 @@ class ProgressContext:
         self.display(progress=end)
 
     def write(self, *args, **kw):
-        if self.progress_bar_type and len(kw) < 3 and not using_notebook():
+        if self.progress_bar_type and len(kw) < 3 and not in_jupyter():
             self.progress_bar_type.write(*args, **kw)
         else:
             print(*args, **kw)
@@ -163,14 +165,6 @@ CURRENT = threading.local()
 CURRENT.context = None
 
 
-def using_notebook():
-    try:
-        get_ipython()
-        return True
-    except NameError:
-        return False
-
-
 def display_wrap(slow_function):
     """Decorator which give the function its own UI context.
     The function will receive an extra argument, 'ui',
@@ -181,7 +175,7 @@ def display_wrap(slow_function):
         if getattr(CURRENT, "context", None) is None:
             if sys.stdout.isatty():
                 klass = tqdm
-            elif using_notebook():
+            elif in_jupyter():
                 klass = notebook.tqdm
             elif isinstance(sys.stdout, io.FileIO):
                 klass = LogFileOutput


Test patch:
diff --git a/tests/test_app/test_app_mpi.py b/tests/test_app/test_app_mpi.py
index f2b542804..1146d3834 100644
--- a/tests/test_app/test_app_mpi.py
+++ b/tests/test_app/test_app_mpi.py
@@ -35,7 +35,6 @@ class MPITests(TestCase):
 
             r = process.apply_to(
                 members,
-                logger=False,
                 show_progress=False,
                 parallel=True,
                 par_kw=dict(use_mpi=True),
diff --git a/tests/test_app/test_composable.py b/tests/test_app/test_composable.py
index c44ca8777..19365746b 100644
--- a/tests/test_app/test_composable.py
+++ b/tests/test_app/test_composable.py
@@ -111,7 +111,7 @@ class TestComposableBase(TestCase):
         self.assertEqual(len(got), len(dstore))
         # should also be able to apply the results to another composable func
         min_length = sample_app.min_length(10)
-        got = min_length.apply_to(got, show_progress=False, logger=True)
+        got = min_length.apply_to(got, show_progress=False)
         self.assertEqual(len(got), len(dstore))
         # should work on a chained function
         proc = reader + min_length
@@ -147,27 +147,30 @@ class TestComposableBase(TestCase):
             self.assertEqual(len(process.data_store.logs), 1)
             process.data_store.close()
 
-    def test_apply_to_logging(self):
-        """correctly creates log file"""
-        dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
+    def test_apply_to_non_unique_identifiers(self):
+        """should fail if non-unique names"""
+        dstore = [
+            "brca1.bats.fasta",
+            "brca1.apes.fasta",
+        ]
         with TemporaryDirectory(dir=".") as dirname:
             reader = io_app.load_aligned(format="fasta", moltype="dna")
             min_length = sample_app.min_length(10)
-            outpath = os.path.join(os.getcwd(), dirname, "delme.tinydb")
-            writer = io_app.write_db(outpath)
-            process = reader + min_length + writer
-            r = process.apply_to(dstore, show_progress=False, logger=False)
-            self.assertEqual(len(process.data_store.logs), 0)
-            process.data_store.close()
+            process = reader + min_length
+            with self.assertRaises(ValueError):
+                process.apply_to(dstore)
 
+    def test_apply_to_logging(self):
+        """correctly creates log file"""
+        dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
         with TemporaryDirectory(dir=".") as dirname:
             reader = io_app.load_aligned(format="fasta", moltype="dna")
-            # trigger creation of notcompleted
             min_length = sample_app.min_length(10)
             outpath = os.path.join(os.getcwd(), dirname, "delme.tinydb")
             writer = io_app.write_db(outpath)
             process = reader + min_length + writer
-            r = process.apply_to(dstore, show_progress=False, logger=True)
+            r = process.apply_to(dstore, show_progress=False)
+            # always creates a log
             self.assertEqual(len(process.data_store.logs), 1)
             process.data_store.close()
 
@@ -185,6 +188,20 @@ class TestComposableBase(TestCase):
             self.assertEqual(len(process.data_store.logs), 1)
             process.data_store.close()
 
+    def test_apply_to_invalid_logger(self):
+        """incorrect logger value raises TypeError"""
+        dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
+        for logger_val in (True, "somepath.log"):
+            with TemporaryDirectory(dir=".") as dirname:
+                reader = io_app.load_aligned(format="fasta", moltype="dna")
+                min_length = sample_app.min_length(10)
+                outpath = os.path.join(os.getcwd(), dirname, "delme.tinydb")
+                writer = io_app.write_db(outpath)
+                process = reader + min_length + writer
+                with self.assertRaises(TypeError):
+                    process.apply_to(dstore, show_progress=False, logger=logger_val)
+                process.data_store.close()
+
     def test_apply_to_not_completed(self):
         """correctly creates notcompleted"""
         dstore = io_app.get_data_store("data", suffix="fasta", limit=3)
diff --git a/tests/test_app/test_data_store.py b/tests/test_app/test_data_store.py
index 9708ae765..ec8f0ac73 100644
--- a/tests/test_app/test_data_store.py
+++ b/tests/test_app/test_data_store.py
@@ -447,8 +447,18 @@ class DirectoryDataStoreReadTests(
             with self.assertRaises(FileNotFoundError):
                 self.WriteClass(path, suffix=".json", create=False)
 
-            # correctly creates tinydb when full path does not exist
-            _ = self.WriteClass(path, suffix=".json", create=True)
+    def test_write_not_completed(self):
+        """directory data store ignores"""
+        with TemporaryDirectory(dir=".") as dirname:
+            # tests the case when the directory has the file with the same suffix to self.suffix
+            from cogent3.app.composable import NotCompleted
+
+            with TemporaryDirectory(dir=".") as dirname:
+                path = Path(dirname) / "subdir"
+                writer = self.WriteClass(path, suffix=".fasta", create=True)
+                nc = NotCompleted("FAIL", "test", "dummy fail", source="blah.json")
+                got = writer.write(nc.source, nc)
+                assert got is nc
 
 
 class ZippedDataStoreReadTests(TestCase, DataStoreBaseReadTests):
@@ -635,7 +645,7 @@ class TinyDBDataStoreTests(TestCase):
             path = os.path.join(dirname, self.basedir)
             dstore = self.WriteClass(path, if_exists="overwrite")
             id_ = dstore.make_relative_identifier(incomplete[0])
-            dstore.write_incomplete(id_, incomplete[1])
+            dstore.write(id_, incomplete[1])
             for k in keys:
                 id_ = dstore.make_relative_identifier(k)
                 dstore.write(id_, self.data[k])
diff --git a/tests/test_app/test_io.py b/tests/test_app/test_io.py
index 1d07ac33c..214584944 100644
--- a/tests/test_app/test_io.py
+++ b/tests/test_app/test_io.py
@@ -239,16 +239,6 @@ class TestIo(TestCase):
             dstore.close()
             self.assertEqual(got, data)
 
-    def test_write_db_invalid(self):
-        """value error if identifier does not match data.info.source"""
-        with TemporaryDirectory(dir=".") as dirname:
-            outpath = join(dirname, "delme")
-            writer = write_db(outpath, create=True, if_exists="ignore")
-            data = UnionDict(a=[1, 2], b="string", source="delme2.json")
-            got = writer(data, identifier=join("blah", "delme.json"))
-            self.assertTrue("ValueError" in got.message)
-            writer.data_store.db.close()
-
     def test_load_db_failure_json_file(self):
         """informative load_db error message when given a json file path"""
         # todo this test has a trapped exception about being unable to delete
@@ -522,7 +512,7 @@ class TestIo(TestCase):
         writer = write_db("delme.tinydb", create=True, if_exists="overwrite")
         process = reader + aligner + writer
 
-        r = process.apply_to(members, logger=False, show_progress=False, parallel=True)
+        r = process.apply_to(members, show_progress=False, parallel=True)
 
         expect = [str(m) for m in process.data_store]
         process.data_store.close()