diff --git a/.github/workflows/delta_spark_ut.yml b/.github/workflows/delta_spark_ut.yml index 3917605354..7d42b16718 100644 --- a/.github/workflows/delta_spark_ut.yml +++ b/.github/workflows/delta_spark_ut.yml @@ -55,6 +55,7 @@ on: paths: - '.github/workflows/delta_spark_ut.yml' - '.github/workflows/util/delta-spark-ut/**' + - 'gluten-ut/src-delta-validation*/**' # Velox revision updates can change Delta offload behavior. - 'ep/build-velox/src/get-velox.sh' - 'gluten-delta/**' @@ -215,6 +216,8 @@ jobs: - uses: actions/checkout@v7 with: ref: ${{ env.DELTA_CHECKOUT_REF }} + - name: Test Delta validation utilities + run: python3 -B -m unittest discover -s gluten-ut/src-delta-validation/test/python - name: Get Ccache from Apache Stash id: ccache-stash uses: apache/infrastructure-actions/stash/restore@69afc125e535c4c41e7f1b7470f583087e0f344b @@ -532,6 +535,14 @@ jobs: delta/**/target/surefire-reports/**/*.xml if-no-files-found: warn + - name: Upload Gluten activation validation + if: always() + uses: actions/upload-artifact@v7 + with: + name: delta-spark-ut-validation-shard-${{ matrix.shard }} + path: gluten-validation/ + if-no-files-found: warn + - name: Upload hang watchdog thread dumps if: always() uses: actions/upload-artifact@v7 diff --git a/.github/workflows/util/delta-spark-ut/README.md b/.github/workflows/util/delta-spark-ut/README.md index cecee7f3ea..4700996456 100644 --- a/.github/workflows/util/delta-spark-ut/README.md +++ b/.github/workflows/util/delta-spark-ut/README.md @@ -45,6 +45,12 @@ starts failing** (a regression). | `setup-delta.sh` | Clones Delta, drops in the Gluten bundle, and applies temporary compatibility patches. Takes ` `. | | `apply-delta-test-patches.sh` | Cherry-picks upstream scan fixes, applies the ordered local patch list, and verifies the resulting test sources. Called by `setup-delta.sh` with ` `. | | `patches/*.patch` | Separate patches for CDF plan assertions, mock-S3 fixture writes, the two Parquet row-group fixtures, and 2B-row DV fail-fast behavior. Each patch documents its rationale and removal condition. | +| `install-validation.py` | Installs CI-only runtime checks in the upstream shared, Hive, Java/JUnit, session-test, and listed direct-factory paths. | +| `gluten-ut/src-delta-validation-sbt/` | sbt helper, copied into Delta's meta-build and explicitly applied by the runner to inventory compiled suites and schedule the per-fork native canary. | +| `check-gluten-validation.py` | Requires matching per-test checks and native proof from every fork before any baseline comparison. | +| `validation-factories.txt` | Known direct Spark-creation suites, shared by source installation and the compiled-suite inventory. | +| `validation-exemptions.txt` | Explicit, reasoned exceptions for tests intentionally starting with Gluten disabled. Empty by default. | +| `gluten-ut/src-delta-validation/` | Scala runtime checks, the native canary, and Scala/Python unit tests. Scala sources are compiled only in the CI-created Delta test project. | ## Temporary Delta test patches @@ -120,6 +126,90 @@ Delta extensions and the Delta catalog remain suite-specific rather than becoming JVM defaults, because some suites test behavior when those settings are absent. +## Mandatory Gluten activation validation + +Defaults alone do not prove that the test's actual session is configured correctly +([#13496](https://github.com/apache/gluten/issues/13496)). After the compatibility +patches, setup aliases upstream `SharedSparkSession` imports to a checked fixture +and installs equivalent hooks for `DeltaHiveTest`, `DeltaSessionQueryTest`, Java's +`DeltaSQLCommandJavaTest`, and the direct factories in `validation-factories.txt`. +The helper sources live under `gluten-ut/src-delta-validation`; no production +Gluten classes or dependency JARs are modified. + +Generated declarations and factory calls respect Delta's 100-character line +limit. Copied helper files keep their original ASF license headers; only Delta's +copyright-specific `header.matches` rule is exempted in those copies. Their +remaining Delta style checks still run, and the original sources remain subject +to Gluten's formatting and license checks. + +Each checked test verifies the actual context's plugin configuration, installed +Gluten columnar rules, `spark.gluten.enabled`, and +`gluten.enabledForCurrentThread`. Direct-factory tests check each wrapped +context/session when it is returned. Checks run again after test cleanup to +detect leaked disabling. Scoped disabling inside a test is allowed when restored. +Every check failure is recorded before it is thrown, so even a test that catches +the exception cannot hide a validation violation. + +`GlutenTestGrouping` inventories the **compiled** suite hierarchies independently of +these hooks. Losing a hook therefore produces missing required evidence, rather +than silently turning that suite into an unchecked success. Suites outside the +recognized fixtures/factories are listed explicitly as **uncovered**, not +automatically assumed to be Spark-free. + +Its sbt listener records lossless framework result identities separately from +JUnit XML. Names are Base64-encoded in completed, counted per-suite records, so +spaces, tabs, newlines, and suite-name prefixes remain distinct. The activation +gate matches those raw results to the JVM evidence. JUnit is cross-checked for +suite presence and test counts, not display-name equality: XML normalizes +attribute whitespace, and sbt's JUnit writer can strip a suite-name prefix. +Missing/truncated execution records and listener errors fail the gate. + +The runner applies these settings with sbt's `set` command after the build loads. +Merely adding a root `.sbt` file is insufficient: Delta's project-level test +settings can override it, leaving the validation inventory and per-fork canary +unscheduled. + +Every fork also runs `GlutenActivationSuite`, which executes and checks a small +native aggregation using the inherited defaults. It does not set the Gluten +plugin or enable flag itself. Its process-specific ScalaTest `suiteId` prevents +forks from overwriting one another's JUnit XML. The gate requires native proof +from the requested number of distinct JVMs and from each JVM that recorded a +test check; a successful smoke test in another fork is insufficient. +At least one upstream test must execute on the shard: canaries and harness +self-tests cannot turn an otherwise empty Delta run green. + +`run-delta-tests.sh` uses a fresh evidence directory and runs +`check-gluten-validation.py` **before** the ordinary result comparison. Missing +or malformed evidence, violations, and failing harness tests always fail the +shard, including with `update_baseline=true` or `fail_on_fixed=false`. Neither +known failures nor flaky quarantine can suppress them. The workflow uploads the +compiled inventory, lossless execution records, per-JVM JSONL records, and coverage +summary as `delta-spark-ut-validation-shard-`. + +Intentional default-disabled tests require an entry in +`validation-exemptions.txt`: `#`. An exact +suite followed by `#*` exempts that whole suite. Exempted tests must still emit +matching lifecycle evidence, and the validation harness cannot be exempted. +An exact exemption whose normalized text matches multiple distinct raw test +names is rejected instead of silently exempting all of them. +Temporary `withSQLConf("spark.gluten.enabled" -> "false")` scopes need no entry +when the enabled default is restored. + +**Coverage limit:** these are fixture/factory checks, not instrumentation of +every Spark session or query. Additional sessions created through unwrapped +paths inside otherwise checked tests remain outside this guarantee. Legitimate +Spark fallback and metadata-only tests are allowed; only the controlled canary +must contain a native stage. + +The Python installer/gate tests run before the native build: + +```bash +python3 -B -m unittest discover -s gluten-ut/src-delta-validation/test/python +``` + +The Scala guard tests are copied into Delta's test sources and run with the +suite. Their failures, like canary failures, are infrastructure failures. + ## How the gate works Each test shard: @@ -322,12 +412,28 @@ configuration reaches the sbt launcher and every forked test JVM: ```bash # from the Delta clone prepared by setup-delta.sh (which has the Gluten bundle): -source /.github/workflows/util/delta-spark-ut/java-test-args.sh -./build/sbt -DsparkVersion=4.1 "++ 2.13.17" spark/test # or a single suite via testOnly +GLUTEN_REPO= +source "$GLUTEN_REPO/.github/workflows/util/delta-spark-ut/java-test-args.sh" +export NUM_SHARDS=1 SHARD_ID=0 TEST_PARALLELISM_COUNT=1 +export GLUTEN_VALIDATION_DIR="$(mktemp -d)" +export GLUTEN_VALIDATION_EXEMPTIONS="$GLUTEN_REPO/.github/workflows/util/delta-spark-ut/validation-exemptions.txt" +./build/sbt -DsparkVersion=4.1 "++ 2.13.17" \ + 'set spark / Test / testOptions += Tests.Argument(TestFrameworks.ScalaTest, "-u", "target/test-reports")' \ + 'set org.apache.gluten.ci.GlutenTestGrouping.results' \ + 'set org.apache.gluten.ci.GlutenTestGrouping.setting' \ + spark/test + +python3 -B "$GLUTEN_REPO/.github/workflows/util/delta-spark-ut/check-gluten-validation.py" \ + --reports-dir "$PWD" \ + --evidence-dir "$GLUTEN_VALIDATION_DIR" \ + --exemptions "$GLUTEN_VALIDATION_EXEMPTIONS" \ + --expected-forks "$TEST_PARALLELISM_COUNT" ``` `run-delta-tests.sh` sources the same file, so CI and local runs use identical -flags. +flags. The documented validation gate is for a full `spark/test` invocation; +a local `testOnly` selection can filter out the canary, so its results must not +be presented as a fully validated shard. Existing `JAVA_TOOL_OPTIONS` are preserved after the shared defaults, so caller-supplied `-D` properties take precedence. diff --git a/.github/workflows/util/delta-spark-ut/check-gluten-validation.py b/.github/workflows/util/delta-spark-ut/check-gluten-validation.py new file mode 100644 index 0000000000..076c398865 --- /dev/null +++ b/.github/workflows/util/delta-spark-ut/check-gluten-validation.py @@ -0,0 +1,459 @@ +#!/usr/bin/env python3 +# +# 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. + +"""Fail closed on missing/failed Gluten activation checks, outside the failure baseline.""" + +import argparse +import base64 +import binascii +from collections import Counter +import importlib.util +import json +import os +from pathlib import Path +import re +import sys +import xml.etree.ElementTree as ET + +SPEC = importlib.util.spec_from_file_location( + "delta_test_results", Path(__file__).with_name("compare-test-results.py") +) +results = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(results) + +INFRASTRUCTURE = "org.apache.spark.sql.gluten." +CANARY = INFRASTRUCTURE + "GlutenActivationSuite" +CANARY_TEST = "native execution with inherited Gluten defaults" +SBT_SUITE_SELECTOR = "(It is not a test it is a sbt.testing.SuiteSelector)" +EXECUTED = {"Success", "Failure", "Error"} +STATUSES = EXECUTED | {"Skipped", "Ignored", "Canceled", "Pending"} + + +def display_test(key): + return "{}#{}".format(key[0], json.dumps(key[1], ensure_ascii=True)) + + +def decode_identity(value): + try: + return base64.b64decode(value, validate=True).decode("utf-8") + except (binascii.Error, UnicodeDecodeError) as error: + raise ValueError("Invalid execution identity encoding") from error + + +def execution_name(suite, name, fingerprint): + # junit-interface 0.11 prefixes method selectors. ScalaTest names must stay untouched. + if fingerprint in { + "annotation:org.junit.Test", + "subclass:junit.framework.TestCase", + }: + for prefix in (suite + ".", suite.rsplit(".", 1)[-1] + "."): + if name.startswith(prefix): + return name[len(prefix) :] + return name + + +def read_executions(directory): + failures = sorted(directory.glob("execution-error-*.txt")) + if failures: + raise ValueError( + "Execution result listener failed:\n" + + "\n".join(path.read_text(encoding="utf-8") for path in failures[:3]) + ) + paths = sorted(directory.glob("execution-*.tsv")) + if not paths: + raise ValueError( + "Missing lossless execution results; the sbt listener did not run" + ) + outcomes = {} + groups = Counter() + aborted = set() + canaries = [] + for path in paths: + text = path.read_text(encoding="utf-8") + if not text.endswith("\n"): + raise ValueError("Truncated execution results: " + str(path)) + lines = text[:-1].split("\n") + if len(lines) < 2: + raise ValueError("Incomplete execution results: " + str(path)) + header, footer = lines[0].split("\t"), lines[-1].split("\t") + if ( + len(header) != 3 + or header[:2] != ["version", "1"] + or len(footer) != 3 + or footer[0] != "complete" + or footer[1] not in {"Passed", "Failed", "Error"} + or not re.fullmatch(r"0|[1-9][0-9]*", footer[2]) + or int(footer[2]) != len(lines) - 2 + ): + raise ValueError("Invalid execution result framing: " + str(path)) + group = decode_identity(header[2]) + if not group: + raise ValueError("Empty execution result group: " + str(path)) + groups[group] += 1 + if groups[group] > 1 and group != CANARY: + raise ValueError("Duplicate execution result group: " + group) + failed_event = False + for line in lines[1:-1]: + parts = line.split("\t") + if ( + len(parts) != 5 + or parts[0] not in {"test", "suite"} + or parts[1] not in STATUSES + ): + raise ValueError("Invalid execution event: " + str(path)) + kind, status = parts[:2] + suite, name, fingerprint = map(decode_identity, parts[2:]) + if ( + not suite + or (kind == "test" and not name) + or (kind == "suite" and name) + or ( + not re.fullmatch(r"(annotation|subclass):[\w.$]+", fingerprint) + and not (kind == "suite" and fingerprint == "") + ) + ): + raise ValueError("Invalid execution event identity: " + str(path)) + failed_event |= status in {"Failure", "Error"} + if kind == "suite": + if status in {"Failure", "Error"}: + aborted.add(suite) + continue + key = (suite, execution_name(suite, name, fingerprint)) + if key == (CANARY, CANARY_TEST): + canaries.append(status) + elif key in outcomes: + raise ValueError("Duplicate execution result: " + display_test(key)) + outcomes[key] = status + if footer[1] != "Passed" and not failed_event: + aborted.add(group) + return outcomes, groups, aborted, canaries + + +def junit_counts(reports): + """Cross-check cardinality, never identities, across duplicate JUnit reporters.""" + counts = Counter() + for path in reports.rglob("TEST-*.xml"): + root = ET.parse(path).getroot() + per_file = Counter() + for suite in results._iter_testsuites(root): + name = suite.get("name") or "" + per_file[name] += 0 + for case in suite: + if case.tag.split("}")[-1] != "testcase": + continue + if case.get("name") == SBT_SUITE_SELECTOR: + continue + per_file[case.get("classname") or name] += 1 + for suite, count in per_file.items(): + counts[suite] = max(counts[suite], count) + return counts + + +def read_exemptions(path): + entries = {} + for line in path.read_text(encoding="utf-8").splitlines(): + line = line.strip() + if not line or line.startswith("#"): + continue + parts = line.split("\t", 1) + if ( + len(parts) != 2 + or not parts[1].strip() + or not re.fullmatch(r"[\w.$]+#.+", parts[0]) + ): + raise ValueError( + "Exemptions require #: " + line + ) + if parts[0].startswith(INFRASTRUCTURE) or parts[0] in entries: + raise ValueError("Duplicate or infrastructure exemption: " + parts[0]) + entries[parts[0]] = parts[1].strip() + return entries + + +def read_inventory(directory): + inventory = {} + for line in ( + (directory / "expected-suites.tsv").read_text(encoding="utf-8").splitlines() + ): + parts = line.split("\t") + if len(parts) != 2 or parts[0] not in { + "required", + "uncovered", + "canary", + "self-test", + }: + raise ValueError("Malformed compiled-suite inventory: " + line) + role, suite = parts + if not re.fullmatch(r"[\w.$]+", suite) or suite in inventory: + raise ValueError("Invalid or duplicate inventory suite: " + suite) + inventory[suite] = role + if inventory.get(CANARY) != "canary": + raise ValueError("Compiled-suite inventory is missing the native canary") + return inventory + + +def read_evidence(directory): + paths = sorted(directory.glob("jvm-*.jsonl")) + if not paths: + raise ValueError( + "No JVM validation evidence; the validation hooks may not have run" + ) + rows = [] + for path in paths: + text = path.read_text(encoding="utf-8") + if not text or not text.endswith("\n"): + raise ValueError("Empty or truncated validation evidence: " + str(path)) + for number, line in enumerate(text[:-1].split("\n"), 1): + try: + row = json.loads(line) + except json.JSONDecodeError as error: + raise ValueError("{}:{}: {}".format(path, number, error)) from error + if ( + not isinstance(row, dict) + or type(row.get("version")) is not int + or row["version"] != 1 + or not isinstance(row.get("pid"), str) + or not re.fullmatch(r"[1-9][0-9]*", row["pid"]) + or row.get("kind") not in {"test", "canary", "violation"} + or not isinstance(row.get("suite"), str) + or not row["suite"] + or not isinstance(row.get("test"), str) + or not row["test"] + ): + raise ValueError("{}:{}: invalid evidence schema".format(path, number)) + rows.append(row) + return rows + + +def validate(reports, evidence, exemptions_path, expected_forks): + if expected_forks < 1: + raise ValueError("Expected fork count must be positive") + inventory = read_inventory(evidence) + planned_forks = int( + (evidence / "expected-forks.txt").read_text(encoding="utf-8").strip() + ) + if planned_forks != expected_forks: + raise ValueError( + "Fork plan {} disagrees with requested parallelism {}".format( + planned_forks, expected_forks + ) + ) + exemptions = read_exemptions(exemptions_path) + xml_passed, xml_failed, xml_skipped, _ = results.parse_reports(str(reports)) + outcomes, groups, aborted, canaries = read_executions(evidence) + executed = {key for key, status in outcomes.items() if status in EXECUTED} + if not any(not suite.startswith(INFRASTRUCTURE) for suite, _ in executed): + raise ValueError( + "No Delta tests executed; validation canaries cannot replace the upstream suite" + ) + reported = set(outcomes) + observed_suites = set(groups) | {suite for suite, _ in reported} | aborted + unknown = observed_suites - inventory.keys() + if unknown: + raise ValueError( + "Execution results contain suites absent from the compiled inventory: " + + ", ".join(sorted(unknown)) + ) + errors = [] + xml_suites = {suite for suite, _ in xml_passed | xml_failed | xml_skipped} + missing_results = xml_suites - observed_suites + if missing_results: + errors.append( + "JUnit suites have no lossless execution results: " + + ", ".join(sorted(missing_results)) + ) + raw_counts = Counter(suite for suite, _ in outcomes) + for suite, count in junit_counts(reports).items(): + if suite != CANARY and raw_counts[suite] != count: + errors.append( + "Execution/JUnit count mismatch for {}: {} raw results, {} JUnit tests".format( + suite, raw_counts[suite], count + ) + ) + required_groups = { + suite + for suite, role in inventory.items() + if role in {"required", "self-test", "canary"} + } + if required_groups - groups.keys(): + errors.append( + "Missing completed execution groups: " + + ", ".join(sorted(required_groups - groups.keys())) + ) + for suite in sorted(aborted): + if inventory[suite] != "uncovered": + errors.append("Required execution suite aborted: " + suite) + exemption_identities = {} + for key in reported: + normalized = results.format_entry(*results.normalize_key(*key)) + exemption_identities.setdefault(normalized, set()).add(key) + for entry in exemptions: + if not entry.endswith("#*") and len(exemption_identities.get(entry, ())) > 1: + errors.append( + "Ambiguous exemption matches distinct raw test names: " + entry + ) + native = set() + checked = {} + for row in read_evidence(evidence): + key = (row["suite"], row["test"]) + if row["kind"] == "violation": + errors.append( + "{} [{}]: {}".format( + display_test(key), + row.get("phase", "unknown"), + row.get("message", "validation failed"), + ) + ) + elif row["kind"] == "canary": + if key != (CANARY, CANARY_TEST) or row.get("status") != "passed": + errors.append("Invalid native canary record: " + repr(row)) + elif row["pid"] in native: + errors.append("Duplicate native canary for JVM " + row["pid"]) + else: + native.add(row["pid"]) + else: + if key in checked: + errors.append("Duplicate validation for " + display_test(key)) + if key not in reported: + errors.append( + "Validation has no matching execution result: " + display_test(key) + ) + expected_reason = exemptions.get( + results.format_entry(*results.normalize_key(*key)), + exemptions.get(key[0] + "#*"), + ) + if row.get("status") == "exempt": + if ( + not expected_reason + or row.get("reason") != expected_reason + or type(row.get("checks")) is not int + or row["checks"] != 0 + ): + errors.append( + "Undeclared validation exemption: " + display_test(key) + ) + elif ( + row.get("status") != "checked" + or type(row.get("checks")) is not int + or row["checks"] < 1 + or expected_reason is not None + ): + errors.append("Invalid completed validation: " + display_test(key)) + checked[key] = row + + if len(native) != expected_forks: + errors.append( + "Expected native proof from {} JVMs, received {}".format( + expected_forks, len(native) + ) + ) + if ( + groups[CANARY] != expected_forks + or len(canaries) != expected_forks + or any(status != "Success" for status in canaries) + ): + errors.append( + "Native canary execution results do not cover every requested fork" + ) + if (CANARY, CANARY_TEST) not in xml_passed: + errors.append("Native canary has no successful JUnit result") + for key, status in outcomes.items(): + if key[0].startswith(INFRASTRUCTURE) and status in {"Failure", "Error"}: + errors.append("Validation harness test failed: " + display_test(key)) + if any(suite.startswith(INFRASTRUCTURE) for suite, _ in xml_failed): + errors.append("Validation harness has a failed JUnit report") + required = {key for key in executed if inventory[key[0]] == "required"} + missing = required - checked.keys() + if missing: + errors.append( + "{} tests are missing required validation:\n{}".format( + len(missing), + "\n".join(display_test(key) for key in sorted(missing)[:20]), + ) + ) + for key, row in checked.items(): + if row["pid"] not in native: + errors.append( + "No native proof for JVM {} running {}".format( + row["pid"], display_test(key) + ) + ) + if errors: + raise ValueError("\n".join(errors[:30])) + uncovered = Counter( + suite + for suite, test in executed + if inventory[suite] == "uncovered" and (suite, test) not in checked + ) + return { + "forks": len(native), + "checked": sum( + key in executed and row["status"] == "checked" + for key, row in checked.items() + ), + "exempt": sum( + key in executed and row["status"] == "exempt" + for key, row in checked.items() + ), + "uncovered": dict(sorted(uncovered.items())), + } + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--reports-dir", required=True, type=Path) + parser.add_argument("--evidence-dir", required=True, type=Path) + parser.add_argument("--exemptions", required=True, type=Path) + parser.add_argument("--expected-forks", required=True, type=int) + args = parser.parse_args() + try: + summary = validate( + args.reports_dir, args.evidence_dir, args.exemptions, args.expected_forks + ) + except (OSError, ValueError, results.NoReportsError) as error: + print( + "::error::Gluten activation validation failed:\n" + str(error), + file=sys.stderr, + ) + return 2 + lines = [ + "## Gluten activation validation", + "", + "{} native-validated JVMs; {} checked tests; {} explicit exemptions.".format( + summary["forks"], summary["checked"], summary["exempt"] + ), + "", + "### Uncovered suites (not automatically classified as Spark-free)", + "", + ] + lines.extend( + "- `{}`: {} executed tests".format(suite, count) + for suite, count in summary["uncovered"].items() + ) + if not summary["uncovered"]: + lines.append("None.") + text = "\n".join(lines) + "\n" + print(text) + (args.evidence_dir / "summary.txt").write_text(text, encoding="utf-8") + if os.environ.get("GITHUB_STEP_SUMMARY"): + with open(os.environ["GITHUB_STEP_SUMMARY"], "a", encoding="utf-8") as output: + output.write(text) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/workflows/util/delta-spark-ut/install-validation.py b/.github/workflows/util/delta-spark-ut/install-validation.py new file mode 100644 index 0000000000..8c1c6dbba3 --- /dev/null +++ b/.github/workflows/util/delta-spark-ut/install-validation.py @@ -0,0 +1,283 @@ +#!/usr/bin/env python3 +# +# 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. + +"""Install CI-only fixture validation into the prepared upstream Delta checkout.""" + +import argparse +from pathlib import Path +import re +import shutil +import sys + +SUPPORT = Path(__file__).resolve().parent +SOURCES = SUPPORT.parents[3] / "gluten-ut/src-delta-validation" +PACKAGE = "org.apache.spark.sql.gluten" +CHECKED = PACKAGE + ".GlutenValidation" +SHARED_IMPORT = ( + "import " + PACKAGE + ".{GlutenSharedSparkSession => SharedSparkSession}" +) +FOREIGN_HEADER_DIRECTIVE = "// scalastyle:off header.matches\n" + + +def replace_once(text, old, new, description): + if new in text: + return text + if text.count(old) != 1: + raise ValueError("{}: expected exactly one {!r}".format(description, old)) + return text.replace(old, new) + + +def shared_imports(text): + """Alias the fixture import, not type names inside test strings or comments.""" + text = re.sub( + r"(?m)^import org\.apache\.spark\.sql\.test\.SharedSparkSession[ \t]*$", + SHARED_IMPORT, + text, + ) + + def grouped(match): + names = [name.strip() for name in match.group(1).split(",")] + if "SharedSparkSession" not in names: + return match.group(0) + names.remove("SharedSparkSession") + original = ( + "import org.apache.spark.sql.test.{" + ", ".join(names) + "}\n" + if names + else "" + ) + return original + SHARED_IMPORT + + return re.sub( + r"(?m)^import org\.apache\.spark\.sql\.test\.\{([^}]+)\}", grouped, text + ) + + +def add_import(text, name): + statement = "import " + name + if statement in text: + return text + pattern = r"(?m)^(package [\w.]+[ \t]*\r?\n)" + text, count = re.subn( + pattern, lambda match: match[0] + "\n" + statement + "\n", text + ) + if count != 1: + raise ValueError("Expected one package declaration when adding " + name) + return text + + +def add_hook(text, class_name): + pattern = r"((?:class|trait)\s+" + re.escape(class_name) + r"\b[^{}]*)(\{)" + matches = list(re.finditer(pattern, text)) + if len(matches) != 1: + raise ValueError("Expected one declaration of " + class_name) + header = re.sub( + r"\s+with\s+" + re.escape(PACKAGE + ".GlutenCheckedTest"), + "", + matches[0][1], + ) + suffix = text[matches[0].end() :] + return ( + text[: matches[0].start()] + + (header.rstrip() + "\n with " + PACKAGE + ".GlutenCheckedTest {") + + ("" if suffix.startswith("\n") else "\n") + + suffix + ) + + +def wrap_validation_calls(text): + """Keep the added calls within Delta's line limit without reformatting test bodies.""" + lines = [] + for line in text.splitlines(): + if len(line) > 100: + split = -1 + if "checkedContext(" in line: + split = line.index("checkedContext(") + len("checkedContext(") + elif ".glutenChecked" in line: + split = line.index(".glutenChecked") + if split >= 0: + indent = re.match(r"[ \t]*", line)[0] + lines.extend((line[:split], indent + " " + line[split:])) + continue + lines.append(line) + return "\n".join(lines) + "\n" + + +def install(delta, sources=SOURCES, support=SUPPORT): + spark_tests = delta / "spark/src/test" + roots = [ + spark_tests, + delta / "spark-unified/src/test", + delta / "spark-shared-tests/src/test", + ] + shared_count = 0 + for root in roots: + for path in root.rglob("*.scala"): + # Runtime sources are copied last and must retain the original Spark fixture. + if PACKAGE.replace(".", "/") in path.as_posix(): + continue + old = path.read_text(encoding="utf-8") + new = shared_imports(old) + if old != new: + path.write_text(new, encoding="utf-8") + shared_count += 1 + + fixture_dir = spark_tests / "scala/org/apache/spark/sql/delta/test" + primary = (fixture_dir / "DeltaSQLCommandTest.scala").read_text(encoding="utf-8") + if SHARED_IMPORT not in primary: + raise ValueError( + "DeltaSQLCommandTest did not receive the checked shared fixture" + ) + + for name, anchor in ( + ("DeltaHiveTest", " override protected def spark: SparkSession"), + ("DeltaSessionQueryTest", " override protected def sparkConf: SparkConf"), + ): + path = fixture_dir / (name + ".scala") + text = add_hook(path.read_text(encoding="utf-8"), name) + member = ( + " override protected def glutenSession: " + "Option[org.apache.spark.sql.SparkSession] = Some(spark)\n\n" + ) + text = replace_once(text, anchor, member + anchor, name) + path.write_text(text, encoding="utf-8") + + factory_names = [ + line.strip() + for line in (support / "validation-factories.txt") + .read_text(encoding="utf-8") + .splitlines() + if line.strip() and not line.lstrip().startswith("#") + ] + for name in factory_names: + path = spark_tests / "scala" / (name.replace(".", "/") + ".scala") + text = add_hook(path.read_text(encoding="utf-8"), name.rsplit(".", 1)[1]) + text = add_import(text, CHECKED + "._") + if name.endswith(".SparkThreadLocalForwardingSuite"): + original = 'SparkContext.getOrCreate(new SparkConf().setAppName("test").setMaster("local"))' + wrapped = "checkedContext(" + original + ")" + wrapped_count = len( + re.findall(r"checkedContext\(\s*" + re.escape(original) + r"\)", text) + ) + if text.count(original) != 4 or wrapped_count not in (0, 4): + raise ValueError( + "Expected four consistently wrapped thread-forwarding SparkContext factories" + ) + if not wrapped_count: + text = text.replace(original, wrapped) + else: + text = re.sub( + r"\.getOrCreate\(\)(?!\s*\.glutenChecked)", + ".getOrCreate().glutenChecked", + text, + ) + if not re.search(r"\.getOrCreate\(\)\s*\.glutenChecked", text): + raise ValueError("No checked SparkSession factory found in " + name) + path.write_text(wrap_validation_calls(text), encoding="utf-8") + + java_root = spark_tests / "java" + java_fixture = java_root / "org/apache/spark/sql/delta/DeltaSQLCommandJavaTest.java" + text = java_fixture.read_text(encoding="utf-8") + if CHECKED + ".checkedSession(" not in text: + text = replace_once( + text, + "return SparkSession.builder()", + "return " + CHECKED + ".checkedSession(SparkSession.builder()", + str(java_fixture), + ) + text = replace_once( + text, ".getOrCreate();", ".getOrCreate());", str(java_fixture) + ) + java_fixture.write_text(text, encoding="utf-8") + + java_count = 0 + for path in java_root.rglob("*.java"): + text = path.read_text(encoding="utf-8") + match = re.search( + r"public class \w+\b[^{}]*\bimplements\b[^{}]*\bDeltaSQLCommandJavaTest\b[^{}]*\{", + text, + ) + if not match: + continue + java_count += 1 + if "new " + PACKAGE + ".GlutenValidationRule()" not in text: + rule = ( + "\n @org.junit.Rule\n" + " public final transient org.junit.rules.TestRule glutenValidation =\n" + " new " + PACKAGE + ".GlutenValidationRule();\n" + ) + text = text[: match.end()] + rule + text[match.end() :] + # Serializable Java suites can be captured by Spark closures; the driver-only rule cannot. + text = replace_once( + text, + "public final org.junit.rules.TestRule glutenValidation", + "public final transient org.junit.rules.TestRule glutenValidation", + str(path), + ) + path.write_text(text, encoding="utf-8") + if not java_count: + raise ValueError("No DeltaSQLCommandJavaTest implementations were guarded") + + destination = delta / "spark-unified/src/test/scala" + copied = 0 + for kind in ("main", "test"): + source_root = sources / kind / "scala" + for path in source_root.rglob("*.scala"): + target = destination / path.relative_to(source_root) + target.parent.mkdir(parents=True, exist_ok=True) + # Delta requires its own copyright. Retain the foreign ASF header and all other checks. + target.write_text( + FOREIGN_HEADER_DIRECTIVE + path.read_text(encoding="utf-8"), + encoding="utf-8", + ) + copied += 1 + if copied != 3: + raise ValueError( + "Expected three Gluten validation Scala sources, found {}".format(copied) + ) + grouping = ( + sources.with_name("src-delta-validation-sbt") + / "main/scala/org/apache/gluten/ci/GlutenTestGrouping.scala" + ) + (delta / "project").mkdir(exist_ok=True) + shutil.copyfile(grouping, delta / "project/GlutenTestGrouping.scala") + shutil.copyfile( + support / "validation-factories.txt", delta / "gluten-validation-factories.txt" + ) + print( + "Installed Gluten validation: {} shared imports, {} Java suites, {} direct factories".format( + shared_count, java_count, len(factory_names) + ) + ) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("delta_dir", type=Path) + args = parser.parse_args() + try: + install(args.delta_dir) + except (OSError, ValueError) as error: + print( + "ERROR: cannot install Gluten runtime validation: " + str(error), + file=sys.stderr, + ) + return 1 + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/.github/workflows/util/delta-spark-ut/run-delta-tests.sh b/.github/workflows/util/delta-spark-ut/run-delta-tests.sh index 982387806f..36ce32778b 100755 --- a/.github/workflows/util/delta-spark-ut/run-delta-tests.sh +++ b/.github/workflows/util/delta-spark-ut/run-delta-tests.sh @@ -47,8 +47,14 @@ if command -v getent > /dev/null; then fi # Gluten defaults and JDK17 test JVM flags, shared with local dev runs. # Sets JAVA_TOOL_OPTIONS so they reach the sbt launcher + forked JVMs. +UTIL_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" # shellcheck source=./java-test-args.sh -source "$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/java-test-args.sh" +source "$UTIL_DIR/java-test-args.sh" +# A fresh directory prevents a local rerun from accepting an older JVM's proof. +mkdir -p "$GITHUB_WORKSPACE/gluten-validation" +GLUTEN_VALIDATION_DIR="$(mktemp -d "$GITHUB_WORKSPACE/gluten-validation/shard-${SHARD_ID}-XXXXXX")" +export GLUTEN_VALIDATION_DIR +export GLUTEN_VALIDATION_EXEMPTIONS="$UTIL_DIR/validation-exemptions.txt" cd "$GITHUB_WORKSPACE/delta" chmod +x build/sbt # Only run the unified `spark` sbt project, NOT `sparkGroup/test` -- @@ -240,6 +246,8 @@ WATCHDOG_PID=$! "++ ${DELTA_SCALA_VERSION}" \ 'set spark / Test / javaOptions ++= Seq("-Xmx2G", "-XX:+HeapDumpOnOutOfMemoryError", "-XX:HeapDumpPath=/tmp/")' \ 'set spark / Test / testOptions += Tests.Argument(TestFrameworks.ScalaTest, "-u", "target/test-reports")' \ + 'set org.apache.gluten.ci.GlutenTestGrouping.results' \ + 'set org.apache.gluten.ci.GlutenTestGrouping.setting' \ "spark/test" 2>&1 | tee "$SBT_LOG" SBT_EXIT=${PIPESTATUS[0]} touch "$SBT_DONE_MARKER" @@ -276,11 +284,17 @@ if [ "${REPORT_COUNT}" -eq 0 ]; then exit 1 fi +# Infrastructure validation must not enter the known-failure/quarantine/seed path. +python3 -B "$UTIL_DIR/check-gluten-validation.py" \ + --reports-dir "$GITHUB_WORKSPACE/delta" \ + --evidence-dir "$GLUTEN_VALIDATION_DIR" \ + --exemptions "$GLUTEN_VALIDATION_EXEMPTIONS" \ + --expected-forks "$TEST_PARALLELISM_COUNT" + # Classify this shard's results against the baseline: seed mode when # UPDATE_BASELINE=true (record failures, never fail) so the baseline can be # (re)generated; otherwise enforce against it. Writes this shard's gate-out/*.txt # for the aggregate job. -UTIL_DIR="$GITHUB_WORKSPACE/.github/workflows/util/delta-spark-ut" GATE_MODE=enforce if [ "${UPDATE_BASELINE}" = "true" ]; then GATE_MODE=seed diff --git a/.github/workflows/util/delta-spark-ut/setup-delta.sh b/.github/workflows/util/delta-spark-ut/setup-delta.sh index f7905174ed..b41474b0da 100755 --- a/.github/workflows/util/delta-spark-ut/setup-delta.sh +++ b/.github/workflows/util/delta-spark-ut/setup-delta.sh @@ -101,3 +101,7 @@ echo "::endgroup::" PATCH_SCRIPT="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)/apply-delta-test-patches.sh" bash "$PATCH_SCRIPT" "$DELTA_REF" "$DELTA_DIR" + +echo "::group::Installing Gluten runtime validation" +python3 -B "$(dirname "$PATCH_SCRIPT")/install-validation.py" "$DELTA_DIR" +echo "::endgroup::" diff --git a/.github/workflows/util/delta-spark-ut/validation-exemptions.txt b/.github/workflows/util/delta-spark-ut/validation-exemptions.txt new file mode 100644 index 0000000000..01a7b5960e --- /dev/null +++ b/.github/workflows/util/delta-spark-ut/validation-exemptions.txt @@ -0,0 +1,19 @@ +# 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. + +# Intentional default-disabled tests only: # +# Use #* only when the entire suite intentionally disables Gluten. +# A scoped disable that restores the enabled default needs no exemption. +# No exemptions are currently needed. The validation harness cannot be exempted. diff --git a/.github/workflows/util/delta-spark-ut/validation-factories.txt b/.github/workflows/util/delta-spark-ut/validation-factories.txt new file mode 100644 index 0000000000..ad412f257b --- /dev/null +++ b/.github/workflows/util/delta-spark-ut/validation-factories.txt @@ -0,0 +1,23 @@ +# 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. + +# Known Spark-creating suites outside the shared SQL/Hive/Java fixtures. +# Used by both source installation and the independent compiled-suite inventory. +io.delta.sql.DeltaExtensionAndCatalogSuite +org.apache.spark.sql.delta.DelegatingLogStoreSuite +org.apache.spark.sql.delta.DeltaCommitLockSuite +org.apache.spark.sql.delta.DeltaRestartSessionSuite +org.apache.spark.sql.delta.LogStoreProviderSuite +org.apache.spark.sql.delta.util.threads.SparkThreadLocalForwardingSuite diff --git a/gluten-ut/src-delta-validation-sbt/main/scala/org/apache/gluten/ci/GlutenTestGrouping.scala b/gluten-ut/src-delta-validation-sbt/main/scala/org/apache/gluten/ci/GlutenTestGrouping.scala new file mode 100644 index 0000000000..3db570d644 --- /dev/null +++ b/gluten-ut/src-delta-validation-sbt/main/scala/org/apache/gluten/ci/GlutenTestGrouping.scala @@ -0,0 +1,182 @@ +/* + * 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. + */ +package org.apache.gluten.ci + +import sbt._ +import sbt.Keys._ +import sbt.testing.{AnnotatedFingerprint, Event, NestedSuiteSelector, NestedTestSelector, SubclassFingerprint, SuiteSelector, TestSelector} + +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.Files +import java.util.{Base64, UUID} + +import scala.collection.mutable.ArrayBuffer +import scala.util.control.NonFatal + +object GlutenTestGrouping { + val results = LocalProject("spark") / Test / testListeners += + new ExecutionResults(file(sys.env("GLUTEN_VALIDATION_DIR"))) + + private class ExecutionResults(directory: File) extends TestReportListener { + private case class Group(name: String) { + val events: ArrayBuffer[Event] = ArrayBuffer.empty + } + + private val active = new InheritableThreadLocal[Group]() + + private def encoded(value: String): String = { + val bytes = value.getBytes(UTF_8) + require(new String(bytes, UTF_8) == value, "Invalid UTF-8 test identity") + Base64.getEncoder.encodeToString(bytes) + } + + private def guarded(body: => Unit): Unit = { + try { + body + } catch { + case NonFatal(error) => + IO.write(directory / s"execution-error-${UUID.randomUUID()}.txt", error.toString) + throw error + } + } + + private def current: Group = + Option(active.get()).getOrElse(sys.error("Missing execution-results group")) + + override def startGroup(name: String): Unit = guarded { + require(active.get() == null, s"Overlapping execution-results groups: $name") + active.set(Group(name)) + } + + override def testEvent(event: TestEvent): Unit = guarded { + current.events ++= event.detail + } + + private def eventLine(group: Group, event: Event): String = { + val fingerprint = event.fingerprint match { + case value: AnnotatedFingerprint => "annotation:" + value.annotationName + case value: SubclassFingerprint => "subclass:" + value.superclassName + case other => sys.error(s"Unsupported execution-results fingerprint: $other") + } + val (kind, suite, test) = event.selector match { + case value: TestSelector => ("test", group.name, value.testName()) + case value: NestedTestSelector => ("test", value.suiteId(), value.testName()) + case _: SuiteSelector => ("suite", group.name, "") + case value: NestedSuiteSelector => ("suite", value.suiteId(), "") + case other => sys.error(s"Unsupported execution-results selector: $other") + } + Seq(kind, event.status.toString, encoded(suite), encoded(test), encoded(fingerprint)) + .mkString("\t") + } + + private def finish(name: String, result: TestResult, aborted: Boolean): Unit = guarded { + val group = current + try { + require(group.name == name, s"Mismatched execution-results group: $name") + val events = group.events.map(event => eventLine(group, event)) + if (aborted) { + events += Seq("suite", "Error", encoded(name), "", "").mkString("\t") + } + // JUnit attribute normalization and sbt's display-name rewriting are intentionally avoided. + val lines = Seq("version\t1\t" + encoded(name)) ++ events ++ + Seq(s"complete\t$result\t${events.size}") + Files.write( + (directory / s"execution-${UUID.randomUUID()}.tsv").toPath, + lines.mkString("", "\n", "\n").getBytes(UTF_8)) + } finally { + active.remove() + } + } + + override def endGroup(name: String, result: TestResult): Unit = + finish(name, result, aborted = false) + + override def endGroup(name: String, error: Throwable): Unit = + finish(name, TestResult.Error, aborted = true) + } + + val setting = LocalProject("spark") / Test / testGrouping := { + val groups = (LocalProject("spark") / Test / testGrouping).value + val definitions = (LocalProject("spark") / Test / definedTests).value + val classpath = (LocalProject("spark") / Test / fullClasspath).value.files + val root = (LocalRootProject / baseDirectory).value + val output = file(sys.env("GLUTEN_VALIDATION_DIR")) + val canaryName = "org.apache.spark.sql.gluten.GlutenActivationSuite" + val canary = definitions.find(_.name == canaryName).getOrElse( + sys.error("Gluten activation canary is missing from compiled tests")) + require(groups.nonEmpty, "No Delta test forks were scheduled") + require( + groups.forall(_.runPolicy.isInstanceOf[Tests.SubProcess]), + "Gluten validation requires forked test groups; set NUM_SHARDS, SHARD_ID and TEST_PARALLELISM_COUNT" + ) + val factories = IO.readLines(root / "gluten-validation-factories.txt") + .map(_.trim).filter(line => line.nonEmpty && !line.startsWith("#")).toSet + val sparkFixtures = Set( + "org.apache.spark.sql.test.SharedSparkSession", + "org.apache.spark.sql.test.SharedSparkSessionBase", + "org.apache.spark.sql.hive.test.TestHiveSingleton", + "org.apache.spark.sql.delta.test.DeltaSQLCommandTest", + "org.apache.spark.sql.delta.test.DeltaHiveTest", + "org.apache.spark.sql.delta.DeltaSQLCommandJavaTest", + "org.apache.spark.sql.SessionQueryTestBase", + "org.apache.spark.LocalSparkContext", + "org.apache.spark.sql.gluten.GlutenCheckedTest" + ) + + // Inspect compiled inheritance independently of the runtime hooks, without class initialization. + val loader = new java.net.URLClassLoader( + classpath.map(_.toURI.toURL).toArray, + ClassLoader.getPlatformClassLoader) + def hierarchy(clazz: Class[_]): Set[String] = { + val pending = scala.collection.mutable.Stack[Class[_]](clazz) + val names = scala.collection.mutable.Set.empty[String] + while (pending.nonEmpty) { + val current = pending.pop() + if (current != null && names.add(current.getName)) { + pending.push(current.getSuperclass) + current.getInterfaces.foreach(parent => pending.push(parent)) + } + } + names.toSet + } + try { + val scheduled = (groups.flatMap(_.tests).filterNot(_.name == canaryName) :+ canary) + .groupBy(_.name).values.map(_.head).toSeq.sortBy(_.name) + val rows = scheduled.map { + test => + val module = test.fingerprint match { + case fingerprint: sbt.testing.SubclassFingerprint => fingerprint.isModule + case fingerprint: sbt.testing.AnnotatedFingerprint => fingerprint.isModule + case other => sys.error(s"Unsupported test fingerprint: $other") + } + val clazz = Class.forName(test.name + (if (module) "$" else ""), false, loader) + val parents = hierarchy(clazz) + val role = + if (test.name == canaryName) "canary" + else if (factories(test.name) || parents.exists(sparkFixtures)) "required" + else if (test.name.startsWith("org.apache.spark.sql.gluten.")) "self-test" + else "uncovered" + s"$role\t${test.name}" + } + IO.write(output / "expected-suites.tsv", rows.mkString("", "\n", "\n")) + IO.write(output / "expected-forks.txt", groups.size.toString + "\n") + } finally { + loader.close() + } + groups.map(group => group.copy(tests = canary +: group.tests.filterNot(_.name == canaryName))) + } +} diff --git a/gluten-ut/src-delta-validation/main/scala/org/apache/spark/sql/gluten/GlutenActivationSuite.scala b/gluten-ut/src-delta-validation/main/scala/org/apache/spark/sql/gluten/GlutenActivationSuite.scala new file mode 100644 index 0000000000..ddd3ee0e2c --- /dev/null +++ b/gluten-ut/src-delta-validation/main/scala/org/apache/spark/sql/gluten/GlutenActivationSuite.scala @@ -0,0 +1,41 @@ +/* + * 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. + */ +package org.apache.spark.sql.gluten + +import org.apache.gluten.execution.WholeStageTransformer + +import org.apache.spark.sql.{QueryTest, Row} +import org.apache.spark.sql.test.SharedSparkSession + +class GlutenActivationSuite extends QueryTest with SharedSparkSession { + // Every fork runs this suite. ScalaTest uses suiteId for the XML file name. + override def suiteId: String = s"${super.suiteId}-${GlutenValidation.processId}" + + test(GlutenValidation.canaryTest) { + GlutenValidation.nativeCanary(spark) { + withSQLConf("spark.sql.adaptive.enabled" -> "false") { + val query = spark.range(20).selectExpr("id % 3 AS k", "id AS v").groupBy("k").sum("v") + val rows = query.collect().toSet + assert(rows == Set(Row(0L, 63L), Row(1L, 70L), Row(2L, 57L))) + assert( + query.queryExecution.executedPlan.exists(_.isInstanceOf[WholeStageTransformer]), + s"No native Gluten stage in the canary plan:\n${query.queryExecution.executedPlan}" + ) + } + } + } +} diff --git a/gluten-ut/src-delta-validation/main/scala/org/apache/spark/sql/gluten/GlutenValidation.scala b/gluten-ut/src-delta-validation/main/scala/org/apache/spark/sql/gluten/GlutenValidation.scala new file mode 100644 index 0000000000..1510198cd5 --- /dev/null +++ b/gluten-ut/src-delta-validation/main/scala/org/apache/spark/sql/gluten/GlutenValidation.scala @@ -0,0 +1,235 @@ +/* + * 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. + */ +package org.apache.spark.sql.gluten + +import org.apache.gluten.extension.injector.InjectorControl + +import org.apache.spark.SparkContext +import org.apache.spark.sql.SparkSession +import org.apache.spark.sql.test.SharedSparkSession + +import com.fasterxml.jackson.databind.ObjectMapper +import org.junit.rules.TestWatcher +import org.junit.runner.Description +import org.scalatest.{BeforeAndAfterEachTestData, Suite, TestData} + +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.{Files, Path, Paths, StandardOpenOption} +import java.util.UUID + +import scala.collection.JavaConverters._ +import scala.collection.mutable.ArrayBuffer +import scala.util.control.NonFatal + +object GlutenValidation { + val canarySuite: String = "org.apache.spark.sql.gluten.GlutenActivationSuite" + val canaryTest: String = "native execution with inherited Gluten defaults" + val processId: String = ProcessHandle.current().pid().toString + + private val plugin = "org.apache.gluten.GlutenPlugin" + private val threadKey = "gluten.enabledForCurrentThread" + private val active = new ThreadLocal[Scope] + private val mapper = new ObjectMapper() + private lazy val output = { + val directory = Paths.get(sys.env("GLUTEN_VALIDATION_DIR")) + require(directory.isAbsolute, "GLUTEN_VALIDATION_DIR must be absolute") + Files.createDirectories(directory) + directory.resolve(s"jvm-$processId-${UUID.randomUUID()}.jsonl") + } + private lazy val exemptions = + readExemptions(Paths.get(sys.env("GLUTEN_VALIDATION_EXEMPTIONS"))) + + private case class Scope(suite: String, test: String, exemption: Option[String]) { + val checks: ArrayBuffer[() => Unit] = ArrayBuffer.empty + } + + // Only the legacy text exemption format is normalized; evidence keeps the raw test identity. + private[gluten] def testKey(suite: String, test: String): String = + s"$suite#${test.replace('\r', ' ').replace('\n', ' ')}".trim + + private[gluten] def readExemptions(path: Path): Map[String, String] = { + val entries = Files.readAllLines(path, UTF_8).asScala + .map(_.trim) + .filter(line => line.nonEmpty && !line.startsWith("#")) + .map { + line => + val parts = line.split("\t", 2) + require(parts.length == 2 && parts(1).trim.nonEmpty, s"Missing exemption reason: $line") + require(parts(0).matches("[\\w.$]+#.+"), s"Invalid exemption key: ${parts(0)}") + require( + !parts(0).startsWith("org.apache.spark.sql.gluten."), + "The validation harness cannot be exempted") + parts(0) -> parts(1).trim + } + require(entries.map(_._1).distinct.size == entries.size, "Duplicate validation exemption") + entries.toMap + } + + private[gluten] def assertSettings( + plugins: String, + enabled: Boolean, + threadEnabled: Option[String], + rulesInstalled: Boolean): Unit = { + require( + plugins.split(",").map(_.trim).contains(plugin), + s"Actual SparkContext is missing $plugin in spark.plugins") + require(rulesInstalled, "Actual SparkSession is missing Gluten columnar rules") + require(enabled, "spark.gluten.enabled is false at the test boundary") + require( + threadEnabled.forall(_.toBoolean), + s"$threadKey is false at the test boundary") + } + + private def checkContext(context: SparkContext): Unit = + assertSettings( + context.getConf.get("spark.plugins", ""), + context.getConf.getBoolean("spark.gluten.enabled", true), + Option(context.getLocalProperty(threadKey)), + rulesInstalled = true + ) + + private def checkSession(session: SparkSession): Unit = + assertSettings( + session.sparkContext.getConf.get("spark.plugins", ""), + session.conf.get("spark.gluten.enabled", "true").toBoolean, + Option(session.sparkContext.getLocalProperty(threadKey)), + session.sessionState.columnarRules.exists(_.isInstanceOf[InjectorControl.DisablerAware]) + ) + + private def record(kind: String, fields: (String, AnyRef)*): Unit = synchronized { + val row = new java.util.LinkedHashMap[String, AnyRef]() + row.put("version", Int.box(1)) + row.put("pid", processId) + row.put("kind", kind) + fields.foreach { case (key, value) => row.put(key, value) } + Files.write( + output, + (mapper.writeValueAsString(row) + "\n").getBytes(UTF_8), + StandardOpenOption.CREATE, + StandardOpenOption.APPEND) + } + + private def verify(scope: Scope, phase: String)(check: => Unit): Unit = { + try { + check + } catch { + case NonFatal(error) => + record( + "violation", + "suite" -> scope.suite, + "test" -> scope.test, + "phase" -> phase, + "message" -> error.toString) + throw error + } + } + + def begin(suite: String, test: String): Unit = { + require(active.get() == null, s"Overlapping Gluten validation scopes: $suite#$test") + val exemption = exemptions.get(testKey(suite, test)).orElse(exemptions.get(s"$suite#*")) + active.set(Scope(suite, test, exemption)) + } + + private def track(check: () => Unit): Unit = { + val scope = Option(active.get()).getOrElse { + throw new IllegalStateException("Missing Gluten validation test lifecycle") + } + if (scope.exemption.isEmpty) { + verify(scope, "before")(check()) + scope.checks += check + } + } + + def checkedSession[S <: SparkSession](session: S): S = { + track(() => checkSession(session)) + session + } + + def checkedContext(context: SparkContext): SparkContext = { + track(() => checkContext(context)) + context + } + + implicit class CheckedSessionOps[S <: SparkSession](session: S) { + def glutenChecked: S = checkedSession(session) + } + + def end(suite: String, test: String): Unit = { + val scope = Option(active.get()).getOrElse(Scope(suite, test, None)) + try { + verify(scope, "after") { + require( + scope.suite == suite && scope.test == test, + s"Mismatched Gluten validation lifecycle: $suite#$test") + if (scope.exemption.isEmpty) { + require(scope.checks.nonEmpty, "No Spark context/session was validated for this test") + scope.checks.foreach(_()) + } + } + record( + "test", + "suite" -> suite, + "test" -> test, + "status" -> (if (scope.exemption.isDefined) "exempt" else "checked"), + "checks" -> Int.box(scope.checks.size), + "reason" -> scope.exemption.getOrElse("") + ) + } finally { + active.remove() + } + } + + def nativeCanary(session: SparkSession)(body: => Unit): Unit = { + val scope = Scope(canarySuite, canaryTest, None) + verify(scope, "canary") { + checkSession(session) + body + checkSession(session) + record("canary", "suite" -> canarySuite, "test" -> canaryTest, "status" -> "passed") + } + } +} + +trait GlutenCheckedTest extends BeforeAndAfterEachTestData { this: Suite => + protected def glutenSession: Option[SparkSession] = None + + override protected def beforeEach(data: TestData): Unit = { + super.beforeEach(data) + GlutenValidation.begin(getClass.getName, data.name) + glutenSession.foreach(GlutenValidation.checkedSession(_)) + } + + override protected def afterEach(data: TestData): Unit = { + try { + super.afterEach(data) + } finally { + GlutenValidation.end(getClass.getName, data.name) + } + } +} + +trait GlutenSharedSparkSession extends SharedSparkSession with GlutenCheckedTest { + override protected def glutenSession: Option[SparkSession] = Some(spark) +} + +class GlutenValidationRule extends TestWatcher { + override protected def starting(description: Description): Unit = + GlutenValidation.begin(description.getClassName, description.getMethodName) + + override protected def finished(description: Description): Unit = + GlutenValidation.end(description.getClassName, description.getMethodName) +} diff --git a/gluten-ut/src-delta-validation/test/python/test_installation.py b/gluten-ut/src-delta-validation/test/python/test_installation.py new file mode 100644 index 0000000000..ad83ae99c2 --- /dev/null +++ b/gluten-ut/src-delta-validation/test/python/test_installation.py @@ -0,0 +1,202 @@ +# 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. + +import importlib.util +from pathlib import Path +import tempfile +import unittest + +SUPPORT = Path(__file__).resolve().parents[4] / ".github/workflows/util/delta-spark-ut" +SPEC = importlib.util.spec_from_file_location( + "installation", SUPPORT / "install-validation.py" +) +installation = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(installation) + + +class InstallationTest(unittest.TestCase): + def test_import_alias_preserves_other_imports_and_test_strings(self): + source = ( + "package example\n" + "import org.apache.spark.sql.test.{SharedSparkSession, SQLTestUtils}\n" + 'class Example extends SharedSparkSession { val name = "SharedSparkSession" }\n' + ) + result = installation.shared_imports(source) + self.assertIn("import org.apache.spark.sql.test.{SQLTestUtils}", result) + self.assertIn(installation.SHARED_IMPORT, result) + self.assertIn('val name = "SharedSparkSession"', result) + self.assertEqual(result, installation.shared_imports(result)) + + def test_plain_import_is_aliased(self): + source = "import org.apache.spark.sql.test.SharedSparkSession\nclass A\n" + self.assertEqual( + installation.SHARED_IMPORT + "\nclass A\n", + installation.shared_imports(source), + ) + + def test_hook_preserves_self_type_and_is_idempotent(self): + source = "trait A extends B { self: C =>\n def method = 1\n}\n" + result = installation.add_hook(source, "A") + self.assertIn("with " + installation.PACKAGE + ".GlutenCheckedTest", result) + self.assertIn("self: C =>", result) + self.assertEqual(result, installation.add_hook(result, "A")) + with self.assertRaises(ValueError): + installation.add_hook(source, "Missing") + + def test_long_hook_declarations_obey_delta_line_limit(self): + source = ( + "class DeltaCommitLockSuite extends SparkFunSuite " + "with LocalSparkSession with SQLHelper {\n}\n" + ) + result = installation.add_hook(source, "DeltaCommitLockSuite") + self.assertTrue(all(len(line) <= 100 for line in result.splitlines())) + self.assertEqual(result, installation.add_hook(result, "DeltaCommitLockSuite")) + + def test_factory_wrapping_obeys_delta_line_limit_and_is_idempotent(self): + source = ( + " withSparkSession(SparkSession.builder.config(sparkConf)" + ".getOrCreate().glutenChecked) { spark =>\n" + " val sc = checkedContext(SparkContext.getOrCreate(" + 'new SparkConf().setAppName("test").setMaster("local")))\n' + ) + result = installation.wrap_validation_calls(source) + self.assertTrue(all(len(line) <= 100 for line in result.splitlines())) + self.assertEqual(result, installation.wrap_validation_calls(result)) + self.assertIn(".getOrCreate()\n .glutenChecked)", result) + self.assertIn("checkedContext(\n SparkContext.getOrCreate", result) + + def test_unexpected_source_context_fails_instead_of_skipping(self): + with self.assertRaises(ValueError): + installation.replace_once( + "changed upstream", "expected", "patched", "fixture" + ) + with self.assertRaises(ValueError): + installation.replace_once( + "expected expected", "expected", "patched", "fixture" + ) + + def test_complete_installation_covers_fixtures_factories_and_java(self): + with tempfile.TemporaryDirectory() as temporary: + root = Path(temporary) + + def write(relative, text): + path = root / relative + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(text, encoding="utf-8") + + fixtures = "spark/src/test/scala/org/apache/spark/sql/delta/test/" + write( + fixtures + "DeltaSQLCommandTest.scala", + "package example\nimport org.apache.spark.sql.test.SharedSparkSession\n" + "trait DeltaSQLCommandTest extends SharedSparkSession {}\n", + ) + write( + fixtures + "DeltaHiveTest.scala", + "package example\ntrait DeltaHiveTest extends SparkFunSuite { self: Other =>\n" + " override protected def spark: SparkSession = session\n}\n", + ) + write( + fixtures + "DeltaSessionQueryTest.scala", + "package example\ntrait DeltaSessionQueryTest extends SessionQueryTest {\n" + " override protected def sparkConf: SparkConf = conf\n}\n", + ) + factories = (SUPPORT / "validation-factories.txt").read_text( + encoding="utf-8" + ) + for name in factories.splitlines(): + if not name or name.startswith("#"): + continue + package, simple = name.rsplit(".", 1) + if simple == "SparkThreadLocalForwardingSuite": + body = "\n".join( + 'val sc{} = SparkContext.getOrCreate(new SparkConf().setAppName("test").setMaster("local"))'.format( + index + ) + for index in range(4) + ) + else: + body = "val spark = SparkSession.builder().getOrCreate()" + write( + "spark/src/test/scala/" + name.replace(".", "/") + ".scala", + "package {}\nclass {} extends SparkFunSuite {{\n{}\n}}\n".format( + package, simple, body + ), + ) + java = "spark/src/test/java/org/apache/spark/sql/delta/" + write( + java + "DeltaSQLCommandJavaTest.java", + "public interface DeltaSQLCommandJavaTest {\n" + " default SparkSession buildSparkSession() {\n" + " return SparkSession.builder().getOrCreate();\n }\n}\n", + ) + write( + java + "Example.java", + "public class Example implements Serializable, DeltaSQLCommandJavaTest {}\n", + ) + + installation.install(root) + first = { + path.relative_to(root).as_posix(): path.read_bytes() + for path in root.rglob("*") + if path.is_file() + } + installation.install(root) + second = { + path.relative_to(root).as_posix(): path.read_bytes() + for path in root.rglob("*") + if path.is_file() + } + self.assertEqual(first, second) + self.assertIn(b"GlutenValidationRule", first[java + "Example.java"]) + self.assertIn( + b"public final transient org.junit.rules.TestRule", + first[java + "Example.java"], + ) + self.assertIn( + b"checkedSession", first[java + "DeltaSQLCommandJavaTest.java"] + ) + self.assertFalse((root / "gluten-validation.sbt").exists()) + self.assertTrue((root / "project/GlutenTestGrouping.scala").is_file()) + copied = root / "spark-unified/src/test/scala/org/apache/spark/sql/gluten" + self.assertEqual(3, len(list(copied.glob("*.scala")))) + self.assertIn( + "import org.apache.spark.sql.test.SharedSparkSession", + (copied / "GlutenActivationSuite.scala").read_text(encoding="utf-8"), + ) + for kind in ("main", "test"): + source_root = installation.SOURCES / kind / "scala" + for source in source_root.rglob("*.scala"): + destination = ( + root + / "spark-unified/src/test/scala" + / source.relative_to(source_root) + ) + self.assertEqual( + installation.FOREIGN_HEADER_DIRECTIVE + + source.read_text(encoding="utf-8"), + destination.read_text(encoding="utf-8"), + ) + for source in (root / "spark/src/test/scala").rglob("*.scala"): + for number, line in enumerate( + source.read_text(encoding="utf-8").splitlines(), 1 + ): + if not line.startswith("import "): + self.assertLessEqual( + len(line), 100, "{}:{}".format(source, number) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/gluten-ut/src-delta-validation/test/python/test_validation.py b/gluten-ut/src-delta-validation/test/python/test_validation.py new file mode 100644 index 0000000000..eb151015ce --- /dev/null +++ b/gluten-ut/src-delta-validation/test/python/test_validation.py @@ -0,0 +1,490 @@ +# 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. + +import base64 +from collections import defaultdict +import importlib.util +import json +from pathlib import Path +import tempfile +import unittest +import xml.etree.ElementTree as ET + +SUPPORT = Path(__file__).resolve().parents[4] / ".github/workflows/util/delta-spark-ut" +SPEC = importlib.util.spec_from_file_location( + "validation", SUPPORT / "check-gluten-validation.py" +) +validation = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(validation) + + +class ValidationTest(unittest.TestCase): + suite = "org.example.SparkSuite" + helper = "org.example.HelperSuite" + name = "test with # and\nnewlines " + + def setUp(self): + self.temporary = tempfile.TemporaryDirectory() + self.addCleanup(self.temporary.cleanup) + self.root = Path(self.temporary.name) + self.reports = self.root / "reports" + self.reports.mkdir() + self.evidence = self.root / "evidence" + self.evidence.mkdir() + self.exemptions = self.root / "exemptions.txt" + self.exemptions.write_text("# none\n", encoding="utf-8") + (self.evidence / "expected-suites.tsv").write_text( + "required\t{}\nuncovered\t{}\ncanary\t{}\n".format( + self.suite, self.helper, validation.CANARY + ), + encoding="utf-8", + ) + (self.evidence / "expected-forks.txt").write_text("1\n", encoding="utf-8") + self.rows = [ + self.row( + "canary", validation.CANARY, validation.CANARY_TEST, status="passed" + ), + self.row( + "test", self.suite, self.name, status="checked", checks=1, reason="" + ), + ] + self.write_reports() + self.write_evidence() + + def row(self, kind, suite, name, **values): + return dict(version=1, pid="100", kind=kind, suite=suite, test=name, **values) + + def write_reports(self, failure=False, skipped=False, canary_failure=False): + root = ET.Element("testsuites") + self.events = [] + for suite, name, failed, skip in ( + (self.suite, self.name, failure, skipped), + (self.helper, "pure helper", False, False), + (validation.CANARY, validation.CANARY_TEST, canary_failure, False), + ): + element = ET.SubElement(root, "testsuite", name=suite) + test = ET.SubElement(element, "testcase", classname=suite, name=name) + if failed: + ET.SubElement(test, "failure", message="known failure") + elif skip: + ET.SubElement(test, "skipped") + self.events.append( + self.event( + suite, + name, + "Failure" if failed else "Skipped" if skip else "Success", + ) + ) + ET.ElementTree(root).write(self.reports / "TEST-results.xml", encoding="utf-8") + self.write_executions() + + def event( + self, + suite, + name, + status="Success", + kind="test", + fingerprint="subclass:org.scalatest.Suite", + ): + return (kind, status, suite, name, fingerprint) + + @staticmethod + def encoded(value): + return base64.b64encode(value.encode("utf-8")).decode("ascii") + + def write_group(self, path, suite, events): + failed = any(event[1] in {"Failure", "Error"} for event in events) + lines = ["version\t1\t" + self.encoded(suite)] + lines.extend( + "\t".join( + ( + kind, + status, + self.encoded(owner), + self.encoded(name), + self.encoded(fingerprint), + ) + ) + for kind, status, owner, name, fingerprint in events + ) + lines.append( + "complete\t{}\t{}".format("Failed" if failed else "Passed", len(events)) + ) + path.write_text("\n".join(lines) + "\n", encoding="utf-8") + + def write_executions(self, forks=1): + for path in self.evidence.glob("execution-*.tsv"): + path.unlink() + groups = defaultdict(list) + for event in self.events: + groups[event[2]].append(event) + for index, (suite, events) in enumerate(groups.items()): + copies = forks if suite == validation.CANARY else 1 + for copy in range(copies): + self.write_group( + self.evidence / "execution-{}-{}.tsv".format(index, copy), + suite, + events, + ) + + def write_evidence(self, ensure_ascii=True): + (self.evidence / "jvm-100.jsonl").write_text( + "".join( + json.dumps(row, ensure_ascii=ensure_ascii) + "\n" for row in self.rows + ), + encoding="utf-8", + ) + + def named_cases(self, names): + self.events = [self.event(self.suite, name) for name in names] + [ + self.event(self.helper, "pure helper"), + self.event(validation.CANARY, validation.CANARY_TEST), + ] + self.rows = [self.rows[0]] + [ + self.row("test", self.suite, name, status="checked", checks=1, reason="") + for name in names + ] + (self.reports / "TEST-results.xml").unlink() + for reporter in ("scalatest", "sbt"): + root = ET.Element("testsuites") + for _, status, suite, name, _ in self.events: + # Both XML readers collapse literal attribute whitespace; sbt also strips prefixes. + rendered = name.replace("\r", " ").replace("\n", " ").replace("\t", " ") + if reporter == "sbt": + for prefix in (suite, suite.rsplit(".", 1)[-1]): + if rendered.startswith(prefix): + rendered = rendered[len(prefix) :] + if rendered.startswith("."): + rendered = rendered[1:] + break + owner = ET.SubElement(root, "testsuite", name=suite) + case = ET.SubElement(owner, "testcase", classname=suite, name=rendered) + if status not in validation.EXECUTED: + ET.SubElement(case, "skipped") + ET.ElementTree(root).write( + self.reports / ("TEST-" + reporter + ".xml"), encoding="utf-8" + ) + self.write_executions() + self.write_evidence(ensure_ascii=False) + + def check(self, forks=1): + return validation.validate(self.reports, self.evidence, self.exemptions, forks) + + def test_checked_tests_and_uncovered_suites_are_counted_separately(self): + summary = self.check() + self.assertEqual(1, summary["forks"]) + self.assertEqual(1, summary["checked"]) + self.assertEqual(0, summary["exempt"]) + self.assertEqual({self.helper: 1}, summary["uncovered"]) + + def test_explicit_checks_can_cover_a_suite_outside_the_known_fixtures(self): + self.rows.append( + self.row( + "test", + self.helper, + "pure helper", + status="checked", + checks=1, + reason="", + ) + ) + self.write_evidence() + self.assertEqual({}, self.check()["uncovered"]) + + def test_canaries_alone_cannot_make_an_empty_delta_run_pass(self): + root = ET.Element("testsuite", name=validation.CANARY) + ET.SubElement( + root, "testcase", classname=validation.CANARY, name=validation.CANARY_TEST + ) + ET.ElementTree(root).write(self.reports / "TEST-results.xml", encoding="utf-8") + self.rows.pop() + self.write_evidence() + self.events = [self.event(validation.CANARY, validation.CANARY_TEST)] + self.write_executions() + with self.assertRaisesRegex(ValueError, "No Delta tests executed"): + self.check() + + def test_guard_success_does_not_change_ordinary_test_failures(self): + self.write_reports(failure=True) + self.assertEqual(1, self.check()["checked"]) + + def test_missing_check_fails_even_for_an_already_failing_test(self): + self.write_reports(failure=True) + self.rows.pop() + self.write_evidence() + with self.assertRaisesRegex(ValueError, "missing required validation"): + self.check() + + def test_missing_instrumentation_and_inventory_fail_closed(self): + (self.evidence / "jvm-100.jsonl").unlink() + with self.assertRaisesRegex(ValueError, "No JVM validation evidence"): + self.check() + self.write_evidence() + (self.evidence / "expected-suites.tsv").unlink() + with self.assertRaises(OSError): + self.check() + + def test_every_requested_fork_requires_native_execution(self): + (self.evidence / "expected-forks.txt").write_text("2\n", encoding="utf-8") + with self.assertRaisesRegex(ValueError, "Expected native proof from 2 JVMs"): + self.check(2) + self.rows.append(dict(self.rows[0], pid="101")) + self.write_evidence() + self.write_executions(forks=2) + self.assertEqual(2, self.check(2)["forks"]) + + def test_proof_from_a_different_jvm_cannot_cover_a_test(self): + self.rows[1]["pid"] = "101" + self.write_evidence() + with self.assertRaisesRegex(ValueError, "No native proof for JVM 101"): + self.check() + + def test_duplicate_native_records_cannot_replace_a_missing_fork(self): + self.rows.append(dict(self.rows[0])) + self.write_evidence() + with self.assertRaisesRegex(ValueError, "Duplicate native canary"): + self.check() + + def test_failed_canary_is_never_an_ordinary_baseline_failure(self): + self.write_reports(canary_failure=True) + with self.assertRaisesRegex(ValueError, "Validation harness test failed"): + self.check() + + def test_caught_violation_is_still_fatal(self): + self.rows.append( + self.row( + "violation", + self.suite, + self.name, + phase="before", + message="plugin missing", + ) + ) + self.write_evidence() + with self.assertRaisesRegex(ValueError, "plugin missing"): + self.check() + + def test_exemption_requires_a_matching_reason_and_a_runtime_record(self): + key = validation.results.format_entry( + *validation.results.normalize_key(self.suite, self.name) + ) + self.exemptions.write_text(key + "\tintentionally disabled\n", encoding="utf-8") + self.rows[1].update(status="exempt", checks=0, reason="intentionally disabled") + self.write_evidence() + self.assertEqual(1, self.check()["exempt"]) + self.rows[1]["reason"] = "unreviewed reason" + self.write_evidence() + with self.assertRaisesRegex(ValueError, "Undeclared validation exemption"): + self.check() + self.rows.pop() + self.write_evidence() + with self.assertRaisesRegex(ValueError, "missing required validation"): + self.check() + + def test_suite_exemptions_are_explicit_and_cannot_exempt_the_canary(self): + self.exemptions.write_text( + self.suite + "#*\tdeliberately vanilla\n", encoding="utf-8" + ) + self.rows[1].update(status="exempt", checks=0, reason="deliberately vanilla") + self.write_evidence() + self.assertEqual(1, self.check()["exempt"]) + self.exemptions.write_text(validation.CANARY + "#*\tignore\n", encoding="utf-8") + with self.assertRaisesRegex(ValueError, "infrastructure exemption"): + self.check() + + def test_missing_or_malformed_exemption_reason_is_rejected(self): + for text in ( + "Suite#test", + "Suite#test\t", + "*#test\treason", + "Suite#test\tone\nSuite#test\ttwo", + ): + with self.subTest(text=text): + self.exemptions.write_text(text, encoding="utf-8") + with self.assertRaises(ValueError): + self.check() + + def test_skipped_tests_do_not_require_a_guard_but_can_have_one(self): + self.write_reports(skipped=True) + self.assertEqual(0, self.check()["checked"]) + self.rows.pop() + self.write_evidence() + self.assertEqual(0, self.check()["checked"]) + + def test_corrupt_or_truncated_evidence_is_rejected(self): + for text in ('{"version":', json.dumps(self.rows[0]), "\n"): + with self.subTest(text=text): + (self.evidence / "jvm-100.jsonl").write_text(text, encoding="utf-8") + with self.assertRaises(ValueError): + self.check() + + def test_malformed_records_are_not_success_shaped_defaults(self): + for field, value in ( + ("version", True), + ("pid", ""), + ("kind", "unknown"), + ("test", None), + ("status", "passed"), + ("checks", 0), + ): + with self.subTest(field=field): + original = self.rows[1][field] + self.rows[1][field] = value + self.write_evidence() + with self.assertRaises(ValueError): + self.check() + self.rows[1][field] = original + + def test_partial_inventory_is_rejected(self): + (self.evidence / "expected-suites.tsv").write_text( + "canary\t" + validation.CANARY + "\n", encoding="utf-8" + ) + with self.assertRaisesRegex(ValueError, "absent from the compiled inventory"): + self.check() + + def test_incorrect_fork_plan_is_rejected(self): + (self.evidence / "expected-forks.txt").write_text("2\n", encoding="utf-8") + with self.assertRaisesRegex(ValueError, "disagrees with requested parallelism"): + self.check() + + def test_aborted_required_suite_cannot_hide_missing_validation(self): + root = ET.Element("testsuites") + ET.SubElement(root, "testsuite", name=self.suite, errors="1") + helper = ET.SubElement(root, "testsuite", name=self.helper) + ET.SubElement(helper, "testcase", classname=self.helper, name="pure helper") + canary = ET.SubElement(root, "testsuite", name=validation.CANARY) + ET.SubElement( + canary, "testcase", classname=validation.CANARY, name=validation.CANARY_TEST + ) + ET.ElementTree(root).write(self.reports / "TEST-results.xml", encoding="utf-8") + self.rows.pop() + self.write_evidence() + self.events = [ + self.event(self.suite, "", status="Error", kind="suite"), + self.event(self.helper, "pure helper"), + self.event(validation.CANARY, validation.CANARY_TEST), + ] + self.write_executions() + with self.assertRaisesRegex(ValueError, "Required execution suite aborted"): + self.check() + + def test_duplicate_test_evidence_is_rejected(self): + self.rows.append(dict(self.rows[1])) + self.write_evidence() + with self.assertRaisesRegex(ValueError, "Duplicate validation"): + self.check() + + def test_lossless_names_survive_both_junit_reporter_transformations(self): + names = [ + "partition column names and values contain '{}'".format(char) + for char in (" ", "\n", "\r", "\t") + ] + [ + "SparkSuite with nested struct and mixed types", + "org.example.SparkSuite.case", + "case", + " leading and trailing ", + "Unicode\u2028separator", + ] + self.named_cases(names) + self.assertEqual(len(names), self.check()["checked"]) + + def test_one_whitespace_variant_cannot_cover_another_missing_check(self): + self.named_cases(["column ' '", "column '\n'", "column '\t'"]) + self.rows.pop(2) + self.write_evidence() + with self.assertRaisesRegex(ValueError, "missing required validation") as error: + self.check() + self.assertIn(r"\n", str(error.exception)) + + def test_duplicate_raw_execution_results_are_rejected(self): + self.events.insert(0, self.events[0]) + self.write_executions() + with self.assertRaisesRegex(ValueError, "Duplicate execution result"): + self.check() + + def test_missing_execution_listener_is_fatal(self): + for path in self.evidence.glob("execution-*.tsv"): + path.unlink() + with self.assertRaisesRegex(ValueError, "Missing lossless execution results"): + self.check() + + def test_listener_errors_cannot_be_silently_discarded(self): + (self.evidence / "execution-error-test.txt").write_text( + "unsupported selector", encoding="utf-8" + ) + with self.assertRaisesRegex(ValueError, "listener failed"): + self.check() + + def test_corrupt_execution_streams_fail_closed(self): + path = self.evidence / "execution-0-0.tsv" + original = path.read_text(encoding="utf-8") + variants = [ + original.rstrip("\n"), + original.replace("version\t1", "version\t2"), + original.replace("complete\tPassed\t1", "complete\tPassed\t2"), + original.replace("\tSuccess\t", "\tUnknown\t"), + original.replace(self.encoded(self.name), "$invalid-base64"), + original.replace( + self.encoded("subclass:org.scalatest.Suite"), self.encoded("unknown") + ), + original.replace(self.encoded("subclass:org.scalatest.Suite"), ""), + ] + for text in variants: + with self.subTest(text=text): + path.write_text(text, encoding="utf-8") + with self.assertRaises(ValueError): + self.check() + path.write_text(original, encoding="utf-8") + + def test_partial_raw_results_cannot_hide_missing_guard_evidence(self): + self.named_cases(["first", "second"]) + self.events = [event for event in self.events if event[3] != "first"] + self.rows = [row for row in self.rows if row["test"] != "first"] + self.write_executions() + self.write_evidence() + with self.assertRaisesRegex(ValueError, "Execution/JUnit count mismatch"): + self.check() + + def test_legacy_junit_method_prefix_is_not_applied_to_scalatest_names(self): + self.name = "javaMethod" + self.write_reports() + self.rows[1]["test"] = self.name + self.write_evidence() + self.events[0] = self.event( + self.suite, + self.suite + "." + self.name, + fingerprint="annotation:org.junit.Test", + ) + self.write_executions() + self.assertEqual(1, self.check()["checked"]) + self.events[0] = self.event(self.suite, self.suite + "." + self.name) + self.write_executions() + with self.assertRaisesRegex(ValueError, "missing required validation"): + self.check() + + def test_ambiguous_exemption_does_not_exempt_distinct_raw_names(self): + self.named_cases(["column ' '", "column '\n'"]) + self.exemptions.write_text( + self.suite + "#column ' '\tintentionally disabled\n", encoding="utf-8" + ) + for row in self.rows[1:]: + row.update(status="exempt", checks=0, reason="intentionally disabled") + self.write_evidence() + with self.assertRaisesRegex(ValueError, "Ambiguous exemption"): + self.check() + + +if __name__ == "__main__": + unittest.main() diff --git a/gluten-ut/src-delta-validation/test/scala/org/apache/spark/sql/gluten/GlutenValidationSuite.scala b/gluten-ut/src-delta-validation/test/scala/org/apache/spark/sql/gluten/GlutenValidationSuite.scala new file mode 100644 index 0000000000..a29f6909df --- /dev/null +++ b/gluten-ut/src-delta-validation/test/scala/org/apache/spark/sql/gluten/GlutenValidationSuite.scala @@ -0,0 +1,110 @@ +/* + * 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. + */ +package org.apache.spark.sql.gluten + +import org.apache.spark.sql.QueryTest + +import org.scalatest.funsuite.AnyFunSuite + +import java.nio.charset.StandardCharsets.UTF_8 +import java.nio.file.Files + +class GlutenValidationSuite extends AnyFunSuite { + private val plugin = "org.apache.gluten.GlutenPlugin" + + test("enabled flag alone does not establish plugin or rule installation") { + intercept[IllegalArgumentException] { + GlutenValidation.assertSettings("", true, None, true) + } + + intercept[IllegalArgumentException] { + GlutenValidation.assertSettings(plugin, true, None, false) + } + } + + test("both session and thread disable switches are checked") { + intercept[IllegalArgumentException] { + GlutenValidation.assertSettings(plugin, false, None, true) + } + intercept[IllegalArgumentException] { + GlutenValidation.assertSettings(plugin, true, Some("false"), true) + } + intercept[IllegalArgumentException] { + GlutenValidation.assertSettings(plugin, true, Some("invalid"), true) + } + } + + test("plugin lists and restored defaults are accepted") { + GlutenValidation.assertSettings(s"other.Plugin, $plugin ", true, Some("true"), true) + GlutenValidation.assertSettings(plugin, true, None, true) + } + + test("exemption keys retain the legacy text normalization") { + assert(GlutenValidation.testKey("Suite", "a\r\nb ") == "Suite#a b") + } + + test("exemptions require explicit identities and reasons") { + val path = Files.createTempFile("gluten-exemptions", ".txt") + try { + Files.write(path, "# comment\nSuite#test\tdeliberately vanilla\n".getBytes(UTF_8)) + assert( + GlutenValidation.readExemptions(path) == Map("Suite#test" -> "deliberately vanilla")) + Seq( + "Suite#test", + "Suite#test\t", + "*#test\treason", + "Suite#test\tone\nSuite#test\ttwo", + "org.apache.spark.sql.gluten.GlutenActivationSuite#*\treason").foreach { + text => + Files.write(path, text.getBytes(UTF_8)) + intercept[IllegalArgumentException] { + GlutenValidation.readExemptions(path) + } + } + } finally { + Files.delete(path) + } + } +} + +class GlutenValidationIdentitySuite extends QueryTest with GlutenSharedSparkSession { + private def checkDefault(): Unit = + assert(spark.conf.get("spark.gluten.enabled", "true").toBoolean) + + Seq(" ", "\n", "\r", "\t", 0x2028.toChar.toString).foreach { + character => + test(s"partition column names and values contain '$character'") { + checkDefault() + } + } + + test("GlutenValidationIdentitySuite with nested struct and mixed types") { + checkDefault() + } + + test("org.apache.spark.sql.gluten.GlutenValidationIdentitySuite.case") { + checkDefault() + } + + test("case") { + checkDefault() + } + + test(" leading and trailing ") { + checkDefault() + } +}