This is an automated email from the ASF dual-hosted git repository.
zhouyuan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/gluten.git
The following commit(s) were added to refs/heads/main by this push:
new 30548fe6df [VL] Add iceberg UT CI whitelist (#13323)
30548fe6df is described below
commit 30548fe6df1a1d781c047179108127a787d6f2b9
Author: inf / Hazmi <[email protected]>
AuthorDate: Thu Oct 15 16:16:51 2026 +0300
[VL] Add iceberg UT CI whitelist (#13323)
---
.github/workflows/util/iceberg-upstream/compare.py | 146 +++++++++----
.github/workflows/util/iceberg-upstream/control.py | 27 ++-
.../util/iceberg-upstream/flaky-tests.csv | 20 ++
.github/workflows/util/iceberg-upstream/flaky.py | 47 ++++
.../workflows/util/iceberg-upstream/summarize.py | 13 +-
docs/developers/velox-backend-CI.md | 22 +-
.../test/python/test_compare.py | 242 +++++++++++++++------
.../src-iceberg-upstream/test/python/test_flaky.py | 215 ++++++++++++++++++
.../test/python/test_summary.py | 2 +
9 files changed, 602 insertions(+), 132 deletions(-)
diff --git a/.github/workflows/util/iceberg-upstream/compare.py
b/.github/workflows/util/iceberg-upstream/compare.py
index 9fe24650ec..5ffbf182c8 100644
--- a/.github/workflows/util/iceberg-upstream/compare.py
+++ b/.github/workflows/util/iceberg-upstream/compare.py
@@ -14,7 +14,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-"""Require each upstream Iceberg test to preserve or improve its base-commit
result."""
+"""Gate native-status regressions and new test failures; highlight other
changes."""
import argparse
import csv
@@ -23,6 +23,7 @@ import os
import re
from pathlib import Path
+from flaky import FLAKY_TESTS, is_flaky, load_flaky_tests
from summarize import REPORT_VERSION, native_data_proven, native_proven
PASSING = {
@@ -31,14 +32,6 @@ PASSING = {
"NO_ICEBERG_EXECUTION",
"PASSED_UNVERIFIED",
}
-ALLOWED = {
- "PASSED_NATIVE": {"PASSED_NATIVE"},
- "PASSED_FALLBACK": {"PASSED_FALLBACK", "PASSED_NATIVE"},
- "FAILED": PASSING | {"FAILED"},
- "NO_ICEBERG_EXECUTION": PASSING,
- "PASSED_UNVERIFIED": PASSING,
- "SKIPPED": PASSING | {"SKIPPED"},
-}
def parameters(row):
@@ -64,7 +57,9 @@ def parameters(row):
)
-def index(report, validate=True):
+def index(report, validate=True, flaky_tests=None):
+ if flaky_tests is None:
+ flaky_tests = load_flaky_tests()
if validate and report.get("version") != REPORT_VERSION:
raise ValueError(
"Incompatible coverage schema; a baseline from the current harness
is required"
@@ -74,19 +69,19 @@ def index(report, validate=True):
key = (row["class"], row["name"])
if key in tests:
raise ValueError(f"Duplicate test result: {key}")
- if validate and (row["status"] not in ALLOWED or
row.get("coverage_error")):
+ tests[key] = row
+ if not validate or is_flaky(row, flaky_tests):
+ continue
+ if row["status"] not in PASSING | {"FAILED", "SKIPPED"} or row.get(
+ "coverage_error"
+ ):
raise ValueError(f"Invalid execution coverage: {key}:
{row['status']}")
- if validate and row["status"] == "PASSED_NATIVE" and not
native_proven(row):
+ if row["status"] == "PASSED_NATIVE" and not native_proven(row):
raise ValueError(f"Unproven native coverage: {key}")
- if (
- validate
- and row["status"] == "FAILED"
- and row.get("failure_origin") != "GLUTEN"
- ):
+ if row["status"] == "FAILED" and row.get("failure_origin") != "GLUTEN":
raise ValueError(
f"Unverified failure attribution: {key}:
{row.get('failure_origin', 'MISSING')}"
)
- tests[key] = row
if not tests:
raise ValueError("Cannot gate an empty test report")
return tests
@@ -108,9 +103,13 @@ def proven_native_nodes(row):
}
-def compare(baseline, current, validate=True):
- before, after = index(baseline, validate), index(current, validate)
- regressions, improvements, changes = [], [], []
+def compare(baseline, current, validate=True, flaky_tests=None):
+ if flaky_tests is None:
+ flaky_tests = load_flaky_tests()
+ before = index(baseline, validate, flaky_tests)
+ after = index(current, validate, flaky_tests)
+ regressions, improvements, changes, highlights, ignored_changes = [], [],
[], [], []
+ ignored_tests = 0
for key in sorted(before.keys() | after.keys()):
old = before.get(key, {}).get("status", "NEW")
new = after.get(key, {}).get("status", "MISSING")
@@ -122,10 +121,9 @@ def compare(baseline, current, validate=True):
"after": new,
"base": previous,
"current": current_row,
+ "ignored": is_flaky(previous or current_row, flaky_tests),
}
- # Newly discovered upstream tests must pass or be disabled upstream.
Missing or
- # newly skipped tests cannot conceal a previous pass (or an unresolved
failure).
- allowed = PASSING | {"SKIPPED"} if old == "NEW" else ALLOWED.get(old,
set())
+ ignored_tests += change["ignored"]
mismatch = bool(
previous
and current_row
@@ -134,18 +132,29 @@ def compare(baseline, current, validate=True):
and parameters(previous) != parameters(current_row)
)
lost_native = proven_native_nodes(previous) -
proven_native_nodes(current_row)
+ if lost_native:
+ change["lost_native"] = sorted(lost_native)
if mismatch:
- change["reason"] = "PARAMETER_MISMATCH"
- regressions.append(change)
- elif new not in allowed:
+ change["parameter_mismatch"] = True
+ # Only a native test becoming non-native or a newly failed test gates.
+ # Node-level evidence changes remain visible even when the status is
unchanged.
+ regression = (old == "PASSED_NATIVE" and new != old) or (
+ new == "FAILED" and old != new
+ )
+ improvement = (
+ not regression
+ and not mismatch
+ and (
+ (new == "PASSED_NATIVE" and old != new)
+ or (old == "FAILED" and new in PASSING)
+ )
+ )
+ if regression:
change["reason"] = "STATUS_REGRESSION"
- regressions.append(change)
+ elif mismatch:
+ change["reason"] = "PARAMETER_MISMATCH"
elif lost_native:
- change["reason"] = "NATIVE_COVERAGE_REGRESSION"
- change["lost_native"] = sorted(lost_native)
- regressions.append(change)
- elif old != new and new != "SKIPPED":
- improvements.append(change)
+ change["reason"] = "NATIVE_COVERAGE_CHANGE"
# Metrics/timings vary between successful runs. Compare their
existence and node types,
# while retaining the actual values in each changed row for
verification.
evidence_changed = any(
@@ -170,10 +179,23 @@ def compare(baseline, current, validate=True):
"reason", "STATUS_CHANGE" if old != new else "EVIDENCE_CHANGE"
)
changes.append(change)
+ if change["ignored"]:
+ ignored_changes.append(change)
+ elif regression:
+ regressions.append(change)
+ elif improvement:
+ improvements.append(change)
+ if lost_native:
+ highlights.append(change)
+ else:
+ highlights.append(change)
return {
"regressions": regressions,
"improvements": improvements,
"changes": changes,
+ "highlights": highlights,
+ "ignored_changes": ignored_changes,
+ "ignored_tests": ignored_tests,
"base_tests": len(before),
"current_tests": len(after),
}
@@ -197,6 +219,9 @@ def write_changes(path, changes):
)
columns = [
"reason",
+ "ignored",
+ "lost_native",
+ "parameter_mismatch",
"class",
"test",
"before",
@@ -216,6 +241,9 @@ def write_changes(path, changes):
old, new = row["base"], row["current"]
values = [
row["reason"],
+ row["ignored"],
+ json.dumps(row.get("lost_native", [])),
+ row.get("parameter_mismatch", False),
row["class"],
row["name"],
row["before"],
@@ -239,6 +267,12 @@ def main():
parser.add_argument("baseline", type=Path)
parser.add_argument("current", type=Path)
parser.add_argument("--output", type=Path, required=True)
+ parser.add_argument(
+ "--flaky-tests",
+ type=Path,
+ default=FLAKY_TESTS,
+ help="class,method CSV whitelist (defaults to the adjacent
flaky-tests.csv)",
+ )
parser.add_argument(
"--bootstrap",
action="store_true",
@@ -246,6 +280,11 @@ def main():
)
args = parser.parse_args()
baseline, current, errors = None, None, []
+ flaky_tests = set()
+ try:
+ flaky_tests = load_flaky_tests(args.flaky_tests)
+ except (ValueError, OSError, csv.Error) as error:
+ errors.append(f"Flaky test whitelist unavailable: {error}")
for label, path in (("Baseline", args.baseline), ("Current report",
args.current)):
if label == "Baseline" and args.bootstrap and not path.exists():
continue
@@ -269,25 +308,30 @@ def main():
"changes": [],
"regressions": [],
"improvements": [],
+ "highlights": [],
+ "ignored_changes": [],
+ "ignored_tests": 0,
"base_tests": len((baseline or {}).get("tests", [])),
"current_tests": len((current or {}).get("tests", [])),
}
try:
if errors:
if current is not None:
- index(current)
+ index(current, flaky_tests=flaky_tests)
elif baseline is None and args.bootstrap:
- index(current)
+ index(current, flaky_tests=flaky_tests)
result["baseline_bootstrap"] = True
else:
- result = compare(baseline, current)
+ result = compare(baseline, current, flaky_tests=flaky_tests)
result["comparison_performed"] = True
except (ValueError, KeyError, TypeError) as error:
# Still export diagnostic differences when complete reports exist, but
invalid
# evidence must fail the gate even if no statuses changed.
if baseline is not None and current is not None:
try:
- result = compare(baseline, current, validate=False)
+ result = compare(
+ baseline, current, validate=False, flaky_tests=flaky_tests
+ )
except (ValueError, KeyError, TypeError):
pass
result["comparison_performed"] = False
@@ -322,15 +366,29 @@ def main():
"Report validation failed; the CSV is diagnostic only: "
+ "; ".join(result["validation_errors"])
)
- for row in result["regressions"][:20]:
- detail = row["reason"]
- if row.get("lost_native"):
- detail += ": " + ", ".join(row["lost_native"])
+ if result["ignored_tests"]:
lines.append(
- f"- `{row['class']}#{row['name']}`: {row['before']} →
{row['after']} ({detail})"
+ f"{result['ignored_tests']} whitelisted test invocations excluded
from gating."
)
- if len(result["regressions"]) > 20:
- lines.append("See the comparison artifact for the remaining
regressions.")
+ for title, rows in (
+ ("Regressions", result["regressions"]),
+ ("Other changes (informational)", result["highlights"]),
+ ("Whitelisted changes (ignored)", result["ignored_changes"]),
+ ):
+ if not rows:
+ continue
+ lines.extend(["", f"### {title}: {len(rows)}", ""])
+ for row in rows[:20]:
+ detail = row["reason"]
+ if row.get("lost_native"):
+ detail += "; native nodes lost: " + ",
".join(row["lost_native"])
+ if row.get("parameter_mismatch"):
+ detail += "; parameters changed"
+ lines.append(
+ f"- `{row['class']}#{row['name']}`: {row['before']} →
{row['after']} ({detail})"
+ )
+ if len(rows) > 20:
+ lines.append("See the comparison artifact for the remaining
changes.")
text = "\n".join(lines) + "\n"
print(text)
if os.environ.get("GITHUB_STEP_SUMMARY"):
diff --git a/.github/workflows/util/iceberg-upstream/control.py
b/.github/workflows/util/iceberg-upstream/control.py
index 2298d82436..9c56d5c273 100644
--- a/.github/workflows/util/iceberg-upstream/control.py
+++ b/.github/workflows/util/iceberg-upstream/control.py
@@ -24,6 +24,7 @@ from pathlib import Path
import xml.etree.ElementTree as ET
from compare import PASSING, parameters
+from flaky import is_flaky, load_flaky_tests
from summarize import classify, write_reports
RUNTIME_FAILURE = re.compile(
@@ -32,22 +33,26 @@ RUNTIME_FAILURE = re.compile(
)
-def selection(results):
+def selection(results, flaky_tests=None):
+ if flaky_tests is None:
+ flaky_tests = load_flaky_tests()
methods = collections.defaultdict(set)
for row in results:
- if row["status"] == "FAILED":
+ if row["status"] == "FAILED" and not is_flaky(row, flaky_tests):
methods[row["class"]].add(row["name"].split("(")[0])
return ",".join(
name + "#" + "+".join(sorted(tests)) for name, tests in
sorted(methods.items())
)
-def attribute(results, controls):
+def attribute(results, controls, flaky_tests=None):
+ if flaky_tests is None:
+ flaky_tests = load_flaky_tests()
index = {(row["class"], row["name"]): row for row in controls}
if len(index) != len(controls):
raise ValueError("Duplicate vanilla controls")
for row in results:
- if row["status"] != "FAILED":
+ if row["status"] != "FAILED" or is_flaky(row, flaky_tests):
continue
control = index.get((row["class"], row["name"]))
if control is None:
@@ -79,13 +84,15 @@ def main():
results = json.loads((args.reports /
"execution-coverage.json").read_text())[
"tests"
]
+ flaky_tests = load_flaky_tests()
+ active = [row for row in results if not is_flaky(row, flaky_tests)]
if args.select is not None:
- args.select.write_text(selection(results))
+ args.select.write_text(selection(results, flaky_tests))
return 0
if args.vanilla is None:
parser.error("Specify --select or --vanilla")
controls = []
- if any(row["status"] == "FAILED" for row in results):
+ if any(row["status"] == "FAILED" for row in active):
for path in args.vanilla.glob("TEST-*.xml"):
props = {
p.get("name"): p.get("value")
@@ -97,10 +104,10 @@ def main():
)
controls = classify(args.vanilla)
write_reports(args.vanilla, controls, label="Vanilla Spark controls")
- results = attribute(results, controls)
+ results = attribute(results, controls, flaky_tests)
broken = [
row
- for row in results
+ for row in active
if row["status"] == "FAILED" and row.get("failure_origin") != "GLUTEN"
]
write_reports(args.reports, results, label="Gluten coverage after vanilla
controls")
@@ -108,9 +115,7 @@ def main():
print(
f"{len(broken)} failures could not be attributed reliably; see
failed.csv"
)
- return int(
- bool(broken) or any(row["status"] == "COVERAGE_ERROR" for row in
results)
- )
+ return int(bool(broken) or any(row["status"] == "COVERAGE_ERROR" for row
in active))
if __name__ == "__main__":
diff --git a/.github/workflows/util/iceberg-upstream/flaky-tests.csv
b/.github/workflows/util/iceberg-upstream/flaky-tests.csv
new file mode 100644
index 0000000000..7f5ec34e6d
--- /dev/null
+++ b/.github/workflows/util/iceberg-upstream/flaky-tests.csv
@@ -0,0 +1,20 @@
+# 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.
+#
+# All invocations of these methods run and are reported, but do not affect the
gate.
+# Use exact class and method names; omit method arguments and invocation
indices.
+class,method
+# Example
+#
org.apache.iceberg.spark.extensions.TestAlterTablePartitionFields,testDropPartitionAndSourceColumnLong
diff --git a/.github/workflows/util/iceberg-upstream/flaky.py
b/.github/workflows/util/iceberg-upstream/flaky.py
new file mode 100644
index 0000000000..8cf347c7fe
--- /dev/null
+++ b/.github/workflows/util/iceberg-upstream/flaky.py
@@ -0,0 +1,47 @@
+# 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.
+
+"""Methods whose results are reported but excluded from the Iceberg gate."""
+
+import csv
+from pathlib import Path
+
+FLAKY_TESTS = Path(__file__).with_name("flaky-tests.csv")
+
+
+def load_flaky_tests(path=FLAKY_TESTS):
+ tests = set()
+ with path.open(newline="") as source:
+ reader = csv.reader(
+ (line for line in source if not line.lstrip().startswith("#")),
strict=True
+ )
+ if next(reader, None) != ["class", "method"]:
+ raise ValueError(
+ f"Expected class,method header in flaky test whitelist: {path}"
+ )
+ for row in reader:
+ if not row:
+ continue
+ key = tuple(value.strip() for value in row)
+ if len(key) != 2 or not all(key):
+ raise ValueError(f"Invalid flaky test whitelist entry: {path}:
{row}")
+ if key in tests:
+ raise ValueError(f"Duplicate flaky test whitelist entry:
{path}: {row}")
+ tests.add(key)
+ return tests
+
+
+def is_flaky(row, tests):
+ return (row.get("class"), row.get("name", "").partition("(")[0]) in tests
diff --git a/.github/workflows/util/iceberg-upstream/summarize.py
b/.github/workflows/util/iceberg-upstream/summarize.py
index 82986427e9..4ddb441c53 100644
--- a/.github/workflows/util/iceberg-upstream/summarize.py
+++ b/.github/workflows/util/iceberg-upstream/summarize.py
@@ -24,6 +24,8 @@ from pathlib import Path
import sys
import xml.etree.ElementTree as ET
+from flaky import is_flaky, load_flaky_tests
+
STATUSES = (
"PASSED_NATIVE",
"PASSED_FALLBACK",
@@ -128,6 +130,8 @@ def classify(directory):
def write_reports(directory, results, label="Iceberg execution coverage"):
+ flaky_tests = load_flaky_tests()
+ results = [dict(row, ignored=is_flaky(row, flaky_tests)) for row in
results]
counts = collections.Counter(row["status"] for row in results)
summary = {status: counts[status] for status in STATUSES}
(directory / "execution-coverage.json").write_text(
@@ -144,6 +148,7 @@ def write_reports(directory, results, label="Iceberg
execution coverage"):
)
columns = (
"status",
+ "ignored",
"class",
"test",
"display_name",
@@ -184,6 +189,7 @@ def write_reports(directory, results, label="Iceberg
execution coverage"):
writer.writerow(
(
row["status"],
+ row["ignored"],
row["class"],
row["name"],
row.get("display_name", ""),
@@ -217,12 +223,17 @@ def write_reports(directory, results, label="Iceberg
execution coverage"):
)
table = f"### {label}\n\n| Iceberg execution result | Tests |\n| --- |
---: |\n"
table += "".join(f"| {status} | {count} |\n" for status, count in
summary.items())
+ ignored = sum(row["ignored"] for row in results)
+ if ignored:
+ table += f"\n{ignored} whitelisted test invocations are reported above
but excluded from gating.\n"
print(table)
if os.environ.get("GITHUB_STEP_SUMMARY"):
with open(os.environ["GITHUB_STEP_SUMMARY"], "a") as output:
output.write(table)
# Control runs and the comparison gate handle upstream failures. Broken
observation fails here.
- return int(bool(counts["COVERAGE_ERROR"]))
+ return int(
+ any(row["status"] == "COVERAGE_ERROR" and not row["ignored"] for row
in results)
+ )
def main(directory):
diff --git a/docs/developers/velox-backend-CI.md
b/docs/developers/velox-backend-CI.md
index 0d67413d7f..c074ec45d2 100644
--- a/docs/developers/velox-backend-CI.md
+++ b/docs/developers/velox-backend-CI.md
@@ -92,22 +92,22 @@ final adaptive plans and subqueries.
| `SKIPPED` | Disabled upstream or aborted by an assumption. |
| `COVERAGE_ERROR` | Missing or invalid execution evidence; reporting fails. |
-Failing methods run again without Gluten. `control.py` attributes a failure to
Gluten only
-when vanilla Spark passes with matching parameters and settings. Missing,
skipped, failing
-or mismatched controls fail the gate. Timeouts, resource exhaustion and
broken-runtime
-failures remain unverified even if the control passes.
+Failing methods run again without Gluten unless whitelisted. `control.py`
attributes a failure
+to Gluten only when vanilla Spark passes with matching parameters and
settings. Missing,
+skipped, failing or mismatched controls fail the gate. Timeouts, resource
exhaustion and
+broken-runtime failures remain unverified even if the control passes.
Parameter/setup randomness is seeded.
`iceberg-upstream-gate (3.5)` applies these rules to each invocation:
-- Existing failures may stay failed or improve; new tests must pass or be
disabled upstream.
-- Fallback passes must remain fallback or improve to native; native passes
must remain native.
-- Executed tests cannot disappear or become skipped, and execution evidence
cannot be lost.
-- Mixed fallback/native passes must preserve every proven native node type,
even if their
- overall status is unchanged. Positive counter fluctuations alone are not
regressions.
-- Missing artifacts, duplicate results, incompatible report schemas,
mismatched parameters,
- observer errors and failed infrastructure checks fail the gate.
+- Native passes must remain native; newly failed invocations fail the gate.
+- Other changes are informational, including missing/skipped non-native tests,
base/head
+ parameter differences, and lost native node coverage without a native-status
regression.
+- Missing artifacts, duplicate results, incompatible report schemas, invalid
execution evidence,
+ unverified failure attribution and failed infrastructure checks fail the
gate.
+
+Tests listed by `class,method` in
[flaky-tests.csv](../../.github/workflows/util/iceberg-upstream/flaky-tests.csv)
still run and appear in reports, but do not affect the gate.
`iceberg-upstream-spark3.5-reports` contains base/head coverage JSON and CSVs,
head raw logs,
comparison validation details and a changes-only CSV. Separate CSVs show
native, fallback,
diff --git a/gluten-ut/src-iceberg-upstream/test/python/test_compare.py
b/gluten-ut/src-iceberg-upstream/test/python/test_compare.py
index 4b87024459..a3acbecb6b 100644
--- a/gluten-ut/src-iceberg-upstream/test/python/test_compare.py
+++ b/gluten-ut/src-iceberg-upstream/test/python/test_compare.py
@@ -18,6 +18,7 @@ import copy
import csv
import sys
import json
+import os
import subprocess
import tempfile
from pathlib import Path
@@ -87,49 +88,61 @@ class ComparisonTest(unittest.TestCase):
self.assertTrue(diagnostic["comparison_performed"])
self.assertEqual(1, len(diagnostic["regressions"]))
- def test_native_fallback_failure_transition_rules(self):
- allowed = {
- ("FAILED", "FAILED"),
- ("FAILED", "PASSED_FALLBACK"),
- ("FAILED", "PASSED_NATIVE"),
- ("PASSED_FALLBACK", "PASSED_FALLBACK"),
- ("PASSED_FALLBACK", "PASSED_NATIVE"),
- ("PASSED_NATIVE", "PASSED_NATIVE"),
- }
- for old in ("FAILED", "PASSED_FALLBACK", "PASSED_NATIVE"):
- for new in ("FAILED", "PASSED_FALLBACK", "PASSED_NATIVE"):
+ def test_only_native_status_loss_and_new_failures_gate(self):
+ statuses = sorted(comparison.PASSING | {"FAILED", "SKIPPED"})
+ for old in statuses:
+ for new in statuses:
with self.subTest(before=old, after=new):
result = comparison.compare(report(old), report(new))
- self.assertEqual(
- (old, new) not in allowed, bool(result["regressions"])
+ fails = (old == "PASSED_NATIVE" and new != old) or (
+ new == "FAILED" and old != new
)
+ self.assertEqual(fails, bool(result["regressions"]))
- def test_disabling_or_removing_tests_does_not_hide_regressions(self):
- for old in (
- "FAILED",
- "PASSED_FALLBACK",
- "PASSED_NATIVE",
- "NO_ICEBERG_EXECUTION",
- ):
+ def test_missing_non_native_tests_are_highlighted(self):
+ for old in sorted(comparison.PASSING | {"FAILED", "SKIPPED"}):
with self.subTest(before=old):
- self.assertTrue(
- comparison.compare(report(old),
report("SKIPPED"))["regressions"]
- )
result = comparison.compare(
- report(old), report("PASSED_NATIVE", "different")
+ report(old), report("PASSED_FALLBACK", "different")
)
- self.assertEqual("MISSING", result["regressions"][0]["after"])
+ self.assertEqual(old == "PASSED_NATIVE",
bool(result["regressions"]))
+ changes = [
+ row for row in result["changes"] if row["after"] ==
"MISSING"
+ ]
+ self.assertEqual(1, len(changes))
+ if old != "PASSED_NATIVE":
+ self.assertIn(changes[0], result["highlights"])
- def test_losing_execution_evidence_is_not_an_improvement(self):
- for old in ("PASSED_FALLBACK", "PASSED_NATIVE"):
- for new in ("NO_ICEBERG_EXECUTION", "PASSED_UNVERIFIED"):
- result = comparison.compare(report(old), report(new))
- self.assertTrue(result["regressions"])
+ def
test_passing_coverage_changes_are_informational_without_special_exceptions(
+ self,
+ ):
+ statuses = ("PASSED_FALLBACK", "PASSED_UNVERIFIED",
"NO_ICEBERG_EXECUTION")
+ for expected_exception in (False, True):
+ for old in statuses:
+ for new in statuses:
+ with self.subTest(
+ expected=expected_exception, before=old, after=new
+ ):
+ base, current = report(old), report(new)
+ for source in (base, current):
+ source["tests"][0].update(
+ body_started=True,
expected_exception=expected_exception
+ )
+ result = comparison.compare(base, current)
+ self.assertFalse(result["regressions"])
+ self.assertFalse(result["improvements"])
+ self.assertEqual(old != new,
bool(result["highlights"]))
+ if old != new:
+ self.assertEqual(
+ base["tests"][0],
result["highlights"][0]["base"]
+ )
+ self.assertEqual(
+ current["tests"][0],
result["highlights"][0]["current"]
+ )
- def test_mixed_execution_must_preserve_proven_native_coverage(self):
+ def test_mixed_execution_coverage_loss_is_highlighted(self):
base = report("PASSED_NATIVE")
- row = base["tests"][0]
- row.update(status="PASSED_FALLBACK", fallback=["AppendDataExec"])
+ base["tests"][0].update(status="PASSED_FALLBACK",
fallback=["AppendDataExec"])
for loss in (
"fallback",
"unverified",
@@ -141,8 +154,7 @@ class ComparisonTest(unittest.TestCase):
current = copy.deepcopy(base)
after = current["tests"][0]
if loss in ("fallback", "unverified"):
- after["native"] = []
- after["native_metrics"] = []
+ after.update(native=[], native_metrics=[])
after.setdefault(loss, []).append("IcebergScanTransformer")
elif loss == "metrics":
after["native_metrics"][0]["metrics"]["rawInputRows"] = 0
@@ -154,27 +166,26 @@ class ComparisonTest(unittest.TestCase):
after["expected_exception"] = True
with self.subTest(loss=loss):
result = comparison.compare(base, current)
- self.assertEqual(1, len(result["regressions"]))
- self.assertEqual([], result["improvements"])
- change = result["changes"][0]
- self.assertEqual("NATIVE_COVERAGE_REGRESSION",
change["reason"])
- self.assertEqual(["IcebergScanTransformer"],
change["lost_native"])
+ self.assertFalse(result["regressions"])
+ self.assertFalse(result["improvements"])
+ self.assertEqual(1, len(result["highlights"]))
+ self.assertEqual(
+ "NATIVE_COVERAGE_CHANGE", result["highlights"][0]["reason"]
+ )
+ self.assertEqual(
+ ["IcebergScanTransformer"],
result["highlights"][0]["lost_native"]
+ )
- def
test_native_node_types_cannot_disappear_even_when_status_improves(self):
+ def test_native_to_native_node_changes_are_highlighted(self):
base = report("PASSED_NATIVE")
- row = base["tests"][0]
- row["native"].append("IcebergWriteExec")
- row["native_metrics"].append(
+ base["tests"][0]["native"].append("IcebergWriteExec")
+ base["tests"][0]["native_metrics"].append(
{"node": "IcebergWriteExec", "metrics": {"numWrittenBytes": 10}}
)
- for status in ("PASSED_NATIVE", "PASSED_FALLBACK",
"PASSED_UNVERIFIED"):
- row["status"] = status
- with self.subTest(status=status):
- result = comparison.compare(base, report("PASSED_NATIVE"))
- self.assertEqual(
- ["IcebergWriteExec"],
result["regressions"][0]["lost_native"]
- )
- self.assertFalse(result["improvements"])
+ result = comparison.compare(base, report("PASSED_NATIVE"))
+ self.assertFalse(result["regressions"])
+ self.assertFalse(result["improvements"])
+ self.assertEqual(["IcebergWriteExec"],
result["highlights"][0]["lost_native"])
def
test_counter_variation_and_unproven_nodes_do_not_gate_mixed_execution(self):
base = report("PASSED_NATIVE")
@@ -186,19 +197,119 @@ class ComparisonTest(unittest.TestCase):
row["native"].append("UnprovenPlanNode")
self.assertFalse(comparison.compare(base, current)["regressions"])
- def test_new_tests_and_previously_skipped_tests_cannot_add_failures(self):
+ def test_whitelist_excludes_all_invocations_from_every_gate_outcome(self):
+ whitelist = {("org.apache.iceberg.TestScan", "read")}
+ statuses = sorted(comparison.PASSING | {"FAILED", "SKIPPED",
"COVERAGE_ERROR"})
+ for old in statuses:
+ for new in statuses:
+ with self.subTest(before=old, after=new):
+ base, current = report(old), report(new)
+ current["tests"][0].update(
+ parameters={"formatVersion": "3"},
+ failure_origin="CONTROL_MISSING",
+ )
+ result = comparison.compare(base, current,
flaky_tests=whitelist)
+ self.assertFalse(result["regressions"])
+ self.assertFalse(result["improvements"])
+ self.assertFalse(result["highlights"])
+ self.assertEqual(1, result["ignored_tests"])
+ self.assertTrue(result["changes"][0]["ignored"])
+ for name in ("read", "read()", "read()[12]", "read(int)[3]"):
+ base, current = report("PASSED_NATIVE", name), report("FAILED",
name)
+ current["tests"][0].pop("failure_origin")
+ result = comparison.compare(base, current, flaky_tests=whitelist)
+ self.assertFalse(result["regressions"])
+ self.assertEqual(1, len(result["ignored_changes"]))
+ removed = comparison.compare(
+ report("PASSED_NATIVE"),
+ report("PASSED_FALLBACK", "other"),
+ flaky_tests=whitelist,
+ )
+ self.assertFalse(removed["regressions"])
+ self.assertEqual("MISSING", removed["ignored_changes"][0]["after"])
+ for fields in (
+ {"class": "org.apache.iceberg.OtherTest"},
+ {"name": "readOther()[1]"},
+ ):
+ base, current = report("PASSED_NATIVE"), report("FAILED")
+ base["tests"][0].update(fields)
+ current["tests"][0].update(fields)
+ self.assertTrue(
+ comparison.compare(base, current,
flaky_tests=whitelist)["regressions"]
+ )
+
+ def test_whitelist_does_not_hide_invalid_report_structure(self):
+ whitelist = {("org.apache.iceberg.TestScan", "read")}
+ for invalid in (
+ {"version": 1, "tests": report("FAILED")["tests"]},
+ {"version": 2, "tests": []},
+ {"version": 2, "tests": report("FAILED")["tests"] * 2},
+ ):
+ with self.subTest(invalid=invalid), self.assertRaises(ValueError):
+ comparison.compare(invalid, report("FAILED"),
flaky_tests=whitelist)
+
+ def
test_csv_whitelist_and_informational_changes_are_visible_in_gate_summary(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+ base, current = report("PASSED_NATIVE"), report("FAILED")
+ current["tests"][0].pop("failure_origin")
+ base["tests"].extend(report("PASSED_FALLBACK",
"anotherMethod")["tests"])
+ current["tests"].extend(
+ report("PASSED_UNVERIFIED", "anotherMethod")["tests"]
+ )
+ (root / "base.json").write_text(json.dumps(base))
+ (root / "head.json").write_text(json.dumps(current))
+ config = root / "flaky.csv"
+ config.write_text(
+ "# Known flaky
test\nclass,method\norg.apache.iceberg.TestScan,read\n"
+ )
+ command = [
+ sys.executable,
+ str(SCRIPT),
+ str(root / "base.json"),
+ str(root / "head.json"),
+ "--output",
+ str(root / "diff.json"),
+ "--flaky-tests",
+ str(config),
+ ]
+ summary = root / "summary.md"
+ result = subprocess.run(
+ command,
+ capture_output=True,
+ text=True,
+ env=dict(os.environ, GITHUB_STEP_SUMMARY=str(summary)),
+ )
+ self.assertEqual(0, result.returncode, result.stderr +
result.stdout)
+ diagnostic = json.loads((root / "diff.json").read_text())
+ self.assertFalse(diagnostic["regressions"])
+ self.assertFalse(diagnostic["improvements"])
+ self.assertEqual(1, len(diagnostic["ignored_changes"]))
+ self.assertEqual(1, len(diagnostic["highlights"]))
+ self.assertIn("Other changes (informational)", summary.read_text())
+ self.assertIn("PASSED_FALLBACK → PASSED_UNVERIFIED",
summary.read_text())
+ self.assertIn("Whitelisted changes (ignored)", summary.read_text())
+ self.assertIn("PASSED_NATIVE → FAILED", summary.read_text())
+ with (root / "diff.csv").open(newline="") as output:
+ rows = {row["test"]: row for row in csv.DictReader(output)}
+ self.assertEqual("True", rows["read()[1]"]["ignored"])
+ self.assertEqual("False", rows["anotherMethod"]["ignored"])
+ for contents in ("class,method,reason\n", None):
+ with self.subTest(contents=contents):
+ if contents is None:
+ config.unlink()
+ else:
+ config.write_text(contents)
+ result = subprocess.run(command, capture_output=True,
text=True)
+ self.assertEqual(1, result.returncode, result.stderr)
+ diagnostic = json.loads((root / "diff.json").read_text())
+ self.assertTrue(diagnostic["validation_errors"])
+
+ def test_new_tests_cannot_add_failures(self):
current = report("PASSED_NATIVE")
current["tests"].extend(report("FAILED", "new")["tests"])
result = comparison.compare(report("PASSED_NATIVE"), current)
self.assertEqual("NEW", result["regressions"][0]["before"])
- self.assertTrue(
- comparison.compare(report("SKIPPED"),
report("FAILED"))["regressions"]
- )
- self.assertFalse(
- comparison.compare(report("SKIPPED"), report("PASSED_NATIVE"))[
- "regressions"
- ]
- )
def test_empty_duplicate_and_broken_reports_fail_closed(self):
duplicate = report("PASSED_NATIVE")
@@ -214,7 +325,7 @@ class ComparisonTest(unittest.TestCase):
with self.assertRaises(ValueError):
comparison.compare(report("PASSED_NATIVE"), invalid)
- def test_parameters_must_match_except_fixture_ports(self):
+ def test_parameter_changes_are_highlighted_except_fixture_ports(self):
old, new = report("PASSED_NATIVE"), report("PASSED_NATIVE")
old["tests"][0]["parameters"] = {
"config": "uri=http://localhost:1234/";,
@@ -228,10 +339,10 @@ class ComparisonTest(unittest.TestCase):
new["tests"][0]["parameters"]["vectorized"] = "false"
self.assertEqual(
"PARAMETER_MISMATCH",
- comparison.compare(old, new)["regressions"][0]["reason"],
+ comparison.compare(old, new)["highlights"][0]["reason"],
)
- def test_random_uuid_literal_is_data_but_other_arguments_must_match(self):
+ def
test_random_uuid_literals_are_ignored_but_other_arguments_are_highlighted(self):
old, new = report("PASSED_FALLBACK"), report("PASSED_FALLBACK")
old["tests"][0][
"display_name"
@@ -243,7 +354,8 @@ class ComparisonTest(unittest.TestCase):
new["tests"][0][
"display_name"
] = "[10] string, f41c0f33-5800-453f-b4d5-0d77ca0d1915"
- self.assertTrue(comparison.compare(old, new)["regressions"])
+ self.assertFalse(comparison.compare(old, new)["regressions"])
+ self.assertTrue(comparison.compare(old, new)["highlights"])
def test_gate_rejects_unproven_native_and_unattributed_failures(self):
invalid_native, invalid_failure = report("PASSED_NATIVE"),
report("FAILED")
diff --git a/gluten-ut/src-iceberg-upstream/test/python/test_flaky.py
b/gluten-ut/src-iceberg-upstream/test/python/test_flaky.py
new file mode 100644
index 0000000000..fa247d2e33
--- /dev/null
+++ b/gluten-ut/src-iceberg-upstream/test/python/test_flaky.py
@@ -0,0 +1,215 @@
+# 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 csv
+import json
+from pathlib import Path
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+import xml.etree.ElementTree as ET
+
+SCRIPTS = (
+ Path(__file__).resolve().parents[4] /
".github/workflows/util/iceberg-upstream"
+)
+sys.path.insert(0, str(SCRIPTS))
+from compare import index
+from control import attribute, selection
+from flaky import is_flaky, load_flaky_tests
+
+
+class FlakyTest(unittest.TestCase):
+ def
test_two_column_whitelist_matches_exact_methods_across_invocations(self):
+ with tempfile.TemporaryDirectory() as temporary:
+ path = Path(temporary) / "flaky.csv"
+ path.write_text("# Confirmed flake\nclass,method\nexample.Test,
read\n\n")
+ tests = load_flaky_tests(path)
+ self.assertEqual({("example.Test", "read")}, tests)
+ for name in ("read", "read()", "read()[12]", "read(int)[1]"):
+ self.assertTrue(
+ is_flaky({"class": "example.Test", "name": name}, tests)
+ )
+ for row in (
+ {"class": "example.Other", "name": "read"},
+ {"class": "example.Test", "name": "reader()"},
+ ):
+ self.assertFalse(is_flaky(row, tests))
+
+ def test_invalid_whitelist_fails_configuration_validation(self):
+ header, row = "class,method\n", "example.Test,read\n"
+ with tempfile.TemporaryDirectory() as temporary:
+ path = Path(temporary) / "flaky.csv"
+ for contents in (
+ "",
+ "method,class\n",
+ header + row + row,
+ header + "example.Test, \n",
+ header + "example.Test\n",
+ header + "example.Test,read,extra\n",
+ header + 'example.Test,"unclosed',
+ ):
+ with self.subTest(contents=contents):
+ path.write_text(contents)
+ with self.assertRaises((ValueError, csv.Error)):
+ load_flaky_tests(path)
+
+ def test_controls_are_only_required_for_unlisted_failures(self):
+ tests = {("example.Test", "flakyMethod")}
+ ignored = {
+ "class": "example.Test",
+ "name": "flakyMethod()[1]",
+ "status": "FAILED",
+ }
+ active = {
+ "class": "example.Test",
+ "name": "otherMethod()[1]",
+ "status": "FAILED",
+ }
+ self.assertEqual(
+ "example.Test#otherMethod", selection([ignored, active], tests)
+ )
+ attribute([ignored, active], [], tests)
+ self.assertNotIn("failure_origin", ignored)
+ self.assertEqual("CONTROL_MISSING", active["failure_origin"])
+
+ def
test_whitelisted_failures_survive_the_complete_reporting_pipeline(self):
+ class_name, method = "org.apache.iceberg.TestScan", "flakyMethod"
+ for status in ("FAILED", "COVERAGE_ERROR"):
+ with self.subTest(
+ status=status
+ ), tempfile.TemporaryDirectory() as temporary:
+ root = Path(temporary)
+ scripts = root / "harness"
+ scripts.mkdir()
+ for script in (
+ "compare.py",
+ "control.py",
+ "flaky.py",
+ "merge.py",
+ "summarize.py",
+ ):
+ shutil.copy2(SCRIPTS / script, scripts / script)
+ (scripts / "flaky-tests.csv").write_text(
+ f"class,method\n{class_name},{method}\n"
+ )
+ shard = root / "shard0"
+ shard.mkdir()
+ name = method + "()[2]"
+ ignored = {
+ "class": class_name,
+ "name": name,
+ "status": "PASSED_FALLBACK",
+ }
+ active = {
+ "class": class_name,
+ "name": "nonFlakyMethod",
+ "status": "PASSED_FALLBACK",
+ }
+ suite = ET.Element("testsuite")
+ case = ET.SubElement(suite, "testcase", classname=class_name,
name=name)
+ ET.SubElement(case, "failure", message="flaky assertion")
+ ET.SubElement(
+ suite, "testcase", classname=class_name,
name=active["name"]
+ )
+ ET.ElementTree(suite).write(shard / "TEST-queries.xml")
+ # Missing per-test observation must also remain non-gating for
this method.
+ observed = [active] + ([ignored] if status == "FAILED" else [])
+ (shard / "coverage-queries.jsonl").write_text(
+ "".join(json.dumps(row) + "\n" for row in observed)
+ )
+ (shard / "shard.json").write_text(json.dumps({"index": 0,
"count": 1}))
+
+ commands = [
+ ["summarize.py", str(shard)],
+ [
+ "control.py",
+ str(shard),
+ "--select",
+ str(shard / "selection.txt"),
+ ],
+ [
+ "control.py",
+ str(shard),
+ "--vanilla",
+ str(shard / "absent-vanilla"),
+ ],
+ ["merge.py", str(root), "--shards", "1"],
+ ]
+ for script, *arguments in commands:
+ result = subprocess.run(
+ [sys.executable, str(scripts / script), *arguments],
+ capture_output=True,
+ text=True,
+ )
+ self.assertEqual(
+ 0, result.returncode, result.stderr + result.stdout
+ )
+ self.assertEqual("", (shard / "selection.txt").read_text())
+ report = json.loads((root /
"execution-coverage.json").read_text())
+ # Baseline record/restore uses the same validator.
+ indexed = index(report, flaky_tests={(class_name, method)})
+ outcome = indexed[class_name, name]
+ self.assertEqual(status, outcome["status"])
+ self.assertTrue(outcome["ignored"])
+ self.assertEqual("flaky assertion", outcome["failure"])
+ self.assertFalse(indexed[class_name,
active["name"]]["ignored"])
+ with (root / "failed.csv").open(newline="") as source:
+ rows = list(csv.DictReader(source))
+ self.assertEqual("True", rows[0]["ignored"])
+
+ base = dict(
+ ignored,
+ status="PASSED_NATIVE",
+ body_started=True,
+ native=["IcebergScanTransformer"],
+ native_metrics=[
+ {
+ "node": "IcebergScanTransformer",
+ "metrics": {"rawInputRows": 1},
+ }
+ ],
+ )
+ (root / "base.json").write_text(
+ json.dumps({"version": 2, "tests": [base, active]})
+ )
+ command = [
+ sys.executable,
+ str(scripts / "compare.py"),
+ str(root / "base.json"),
+ str(root / "execution-coverage.json"),
+ "--output",
+ str(root / "diff.json"),
+ ]
+ result = subprocess.run(command, capture_output=True,
text=True)
+ self.assertEqual(0, result.returncode, result.stderr +
result.stdout)
+ diff = json.loads((root / "diff.json").read_text())
+ self.assertFalse(diff["regressions"])
+ self.assertFalse(diff["improvements"])
+ self.assertEqual(status, diff["ignored_changes"][0]["after"])
+
+ # The identical failure in an unlisted method must still fail
validation.
+ outcome["name"] = "notWhitelisted"
+ (root /
"execution-coverage.json").write_text(json.dumps(report))
+ result = subprocess.run(command, capture_output=True,
text=True)
+ self.assertEqual(1, result.returncode, result.stderr)
+ self.assertTrue(
+ json.loads((root /
"diff.json").read_text())["validation_errors"]
+ )
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/gluten-ut/src-iceberg-upstream/test/python/test_summary.py
b/gluten-ut/src-iceberg-upstream/test/python/test_summary.py
index 12d454dae8..7440577bef 100644
--- a/gluten-ut/src-iceberg-upstream/test/python/test_summary.py
+++ b/gluten-ut/src-iceberg-upstream/test/python/test_summary.py
@@ -17,6 +17,7 @@ import importlib.util
import csv
import json
from pathlib import Path
+import sys
import tempfile
import unittest
import xml.etree.ElementTree as ET
@@ -25,6 +26,7 @@ SCRIPT = (
Path(__file__).resolve().parents[4]
/ ".github/workflows/util/iceberg-upstream/summarize.py"
)
+sys.path.insert(0, str(SCRIPT.parent))
SPEC = importlib.util.spec_from_file_location("summary", SCRIPT)
summary = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(summary)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
