Adding a metadata flag for signaling successful ingestion (#140)

* Adding a metadata flag for signaling successful ingestion

* Removing the valid metadata KV on each run before termination

Author: ktsitsi

tests/integration/converters/test_ome_tiff.py

@@ -297,3 +297,28 @@ def test_ome_tiff_converter_different_compressors(
             with open_bioimg(str(tmp_path / "l_1.tdb")) as A:
                 assert len(A.schema.attr(0).filters) == 1
                 assert isinstance(A.schema.attr(0).filters[0], type(single_compressor))
+
+
+@pytest.mark.parametrize(
+    "filename", ["CMU-1-Small-Region.ome.tiff", "CMU-1-Small-Region-rgb.ome.tiff"]
+)
+def test_valid_ingestion_check(tmp_path, filename):
+    input_path = str(get_path(filename))
+    output_path = str(tmp_path)
+    OMETiffConverter.to_tiledb(input_path, output_path, compressor=None)
+
+    with tiledb.Group(output_path, "r") as grp:
+        # Check that ingestion was successful
+        assert grp.meta.get("valid")
+
+    # Manually change the valid KV simulating ingestion was not completed
+    with tiledb.Group(output_path, "w") as grp:
+        grp.meta.update(valid=False)
+
+    # Initialize ingestion again - Simulating a retry on the same destination path
+    OMETiffConverter.to_tiledb(input_path, output_path, compressor=None)
+
+    # Check that ingestion was successful after the group was overwritten
+    with tiledb.Group(output_path, "r") as grp:
+        # Check that ingestion was successful
+        assert grp.meta.get("valid")

tiledb/bioimg/converters/base.py

@@ -410,6 +410,26 @@ def to_tiledb(
         metadata = {}
         original_metadata = {}
 
+        # Checks group validity
+        overwrite = False
+        with rw_group:
+            if not rw_group.valid_group:
+                logger.debug(
+                    f"Already existing on disk Group at {output_path} evaluated as corrupted."
+                )
+                # Overwrite destination group
+                rw_group.m_group.delete(recursive=True)
+                overwrite = True
+            else:
+                # If group is valid we remove the KV and we will re-introduce it
+                # after the current ingestion process as the final step
+                del rw_group.w_group.meta["valid"]
+
+        if overwrite:
+            # Re-initializes the destination Group for re-ingestion
+            logger.debug(f"Group {output_path} will be overwritten.")
+            rw_group = ReadWriteGroup(output_path, ctx=reader.dest_ctx)
+
         with rw_group, reader:
             # Update MIME type
             rw_group.w_group.meta.update(dataset_type=DATASET_TYPE)
@@ -541,6 +561,11 @@ def to_tiledb(
                 metadata=jsonpickle.encode(metadata, unpicklable=False),
                 original_metadata=jsonpickle.encode(original_metadata),
             )
+
+        # This is the last step of the ingestion unless the following metadata KV
+        # is not stored then we consider the ingested asset as corrupted
+        with rw_group:
+            rw_group.w_group.meta.update(valid=True)
         return cls
 
 

tiledb/bioimg/helpers.py

@@ -41,16 +41,29 @@ def __init__(self, uri: str, ctx: Optional[Ctx] = None):
         # Windows paths produce single letter scheme matching the drive letter
         # Unix absolute path produce an empty scheme
         self._ctx = ctx
+        self._group_exists = False
+
+        # At construction time we consider it as valid, this value will be valuated upon
+        # opening the Group under a context
+        self._valid_group = True
         if len(parsed_uri.scheme) < 2 or parsed_uri.scheme == "file":
             uri = str(Path(parsed_uri.path).resolve()).replace("\\", "/")
         if tiledb.object_type(uri, ctx=ctx) != "group":
             tiledb.group_create(uri, ctx=ctx)
+        else:
+            # The destination path is already a group thus we check the validity of
+            # a previously successful ingestion by checking the existence of
+            # the ACK metadata KV
+            self._group_exists = True
         self._uri = uri if uri.endswith("/") else uri + "/"
         self._is_cloud = parsed_uri.scheme == "tiledb"
 
     def __enter__(self) -> ReadWriteGroup:
         self.r_group = tiledb.Group(self._uri, "r", ctx=self._ctx)
         self.w_group = tiledb.Group(self._uri, "w", ctx=self._ctx)
+        self.m_group = tiledb.Group(self._uri, "m", ctx=self._ctx)
+        if self._group_exists:
+            self._valid_group = self.r_group.meta.get("valid", False)
         return self
 
     def __exit__(self, exc_type: Any, exc_val: Any, exc_tb: Any) -> None:
@@ -93,6 +106,10 @@ def get_or_create(self, name: str, schema: tiledb.ArraySchema) -> Tuple[str, boo
                 self.w_group.add(name, name, relative=True)
         return uri, create
 
+    @property
+    def valid_group(self) -> bool:
+        return self._valid_group
+
 
 def validate_ingestion(uri: str, ctx: tiledb.Ctx = None) -> bool:
     """