This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12241-febe1f947a37dfee43064a5c49b34bd4ff7cf6ff in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit c24e46adfd2efc32e2aa6b37caf8eb808dc491ed Author: Goutam Adwant <[email protected]> AuthorDate: Fri Sep 18 14:54:08 2026 +0000 [Test][CLI] Add opt-in paraphrase benchmark suite (#12241) Signed-off-by: Goutam Adwant <[email protected]> --- docs/en/ai-cli/benchmark.md | 37 +++ docs/zh/ai-cli/benchmark.md | 32 ++ seatunnel-cli/benchmark/README.md | 50 +++ seatunnel-cli/benchmark/compare.py | 18 + seatunnel-cli/benchmark/paraphrases.py | 73 ++++ seatunnel-cli/benchmark/runner.py | 58 +++- seatunnel-cli/benchmark/tasks/paraphrase.json | 81 +++++ seatunnel-cli/tests/test_benchmark_comparison.py | 44 +++ seatunnel-cli/tests/test_benchmark_paraphrases.py | 386 ++++++++++++++++++++++ 9 files changed, 775 insertions(+), 4 deletions(-) diff --git a/docs/en/ai-cli/benchmark.md b/docs/en/ai-cli/benchmark.md index 4c0c3e6d35..173397d17c 100644 --- a/docs/en/ai-cli/benchmark.md +++ b/docs/en/ai-cli/benchmark.md @@ -8,6 +8,43 @@ The AI CLI's accuracy is measured — not assumed — by a dedicated benchmark: ## Comparing CLI Revisions +### Optional alternative-wording checks + +The default 100-task benchmark is unchanged. To run the separate public +paraphrase suite of 12 tasks: + +```bash +cd seatunnel-cli +python -m benchmark.runner --provider openai --model gpt-4o \ + --suite paraphrase --out benchmark/paraphrase-baseline +``` + +The suite covers routing, CDC prerequisites and connector options/mode selection, +including one Chinese routing prompt. It runs only variants, without adding the +original tasks to the denominator. `--tiers` filters inherited tiers; `--tasks` +uses distinct IDs such as `t2_cdc_pg_kafka_p1`. Invalid or duplicate variant +selections fail before provider setup. The usual generation, repair and gate +pipeline applies: this command calls a model, even with `--level l1`. + +Each variant inherits all assertions and execution fixtures from a fingerprinted +baseline task. Changed parents require review and explicit repinning, not silent +expectation updates. Saved results retain parent provenance and fingerprint the +complete expanded variant. Run the same suite in a separate candidate directory +and use the comparison command below; do not compare parent IDs against variant +IDs or combine their rates as independent evidence. +New results record the run-level `suite`; cross-suite comparisons are rejected +before task pairing. Older unmarked baseline results remain compatible, but +unmarked paraphrase results need a fresh run. This initial corpus intentionally +has one reviewed alternative wording per parent; additional wordings are a +separate corpus expansion. + +This public suite is a regression tool, **not an unseen holdout**. Offline tests +validate its harness contracts, not generation accuracy or full semantic +equivalence of output data. Existing default prompts, scoring and report formats +are unchanged. See the benchmark README for the corpus contract. + +### Saved-result comparison + Save baseline and candidate benchmark runs in separate directories, then compare their `results.json` files without making additional model calls: diff --git a/docs/zh/ai-cli/benchmark.md b/docs/zh/ai-cli/benchmark.md index e2f14bd16c..aa74ae4d9a 100644 --- a/docs/zh/ai-cli/benchmark.md +++ b/docs/zh/ai-cli/benchmark.md @@ -4,6 +4,38 @@ sidebar_position: 4 # 模型基准测试 +## 可选的改写提示词测试集 + +默认的 100 个基准任务保持不变。使用以下命令单独运行包含 12 个任务的公开改写测试集： + +```bash +cd seatunnel-cli +python -m benchmark.runner --provider openai --model gpt-4o \ + --suite paraphrase --out benchmark/paraphrase-baseline +``` + +该测试集覆盖路由、CDC 前置条件以及连接器选项和模式选择，其中包含一个中文路由提示词。 +它只运行改写任务，不会把原始任务一起计入分母。`--tiers` 按继承的复杂度层级筛选， +`--tasks` 使用独立 ID，例如 `t2_cdc_pg_kafka_p1`。无效、重复、空或不属于所选层级的 +改写任务选择会在初始化模型提供方之前报错。生成、修复和验证流程仍沿用现有实现； +即使指定 `--level l1`，上述命令也会调用模型。 + +每个改写任务继承原始任务的全部断言和执行测试数据，并通过 `parent_sha256` 固定完整的 +原始任务定义。原始任务发生变化时，必须重新检查语义等价性并显式更新指纹，不能静默 +改变期望结果。保存的结果包含 `parent_id` 和 `parent_sha256`，`task_sha256` 则覆盖 +展开后的完整改写任务，包括提示词和来源信息。 + +在不同输出目录中运行同一测试集的基准和候选版本，再使用 `benchmark.compare` 比较相同 +改写任务 ID 的结果。不要把原始任务与改写任务直接配对，也不要将两者的通过率合并为独立证据。 +新结果在运行级别记录 `suite`，不同测试集的比较会在任务配对前被拒绝。旧版未标记的 +基准结果仍兼容，但未标记的改写结果需要重新运行。首个版本有意为每个原始任务保留 +一个经过审查的改写提示词，更多改写属于后续测试集扩展。 +现有默认提示词、评分、报告格式以及跳过验证层的排除规则均保持不变。 + +这是一个**公开回归测试集，而不是未见过的保留测试集**。离线测试只能验证任务选择、 +测试数据继承、评分和报告契约，不能证明模型准确率、泛化能力或输出数据的完整语义等价性。 +模型质量结论仍需要实际模型运行；生产提示词调整应与测试集变更分开进行。 + AI CLI 的准确率靠实测而非假设：专门的基准测试包含 100 个任务、三档复杂度，判定门层层递进直至**真实作业执行**（连接 Docker 化的真实数据源），覆盖 7 个主流大模型。本页汇总方法论、测试结果与选型建议。 ## 比较 CLI 版本 diff --git a/seatunnel-cli/benchmark/README.md b/seatunnel-cli/benchmark/README.md index d3f32b3cf4..e4d877da6f 100644 --- a/seatunnel-cli/benchmark/README.md +++ b/seatunnel-cli/benchmark/README.md @@ -109,6 +109,56 @@ Task format: } ``` +### Optional public paraphrase suite + +The default remains the same 100 baseline tasks. Select `--suite paraphrase` +to run **only 12 alternative-wording tasks**, not the baseline plus its variants: + +```bash +cd seatunnel-cli +python -m benchmark.runner --provider openai --model gpt-4o \ + --suite paraphrase --out benchmark/paraphrase-baseline + +# Filter using the inherited tier and the distinct variant ID. +python -m benchmark.runner --provider openai --model gpt-4o \ + --suite paraphrase --tiers 2 --tasks t2_cdc_pg_kafka_p1 \ + --out benchmark/paraphrase-postgres +``` + +These commands call the selected model and use the normal gate and repair +pipeline. `--level l1` removes engine execution, **not model calls**. The wrapper +`run_benchmark.sh` also forwards `--suite`. Invalid, duplicate, empty, or +out-of-tier task selections in the paraphrase suite fail before provider setup. +Baseline selection behavior is unchanged. + +`tasks/paraphrase.json` contains four routing variants, four CDC variants, and +four connector-option/mode variants, including one Chinese routing prompt. +Each record provides only `parent_id`, `parent_sha256`, and alternative `prompt`. +The loader copies the entire canonical task, including assertions and execution +fixtures, and uses `<parent_id>_p1` as its stable task ID. It refuses unknown or +duplicate parents, assertion overrides, unchanged wording, and parent fingerprint +mismatches. A parent edit requires reviewing semantic equivalence and explicitly +repinning the full parent contract; do not automatically refresh pins. +This first slice intentionally uses one reviewed wording per parent; multiple +wordings and their ID/weighting policy are left for a separate corpus expansion. + +Saved variant entries include `parent_id` and `parent_sha256`. The existing +`task_sha256` covers the expanded variant, including its wording and provenance. +New results also record `suite` at run level. Cross-suite comparisons are rejected +before task pairing. Older baseline results without a suite marker remain +compatible; unmarked paraphrase results must be collected again with this marker. +Keep baseline and candidate runs in separate directories and compare the same +variant IDs using `benchmark.compare`; parent tasks and variants are different +tasks, not directly paired samples. Changed wording/contracts are excluded from +cross-revision comparisons. Existing report formats and skipped-gate exclusions +are unchanged. + +This is a **public regression corpus**, not an unseen holdout or a measurement of +generalization. A wording-sensitive regression can only be established by actual +model runs. Offline tests verify selection, fixture inheritance, scoring and +reporting contracts; they do not prove model accuracy or full output-data semantic +equivalence. Keep production prompt tuning separate from corpus changes. + ## Docker data environment `docker/docker-compose.yml` provides sources and sinks with pre-seeded data diff --git a/seatunnel-cli/benchmark/compare.py b/seatunnel-cli/benchmark/compare.py index 3be9d0af38..949eb2f4d4 100644 --- a/seatunnel-cli/benchmark/compare.py +++ b/seatunnel-cli/benchmark/compare.py @@ -49,6 +49,22 @@ def _run_issues(baseline: dict, candidate: dict) -> list[str]: for name, run in (("baseline", baseline), ("candidate", candidate)): if not isinstance(run, dict): raise ValueError(f"{name} results must be an object") + # Results predating suite selection are baseline runs. Older paraphrase + # results carry task provenance, so require a marker instead of guessing. + suite = run.get("suite", "baseline") + if suite not in ("baseline", "paraphrase"): + issues.append(f"{name}: suite metadata invalid") + models = run.get("models", []) + if isinstance(models, list): + for model in models: + tasks = model.get("tasks", []) if isinstance(model, dict) else [] + if isinstance(tasks, list) and any( + isinstance(task, dict) + and ("parent_id" in task) != (suite == "paraphrase") + for task in tasks + ): + issues.append(f"{name}: suite metadata does not match task provenance") + break if ( not isinstance(run.get("cli"), dict) or not isinstance(run["cli"].get("cli_commit"), str) @@ -65,6 +81,8 @@ def _run_issues(baseline: dict, candidate: dict) -> list[str]: for field in ("levels", "trials", "max_repairs"): if baseline.get(field) != candidate.get(field): issues.append(f"{field} differs between runs") + if baseline.get("suite", "baseline") != candidate.get("suite", "baseline"): + issues.append("suite differs between runs") return issues diff --git a/seatunnel-cli/benchmark/paraphrases.py b/seatunnel-cli/benchmark/paraphrases.py new file mode 100644 index 0000000000..2c9296209d --- /dev/null +++ b/seatunnel-cli/benchmark/paraphrases.py @@ -0,0 +1,73 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Public alternative wording, pinned to reviewed baseline task contracts.""" + +import copy +import hashlib +import json +from pathlib import Path + +PARAPHRASES_PATH = Path(__file__).resolve().parent / "tasks" / "paraphrase.json" + + +def load_paraphrases(parents: list[dict]) -> list[dict]: + """Inherit a reviewed task without silently accepting contract drift.""" + by_id = {task["id"]: task for task in parents} + if len(by_id) != len(parents): + raise ValueError("Duplicate baseline task IDs") + data = json.loads(PARAPHRASES_PATH.read_text(encoding="utf-8")) + variants = data.get("paraphrases") if isinstance(data, dict) else None + if not isinstance(variants, list) or not variants: + raise ValueError("Expected a nonempty paraphrases list") + tasks = [] + seen = set(by_id) + for variant in variants: + if not isinstance(variant, dict) or set(variant) != { + "parent_id", + "parent_sha256", + "prompt", + }: + raise ValueError( + "Paraphrases may only specify parent_id, parent_sha256 and prompt" + ) + parent_id = variant["parent_id"] + if not isinstance(parent_id, str) or parent_id not in by_id: + raise ValueError(f"Unknown paraphrase parent: {parent_id!r}") + task_id = parent_id + "_p1" + if task_id in seen: + raise ValueError(f"Duplicate paraphrase task ID: {task_id}") + seen.add(task_id) + parent = by_id[parent_id] + fingerprint = hashlib.sha256( + json.dumps( + parent, sort_keys=True, ensure_ascii=False, separators=(",", ":") + ).encode("utf-8") + ).hexdigest() + if variant["parent_sha256"] != fingerprint: + raise ValueError(f"Paraphrase parent changed; review and repin {parent_id}") + prompt = variant["prompt"] + if ( + not isinstance(prompt, str) + or not prompt.strip() + or prompt.strip() == parent["prompt"].strip() + ): + raise ValueError(f"Paraphrase needs distinct, nonempty wording: {task_id}") + task = copy.deepcopy(parent) + task.update(variant, id=task_id) + tasks.append(task) + return tasks diff --git a/seatunnel-cli/benchmark/runner.py b/seatunnel-cli/benchmark/runner.py index 2775df8756..811602b203 100644 --- a/seatunnel-cli/benchmark/runner.py +++ b/seatunnel-cli/benchmark/runner.py @@ -67,7 +67,38 @@ CLARIFICATION_REPLY = ( ) -def load_tasks(tiers: list[int], task_ids: list[str] | None = None) -> list[dict]: +def load_tasks( + tiers: list[int], task_ids: list[str] | None = None, suite: str = "baseline" +) -> list[dict]: + if suite not in ("baseline", "paraphrase"): + raise ValueError(f"Unknown task suite: {suite}") + if suite == "paraphrase": + from benchmark.paraphrases import load_paraphrases + + if ( + not tiers + or len(tiers) != len(set(tiers)) + or any(t not in TIER_FILES for t in tiers) + ): + raise ValueError("Select distinct tiers from 1, 2 and 3") + tasks = [ + task + for task in load_paraphrases(load_tasks(list(TIER_FILES))) + if task["tier"] in tiers + ] + if task_ids is not None: + if not task_ids or len(task_ids) != len(set(task_ids)): + raise ValueError("Select one or more distinct paraphrase task IDs") + unknown = set(task_ids) - {task["id"] for task in tasks} + if unknown: + raise ValueError( + "Unknown paraphrase tasks in selected tiers: " + + ", ".join(sorted(unknown)) + ) + tasks = [task for task in tasks if task["id"] in task_ids] + if not tasks: + raise ValueError("No paraphrase tasks selected") + return tasks tasks = [] for tier in tiers: path = TASKS_DIR / TIER_FILES[tier] @@ -319,7 +350,12 @@ def collect_cli_fingerprint() -> dict: def run_benchmark(models: list[dict], tasks: list[dict], levels: list[str], - max_repairs: int, trials: int, out_dir: Path) -> dict: + max_repairs: int, trials: int, out_dir: Path, + suite: str = "baseline") -> dict: + if suite not in ("baseline", "paraphrase"): + raise ValueError("Unknown benchmark suite") + if any(("parent_id" in task) != (suite == "paraphrase") for task in tasks): + raise ValueError("Task provenance does not match the selected suite") out_dir.mkdir(parents=True, exist_ok=True) configs_dir = out_dir / "configs" configs_dir.mkdir(exist_ok=True) @@ -333,6 +369,7 @@ def run_benchmark(models: list[dict], tasks: list[dict], levels: list[str], } all_results = { + "suite": suite, "levels": levels, "max_repairs": max_repairs, "trials": trials, @@ -369,6 +406,9 @@ def run_benchmark(models: list[dict], tasks: list[dict], levels: list[str], "task_sha256": task_fingerprints[task["id"]], "trials": [], } + if "parent_id" in task: + task_entry.update(parent_id=task["parent_id"], + parent_sha256=task["parent_sha256"]) for trial in range(trials): label = task["id"] + (f" trial {trial + 1}/{trials}" if trials > 1 else "") print(f" [{label}] ...", end="", flush=True) @@ -541,6 +581,10 @@ def main() -> None: "(DeepSeek, Azure, local vLLM, ...)") parser.add_argument("--tiers", type=int, nargs="+", default=[1, 2, 3], choices=[1, 2, 3]) + parser.add_argument("--suite", choices=["baseline", "paraphrase"], + default="baseline", + help="Task suite: baseline (100 tasks, default) or " + "paraphrase (12 alternative-wording tasks only)") parser.add_argument("--tasks", nargs="*", default=None, help="Optional task id filter") parser.add_argument("--level", default="l3", choices=["l1", "l2", "l3"], @@ -555,13 +599,19 @@ def main() -> None: parser.add_argument("--out", default="benchmark/results") args = parser.parse_args() + # Reject invalid task selections before provider setup can require credentials. + try: + tasks = load_tasks(args.tiers, args.tasks, args.suite) + except (ValueError, OSError) as error: + parser.error(str(error)) models = build_models_from_args(args) - tasks = load_tasks(args.tiers, args.tasks) if not tasks: print("No tasks selected.", file=sys.stderr) sys.exit(1) levels = resolve_levels(args.level, tasks) + if args.suite == "paraphrase": + print("Task suite: paraphrase (public alternative wording)") print(f"Running {len(tasks)} tasks × {len(models)} models × " f"{args.trials} trial(s), gates: {' → '.join(levels)}, " f"max repairs: {args.max_repairs}") @@ -578,7 +628,7 @@ def main() -> None: os.environ.setdefault(key, value) results = run_benchmark(models, tasks, levels, args.max_repairs, - args.trials, Path(args.out)) + args.trials, Path(args.out), suite=args.suite) from benchmark.report import print_summary, write_reports print_summary(results) diff --git a/seatunnel-cli/benchmark/tasks/paraphrase.json b/seatunnel-cli/benchmark/tasks/paraphrase.json new file mode 100644 index 0000000000..cbd44618c8 --- /dev/null +++ b/seatunnel-cli/benchmark/tasks/paraphrase.json @@ -0,0 +1,81 @@ +{ + "_license": [ + "Licensed to the Apache Software Foundation (ASF) under one or more", + "contributor license agreements. See the NOTICE file distributed with", + "this work for additional information regarding copyright ownership.", + "The ASF licenses this file to You under the Apache License, Version 2.0", + "(the \"License\"); you may not use this file except in compliance with", + "the License. You may obtain a copy of the License at", + "", + " http://www.apache.org/licenses/LICENSE-2.0";, + "", + "Unless required by applicable law or agreed to in writing, software", + "distributed under the License is distributed on an \"AS IS\" BASIS,", + "WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.", + "See the License for the specific language governing permissions and", + "limitations under the License." + ], + "description": "Public paraphrases; inherit reviewed baseline task contracts without changing assertions or fixtures.", + "paraphrases": [ + { + "parent_id": "t1_probe_parquet_conditional", + "parent_sha256": "af32bf2a72fda5b103550a02f4f3519daedd4d049d5c909359cbddb678860682", + "prompt": "I need a batch export of shop.products from MySQL at localhost:3306, using root. Save the data locally in /tmp/st_bench/products_parquet as Parquet, compressed with snappy." + }, + { + "parent_id": "t1_probe_streaming_checkpoint", + "parent_sha256": "374f09790dfc69d2ed9588d8c61c39fb2b03a7f3580785ba11ef031caf55fdeb", + "prompt": "Keep producing synthetic records indefinitely and display them on the console. Each record has id of type bigint and name of type string; the source must not stop after a fixed number of rows." + }, + { + "parent_id": "t2_cdc_mysql_initial_snapshot", + "parent_sha256": "58380d160e4479a06dce4f27ebcc0a54b9ce4e62652ad0ee0dc6b3bc74ae3360", + "prompt": "For shop.users in MySQL at localhost:3306 (user root), print all existing rows to the console first, then keep printing incremental changes through CDC. Include the full initial snapshot before following changes." + }, + { + "parent_id": "t2_cdc_pg_kafka", + "parent_sha256": "cb054552e1319c3b09c8abd7379e23b71cbed5c6a5a611caa13fc3ae1ba747be", + "prompt": "Send a live CDC feed from the inventory table in PostgreSQL database analytics to Kafka topic wms.inventory at localhost:9092. PostgreSQL is at localhost:5432, the login is bench, logical replication is already enabled, and the decoding plugin is pgoutput." + }, + { + "parent_id": "t2_cdc_probe_batch_mode", + "parent_sha256": "20cf05311949730f71d1778ee3ce496e821e8fdd7dbdc70985b504ffaa2bfa9a", + "prompt": "Whenever orders change in MySQL database shop, show the changes on the console and keep listening for later changes. Connect as root to localhost:3306. This replication should continue without ending." + }, + { + "parent_id": "t2_cdc_probe_privileges", + "parent_sha256": "1f018216caa10640944adfc73e3c0459cc74e39c18e70c4fd465a0bdf6d0426b", + "prompt": "Monitor shop.orders with MySQL CDC and send the changes to Console, connecting to localhost:3306 as root. Other CDC jobs share this server: allocate this job a separate, unique server id range so they do not conflict." + }, + { + "parent_id": "t2_si_probe_group_offsets", + "parent_sha256": "5f4c435a7875f430d3256795e492a22d843b3debfdd515a966c7a1ed34d09a12", + "prompt": "Resume Kafka consumption at the offsets already committed for group st_bench_group, then stream messages to Console. Use topic events at localhost:9092 with fields id bigint, name string, and ts string." + }, + { + "parent_id": "t2_bs_probe_sink_options_on_source", + "parent_sha256": "ab96a591e60f173ef1fbe9a039a63abc04cae93bc31a5e3ae21ea33105a5923c", + "prompt": "Run a batch copy from shop.users to the existing shop.users_replica table on the same MySQL server, localhost:3306, with user root. Have the sink derive its INSERT statements from the destination database and table; do not supply handwritten INSERT SQL." + }, + { + "parent_id": "t3_cr_amount_split", + "parent_sha256": "26eca94cabaac65274f42f5965918dafde06fbdf00226e66d4fb6fd8a3e4f6ee", + "prompt": "Make a batch job reading shop.transactions from MySQL at localhost:3306 as root. Feed that one source output into two SQL transforms: amount >= 1000 goes to JSON files in /tmp/st_bench/tx_large, while amount < 1000 goes to JSON files in /tmp/st_bench/tx_small." + }, + { + "parent_id": "t3_cr_status_routing", + "parent_sha256": "d03acd9c710a5f3853b7d2f1033333619055fccecff784b4e24dbc68ad439df8", + "prompt": "In one batch job, read shop.orders once from MySQL localhost:3306 using root. Branch through two SQL transforms: send status = 'PAID' to Console and status <> 'PAID' to local JSON files at /tmp/st_bench/unpaid." + }, + { + "parent_id": "t3_cr_log_level_split_zh", + "parent_sha256": "9e11eee428d592410fe74e3e06176389ae85a6eabaf0b89673071017af2ec810", + "prompt": "请创建批处理任务，连接 localhost:5432 上的 PostgreSQL，以 bench 用户读取 analytics 数据库的 app_logs 表。通过两个 SQL transform 分流：level = 'ERROR' 的记录输出到控制台，level <> 'ERROR' 的记录以 JSON 格式保存到本地 /tmp/st_bench/non_error_logs。" + }, + { + "parent_id": "t3_cr_stream_split", + "parent_sha256": "37f1ef729bb42f0f0c7171070b6cc982c6605e372ae6a61bf67db4100bd0705e", + "prompt": "Use one Kafka source to continuously read JSON from clicks on localhost:9092, starting at the earliest offset. Its schema is event_id string, user_id bigint, url string. With two SQL transforms, publish user_id = 1 events to vip_clicks and all other events to other_clicks on that same broker." + } + ] +} diff --git a/seatunnel-cli/tests/test_benchmark_comparison.py b/seatunnel-cli/tests/test_benchmark_comparison.py index ca5aadc3b2..dec249bc3f 100644 --- a/seatunnel-cli/tests/test_benchmark_comparison.py +++ b/seatunnel-cli/tests/test_benchmark_comparison.py @@ -117,6 +117,49 @@ def test_identical_runs_are_unchanged_and_inputs_are_immutable(): ) +def test_legacy_baseline_results_compare_with_explicit_baseline_suite(): + before = results({"a": True}) + after = copy.deepcopy(before) + after["suite"] = "baseline" + comparison = compare_results(before, after) + assert not comparison["issues"] + assert len(comparison["rows"]) == 1 + + +def test_cross_suite_comparison_is_rejected_before_task_pairing(): + before = results({"a": True}) + before["suite"] = "baseline" + after = results({"a_p1": True}) + after["suite"] = "paraphrase" + after["models"][0]["tasks"][0]["parent_id"] = "a" + comparison = compare_results(before, after) + assert comparison["issues"] == ["suite differs between runs"] + assert not comparison["rows"] + + [email protected]("suite", [None, "unknown", [], {}]) +def test_invalid_suite_metadata_is_not_scored(suite): + before = results({"a": True}) + after = copy.deepcopy(before) + after["suite"] = suite + comparison = compare_results(before, after) + assert "candidate: suite metadata invalid" in comparison["issues"] + assert not comparison["rows"] + + [email protected]("suite", [None, "baseline", "paraphrase"]) +def test_suite_and_task_provenance_must_agree(suite): + before = results({"a": True}) + after = copy.deepcopy(before) + if suite is not None: + after["suite"] = suite + if suite != "paraphrase": + after["models"][0]["tasks"][0]["parent_id"] = "parent" + comparison = compare_results(before, after) + assert "candidate: suite metadata does not match task provenance" in comparison["issues"] + assert not comparison["rows"] + + @pytest.mark.parametrize( "key,value", [ @@ -280,6 +323,7 @@ def test_runner_saved_results_compare_without_changing_legacy_reports( models = [{"name": "model-a", "provider": "openai", "model": "model-a"}] original_tasks = copy.deepcopy(tasks) first = run_benchmark(models, tasks, ["l1"], 3, 2, tmp_path / "before") + assert first["suite"] == "baseline" second = run_benchmark(models, tasks, ["l1"], 3, 2, tmp_path / "after") before = json.loads((tmp_path / "before/results.json").read_text()) after = json.loads((tmp_path / "after/results.json").read_text()) diff --git a/seatunnel-cli/tests/test_benchmark_paraphrases.py b/seatunnel-cli/tests/test_benchmark_paraphrases.py new file mode 100644 index 0000000000..b9d3dee3ce --- /dev/null +++ b/seatunnel-cli/tests/test_benchmark_paraphrases.py @@ -0,0 +1,386 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one or more +# contributor license agreements. See the NOTICE file distributed with +# this work for additional information regarding copyright ownership. +# The ASF licenses this file to You under the Apache License, Version 2.0 +# (the "License"); you may not use this file except in compliance with +# the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +# + +"""Offline contracts for the optional, public alternative-wording suite.""" + +import copy +import hashlib +import json +import re +import subprocess +import sys +from pathlib import Path +from unittest import mock + +import pytest + +sys.path.insert(0, str(Path(__file__).parent.parent)) + +from benchmark import paraphrases, runner # noqa: E402 +from benchmark.compare import compare_results # noqa: E402 +from benchmark.report import write_reports # noqa: E402 +from benchmark.scoring import score_task # noqa: E402 + +VARIANTS = json.loads(paraphrases.PARAPHRASES_PATH.read_text(encoding="utf-8"))[ + "paraphrases" +] + + +def replace_variants(monkeypatch, tmp_path, variants): + path = tmp_path / "paraphrase.json" + path.write_text(json.dumps({"paraphrases": variants}), encoding="utf-8") + monkeypatch.setattr(paraphrases, "PARAPHRASES_PATH", path) + + +def fingerprint(task): + return hashlib.sha256( + json.dumps( + task, sort_keys=True, ensure_ascii=False, separators=(",", ":") + ).encode("utf-8") + ).hexdigest() + + +STREAM_CONFIG = """ +env { + parallelism = 1 + job.mode = "STREAMING" + checkpoint.interval = 10000 +} +source { + FakeSource { + schema { fields { id = "bigint", name = "string" } } + } +} +sink { Console {} } +""" + + +def test_baseline_stays_identical_and_does_not_load_paraphrases(monkeypatch, tmp_path): + replace_variants(monkeypatch, tmp_path, [{"invalid": "ignored"}]) + tasks = runner.load_tasks([1, 2, 3]) + assert len(tasks) == 100 + assert tasks == runner.load_tasks([1, 2, 3], suite="baseline") + # Baseline loading must preserve every canonical field and its ordering, + # without adding variant metadata that changes saved-result fingerprints. + canonical = [] + for tier, filename in runner.TIER_FILES.items(): + data = json.loads((runner.TASKS_DIR / filename).read_text(encoding="utf-8")) + canonical.extend(dict(task, tier=tier) for task in data["tasks"]) + assert tasks == canonical + assert fingerprint(tasks) == fingerprint(canonical) + assert all("parent_id" not in task for task in tasks) + assert runner.load_tasks([1], ["t1_probe_parquet_conditional_p1"]) == [] + assert len(runner.load_tasks([1], [])) == 20 + + [email protected]("variant", VARIANTS, ids=lambda item: item["parent_id"]) +def test_variants_inherit_the_complete_task_without_aliasing(variant): + parents = runner.load_tasks([1, 2, 3]) + original = copy.deepcopy(parents) + parent = next(task for task in parents if task["id"] == variant["parent_id"]) + task = next( + task + for task in paraphrases.load_paraphrases(parents) + if task["parent_id"] == parent["id"] + ) + assert task["id"] == parent["id"] + "_p1" + assert task["prompt"] != parent["prompt"] + assert task["parent_sha256"] == fingerprint(parent) + assert fingerprint(task) != fingerprint(parent) + inherited = { + key: value + for key, value in task.items() + if key not in ("id", "prompt", "parent_id", "parent_sha256") + } + assert inherited == { + key: value for key, value in parent.items() if key not in ("id", "prompt") + } + assert task["expect"]["source"] and task["expect"]["sink"] + assert task["execution"]["l3"] in ("run", "skip") + assert (task["expect"]["job_mode"] == "STREAMING") == ( + task["execution"]["mode"] == "streaming" + ) + for pattern in task["expect"].get("must_match", []) + task["expect"].get( + "must_not_match", [] + ): + re.compile(pattern) + task["expect"]["source"].clear() + task["execution"]["services"].append("additional-service") + assert parents == original + + +def test_selection_is_opt_in_and_tier_filtered(): + tasks = runner.load_tasks([1, 2, 3], suite="paraphrase") + assert len(tasks) == len({task["id"] for task in tasks}) == 12 + assert [ + len(runner.load_tasks([tier], suite="paraphrase")) for tier in (1, 2, 3) + ] == [2, 6, 4] + assert not ( + {task["id"] for task in tasks} + & {task["id"] for task in runner.load_tasks([1, 2, 3])} + ) + assert runner.load_tasks([1], [tasks[0]["id"]], "paraphrase") == [tasks[0]] + + [email protected]( + "tiers,ids", + [ + ([1], []), + ([1], ["unknown"]), + ([1], ["t1_probe_parquet_conditional"]), + ([1], ["t1_probe_parquet_conditional_p1", "unknown"]), + ([1], ["t1_probe_parquet_conditional_p1"] * 2), + ([2], ["t1_probe_parquet_conditional_p1"]), + ([1, 1], None), + ([], None), + ([4], None), + ], +) +def test_invalid_paraphrase_selection_fails_closed(tiers, ids): + with pytest.raises(ValueError): + runner.load_tasks(tiers, ids, "paraphrase") + + [email protected]( + "field", ["prompt", "expect", "execution", "tier", "category", "name"] +) +def test_parent_changes_require_review_and_repinning(field): + parents = runner.load_tasks([1, 2, 3]) + parent = next(task for task in parents if task["id"] == VARIANTS[0]["parent_id"]) + parent[field] = "changed" + with pytest.raises(ValueError, match="review and repin"): + paraphrases.load_paraphrases(parents) + + [email protected]( + "change", + [ + lambda variants: variants.append(copy.deepcopy(variants[0])), + lambda variants: variants[0].update(parent_id="unknown"), + lambda variants: variants[0].update(parent_id=[]), + lambda variants: variants[0].update(parent_sha256="0" * 64), + lambda variants: variants[0].update(prompt=" "), + lambda variants: variants[0].update(prompt=None), + lambda variants: variants[0].update(expect={}), + lambda variants: variants[0].pop("parent_sha256"), + ], +) +def test_invalid_variant_definitions_are_rejected(change, monkeypatch, tmp_path): + variants = copy.deepcopy(VARIANTS) + change(variants) + replace_variants(monkeypatch, tmp_path, variants) + with pytest.raises(ValueError): + runner.load_tasks([1, 2, 3], suite="paraphrase") + + +def test_identical_prompt_and_duplicate_parent_are_rejected(monkeypatch, tmp_path): + parents = runner.load_tasks([1, 2, 3]) + with pytest.raises(ValueError, match="Duplicate baseline"): + paraphrases.load_paraphrases(parents + [parents[0]]) + variants = copy.deepcopy(VARIANTS) + variants[0]["prompt"] = next( + task["prompt"] for task in parents if task["id"] == variants[0]["parent_id"] + ) + replace_variants(monkeypatch, tmp_path, variants) + with pytest.raises(ValueError, match="distinct"): + paraphrases.load_paraphrases(parents) + + [email protected]( + "content", + [ + "{", + "[]", + "{}", + '{"paraphrases": []}', + '{"paraphrases": {}}', + '{"paraphrases": [null]}', + ], +) +def test_malformed_or_empty_corpus_fails_closed(content, monkeypatch, tmp_path): + path = tmp_path / "paraphrase.json" + path.write_text(content, encoding="utf-8") + monkeypatch.setattr(paraphrases, "PARAPHRASES_PATH", path) + with pytest.raises(ValueError): + runner.load_tasks([1, 2, 3], suite="paraphrase") + + [email protected]( + "config", [STREAM_CONFIG, STREAM_CONFIG.replace("STREAMING", "BATCH"), "invalid {"] +) +def test_same_config_receives_same_verdict_for_every_family(config): + parents = {task["id"]: task for task in runner.load_tasks([1, 2, 3])} + for task in runner.load_tasks([1, 2, 3], suite="paraphrase"): + original = score_task(parents[task["parent_id"]], config).to_dict() + alternate = score_task(task, config).to_dict() + original.pop("task_id") + alternate.pop("task_id") + assert original == alternate + assert score_task(parents["t1_probe_streaming_checkpoint"], STREAM_CONFIG).passed + + +def test_only_variant_wording_reaches_generation_and_skipped_gate_is_not_pass(): + task = runner.load_tasks([1], ["t1_probe_streaming_checkpoint_p1"], "paraphrase")[0] + with mock.patch("seatunnel_cli.agents.Orchestrator") as factory, mock.patch( + "benchmark.execution.run_execute", + return_value={"passed": None, "detail": "SKIPPED: fixture"}, + ): + orchestrator = factory.return_value + orchestrator.process_user_input.return_value = { + "type": "config", + "config": STREAM_CONFIG, + } + record = runner.run_task_with_repairs(object(), task, ["l1", "l3"], 2) + orchestrator.process_user_input.assert_called_once_with(task["prompt"]) + orchestrator._run_fix.assert_not_called() + assert record["first_pass_round"] is None + assert record["attempts"][0]["layers"]["skipped_layers"] == ["l3"] + + +def test_cli_dispatches_selected_suite_before_provider_setup(monkeypatch, tmp_path): + monkeypatch.setattr( + sys, + "argv", + [ + "benchmark.runner", + "--suite", + "paraphrase", + "--tiers", + "1", + "--tasks", + "t1_probe_streaming_checkpoint_p1", + "--level", + "l1", + ], + ) + with mock.patch.object( + runner, "build_models_from_args", return_value=[] + ) as models, mock.patch.object( + runner, "resolve_levels", return_value=["l1"] + ), mock.patch.object( + runner, "run_benchmark", return_value={} + ) as run, mock.patch( + "benchmark.report.print_summary" + ), mock.patch( + "benchmark.report.write_reports" + ), mock.patch.object( + runner.tempfile, "mkdtemp", return_value=str(tmp_path) + ): + runner.main() + selected = run.call_args.args[1] + assert [task["id"] for task in selected] == ["t1_probe_streaming_checkpoint_p1"] + assert run.call_args.kwargs["suite"] == "paraphrase" + models.assert_called_once() + sys.argv[-1] = "l3" + sys.argv[6] = "unknown" + models.reset_mock() + with pytest.raises(SystemExit) as error: + runner.main() + assert error.value.code == 2 + models.assert_not_called() + + [email protected]( + "args", [["--suite", "unknown"], ["--suite", "paraphrase", "--tasks", "unknown"]] +) +def test_invalid_suite_cli_needs_no_third_party_packages_or_provider(args): + completed = subprocess.run( + [sys.executable, "-S", "-m", "benchmark.runner", *args], + capture_output=True, + text=True, + timeout=10, + check=False, + ) + assert completed.returncode == 2 + assert "Traceback" not in completed.stderr + assert "error:" in completed.stderr + + +def test_saved_variants_compare_by_identity_and_keep_provenance(tmp_path, monkeypatch): + monkeypatch.setattr( + runner, "collect_cli_fingerprint", lambda: {"cli_commit": "test-revision"} + ) + monkeypatch.setattr( + "seatunnel_cli.llm_provider.create_provider", lambda _: object() + ) + monkeypatch.setattr( + runner, + "run_task_with_repairs", + lambda *args: { + "attempts": [ + { + "round": 0, + "seconds": 0.1, + "layers": { + "l1": {"passed": True}, + "l2": None, + "l3": None, + "skipped_layers": [], + "all_gates_executed": True, + }, + } + ], + "first_pass_round": 0, + }, + ) + tasks = runner.load_tasks([1, 2, 3], suite="paraphrase") + original = copy.deepcopy(tasks) + models = [{"name": "fixture", "provider": "openai", "model": "fixture"}] + first = runner.run_benchmark( + models, tasks, ["l1"], 0, 1, tmp_path / "before", suite="paraphrase" + ) + runner.run_benchmark( + models, tasks, ["l1"], 0, 1, tmp_path / "after", suite="paraphrase" + ) + saved = json.loads((tmp_path / "after/results.json").read_text()) + assert first["suite"] == saved["suite"] == "paraphrase" + rows = compare_results(first, saved)["rows"] + assert len(rows) == 12 and all(row["first"] == "pass→pass" for row in rows) + assert tasks == original + for task, entry in zip(tasks, saved["models"][0]["tasks"]): + assert entry["task_sha256"] == fingerprint(task) + assert entry["parent_id"] == task["parent_id"] + assert entry["parent_sha256"] == task["parent_sha256"] + without_provenance = copy.deepcopy(saved) + for entry in without_provenance["models"][0]["tasks"]: + del entry["parent_id"], entry["parent_sha256"] + write_reports(saved, tmp_path / "reports") + write_reports(without_provenance, tmp_path / "legacy-reports") + for filename in ("summary.md", "summary.csv"): + assert (tmp_path / "reports" / filename).read_bytes() == ( + tmp_path / "legacy-reports" / filename + ).read_bytes() + tasks[0]["prompt"] += " Revised wording." + changed = runner.run_benchmark( + models, tasks, ["l1"], 0, 1, tmp_path / "changed", suite="paraphrase" + ) + assert ( + sum( + row["reason"] == "task definition differs" + for row in compare_results(first, changed)["rows"] + ) + == 1 + ) + + [email protected]("suite", ["unknown", "baseline"]) +def test_runner_rejects_invalid_suite_before_creating_results(tmp_path, suite): + tasks = runner.load_tasks([1], suite="paraphrase") + output = tmp_path / "invalid" + with pytest.raises(ValueError, match="suite"): + runner.run_benchmark([], tasks, ["l1"], 0, 1, output, suite=suite) + assert not output.exists()
