remove parquet output support

Nate has given up on getting wikiq to output parquet directly because it
makes too many unpredictable large memory allocations. The workflow now
outputs JSONL in a single pass and then uses spark (wikiq_spark) to index
it as parquet in a second pass.

Remove the parquet output path along with the machinery that existed only
to support it: checkpoint files, resume temp-file merging, namespace
partitioning, and file rotation (--partition-namespaces,
--max-revisions-per-file). Resume support remains for JSONL output, where
the resume point is derived from the last complete line of the output
file. Also remove the parquet tests and baseline files.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
(cherry picked from commit 72a8851373)

test/Wikiq_Unit_Test.py

@@ -440,7 +440,7 @@ def test_external_links_only():
     # Verify citations column does NOT exist
     assert "citations" not in test.columns, "citations column should NOT exist when only --external-links is used"
 
-    # Verify column has list/array type (pandas reads parquet lists as numpy arrays)
+    # Verify column has list/array type (pandas reads pyarrow lists as numpy arrays)
     assert test["external_links"].apply(lambda x: x is None or hasattr(x, '__len__')).all(), \
         "external_links should be a list/array type or None"
 
@@ -495,7 +495,7 @@ def test_citations_only():
     # Verify external_links column does NOT exist
     assert "external_links" not in test.columns, "external_links column should NOT exist when only --citations is used"
 
-    # Verify column has list/array type (pandas reads parquet lists as numpy arrays)
+    # Verify column has list/array type (pandas reads pyarrow lists as numpy arrays)
     assert test["citations"].apply(lambda x: x is None or hasattr(x, '__len__')).all(), \
         "citations should be a list/array type or None"
 
@@ -545,7 +545,7 @@ def test_external_links_and_citations():
     assert "external_links" in test.columns, "external_links column should exist"
     assert "citations" in test.columns, "citations column should exist"
 
-    # Verify both columns have list/array types (pandas reads parquet lists as numpy arrays)
+    # Verify both columns have list/array types (pandas reads pyarrow lists as numpy arrays)
     assert test["external_links"].apply(lambda x: x is None or hasattr(x, '__len__')).all(), \
         "external_links should be a list/array type or None"
     assert test["citations"].apply(lambda x: x is None or hasattr(x, '__len__')).all(), \
@@ -753,37 +753,4 @@ def test_headings():
     print(f"Headings test passed! {len(test)} rows processed")
 
 
-def test_parquet_output():
-    """Test that Parquet output format works correctly."""
-    tester = WikiqTester(SAILORMOON, "parquet_output", in_compression="7z", out_format="parquet")
-
-    try:
-        tester.call_wikiq("--fandom-2020")
-    except subprocess.CalledProcessError as exc:
-        pytest.fail(exc.stderr.decode("utf8"))
-
-    # Verify output file exists
-    assert os.path.exists(tester.output), f"Parquet output file should exist at {tester.output}"
-
-    # Read and verify content
-    test = pd.read_parquet(tester.output)
-
-    # Verify expected columns exist
-    assert "revid" in test.columns
-    assert "articleid" in test.columns
-    assert "title" in test.columns
-    assert "namespace" in test.columns
-
-    # Verify row count matches JSONL output
-    tester_jsonl = WikiqTester(SAILORMOON, "parquet_compare", in_compression="7z", out_format="jsonl")
-    try:
-        tester_jsonl.call_wikiq("--fandom-2020")
-    except subprocess.CalledProcessError as exc:
-        pytest.fail(exc.stderr.decode("utf8"))
-
-    test_jsonl = pd.read_json(tester_jsonl.output, lines=True)
-    assert len(test) == len(test_jsonl), f"Parquet and JSONL should have same row count: {len(test)} vs {len(test_jsonl)}"
-
-    print(f"Parquet output test passed! {len(test)} rows")
-
 

test/test_resume.py

@@ -9,10 +9,6 @@ import time
 
 import pytest
 
-from wikiq.resume import (
-    get_checkpoint_path,
-    read_checkpoint,
-)
 from wikiq_test_utils import (
     SAILORMOON,
     TEST_DIR,
@@ -33,7 +29,7 @@ def read_jsonl(filepath):
 
 
 def test_resume():
-    """Test that --resume properly resumes processing from the last checkpoint."""
+    """Test that --resume properly resumes processing from the last line of output."""
     import pandas as pd
     from pandas.testing import assert_frame_equal
 
@@ -57,10 +53,6 @@ def test_resume():
         for row in full_rows[:middle_idx + 1]:
             f.write(json.dumps(row) + "\n")
 
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": full_rows[middle_idx]["articleid"], "revid": resume_revid}, f)
-
     try:
         tester_partial.call_wikiq("--fandom-2020", "--resume")
     except subprocess.CalledProcessError as exc:
@@ -103,10 +95,6 @@ def test_resume_with_diff():
         for row in full_rows[:resume_idx + 1]:
             f.write(json.dumps(row) + "\n")
 
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": full_rows[resume_idx]["articleid"], "revid": resume_revid}, f)
-
     try:
         tester_partial.call_wikiq("--diff", "--fandom-2020", "--resume")
     except subprocess.CalledProcessError as exc:
@@ -170,10 +158,6 @@ def test_resume_simple():
         for row in full_rows[:resume_idx + 1]:
             f.write(json.dumps(row) + "\n")
 
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": full_rows[resume_idx]["articleid"], "revid": resume_revid}, f)
-
     try:
         tester_partial.call_wikiq("--resume")
     except subprocess.CalledProcessError as exc:
@@ -186,39 +170,6 @@ def test_resume_simple():
     assert_frame_equal(df_full, df_resumed)
 
 
-def test_checkpoint_read():
-    """Test that read_checkpoint correctly reads checkpoint files."""
-    with tempfile.TemporaryDirectory() as tmpdir:
-        checkpoint_path = os.path.join(tmpdir, "test.jsonl.checkpoint")
-
-        # Test reading valid checkpoint
-        with open(checkpoint_path, 'w') as f:
-            json.dump({"pageid": 100, "revid": 200}, f)
-
-        result = read_checkpoint(checkpoint_path)
-        assert result == (100, 200), f"Expected (100, 200), got {result}"
-
-        # Test reading non-existent checkpoint
-        result = read_checkpoint(os.path.join(tmpdir, "nonexistent.checkpoint"))
-        assert result is None, f"Expected None for non-existent file, got {result}"
-
-        # Test reading empty checkpoint
-        empty_path = os.path.join(tmpdir, "empty.checkpoint")
-        with open(empty_path, 'w') as f:
-            f.write("{}")
-        result = read_checkpoint(empty_path)
-        assert result is None, f"Expected None for empty checkpoint, got {result}"
-
-        # Test reading corrupted checkpoint
-        corrupt_path = os.path.join(tmpdir, "corrupt.checkpoint")
-        with open(corrupt_path, 'w') as f:
-            f.write("not valid json")
-        result = read_checkpoint(corrupt_path)
-        assert result is None, f"Expected None for corrupted checkpoint, got {result}"
-
-    print("Checkpoint read test passed!")
-
-
 def test_resume_with_interruption():
     """Test that resume works correctly after interruption."""
     import pandas as pd
@@ -245,9 +196,6 @@ def test_resume_with_interruption():
     # Clean up for interrupted run
     if os.path.exists(output_file):
         os.remove(output_file)
-    checkpoint_path = get_checkpoint_path(output_file)
-    if os.path.exists(checkpoint_path):
-        os.remove(checkpoint_path)
 
     # Start wikiq and interrupt it
     cmd_partial = [
@@ -301,48 +249,6 @@ def test_resume_with_interruption():
     assert_frame_equal(df_full, df_resumed)
 
 
-def test_resume_parquet():
-    """Test that --resume works correctly with Parquet output format."""
-    import pandas as pd
-    from pandas.testing import assert_frame_equal
-    import pyarrow.parquet as pq
-
-    tester_full = WikiqTester(SAILORMOON, "resume_parquet_full", in_compression="7z", out_format="parquet")
-
-    try:
-        tester_full.call_wikiq("--fandom-2020")
-    except subprocess.CalledProcessError as exc:
-        pytest.fail(exc.stderr.decode("utf8"))
-
-    full_output_path = tester_full.output
-    full_table = pq.read_table(full_output_path)
-
-    # Use unsorted indices consistently - slice the table and get checkpoint from same position
-    resume_idx = len(full_table) // 3
-    resume_revid = int(full_table.column("revid")[resume_idx].as_py())
-    resume_pageid = int(full_table.column("articleid")[resume_idx].as_py())
-
-    tester_partial = WikiqTester(SAILORMOON, "resume_parquet_partial", in_compression="7z", out_format="parquet")
-    partial_output_path = tester_partial.output
-
-    # Write partial Parquet file using the SAME schema as the full file
-    partial_table = full_table.slice(0, resume_idx + 1)
-    pq.write_table(partial_table, partial_output_path)
-
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": resume_pageid, "revid": resume_revid}, f)
-
-    try:
-        tester_partial.call_wikiq("--fandom-2020", "--resume")
-    except subprocess.CalledProcessError as exc:
-        pytest.fail(exc.stderr.decode("utf8"))
-
-    df_full = full_table.to_pandas()
-    df_resumed = pd.read_parquet(partial_output_path)
-    assert_frame_equal(df_full, df_resumed)
-
-
 def test_resume_tsv_error():
     """Test that --resume with TSV output produces a proper error message."""
     tester = WikiqTester(SAILORMOON, "resume_tsv_error", in_compression="7z", out_format="tsv")
@@ -352,7 +258,7 @@ def test_resume_tsv_error():
         pytest.fail("Expected error for --resume with TSV output")
     except subprocess.CalledProcessError as exc:
         stderr = exc.stderr.decode("utf8")
-        assert "Error: --resume only works with JSONL or Parquet" in stderr, \
+        assert "Error: --resume only works with JSONL output" in stderr, \
             f"Expected proper error message, got: {stderr}"
 
     print("TSV resume error test passed!")
@@ -387,10 +293,6 @@ def test_resume_data_equivalence():
         for row in full_rows[:resume_idx + 1]:
             f.write(json.dumps(row) + "\n")
 
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": full_rows[resume_idx]["articleid"], "revid": resume_revid}, f)
-
     try:
         tester_partial.call_wikiq("--fandom-2020", "--resume")
     except subprocess.CalledProcessError as exc:
@@ -433,10 +335,6 @@ def test_resume_with_persistence():
         for row in full_rows[:resume_idx + 1]:
             f.write(json.dumps(row) + "\n")
 
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": full_rows[resume_idx]["articleid"], "revid": resume_revid}, f)
-
     try:
         tester_partial.call_wikiq("--persistence wikidiff2", "--fandom-2020", "--resume")
     except subprocess.CalledProcessError as exc:
@@ -487,10 +385,6 @@ def test_resume_corrupted_jsonl_last_line():
     # Record file size before resume
     size_before = os.path.getsize(corrupt_output_path)
 
-    # NO checkpoint file - JSONL resume works from last valid line in the file
-    checkpoint_path = get_checkpoint_path(corrupt_output_path)
-    assert not os.path.exists(checkpoint_path), "Test setup error: checkpoint should not exist"
-
     # Resume should detect corrupted line, truncate it, then append new data
     try:
         tester_corrupt.call_wikiq("--fandom-2020", "--resume")
@@ -537,10 +431,6 @@ def test_resume_diff_persistence_combined():
         for row in full_rows[:resume_idx + 1]:
             f.write(json.dumps(row) + "\n")
 
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": full_rows[resume_idx]["articleid"], "revid": resume_revid}, f)
-
     try:
         tester_partial.call_wikiq("--diff", "--persistence wikidiff2", "--fandom-2020", "--resume")
     except subprocess.CalledProcessError as exc:
@@ -588,7 +478,7 @@ def test_resume_mid_page():
     resume_revid = int(resume_rev["revid"])
     resume_pageid = int(resume_rev["articleid"])
 
-    # Find global index for checkpoint
+    # Find global index for the resume point
     global_idx = df_full[df_full["revid"] == resume_revid].index[0]
 
     tester_partial = WikiqTester(SAILORMOON, "resume_midpage_partial", in_compression="7z", out_format="jsonl")
@@ -600,10 +490,6 @@ def test_resume_mid_page():
         for row in rows_to_write:
             f.write(json.dumps(row) + "\n")
 
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": resume_pageid, "revid": resume_revid}, f)
-
     try:
         tester_partial.call_wikiq("--diff", "--fandom-2020", "--resume")
     except subprocess.CalledProcessError as exc:
@@ -653,10 +539,6 @@ def test_resume_page_boundary():
         for row in rows_to_write:
             f.write(json.dumps(row) + "\n")
 
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": resume_pageid, "revid": resume_revid}, f)
-
     try:
         tester_partial.call_wikiq("--diff", "--fandom-2020", "--resume")
     except subprocess.CalledProcessError as exc:
@@ -790,10 +672,6 @@ def test_resume_revert_detection():
         for row in full_rows[:resume_idx + 1]:
             f.write(json.dumps(row) + "\n")
 
-    checkpoint_path = get_checkpoint_path(partial_output_path)
-    with open(checkpoint_path, 'w') as f:
-        json.dump({"pageid": resume_pageid, "revid": resume_revid}, f)
-
     try:
         tester_partial.call_wikiq("--fandom-2020", "--resume")
     except subprocess.CalledProcessError as exc:

test/wikiq_test_utils.py

@@ -42,17 +42,8 @@ class WikiqTester:
             else:
                 shutil.rmtree(self.output)
 
-        # Also clean up resume-related files
-        for temp_suffix in [".resume_temp", ".checkpoint", ".merged"]:
-            temp_path = self.output + temp_suffix
-            if os.path.exists(temp_path):
-                if os.path.isfile(temp_path):
-                    os.remove(temp_path)
-                else:
-                    shutil.rmtree(temp_path)
-
-        # For JSONL and Parquet, self.output is a file path. Create parent directory if needed.
-        if out_format in ("jsonl", "parquet"):
+        # For JSONL, self.output is a file path. Create parent directory if needed.
+        if out_format == "jsonl":
             parent_dir = os.path.dirname(self.output)
             if parent_dir:
                 os.makedirs(parent_dir, exist_ok=True)