Add dedup output validation steps to integration test

Validate exact paragraph and fuzzy document dedup outputs after they run:
exact dedup must produce vortex files with 2 rows/dups, fuzzy dedup must
produce jsonl.gz files with 2 documents marked as fuzzy duplicates.

Author: ravwojdyla

tests/integration_test.py

@@ -32,6 +32,7 @@
 from marin.schemas.web.convert import ResiliparseConfig
 from marin.training.training import TrainLmOnPodConfig, run_levanter_train_lm
 from marin.transform.simple_html_to_md.process import SimpleHtmlToMdConfig, html_to_md
+from marin.utils import fsspec_glob
 
 from iris.logging import configure_logging
 
@@ -68,6 +69,61 @@ def _setup_log_tee(log_path: str) -> None:
     _tee_fd(sys.stderr.fileno(), log_fh)
 
 
+@dataclasses.dataclass(frozen=True)
+class ValidateExactDedupConfig:
+    data_path: str
+    expected_rows: int
+    expected_dups: int
+
+
+def validate_exact_dedup_output(config: ValidateExactDedupConfig):
+    """Validate exact dedup output: directory exists, files are vortex, has expected row/dup counts."""
+    import vortex
+
+    vortex_files = fsspec_glob(os.path.join(config.data_path, "**/*.vortex"))
+    assert vortex_files, f"No vortex files found in {config.data_path}"
+
+    total_rows = 0
+    for vf_path in vortex_files:
+        table = vortex.open(vf_path).to_arrow().read_all()
+        total_rows += len(table)
+
+    # Every row in exact paragraph dedup output has dup_spans (non-dups are omitted)
+    assert total_rows == config.expected_rows, f"Expected {config.expected_rows} rows, got {total_rows}"
+    assert total_rows == config.expected_dups, f"Expected {config.expected_dups} dups, got {total_rows}"
+    logger.info(f"Validated exact dedup: {total_rows} rows in {len(vortex_files)} vortex files")
+
+
+@dataclasses.dataclass(frozen=True)
+class ValidateFuzzyDedupConfig:
+    data_path: str
+    expected_dups: int
+
+
+def validate_fuzzy_dedup_output(config: ValidateFuzzyDedupConfig):
+    """Validate fuzzy dedup output: directory exists, files are jsonl.gz, has expected dup counts."""
+    import gzip
+    import json
+
+    jsonl_files = fsspec_glob(os.path.join(config.data_path, "**/*.jsonl.gz"))
+    assert jsonl_files, f"No jsonl.gz files found in {config.data_path}"
+
+    total_rows = 0
+    total_dups = 0
+    for jf_path in jsonl_files:
+        with gzip.open(jf_path, "rt") as f:
+            for line in f:
+                row = json.loads(line)
+                total_rows += 1
+                if row.get("attributes", {}).get(str(DedupMode.FUZZY_DOCUMENT), False):
+                    total_dups += 1
+
+    assert total_dups == config.expected_dups, f"Expected {config.expected_dups} dups, got {total_dups}"
+    logger.info(
+        f"Validated fuzzy dedup: {total_dups} dups out of {total_rows} rows in {len(jsonl_files)} jsonl.gz files"
+    )
+
+
 def create_steps(prefix: str, synth_data: str) -> list[ExecutorStep]:
     # ############################################################
     # Transform HTML to text
@@ -176,6 +232,28 @@ def create_steps(prefix: str, synth_data: str) -> list[ExecutorStep]:
         ),
     )
 
+    ############################################################
+    # Validate dedup outputs
+
+    validate_exact_dedup_step = ExecutorStep(
+        name=os.path.join(prefix, "validate_exact_dedup"),
+        fn=validate_exact_dedup_output,
+        config=ValidateExactDedupConfig(
+            data_path=dedup_exact_paragraph_step.cd("data"),
+            expected_rows=2,
+            expected_dups=2,
+        ),
+    )
+
+    validate_fuzzy_dedup_step = ExecutorStep(
+        name=os.path.join(prefix, "validate_fuzzy_dedup"),
+        fn=validate_fuzzy_dedup_output,
+        config=ValidateFuzzyDedupConfig(
+            data_path=dedup_fuzzy_document_step.cd("data"),
+            expected_dups=2,
+        ),
+    )
+
     ############################################################
     # Consolidate
 
@@ -267,6 +345,8 @@ def create_steps(prefix: str, synth_data: str) -> list[ExecutorStep]:
         inference_lq_step,
         dedup_exact_paragraph_step,
         dedup_fuzzy_document_step,
+        validate_exact_dedup_step,
+        validate_fuzzy_dedup_step,
         consolidate_step,
         tokenize_step,
         train_step,