improved metrics display and aggregate triple duplication bug fix

Author: ecrum19

src/run_conversion.sh

@@ -35,7 +35,7 @@ mkdir -p "$TIME_LOG_DIR" "$METRICS_JSON_DIR"
 TIME_LOG="$TIME_LOG_DIR/${RUN_ID}.txt"
 METRICS_JSON="$METRICS_JSON_DIR/${RUN_ID}.json"
 METRICS_CSV="$LOGDIR/metrics.csv"
-METRICS_HEADER="run_id,timestamp,output_name,output_dir,exit_code_java,wall_seconds_java,user_seconds_java,sys_seconds_java,max_rss_kb_java,input_mapping_size_bytes,input_vcf_size_bytes,output_dir_size_bytes,output_triples,jar,mapping_file,output_path,combined_rdf_size_bytes,gzip_size_bytes,brotli_size_bytes,hdt_size_bytes,exit_code_gzip,exit_code_brotli,exit_code_hdt,wall_seconds_gzip,user_seconds_gzip,sys_seconds_gzip,max_rss_kb_gzip,wall_seconds_brotli,user_seconds_brotli,sys_seconds_brotli,max_rss_kb_brotli,wall_seconds_hdt,user_seconds_hdt,sys_seconds_hdt,max_rss_kb_hdt,compression_methods,hdt_source,gzip_on_hdt_size_bytes,brotli_on_hdt_size_bytes,exit_code_gzip_on_hdt,exit_code_brotli_on_hdt,wall_seconds_gzip_on_hdt,user_seconds_gzip_on_hdt,sys_seconds_gzip_on_hdt,max_rss_kb_gzip_on_hdt,wall_seconds_brotli_on_hdt,user_seconds_brotli_on_hdt,sys_seconds_brotli_on_hdt,max_rss_kb_brotli_on_hdt"
+METRICS_HEADER="run_id,timestamp,output_name,output_dir,exit_code_java,wall_seconds_java,user_seconds_java,sys_seconds_java,max_rss_kb_java,input_mapping_size_bytes,input_vcf_size_bytes,output_dir_size_bytes,output_triples,jar,mapping_file,output_path"
 
 
 # Return byte size for file or directory (GNU + BSD compatible).
@@ -76,6 +76,21 @@ stat_size() {
 
 have_gnu_time() { [[ -x /usr/bin/time ]] && /usr/bin/time --version >/dev/null 2>&1; }
 
+# Return stable content hash for duplicate part detection.
+hash_file_sha256() {
+  local path="$1"
+  if command -v sha256sum >/dev/null 2>&1; then
+    sha256sum "$path" | awk '{print $1}'
+    return
+  fi
+  if command -v shasum >/dev/null 2>&1; then
+    shasum -a 256 "$path" | awk '{print $1}'
+    return
+  fi
+  # Last-resort fallback when SHA utilities are unavailable.
+  cksum "$path" | awk '{print $1":"$2}'
+}
+
 # Count triples via non-comment RDF lines ending in '.'.
 count_triples_json() {
   local path="$1"
@@ -167,13 +182,30 @@ if [[ "$AGGREGATE_RDF" == "1" ]]; then
   PART_FILES=("$OUT_DIR/$OUT_NAME"/*.nt)
   if (( ${#PART_FILES[@]} > 0 )); then
     : > "$MERGED_NT"
+    # Defensive dedupe: some Spark/RMLStreamer runs can emit identical part
+    # files for the same dataset. Skip exact duplicate part payloads to avoid
+    # doubling every triple in the merged output.
+    SEEN_HASH_FILE="$OUT_DIR/$OUT_NAME/.seen_part_hashes.$$"
+    SEEN_MAP_FILE="$OUT_DIR/$OUT_NAME/.seen_part_hash_map.$$"
+    : > "$SEEN_HASH_FILE"
+    : > "$SEEN_MAP_FILE"
     for PART_NT in "${PART_FILES[@]}"; do
       if [[ "$PART_NT" == "$MERGED_NT" ]]; then
         continue
       fi
+      PART_HASH=$(hash_file_sha256 "$PART_NT")
+      if grep -Fqx "$PART_HASH" "$SEEN_HASH_FILE"; then
+        FIRST_SEEN=$(awk -F'\t' -v hash="$PART_HASH" '$1 == hash { print $2; exit }' "$SEEN_MAP_FILE")
+        echo "WARNING: skipping duplicate RDF part '$PART_NT' (same content as '$FIRST_SEEN')." >&2
+        rm -f "$PART_NT"
+        continue
+      fi
+      printf "%s\n" "$PART_HASH" >> "$SEEN_HASH_FILE"
+      printf "%s\t%s\n" "$PART_HASH" "$PART_NT" >> "$SEEN_MAP_FILE"
       cat "$PART_NT" >> "$MERGED_NT"
       rm -f "$PART_NT"
     done
+    rm -f "$SEEN_HASH_FILE" "$SEEN_MAP_FILE"
   else
     : > "$MERGED_NT"
   fi
@@ -265,13 +297,7 @@ csv_fields=(
   "$JAR"
   "$IN"
   "$OUTPUT_PATH"
-  "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" "" ""
 )
-
-# Compression-related fields are initialized as empty from conversion step output.
-for _ in $(seq 1 13); do
-  csv_fields+=("")
-done
 ( IFS=,; echo "${csv_fields[*]}" ) >> "$METRICS_CSV"
 
 echo "Done."

test/helpers.py

@@ -56,6 +56,25 @@
     "max_rss_kb_brotli_on_hdt",
 ]
 
+CONVERSION_METRICS_HEADER = [
+    "run_id",
+    "timestamp",
+    "output_name",
+    "output_dir",
+    "exit_code_java",
+    "wall_seconds_java",
+    "user_seconds_java",
+    "sys_seconds_java",
+    "max_rss_kb_java",
+    "input_mapping_size_bytes",
+    "input_vcf_size_bytes",
+    "output_dir_size_bytes",
+    "output_triples",
+    "jar",
+    "mapping_file",
+    "output_path",
+]
+
 
 def make_executable(path: Path, content: str) -> None:
     path.write_text(content)

test/test_run_conversion_unit.py

@@ -4,7 +4,7 @@
 import unittest
 from pathlib import Path
 
-from test.helpers import METRICS_HEADER, VerboseTestCase, env_with_path, make_executable
+from test.helpers import CONVERSION_METRICS_HEADER, VerboseTestCase, env_with_path, make_executable
 
 
 REPO_ROOT = Path(__file__).resolve().parents[1]
@@ -76,11 +76,10 @@ def test_run_conversion_writes_nt_and_metrics_without_real_java(self):
                 rows = list(csv.DictReader(f))
             self.assertTrue(rows)
             row = rows[0]
-            self.assertEqual(list(row.keys()), METRICS_HEADER)
+            self.assertEqual(list(row.keys()), CONVERSION_METRICS_HEADER)
             self.assertEqual(row["run_id"], "run123")
             self.assertEqual(row["output_name"], "rdf")
             self.assertEqual(row["exit_code_java"], "0")
-            self.assertEqual(row["compression_methods"], "")
 
     def test_run_conversion_exits_non_zero_when_java_fails(self):
         """Conversion script returns non-zero and records exit_code_java when Java command fails."""
@@ -355,6 +354,67 @@ def test_run_conversion_batch_mode_keeps_individual_nt_parts(self):
             self.assertTrue((out_dir / "rdf" / "part-00001.nt").exists())
             self.assertFalse((out_dir / "rdf" / "rdf.nt").exists())
 
+    def test_run_conversion_aggregate_skips_duplicate_part_payloads(self):
+        """Aggregate merge skips exact duplicate part files to avoid duplicated triples."""
+        with tempfile.TemporaryDirectory() as td:
+            tmp_path = Path(td)
+            fake_bin = tmp_path / "bin"
+            fake_bin.mkdir()
+            make_executable(
+                fake_bin / "java",
+                """#!/usr/bin/env bash
+set -euo pipefail
+if [[ "${1:-}" == "-version" ]]; then
+  echo 'openjdk version "11.0.0"' >&2
+  exit 0
+fi
+out=""
+while [[ $# -gt 0 ]]; do
+  if [[ "$1" == "-o" ]]; then
+    out="$2"
+    shift 2
+    continue
+  fi
+  shift
+done
+mkdir -p "$out"
+printf '<dup> <p> <o> .\\n' > "$out/part-00000"
+printf '<dup> <p> <o> .\\n' > "$out/part-00001"
+printf '<uniq> <p> <o> .\\n' > "$out/part-00002"
+""",
+            )
+
+            out_dir = tmp_path / "out"
+            metrics_dir = tmp_path / "metrics"
+            rules = tmp_path / "rules.ttl"
+            rules.write_text("@prefix ex: <http://example.org/> .\n")
+            vcf = tmp_path / "input.vcf"
+            vcf.write_text("##fileformat=VCFv4.2\n#CHROM\tPOS\n1\t5\n")
+
+            env = env_with_path(fake_bin)
+            env.update(
+                {
+                    "JAR": "fake.jar",
+                    "IN": str(rules),
+                    "IN_VCF": str(vcf),
+                    "OUT_DIR": str(out_dir),
+                    "OUT_NAME": "rdf",
+                    "LOGDIR": str(metrics_dir),
+                    "RUN_ID": "run-dedupe",
+                    "TIMESTAMP": "2026-01-01T00:00:00",
+                }
+            )
+
+            result = subprocess.run(["bash", str(SCRIPT)], env=env, capture_output=True, text=True)
+            self.assertEqual(result.returncode, 0, msg=result.stderr)
+            self.assertIn("skipping duplicate RDF part", result.stderr)
+
+            merged_nt = out_dir / "rdf" / "rdf.nt"
+            self.assertTrue(merged_nt.exists())
+            lines = [line.strip() for line in merged_nt.read_text().splitlines() if line.strip()]
+            self.assertEqual(lines.count("<dup> <p> <o> ."), 1)
+            self.assertEqual(lines.count("<uniq> <p> <o> ."), 1)
+
 
 if __name__ == "__main__":
     unittest.main()

test/test_vcf_rdfizer_unit.py

@@ -163,11 +163,13 @@ def test_update_metrics_csv_keeps_raw_and_hdt_compound_metrics_separate(self):
             )
 
             with metrics_csv.open() as handle:
-                row = next(csv.DictReader(handle))
+                reader = csv.DictReader(handle)
+                row = next(reader)
+                fieldnames = reader.fieldnames or []
 
-            self.assertEqual(row["gzip_size_bytes"], "0")
+            self.assertNotIn("gzip_size_bytes", fieldnames)
+            self.assertIn("gzip_on_hdt_size_bytes", fieldnames)
             self.assertEqual(row["gzip_on_hdt_size_bytes"], "12")
-            self.assertEqual(row["exit_code_gzip"], "0")
             self.assertEqual(row["exit_code_gzip_on_hdt"], "0")
             self.assertEqual(row["hdt_source"], "existing")
             self.assertEqual(row["user_seconds_hdt"], "1.100000")
@@ -630,6 +632,8 @@ def fake_run(cmd, cwd=None, env=None):
             output = out_buf.getvalue()
             self.assertIn("Triples produced: 17", output)
             self.assertIn("Total triples produced (full run): 17", output)
+            self.assertIn("Final RDF size (no compression):", output)
+            self.assertIn("- N-Triples (.nt):", output)
             self.assertIn("Run time (full mode):", output)
 
             run_metrics_dir = latest_metrics_run_dir(metrics_dir)
@@ -1624,11 +1628,12 @@ def fake_run(cmd, cwd=None, env=None):
             self.assertEqual(rc, 0)
 
     def test_main_removes_tsv_when_wrapper_created_it(self):
-        """Wrapper removes TSV directory when it created it and --keep-tsv is not set."""
+        """Wrapper removes hidden .intermediate directory when --keep-tsv is not set."""
         with tempfile.TemporaryDirectory() as td:
             tmp_path = Path(td)
             input_dir, rules_path = prepare_inputs(tmp_path)
-            tsv_dir = tmp_path / "tsv-out"
+            intermediate_dir = tmp_path / "out" / ".intermediate"
+            tsv_dir = intermediate_dir / "tsv"
 
             def fake_run(cmd, cwd=None, env=None):
                 return 0
@@ -1649,23 +1654,23 @@ def fake_run(cmd, cwd=None, env=None):
                             str(input_dir),
                             "--rules",
                             str(rules_path),
-                            "--tsv",
-                            str(tsv_dir),
                         ]
                     )
             finally:
                 os.chdir(old_cwd)
 
             self.assertEqual(rc, 0)
             self.assertFalse(tsv_dir.exists())
+            self.assertFalse(intermediate_dir.exists())
 
-    def test_main_keeps_preexisting_tsv_directory(self):
-        """Wrapper preserves preexisting TSV directory to avoid deleting user-managed files."""
+    def test_main_keep_tsv_preserves_hidden_intermediate_directory(self):
+        """Wrapper preserves hidden intermediates when --keep-tsv is set."""
         with tempfile.TemporaryDirectory() as td:
             tmp_path = Path(td)
             input_dir, rules_path = prepare_inputs(tmp_path)
-            tsv_dir = tmp_path / "tsv-out"
-            tsv_dir.mkdir()
+            intermediate_dir = tmp_path / "out" / ".intermediate"
+            tsv_dir = intermediate_dir / "tsv"
+            tsv_dir.mkdir(parents=True, exist_ok=True)
             sentinel = tsv_dir / "keep.me"
             sentinel.write_text("x")
 
@@ -1688,15 +1693,15 @@ def fake_run(cmd, cwd=None, env=None):
                             str(input_dir),
                             "--rules",
                             str(rules_path),
-                            "--tsv",
-                            str(tsv_dir),
+                            "--keep-tsv",
                         ]
                     )
             finally:
                 os.chdir(old_cwd)
 
             self.assertEqual(rc, 0)
             self.assertTrue(sentinel.exists())
+            self.assertTrue(intermediate_dir.exists())
 
     def test_main_fails_when_tsv_step_fails(self):
         """Wrapper stops when TSV conversion command returns non-zero."""