added compression completion check and fixed cottas docker error

Author: ecrum19

COMPRESSION_CHANGELOG.md

@@ -146,6 +146,17 @@ The Docker image installs `pycottas` in `/opt/pycottas-venv` and invokes the
 small adapter at `/opt/vcf-rdfizer/cottas_tool.py`. COTTAS is therefore part of
 the Docker workflow, not a dependency of the lightweight pip/conda wrapper.
 
+The adapter runs every `pycottas.rdf2cottas` and `pycottas.cat` operation in a
+new temporary subdirectory of the container-local `/work` filesystem. pycottas
+uses a default `pycottas.duckdb` database in its current working directory;
+without isolation, a later chunk can reopen the prior database and fail while
+creating its existing `quads` table. Input and output paths are resolved before
+entering the temporary directory, so COTTAS artifacts remain in the designated
+workspace/output mount while the DuckDB database and related scratch files are
+removed immediately after each operation. `COTTAS_SCRATCH_DIR` can override
+`/work` for image-internal deployments, but the wrapper does not require users
+to set it.
+
 ## Metrics and Cleanup
 
 Every chunk conversion, merge, index-generation, and final representation
@@ -162,6 +173,30 @@ default; `--remove-rdf-storage-output` makes aggregate removal explicit, while
 `--keep-rmlstreamer-rdf-output` retains the aggregate produced by RMLStreamer.
 The two flags are mutually exclusive.
 
+### Representation Validation
+
+Every generated base HDT or COTTAS representation is validated before any
+packaging stage and before raw RDF cleanup. The validation has two parts:
+
+1. The artifact is opened and decoded with its native reader. HDT is loaded
+   through `ensure_hdt_index.sh`/HDT Java and streamed through `hdt2rdf`;
+   COTTAS is streamed through `pycottas.cottas2rdf`.
+2. The decoded N-Triples stream is counted and compared with the source count.
+
+Full mode passes the `output_triples` count recorded by RMLStreamer whenever
+it is available. The partition planner already reads the source once, so its
+record count is reused for validation. Compression-only mode has no upstream
+conversion metric; in that case the validator counts plain or gzip-compressed
+N-Triples directly as a fallback. No decoded RDF file is written during this
+check.
+
+A missing decoder, unreadable artifact, or count mismatch makes the
+compression stage fail and prevents RDF cleanup. Reports are stored in the
+method details of the raw compression JSON and expose `source_triples`,
+`decoded_triples`, `count_match`, and `valid`. The aggregate `metrics.csv`
+also records the source count, decoded count, and validation status for HDT
+and COTTAS.
+
 ## Compatibility and Limits
 
 Full mode requires `--rdf-storage-mode plain` or

README.md

@@ -353,6 +353,16 @@ Compression metrics now include per-method:
 - `sys_seconds_*`
 - `max_rss_kb_*`
 
+When HDT or COTTAS is selected, compression also validates the final base
+artifact before packaging or RDF cleanup. The validator reads the source
+triple count, streams the artifact back through the native decoder, and
+requires equal counts. HDT validation also initializes the `.hdt.index`
+sidecar. Validation results and `source_triples`/`decoded_triples` are stored
+in the per-run compression JSON and in the HDT/COTTAS columns of `metrics.csv`.
+Compression fails closed if the artifact cannot be decoded or the counts do
+not match. In compression-only mode, the source count is obtained by a
+streaming fallback when no upstream conversion metrics are available.
+
 For partitioned HDT/COTTAS runs, the final method metric reports one
 sample-level result, while raw metrics also include a sample-scoped
 `__partitioned_compression__` artifact describing chunk conversion, merge
@@ -377,11 +387,20 @@ HDT index is generated after merging through HDT Java's indexed search loader,
 and COTTAS chunks are merged with `pycottas.cat`, which rebuilds the query
 indexes for the merged representation.
 
+After each final HDT/COTTAS base artifact is produced, VCF-RDFizer performs a
+streaming decode/count check. This verifies both readability and that the
+decoded artifact contains exactly the number of source triples. The check is
+performed before `.hdt.gz`, `.hdt.br`, `.cottas.gz`, or `.cottas.br` packaging,
+and before raw RDF cleanup.
+
 Partitioned HDT/COTTAS compression runs in an ephemeral Docker-managed
 workspace. Temporary RDF chunks, COTTAS/DuckDB scratch data, intermediate
 representations, and merge files are not written to the output directory.
 After a successful or failed run, the temporary workspace is removed; only
 the selected final artifacts and normal run metrics remain on the host.
+Each COTTAS conversion and merge also receives a fresh container-local DuckDB
+workspace, which is removed as soon as that operation completes. This prevents
+state from one chunk being reused by another and requires no user configuration.
 
 HDT Java 3.0.10 does not provide a standalone `hdtGenerateIndex` executable.
 VCF-RDFizer sends an `exit` command to the supported `hdtSearch.sh` launcher;

src/cottas_tool.py

@@ -2,7 +2,29 @@
 """Small Docker-side adapter for the pycottas conversion and merge API."""
 
 import argparse
+import os
 import sys
+import tempfile
+from contextlib import contextmanager
+from pathlib import Path
+
+
+@contextmanager
+def cottas_scratch_workspace():
+    """Run one pycottas operation with an isolated DuckDB working directory."""
+    scratch_root = Path(os.environ.get("COTTAS_SCRATCH_DIR", "/work")).resolve()
+    scratch_root.mkdir(parents=True, exist_ok=True)
+    original_working_directory = Path.cwd()
+    # pycottas defaults to ``pycottas.duckdb`` in the current directory.
+    # A fresh directory prevents one chunk from reusing another chunk's
+    # database, whose ``quads`` table already exists.
+    with tempfile.TemporaryDirectory(prefix="vcf-rdfizer-cottas-", dir=scratch_root) as directory:
+        try:
+            os.chdir(directory)
+            yield
+        finally:
+            # Leave the directory before TemporaryDirectory removes it.
+            os.chdir(original_working_directory)
 
 
 def main() -> int:
@@ -28,22 +50,29 @@ def main() -> int:
         return 127
 
     if args.command == "convert":
+        rdf_path = str(Path(args.rdf_path).resolve())
+        cottas_path = str(Path(args.cottas_path).resolve())
         # disk=True keeps parser/index construction from requiring the whole
         # RDF chunk in Python memory.
-        pycottas.rdf2cottas(
-            args.rdf_path,
-            args.cottas_path,
-            index=args.index,
-            disk=True,
-        )
+        with cottas_scratch_workspace():
+            pycottas.rdf2cottas(
+                rdf_path,
+                cottas_path,
+                index=args.index,
+                disk=True,
+            )
         return 0
 
-    pycottas.cat(
-        [args.left_path, args.right_path],
-        args.cottas_path,
-        index=args.index,
-        remove_input_files=True,
-    )
+    left_path = str(Path(args.left_path).resolve())
+    right_path = str(Path(args.right_path).resolve())
+    cottas_path = str(Path(args.cottas_path).resolve())
+    with cottas_scratch_workspace():
+        pycottas.cat(
+            [left_path, right_path],
+            cottas_path,
+            index=args.index,
+            remove_input_files=True,
+        )
     return 0
 
 

src/partitioned_compression.py

@@ -31,6 +31,12 @@
 )
 
 
+def is_triple_line(line: bytes) -> bool:
+    """Identify an N-Triples record without parsing or copying its terms."""
+    stripped = line.strip()
+    return bool(stripped) and not stripped.startswith(b"#") and stripped.endswith(b".")
+
+
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="VCF-RDFizer partitioned compression runner")
     parser.add_argument("--source", required=True, help="plain or gzip-compressed N-Triples input")
@@ -40,6 +46,7 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--target-chunk-bytes", required=True, type=int)
     parser.add_argument("--min-chunk-bytes", required=True, type=int)
     parser.add_argument("--max-chunk-bytes", required=True, type=int)
+    parser.add_argument("--expected-triples", type=int)
     parser.add_argument("--result-path", required=True)
     return parser.parse_args()
 
@@ -124,7 +131,8 @@ def close_chunk():
             handle.write(line)
             chunk_size += line_size
             logical_offset += line_size
-            record_count += 1
+            if is_triple_line(line):
+                record_count += 1
     finally:
         close_chunk()
 
@@ -306,6 +314,12 @@ def main() -> int:
         )
         if not chunks:
             raise ValueError("RDF source contains no complete records")
+        source_triples = int(plan["record_count"])
+        if args.expected_triples is not None and source_triples != args.expected_triples:
+            raise ValueError(
+                "source triple count does not match the upstream conversion count: "
+                f"source={source_triples}, expected={args.expected_triples}"
+            )
         hdt_paths: list[Path] = []
         cottas_paths: list[Path] = []
         needs_hdt = any(method in HDT_METHODS for method in methods)
@@ -326,6 +340,65 @@ def main() -> int:
         if any(method in COTTAS_METHODS for method in methods) and not cottas_python:
             raise RuntimeError("Missing Python runtime for COTTAS")
 
+        def validate_artifact(
+            *,
+            name: str,
+            artifact: Path,
+            artifact_format: str,
+            python_bin: str,
+            skip_index_check: bool = False,
+        ) -> dict:
+            """Decode/count one final artifact and return its validation report."""
+            validation_path = work_dir / f".{name}.validation.json"
+            command = [
+                python_bin,
+                "/opt/vcf-rdfizer/validate_compression.py",
+                "--source",
+                str(source),
+                "--artifact",
+                str(artifact),
+                "--format",
+                artifact_format,
+                "--source-triples",
+                str(source_triples),
+                "--result-path",
+                str(validation_path),
+            ]
+            if args.expected_triples is not None:
+                command.extend(["--expected-triples", str(args.expected_triples)])
+            if skip_index_check:
+                command.append("--skip-index-check")
+            stage = runner.run(name, command, validation_path)
+            try:
+                report = json.loads(validation_path.read_text(encoding="utf-8"))
+            except (FileNotFoundError, json.JSONDecodeError) as exc:
+                report = {
+                    "valid": False,
+                    "count_match": False,
+                    "error": f"validator did not produce a valid report: {exc}",
+                }
+            finally:
+                validation_path.unlink(missing_ok=True)
+            report.setdefault(
+                "timing",
+                {
+                    "wall_seconds": stage.get("wall_seconds"),
+                    "user_seconds": stage.get("user_seconds"),
+                    "sys_seconds": stage.get("sys_seconds"),
+                    "max_rss_kb": stage.get("max_rss_kb"),
+                },
+            )
+            if (
+                stage["exit_code"] != 0
+                or not report.get("valid")
+                or not report.get("count_match")
+            ):
+                raise RuntimeError(
+                    f"{artifact_format.upper()} validation failed for {artifact.name}: "
+                    f"{report.get('error', 'decoded triple count mismatch')}"
+                )
+            return report
+
         for index, chunk in enumerate(chunks):
             if needs_hdt:
                 chunk_hdt = work_dir / f"chunk-{index:05d}.hdt"
@@ -369,6 +442,13 @@ def main() -> int:
             add_totals(hdt_total, index_stage)
             if index_stage["exit_code"] != 0:
                 raise RuntimeError("final HDT index initialization failed")
+            hdt_validation = validate_artifact(
+                name="hdt-validate",
+                artifact=output_hdt,
+                artifact_format="hdt",
+                python_bin=sys.executable,
+                skip_index_check=True,
+            )
             results["hdt"] = {
                 **finalize_totals(hdt_total),
                 "output_path": str(output_hdt),
@@ -379,6 +459,7 @@ def main() -> int:
                     "merge_rounds": hdt_rounds,
                     "index_path": str(output_index),
                     "index_size_bytes": output_index.stat().st_size,
+                    "validation": hdt_validation,
                 },
             }
 
@@ -395,12 +476,23 @@ def main() -> int:
                 raise RuntimeError("COTTAS merge failed")
             shutil.copyfile(final_cottas, output_cottas)
             final_cottas.unlink(missing_ok=True)
+            cottas_validation = validate_artifact(
+                name="cottas-validate",
+                artifact=output_cottas,
+                artifact_format="cottas",
+                python_bin=cottas_python,
+            )
             results["cottas"] = {
                 **finalize_totals(cottas_total),
                 "output_path": str(output_cottas),
                 "output_size_bytes": output_cottas.stat().st_size,
                 "source": "partitioned_generated",
-                "details": {**plan, "merge_rounds": cottas_rounds, "index": "spo"},
+                "details": {
+                    **plan,
+                    "merge_rounds": cottas_rounds,
+                    "index": "spo",
+                    "validation": cottas_validation,
+                },
             }
 
         for method in methods: