[ci]: add preprocessing integration coverage

.buildkite/pipeline.yml

@@ -77,6 +77,17 @@ steps:
             limit: 2
       agents:
         queue: "default"
+    - label: ":microscope: Preprocessing Tests"
+      if: build.env("TEST_SCOPE") == "direct" && build.env("TEST_TYPE") == "preprocessing"
+      command: "timeout 90m .buildkite/scripts/pr_test.sh"
+      retry:
+        automatic:
+          - exit_status: 128
+            limit: 3
+          - exit_status: -1
+            limit: 2
+      agents:
+        queue: "default"
     - label: ":microscope: DreamVerse App Tests"
       if: build.env("TEST_SCOPE") == "direct" && build.env("TEST_TYPE") == "dreamverse_app"
       command: "timeout 90m .buildkite/scripts/pr_test.sh"
@@ -322,6 +333,29 @@ steps:
                   - TEST_TYPE=unit_test
                 agents:
                   queue: "default"
+            - path:
+                - "fastvideo/pipelines/preprocess/**"
+                - "fastvideo/workflow/preprocess/**"
+                - "fastvideo/dataset/dataloader/parquet_io.py"
+                - "fastvideo/dataset/dataloader/record_schema.py"
+                - "fastvideo/dataset/dataloader/schema.py"
+                - "fastvideo/configs/configs.py"
+                - "fastvideo/fastvideo_args.py"
+                - "fastvideo/tests/workflow/test_t2v_preprocessing_e2e.py"
+                - "fastvideo/tests/nightly/reference_video_1_sample_v0.mp4"
+                - "fastvideo/tests/modal/pr_test.py"
+                - ".buildkite/pipeline.yml"
+                - ".buildkite/scripts/pr_test.sh"
+                - ".github/workflows/ci-slash-commands.yml"
+                - "pyproject.toml"
+                - "docker/Dockerfile"
+              config:
+                command: "timeout 20m .buildkite/scripts/pr_test.sh"
+                label: ":microscope: Preprocessing Tests"
+                env:
+                  - TEST_TYPE=preprocessing
+                agents:
+                  queue: "default"
             - path:
                 - "apps/dreamverse/**"
                 - "pyproject.toml"

.buildkite/scripts/pr_test.sh

@@ -233,6 +233,10 @@ case "$TEST_TYPE" in
         log "Running unit tests..."
         MODAL_COMMAND="$MODAL_ENV python3 -m modal run $MODAL_TEST_FILE::run_unit_test"
         ;;
+    "preprocessing")
+        log "Running preprocessing integration test..."
+        MODAL_COMMAND="$MODAL_ENV HF_API_KEY=$HF_API_KEY python3 -m modal run $MODAL_TEST_FILE::run_preprocessing_tests"
+        ;;
     "dreamverse_app")
         log "Running DreamVerse app tests..."
         MODAL_COMMAND="$MODAL_ENV python3 -m modal run $MODAL_TEST_FILE::run_dreamverse_app_tests"

.github/workflows/ci-slash-commands.yml

@@ -129,7 +129,7 @@ jobs:
           set -euo pipefail
           TEST_NAME=$(echo "$COMMENT" | grep -oP '(?<=/test\s)\S+' | head -1 || true)
 
-          VALID="encoder vae transformer kernel unit dreamverse ssim training lora-inference lora-training lora-extraction distillation self-forcing vsa vmoba performance api train-framework eval full fastcheck pre-commit"
+          VALID="encoder vae transformer kernel unit preprocessing dreamverse ssim training lora-inference lora-training lora-extraction distillation self-forcing vsa vmoba performance api train-framework eval full fastcheck pre-commit"
           if [ -z "$TEST_NAME" ] || ! echo "$VALID" | grep -qw "$TEST_NAME"; then
             echo "Unknown test: '$TEST_NAME'. Valid: $VALID"
             exit 1
@@ -137,7 +137,8 @@ jobs:
 
           declare -A MAP=(
             [encoder]=encoder [vae]=vae [transformer]=transformer
-            [kernel]=kernel_tests [unit]=unit_test [dreamverse]=dreamverse_app
+            [kernel]=kernel_tests [unit]=unit_test [preprocessing]=preprocessing
+            [dreamverse]=dreamverse_app
             [ssim]=ssim [training]=training
             [lora-inference]=inference_lora [lora-training]=training_lora
             [lora-extraction]=lora_extraction

docs/contributing/ci_architecture.md

@@ -85,6 +85,7 @@ status.
 | Transformer Tests | `transformer` | `fastvideo/models/dits/**`, `fastvideo/models/loader/**`, `fastvideo/tests/transformers/**`, `fastvideo/layers/**`, `fastvideo/attention/**`, `pyproject.toml`, `docker/Dockerfile` |
 | Kernel Tests | `kernel_tests` | `fastvideo-kernel/**`, `pyproject.toml`, `docker/Dockerfile` |
 | Unit Tests | `unit_test` | `fastvideo/**`, `.buildkite/**`, `.github/**`, `pyproject.toml`, `docker/Dockerfile` |
+| Preprocessing Tests | `preprocessing` | Preprocessing pipelines/workflows, Parquet schema/writer, the integration test, and its CI entrypoints |
 | DreamVerse App Tests | `dreamverse_app` | `apps/dreamverse/**`, `pyproject.toml` |
 
 ### Tier 3: Full Suite
@@ -147,6 +148,7 @@ Valid direct test names:
 | `/test transformer` | `transformer` |
 | `/test kernel` | `kernel_tests` |
 | `/test unit` | `unit_test` |
+| `/test preprocessing` | `preprocessing` |
 | `/test dreamverse` | `dreamverse_app` |
 | `/test ssim` | `ssim` |
 | `/test training` | `training` |

fastvideo/tests/modal/pr_test.py

@@ -375,6 +375,21 @@ def run_unit_test():
     )
 
 
+@app.function(gpu="L40S:1",
+              image=image,
+              timeout=1200,
+              secrets=[hf_secret, ci_env_secret],
+              volumes={"/root/data": model_vol})
+def run_preprocessing_tests():
+    run_test_command(
+        "export HF_HOME='/root/data/.cache' && "
+        "hf auth login --token $HF_API_KEY && "
+        "FASTVIDEO_PREPROCESSING_E2E=1 FASTVIDEO_FA4=0 "
+        "pytest ./fastvideo/tests/workflow/test_t2v_preprocessing_e2e.py -vs",
+        build_kernel=False,
+    )
+
+
 # TODO: David: GPU only used to resolve import time requirement (not needed for this test). Maybe make those imports lazy?
 @app.function(gpu="L40S:1",
               image=dreamverse_image,

fastvideo/tests/workflow/test_t2v_preprocessing_e2e.py

@@ -0,0 +1,118 @@
+import json
+import os
+import shutil
+import subprocess
+import sys
+from pathlib import Path
+
+import pyarrow.parquet as pq
+import pytest
+import torch
+
+from fastvideo.dataset.dataloader.schema import pyarrow_schema_t2v
+
+CAPTION = "a deterministic preprocessing smoke test"
+MODEL_PATH = "Wan-AI/Wan2.1-T2V-1.3B-Diffusers"
+
+pytestmark = [
+    pytest.mark.skipif(os.environ.get("FASTVIDEO_PREPROCESSING_E2E") != "1",
+                       reason="set FASTVIDEO_PREPROCESSING_E2E=1 to run the GPU integration test"),
+    pytest.mark.skipif(not torch.cuda.is_available(), reason="requires CUDA"),
+]
+
+
+def test_t2v_preprocessing_writes_valid_parquet(tmp_path: Path) -> None:
+    raw_data_dir = tmp_path / "raw"
+    video_dir = raw_data_dir / "videos"
+    video_dir.mkdir(parents=True)
+
+    video_name = "sample.mp4"
+    source_video = Path(__file__).parents[1] / "nightly" / "reference_video_1_sample_v0.mp4"
+    shutil.copy2(source_video, video_dir / video_name)
+    (raw_data_dir / "videos2caption.json").write_text(
+        json.dumps([{
+            "path": video_name,
+            "cap": CAPTION,
+            "fps": 16.0,
+            "num_frames": 29,
+            "resolution": {
+                "height": 480,
+                "width": 832,
+            },
+        }]),
+        encoding="utf-8",
+    )
+
+    output_dir = tmp_path / "preprocessed"
+    subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "torch.distributed.run",
+            "--standalone",
+            "--nproc-per-node=1",
+            "-m",
+            "fastvideo.pipelines.preprocess.v1_preprocessing_new",
+            "--model-path",
+            MODEL_PATH,
+            "--mode",
+            "preprocess",
+            "--workload-type",
+            "t2v",
+            "--preprocess.video-loader-type",
+            "torchvision",
+            "--preprocess.dataset-type",
+            "merged",
+            "--preprocess.dataset-path",
+            str(raw_data_dir),
+            "--preprocess.dataset-output-dir",
+            str(output_dir),
+            "--preprocess.preprocess-video-batch-size",
+            "1",
+            "--preprocess.dataloader-num-workers",
+            "0",
+            "--preprocess.max-height",
+            "64",
+            "--preprocess.max-width",
+            "64",
+            "--preprocess.num-frames",
+            "17",
+            "--preprocess.train-fps",
+            "16",
+            "--preprocess.samples-per-file",
+            "1",
+            "--preprocess.flush-frequency",
+            "1",
+            "--preprocess.video-length-tolerance-range",
+            "5",
+        ],
+        check=True,
+    )
+
+    rank_output_dir = output_dir / "combined_parquet_dataset" / "worker_0"
+    parquet_file = rank_output_dir / "worker_0" / "data_chunk_0.parquet"
+    assert parquet_file.is_file()
+    assert not (rank_output_dir / "worker_0" / "data_chunk_0.parquet.tmp").exists()
+
+    table = pq.read_table(parquet_file)
+    assert table.schema.equals(pyarrow_schema_t2v, check_metadata=False)
+    assert table.num_rows == 1
+
+    row = table.to_pylist()[0]
+    assert row["id"] == video_name
+    assert row["file_name"] == video_name
+    assert row["caption"] == CAPTION
+    assert row["media_type"] == "video"
+    assert row["width"] == 64
+    assert row["height"] == 64
+    assert row["fps"] == 16.0
+    assert row["duration_sec"] == pytest.approx(17 / 16)
+    assert row["vae_latent_bytes"]
+    assert len(row["vae_latent_shape"]) == 4
+    assert all(size > 0 for size in row["vae_latent_shape"])
+    assert row["num_frames"] == row["vae_latent_shape"][1]
+    assert row["vae_latent_dtype"]
+    assert row["text_embedding_bytes"]
+    assert len(row["text_embedding_shape"]) == 2
+    assert all(size > 0 for size in row["text_embedding_shape"])
+    assert row["text_embedding_dtype"]

fastvideo/workflow/preprocess/preprocess_workflow.py

@@ -95,12 +95,13 @@ class PreprocessWorkflow(WorkflowBase):
         dataset_output_dir = self.fastvideo_args.preprocess_config.dataset_output_dir
         os.makedirs(dataset_output_dir, exist_ok=True)
 
-        validation_dataset_output_dir = os.path.join(dataset_output_dir, "validation_dataset",
+        validation_dataset_output_dir = os.path.join(dataset_output_dir, "validation_parquet_dataset",
                                                      f"worker_{get_world_rank()}")
         os.makedirs(validation_dataset_output_dir, exist_ok=True)
         self.validation_dataset_output_dir = validation_dataset_output_dir
 
-        training_dataset_output_dir = os.path.join(dataset_output_dir, "training_dataset", f"worker_{get_world_rank()}")
+        training_dataset_output_dir = os.path.join(dataset_output_dir, "combined_parquet_dataset",
+                                                   f"worker_{get_world_rank()}")
         os.makedirs(training_dataset_output_dir, exist_ok=True)
         self.training_dataset_output_dir = training_dataset_output_dir