You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/26 06:38:29 UTC
(datafusion-comet) 01/01: test: report which writer ran each Iceberg write in the Iceberg CI jobs (#6155)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-6155-81f2574d5e92028d6a0c8aebd39e6e0b5b4c2fd2
in repository https://gitbox.apache.org/repos/asf/datafusion-comet.git
commit 14f0f59f74dbcf273ff3cbdb1721a5ee535bbe84
Author: Andy Grove <ag...@apache.org>
AuthorDate: Sat Sep 26 06:38:07 2026 +0000
test: report which writer ran each Iceberg write in the Iceberg CI jobs (#6155)
* test: report which writer ran each Iceberg write in the Iceberg CI jobs
The Iceberg Spark test jobs run with the native Iceberg writer enabled, but
a passing job cannot tell a native write from a silent fallback: the
fallback warnings never reach the job log and no upstream test asserts
which writer ran.
Add a test-only config, spark.comet.testing.icebergWriteReport.dir, backed
by the COMET_ICEBERG_WRITE_REPORT_DIR environment variable. When it is set,
the driver plugin registers IcebergWriteReportListener, which appends one
JSON line per Iceberg write: native, JVM under the split operator (with
Comet's fallback reasons), or Spark's own V2 write. The core and extensions
jobs set the variable, so the Iceberg diffs are unchanged, and
dev/ci/summarize-iceberg-writes.py renders a per-job and all-shards table
on the job summary page. dev/local-ci.sh prints the same summary.
Closes #6148.
* test: report streaming and Spark 3.4 CTAS/RTAS Iceberg writes, and count only each shard's latest attempt
The write report missed two kinds of Iceberg write that Spark runs itself:
- A streaming micro-batch runs as `WriteToDataSourceV2Exec` over a `MicroBatchWrite`, not as a
`V2ExistingTableWriteExec`.
- On Spark 3.4, CTAS and RTAS write the table from the create or replace exec through
`TableWriteExecHelper.writeWithV2`. Spark 3.5+ runs that write as a nested append or overwrite,
which the listener already sees, so `IcebergTableAsSelectShim` recognizes the execs on 3.4 only.
Both went unreported, so the Spark count and the total left them out and the native share read
high.
The summarizer took each shard's latest attempt from its report files, so a retry that recorded no
writes was hidden behind the earlier attempt's records. It now takes the latest attempt from the
shard artifact directories and names a shard whose latest attempt recorded nothing.
`dev/ci/test-summarize-iceberg-writes.py` covers that and runs in Preflight.
---
.github/workflows/ci.yml | 3 +
.github/workflows/iceberg_spark_test_reusable.yml | 32 +++-
dev/ci/summarize-iceberg-writes.py | 171 +++++++++++++++++++++
dev/ci/test-summarize-iceberg-writes.py | 105 +++++++++++++
dev/local-ci.sh | 11 +-
.../contributor-guide/iceberg-spark-tests.md | 29 +++-
.../main/scala/org/apache/comet/CometConf.scala | 13 ++
.../comet/iceberg/IcebergWriteReportListener.scala | 135 ++++++++++++++++
.../src/main/scala/org/apache/spark/Plugins.scala | 46 ++++--
.../comet/iceberg/IcebergTableAsSelectShim.scala | 40 +++++
.../comet/iceberg/IcebergTableAsSelectShim.scala | 34 ++++
.../comet/iceberg/IcebergTableAsSelectShim.scala | 34 ++++
.../comet/CometIcebergWriteActionSuite.scala | 137 ++++++++++++++++-
.../scala/org/apache/spark/CometPluginsSuite.scala | 16 ++
14 files changed, 784 insertions(+), 22 deletions(-)
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 7f87d59199..d153ff3c63 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -186,6 +186,9 @@ jobs:
- name: Check Iceberg shard inventory validation
run: python3 dev/ci/test-iceberg-shards.py
+ - name: Check Iceberg write report summary
+ run: python3 dev/ci/test-summarize-iceberg-writes.py
+
- name: Check CI config invariants
run: python3 dev/ci/check-ci-config.py
diff --git a/.github/workflows/iceberg_spark_test_reusable.yml b/.github/workflows/iceberg_spark_test_reusable.yml
index ae20b6d768..9b4a5ae300 100644
--- a/.github/workflows/iceberg_spark_test_reusable.yml
+++ b/.github/workflows/iceberg_spark_test_reusable.yml
@@ -154,12 +154,20 @@ jobs:
run: |
cd apache-iceberg
rm -rf /root/.m2/repository/org/apache/parquet # somehow parquet cache requires cleanups
- ENABLE_COMET=true ENABLE_COMET_ONHEAP=true ./gradlew -DsparkVersions=${{ inputs.spark-short }} -DscalaVersion=${{ inputs.scala }} -DflinkVersions= -DkafkaVersions= \
+ # COMET_ICEBERG_WRITE_REPORT_DIR records which writer ran each Iceberg
+ # write; see dev/ci/summarize-iceberg-writes.py.
+ ENABLE_COMET=true ENABLE_COMET_ONHEAP=true COMET_ICEBERG_WRITE_REPORT_DIR="$PWD/build/comet-iceberg-writes" \
+ ./gradlew -DsparkVersions=${{ inputs.spark-short }} -DscalaVersion=${{ inputs.scala }} -DflinkVersions= -DkafkaVersions= \
:iceberg-spark:iceberg-spark-${{ inputs.spark-short }}_${{ inputs.scala }}:test \
--init-script ../dev/ci/iceberg-test-shards.gradle \
-PcometShardTask=:iceberg-spark:iceberg-spark-${{ inputs.spark-short }}_${{ inputs.scala }}:test \
-PcometShardIndex=${{ matrix.shard }} -PcometShardCount=${{ needs.build-native.outputs.shard-count }} \
-Pquick=true -x javadoc
+ - name: Summarize Iceberg writes
+ if: ${{ !cancelled() }}
+ run: |
+ python3 dev/ci/summarize-iceberg-writes.py --title "iceberg-spark shard ${{ matrix.shard }}" \
+ apache-iceberg/build/comet-iceberg-writes
- name: Upload Iceberg shard inventory and test reports
if: ${{ !cancelled() }}
# iceberg-spark-shard-coverage downloads the inventory, so a flaky
@@ -170,6 +178,7 @@ jobs:
path: |
apache-iceberg/**/build/comet-shards/*.json
apache-iceberg/**/build/test-results/test/*.xml
+ apache-iceberg/build/comet-iceberg-writes/*.jsonl
retention-days: 7
iceberg-spark-shard-coverage:
@@ -190,6 +199,11 @@ jobs:
run: |
python3 dev/ci/check-iceberg-shards.py --manifests iceberg-shard-reports \
--task :iceberg-spark:iceberg-spark-${{ inputs.spark-short }}_${{ inputs.scala }}:test
+ - name: Summarize Iceberg writes across shards
+ if: ${{ !cancelled() }}
+ run: |
+ python3 dev/ci/summarize-iceberg-writes.py --title "iceberg-spark, all shards" \
+ iceberg-shard-reports
iceberg-spark-extensions:
needs: build-native
@@ -222,9 +236,23 @@ jobs:
run: |
cd apache-iceberg
rm -rf /root/.m2/repository/org/apache/parquet # somehow parquet cache requires cleanups
- ENABLE_COMET=true ENABLE_COMET_ONHEAP=true ./gradlew -DsparkVersions=${{ inputs.spark-short }} -DscalaVersion=${{ inputs.scala }} -DflinkVersions= -DkafkaVersions= \
+ ENABLE_COMET=true ENABLE_COMET_ONHEAP=true COMET_ICEBERG_WRITE_REPORT_DIR="$PWD/build/comet-iceberg-writes" \
+ ./gradlew -DsparkVersions=${{ inputs.spark-short }} -DscalaVersion=${{ inputs.scala }} -DflinkVersions= -DkafkaVersions= \
:iceberg-spark:iceberg-spark-extensions-${{ inputs.spark-short }}_${{ inputs.scala }}:test \
-Pquick=true -x javadoc
+ - name: Summarize Iceberg writes
+ if: ${{ !cancelled() }}
+ run: |
+ python3 dev/ci/summarize-iceberg-writes.py --title "iceberg-spark-extensions" \
+ apache-iceberg/build/comet-iceberg-writes
+ - name: Upload Iceberg write report
+ if: ${{ !cancelled() }}
+ uses: ./.github/actions/upload-artifact-retry
+ with:
+ name: iceberg-spark-extensions-writes-${{ inputs.iceberg-full }}-spark-${{ inputs.spark-full }}-scala-${{ inputs.scala }}-jdk${{ inputs.java }}-attempt-${{ github.run_attempt }}
+ path: apache-iceberg/build/comet-iceberg-writes/*.jsonl
+ if-no-files-found: ignore
+ retention-days: 7
iceberg-spark-runtime:
needs: build-native
diff --git a/dev/ci/summarize-iceberg-writes.py b/dev/ci/summarize-iceberg-writes.py
new file mode 100644
index 0000000000..c94638b4e1
--- /dev/null
+++ b/dev/ci/summarize-iceberg-writes.py
@@ -0,0 +1,171 @@
+# 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.
+
+"""Summarize which writer ran the Iceberg writes of an Iceberg Spark test run.
+
+The Iceberg Spark test jobs set COMET_ICEBERG_WRITE_REPORT_DIR, so Comet's
+IcebergWriteReportListener appends one JSON line per Iceberg write to a file
+in that directory. This prints how many writes ran on Comet's native writer,
+how many Comet's split operator left on Iceberg's JVM writer and why, and how
+many Spark planned without Comet's split operator:
+
+ python3 dev/ci/summarize-iceberg-writes.py --title "iceberg-spark shard 1" DIR...
+
+Files under a directory named like a shard artifact (...-shard-N-attempt-M)
+are counted only for the latest attempt of each shard, so a rerun of failed
+jobs does not count a shard twice. The latest attempt is the newest such
+directory, whether or not it holds any report files, so a rerun that recorded
+no writes is reported as missing rather than replaced by an earlier attempt.
+The summary is also appended to $GITHUB_STEP_SUMMARY when that is set. It
+never fails the job.
+"""
+
+import argparse
+from collections import Counter
+import json
+import os
+from pathlib import Path
+import re
+
+
+SHARD_ATTEMPT = re.compile(r"-shard-(\d+)-attempt-(\d+)$")
+WRITERS = [
+ ("native", "Comet native writer"),
+ ("jvm", "Iceberg JVM writer under Comet's split operator"),
+ ("spark", "Spark V2 write, not planned by Comet's split operator"),
+]
+TOP_REASONS = 20
+
+
+def shard_attempt(path):
+ """The (shard, attempt) of the shard artifact directory holding path, or None."""
+ for part in path.parts:
+ match = SHARD_ATTEMPT.search(part)
+ if match:
+ return int(match.group(1)), int(match.group(2))
+ return None
+
+
+def report_files(roots):
+ """Every report file under roots, keeping only the latest attempt of each shard.
+
+ Returns the files and the (shard, attempt) pairs whose latest attempt holds no report file.
+ The latest attempt comes from the artifact directories rather than the report files, because
+ an attempt that recorded no writes still uploads its shard inventory and test reports.
+ """
+ latest = {}
+ files = []
+ for root in map(Path, roots):
+ for path in [root, *sorted(root.rglob("*"))]:
+ key = shard_attempt(path)
+ if key:
+ latest[key[0]] = max(latest.get(key[0], 0), key[1])
+ if path.suffix == ".jsonl" and path.is_file():
+ files.append((path, key))
+ kept = [(path, key) for path, key in files if key is None or latest[key[0]] == key[1]]
+ reported = {key for _, key in kept}
+ missing = sorted(key for key in latest.items() if key not in reported)
+ return [path for path, _ in kept], missing
+
+
+def load(roots):
+ files, missing = report_files(roots)
+ writes = []
+ for path in files:
+ for line in path.read_text(encoding="utf-8").splitlines():
+ if line.strip():
+ writes.append(json.loads(line))
+ return writes, missing
+
+
+def cell(text):
+ return " ".join(text.split()).replace("|", "\\|")
+
+
+def summarize(title, writes, missing=()):
+ lines = [f"### Iceberg writes: {title}", ""]
+ for shard, attempt in missing:
+ lines += [
+ f"Shard {shard} recorded no Iceberg writes in its latest attempt ({attempt}), "
+ "so none of its writes are counted below.",
+ "",
+ ]
+ if not writes:
+ lines.append(
+ "No Iceberg writes were recorded. Either the target ran none or "
+ "COMET_ICEBERG_WRITE_REPORT_DIR did not reach the test JVMs."
+ )
+ return "\n".join(lines) + "\n"
+
+ total = len(writes)
+ by_writer = Counter(w["writer"] for w in writes)
+ lines += ["| Writer | Writes | Share |", "| --- | ---: | ---: |"]
+ for key, label in WRITERS:
+ count = by_writer.get(key, 0)
+ lines.append(f"| {label} | {count} | {100.0 * count / total:.1f}% |")
+ lines += [f"| Total | {total} | |", ""]
+ failed = sum(1 for w in writes if w.get("failed"))
+ if failed:
+ lines += [f"{failed} of the {total} writes ran in queries that failed.", ""]
+
+ reasons = Counter()
+ for w in writes:
+ if w["writer"] == "jvm":
+ for reason in w.get("reasons") or ["(no reason recorded)"]:
+ reasons[reason] += 1
+ if reasons:
+ lines += [
+ "#### Why the split operator kept the JVM writer",
+ "",
+ "A write can have several reasons, so the counts can add up to more than the "
+ "JVM writes.",
+ "",
+ "| Writes | Reason |",
+ "| ---: | --- |",
+ ]
+ for reason, count in reasons.most_common(TOP_REASONS):
+ lines.append(f"| {count} | {cell(reason)} |")
+ if len(reasons) > TOP_REASONS:
+ lines.append(f"| | and {len(reasons) - TOP_REASONS} more reasons |")
+ lines.append("")
+
+ operators = Counter(w["node"] for w in writes if w["writer"] == "spark")
+ if operators:
+ lines += ["#### Spark V2 writes by operator", "", "| Writes | Operator |", "| ---: | --- |"]
+ for node, count in operators.most_common():
+ lines.append(f"| {count} | {cell(node)} |")
+ lines.append("")
+
+ return "\n".join(lines) + "\n"
+
+
+def main():
+ parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+ parser.add_argument("--title", required=True, help="heading for the summary")
+ parser.add_argument("roots", nargs="+", help="directories holding the report files")
+ args = parser.parse_args()
+
+ summary = summarize(args.title, *load(args.roots))
+ print(summary)
+ step_summary = os.environ.get("GITHUB_STEP_SUMMARY")
+ if step_summary:
+ with open(step_summary, "a", encoding="utf-8") as out:
+ out.write(summary + "\n")
+
+
+if __name__ == "__main__":
+ main()
diff --git a/dev/ci/test-summarize-iceberg-writes.py b/dev/ci/test-summarize-iceberg-writes.py
new file mode 100644
index 0000000000..cce4ada663
--- /dev/null
+++ b/dev/ci/test-summarize-iceberg-writes.py
@@ -0,0 +1,105 @@
+#!/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.
+
+"""Fast regression tests for the Iceberg write report summary."""
+
+import importlib.util
+import json
+from pathlib import Path
+import tempfile
+import unittest
+
+
+SPEC = importlib.util.spec_from_file_location(
+ "summarize_iceberg_writes", Path(__file__).with_name("summarize-iceberg-writes.py"))
+SUMMARIZE = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(SUMMARIZE)
+
+ARTIFACT = "iceberg-spark-1.11.0-spark-4.1.3-scala-2.13-jdk17-shard-{}-attempt-{}"
+
+
+def records(*writers):
+ return "".join(
+ json.dumps({"writer": w, "node": "AppendData", "reasons": [], "failed": False}) + "\n"
+ for w in writers)
+
+
+class SummarizeIcebergWritesTest(unittest.TestCase):
+ def setUp(self):
+ self.temp = tempfile.TemporaryDirectory(prefix="comet-iceberg-writes-test-")
+ self.addCleanup(self.temp.cleanup)
+ self.root = Path(self.temp.name)
+
+ def attempt(self, shard, attempt, *writers):
+ """A shard attempt's artifact as the coverage job downloads it.
+
+ Every attempt uploads its test reports. Only an attempt that recorded writes has a
+ report file.
+ """
+ artifact = self.root / ARTIFACT.format(shard, attempt)
+ reports = artifact / "build/test-results/test"
+ reports.mkdir(parents=True)
+ (reports / "TEST-org.example.TestFixture.xml").write_text("<testsuite/>")
+ if writers:
+ writes = artifact / "build/comet-iceberg-writes"
+ writes.mkdir(parents=True)
+ (writes / "iceberg-writes-fixture.jsonl").write_text(records(*writers))
+
+ def load(self):
+ writes, missing = SUMMARIZE.load([self.root])
+ return sorted(w["writer"] for w in writes), missing
+
+ def test_every_shard_counts_once(self):
+ self.attempt(1, 1, "native")
+ self.attempt(2, 1, "jvm", "spark")
+ self.assertEqual(self.load(), (["jvm", "native", "spark"], []))
+
+ def test_latest_attempt_replaces_an_earlier_one(self):
+ self.attempt(1, 1, "native", "native")
+ self.attempt(1, 2, "jvm")
+ self.assertEqual(self.load(), (["jvm"], []))
+
+ def test_retry_without_writes_is_reported_missing_instead_of_stale(self):
+ self.attempt(1, 1, "native")
+ self.attempt(1, 2)
+ self.attempt(2, 1, "spark")
+ self.assertEqual(self.load(), (["spark"], [(1, 2)]))
+ summary = SUMMARIZE.summarize("fixture", *SUMMARIZE.load([self.root]))
+ self.assertIn("Shard 1 recorded no Iceberg writes in its latest attempt (2)", summary)
+ self.assertIn("| Comet native writer | 0 | 0.0% |", summary)
+ self.assertIn("| Spark V2 write, not planned by Comet's split operator | 1 | 100.0% |",
+ summary)
+
+ def test_root_that_is_itself_a_shard_artifact(self):
+ self.attempt(1, 3)
+ writes, missing = SUMMARIZE.load([self.root / ARTIFACT.format(1, 3)])
+ self.assertEqual((writes, missing), ([], [(1, 3)]))
+
+ def test_files_outside_shard_artifacts_always_count(self):
+ # A shard job and dev/local-ci.sh summarize their own report directory directly.
+ (self.root / "iceberg-writes-local.jsonl").write_text(records("native", "jvm"))
+ self.assertEqual(self.load(), (["jvm", "native"], []))
+
+ def test_no_writes_at_all(self):
+ summary = SUMMARIZE.summarize("fixture", *SUMMARIZE.load([self.root / "missing"]))
+ self.assertIn("No Iceberg writes were recorded", summary)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/dev/local-ci.sh b/dev/local-ci.sh
index 8e3267302d..a6a9d57d83 100755
--- a/dev/local-ci.sh
+++ b/dev/local-ci.sh
@@ -389,16 +389,25 @@ run_iceberg() {
gradlew ":iceberg-spark:iceberg-spark-runtime-${SPARK}_${SCALA}:integrationTest"
;;
esac
+ case "$target" in
+ shard-* | extensions)
+ python3 "$REPO/dev/ci/summarize-iceberg-writes.py" --title "$target" \
+ "$dest/build/comet-iceberg-writes/$target"
+ ;;
+ esac
ok "$target took $(hms $((SECONDS - started)))"
done
}
-# Reads $dest, $spark and $SCALA from run_iceberg.
+# Reads $dest, $spark, $SCALA and $target from run_iceberg.
gradlew() {
(
cd "$dest"
# shellcheck disable=SC2031
export SPARK_LOCAL_IP=localhost ENABLE_COMET=true ENABLE_COMET_ONHEAP=true
+ # One directory per target, emptied first, so a rerun reports only its own writes.
+ export COMET_ICEBERG_WRITE_REPORT_DIR="$dest/build/comet-iceberg-writes/$target"
+ rm -rf "$COMET_ICEBERG_WRITE_REPORT_DIR"
./gradlew "-DsparkVersions=$SPARK" "-DscalaVersion=$SCALA" \
-DflinkVersions= -DkafkaVersions= "$@" -Pquick=true -x javadoc
)
diff --git a/docs/source/contributor-guide/iceberg-spark-tests.md b/docs/source/contributor-guide/iceberg-spark-tests.md
index fb0e262e34..0db54feeff 100644
--- a/docs/source/contributor-guide/iceberg-spark-tests.md
+++ b/docs/source/contributor-guide/iceberg-spark-tests.md
@@ -43,7 +43,9 @@ Here is an overview of the changes that the diffs make to Iceberg:
`LocalTableScanExec`, the conversion is declined, and the write silently runs on the JVM writer. Many Iceberg
suites seed their data that way, so leaving it off hides the native writer from most of the write surface.
- Enable fallback logging (`spark.comet.explainFallback.enabled`) so that every operator Comet declines is
- reported in the test output together with the reason it was declined.
+ reported in the test output together with the reason it was declined. The output goes to the JUnit XML
+ reports rather than the CI job log; see [Which writer ran each Iceberg write](#which-writer-ran-each-iceberg-write)
+ for how CI reports native write coverage.
[#3739]: https://github.com/apache/datafusion-comet/pull/3739
[#5259]: https://github.com/apache/datafusion-comet/issues/5259
@@ -154,3 +156,28 @@ path, reflection code (`org.apache.comet.iceberg.IcebergReflection`), or other l
can differ across Iceberg versions. The Comet test suites in the Linux build do not exercise Iceberg's
own Spark tests, so without the label the first Iceberg 1.11 verdict is the merge queue's, and the
first verdict on the older Iceberg versions is the nightly run's, after the change has landed.
+
+### Which writer ran each Iceberg write
+
+A passing Iceberg job does not show that Comet's native writer ran. `CometIcebergNativeWrite` falls back
+to Iceberg's JVM writer without failing the write, no upstream test asserts which writer ran, and Gradle
+does not copy the fallback warnings into the job log. So the core and extensions jobs set
+`COMET_ICEBERG_WRITE_REPORT_DIR`, the environment variable behind the test-only config
+`spark.comet.testing.icebergWriteReport.dir`. When it is set, the Comet driver plugin registers
+`IcebergWriteReportListener`, which writes one JSON line for each Iceberg write the tests run. Each line
+records one of three writers:
+
+- `native`: Comet's native writer (`CometIcebergWriteExec`).
+- `jvm`: Comet's split operator planned the write but kept Iceberg's JVM writer (`IcebergWriteExec`).
+ The line includes the reasons Comet recorded for not converting it.
+- `spark`: Spark's own V2 write operator ran the write, so Comet's split operator never saw it. Examples
+ are `WriteDelta` for merge-on-read, `WriteToDataSourceV2` for a streaming micro-batch, and on Spark
+ 3.4 the CTAS and RTAS execs, which write the table themselves.
+
+`dev/ci/summarize-iceberg-writes.py` turns these records into a table on the job's summary page. It
+shows the count and share of each writer, the most common fallback reasons, and the Spark write
+operators. Each shard and the extensions job gets its own table. The shard coverage job adds one for
+all shards together, counting only the latest attempt of each shard. A shard whose latest attempt
+recorded no writes is named above the table rather than counted from an earlier attempt. The raw
+records are uploaded with the job's other reports. The summary never fails a job.
+`dev/local-ci.sh iceberg` prints the same summary after each shard and after the extensions target.
diff --git a/spark/src/main/scala/org/apache/comet/CometConf.scala b/spark/src/main/scala/org/apache/comet/CometConf.scala
index c15e391c94..eeb5cf8045 100644
--- a/spark/src/main/scala/org/apache/comet/CometConf.scala
+++ b/spark/src/main/scala/org/apache/comet/CometConf.scala
@@ -951,6 +951,19 @@ object CometConf extends ShimCometConf {
.booleanConf
.createWithDefault(true)
+ val COMET_ICEBERG_WRITE_REPORT_DIR: ConfigEntry[String] =
+ conf("spark.comet.testing.icebergWriteReport.dir")
+ .internal()
+ .category(CATEGORY_TESTING)
+ .doc("Test-only. When set, the Comet driver plugin registers a query listener that " +
+ "records every Iceberg write the application runs, the writer that ran it (Comet's " +
+ "native writer, the JVM writer behind Comet's split operator, or Spark's own V2 write) " +
+ "and the reasons Comet recorded for not writing natively, as JSON lines in this " +
+ "directory. `dev/ci/summarize-iceberg-writes.py` summarizes them. The Iceberg Spark " +
+ "test jobs set it through the environment variable so the Iceberg diffs need no change.")
+ .stringConf
+ .createWithEnvVarOrDefault("COMET_ICEBERG_WRITE_REPORT_DIR", "")
+
val COMET_SPARK_TO_ARROW_ENABLED: ConfigEntry[Boolean] =
conf("spark.comet.sparkToColumnar.enabled")
.category(CATEGORY_EXEC)
diff --git a/spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteReportListener.scala b/spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteReportListener.scala
new file mode 100644
index 0000000000..d5f3663eb1
--- /dev/null
+++ b/spark/src/main/scala/org/apache/comet/iceberg/IcebergWriteReportListener.scala
@@ -0,0 +1,135 @@
+/*
+ * 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.comet.iceberg
+
+import java.io.File
+import java.nio.charset.StandardCharsets.UTF_8
+import java.nio.file.{Files, StandardOpenOption}
+import java.util.UUID
+
+import scala.util.control.NonFatal
+
+import org.json4s.JsonDSL._
+import org.json4s.jackson.JsonMethods._
+
+import org.apache.spark.SparkConf
+import org.apache.spark.internal.Logging
+import org.apache.spark.sql.comet.{CometIcebergWriteExec, IcebergWriteExec}
+import org.apache.spark.sql.connector.write.BatchWrite
+import org.apache.spark.sql.execution.{CommandResultExec, QueryExecution, SparkPlan}
+import org.apache.spark.sql.execution.adaptive.{AdaptiveSparkPlanExec, QueryStageExec}
+import org.apache.spark.sql.execution.datasources.v2.{V2ExistingTableWriteExec, WriteToDataSourceV2Exec}
+import org.apache.spark.sql.execution.streaming.sources.MicroBatchWrite
+import org.apache.spark.sql.util.QueryExecutionListener
+
+import org.apache.comet.CometConf.COMET_ICEBERG_WRITE_REPORT_DIR
+import org.apache.comet.CometExplainInfo
+
+/**
+ * Test-only listener that records which writer ran each Iceberg write, so a CI job running
+ * Iceberg's own Spark suites can tell a native write from a silent fallback. The Comet driver
+ * plugin registers it when `spark.comet.testing.icebergWriteReport.dir` is set. Each write is
+ * appended as one JSON line to a file of its own in that directory, which
+ * `dev/ci/summarize-iceberg-writes.py` reads.
+ */
+class IcebergWriteReportListener(conf: SparkConf) extends QueryExecutionListener with Logging {
+
+ private val reportFile: File = {
+ val dir = new File(
+ conf
+ .get(COMET_ICEBERG_WRITE_REPORT_DIR.key, COMET_ICEBERG_WRITE_REPORT_DIR.defaultValue.get))
+ dir.mkdirs()
+ // One file per listener: Gradle runs several test JVMs, each possibly with several sessions.
+ new File(dir, s"iceberg-writes-${UUID.randomUUID()}.jsonl")
+ }
+
+ override def onSuccess(funcName: String, qe: QueryExecution, durationNs: Long): Unit =
+ record(qe, failed = false)
+
+ override def onFailure(funcName: String, qe: QueryExecution, exception: Exception): Unit =
+ record(qe, failed = true)
+
+ private def record(qe: QueryExecution, failed: Boolean): Unit = {
+ try {
+ val lines = IcebergWriteReportListener.writes(qe.executedPlan).map { w =>
+ compact(
+ render(("writer" -> w.writer) ~ ("node" -> w.node) ~ ("reasons" -> w.reasons.toList) ~
+ ("failed" -> failed))) + "\n"
+ }
+ if (lines.nonEmpty) {
+ synchronized {
+ Files.write(
+ reportFile.toPath,
+ lines.mkString.getBytes(UTF_8),
+ StandardOpenOption.CREATE,
+ StandardOpenOption.APPEND)
+ }
+ }
+ } catch {
+ // A query that failed during planning has no executed plan; nothing was written.
+ case NonFatal(e) => logWarning(s"Could not record Iceberg writes for a query: $e")
+ }
+ }
+}
+
+object IcebergWriteReportListener {
+
+ /** Comet's native (iceberg-rust) writer ran the write. */
+ val Native = "native"
+
+ /** Comet's split operator planned the write, but Iceberg's JVM writer ran it. */
+ val Jvm = "jvm"
+
+ /** Spark's own V2 write operator ran the write; Comet's split operator did not plan it. */
+ val Spark = "spark"
+
+ case class IcebergWrite(writer: String, node: String, reasons: Seq[String])
+
+ /** The Iceberg writes in an executed plan, with the reasons Comet did not write natively. */
+ def writes(plan: SparkPlan): Seq[IcebergWrite] = plan match {
+ // A command's writes are reported by the command's own execution, which runs eagerly before
+ // the query wrapping its result. Descending here would count them twice.
+ case _: CommandResultExec => Nil
+ case a: AdaptiveSparkPlanExec => writes(a.executedPlan)
+ case s: QueryStageExec => writes(s.plan)
+ case w: CometIcebergWriteExec => Seq(IcebergWrite(Native, w.nodeName, Nil))
+ case w: IcebergWriteExec =>
+ val reasons = w.getTagValue(CometExplainInfo.FALLBACK_REASONS).getOrElse(Set.empty)
+ Seq(IcebergWrite(Jvm, w.nodeName, reasons.toSeq.sorted))
+ case w: V2ExistingTableWriteExec if isIceberg(w.write) =>
+ Seq(IcebergWrite(Spark, w.nodeName, Nil))
+ // A streaming micro-batch, which Comet's split operator never plans.
+ case w: WriteToDataSourceV2Exec if isIcebergMicroBatch(w.batchWrite) =>
+ Seq(IcebergWrite(Spark, w.nodeName, Nil))
+ // Spark 3.4 writes a CTAS or RTAS from the create or replace exec itself. Later versions run
+ // that write as a nested append or overwrite, which this listener sees as a query of its own.
+ case w if IcebergTableAsSelectShim.writeCatalog(w).exists(isIceberg) =>
+ Seq(IcebergWrite(Spark, w.nodeName, Nil))
+ case p => p.children.flatMap(writes)
+ }
+
+ private def isIcebergMicroBatch(write: BatchWrite): Boolean = write match {
+ case m: MicroBatchWrite => isIceberg(m.writeSupport)
+ case _ => false
+ }
+
+ private def isIceberg(obj: AnyRef): Boolean =
+ obj.getClass.getName.startsWith("org.apache.iceberg.")
+}
diff --git a/spark/src/main/scala/org/apache/spark/Plugins.scala b/spark/src/main/scala/org/apache/spark/Plugins.scala
index 65fb606edc..2fe6cceff2 100644
--- a/spark/src/main/scala/org/apache/spark/Plugins.scala
+++ b/spark/src/main/scala/org/apache/spark/Plugins.scala
@@ -31,9 +31,10 @@ import org.apache.spark.sql.internal.StaticSQLConf
import org.apache.comet.{COMET_VERSION, CometSparkSessionExtensions, NativeBase}
import org.apache.comet.{CometConf, ConfigEntry}
-import org.apache.comet.CometConf.{COMET_METRICS_ENABLED, COMET_ONHEAP_ENABLED}
+import org.apache.comet.CometConf.{COMET_ICEBERG_WRITE_REPORT_DIR, COMET_METRICS_ENABLED, COMET_ONHEAP_ENABLED}
import org.apache.comet.CometKryoRegistrator
import org.apache.comet.annotation.Public
+import org.apache.comet.iceberg.IcebergWriteReportListener
/**
* Comet driver plugin. This class is loaded by Spark's plugin framework. It will be instantiated
@@ -73,6 +74,7 @@ class CometDriverPlugin extends DriverPlugin with Logging {
// Register Comet metrics
CometDriverPlugin.registerCometMetrics(sc)
+ CometDriverPlugin.registerIcebergWriteReport(sc.conf)
CometDriverPlugin.warnIfExecutorMemoryOverheadUnset(sc.getConf)
CometDriverPlugin.warnIfMemoryPoolFractionSet(sc.getConf)
@@ -229,27 +231,39 @@ object CometDriverPlugin extends Logging {
COMET_METRICS_ENABLED.key,
COMET_METRICS_ENABLED.defaultValue.get)) {
sc.env.metricsSystem.registerSource(CometSource)
-
- val listenerKey = "spark.sql.queryExecutionListeners"
- val listenerClass = "org.apache.comet.CometMetricsListener"
- val listeners = sc.conf.get(listenerKey, "")
- if (listeners.isEmpty) {
- logInfo(s"Setting $listenerKey=$listenerClass")
- sc.conf.set(listenerKey, listenerClass)
- } else {
- val currentListeners = listeners.split(",").map(_.trim)
- if (!currentListeners.contains(listenerClass)) {
- val newValue = s"$listeners,$listenerClass"
- logInfo(s"Setting $listenerKey=$newValue")
- sc.conf.set(listenerKey, newValue)
- }
- }
+ registerQueryExecutionListener(sc.conf, "org.apache.comet.CometMetricsListener")
} else {
logInfo(
"Comet metrics reporting is disabled. Set spark.comet.metrics.enabled=true to enable.")
}
}
+ // Test-only: see COMET_ICEBERG_WRITE_REPORT_DIR. The value may come from the environment, which
+ // lets the Iceberg Spark test jobs turn the report on without changing the Iceberg diffs.
+ def registerIcebergWriteReport(conf: SparkConf): Unit = {
+ if (conf
+ .get(COMET_ICEBERG_WRITE_REPORT_DIR.key, COMET_ICEBERG_WRITE_REPORT_DIR.defaultValue.get)
+ .nonEmpty) {
+ registerQueryExecutionListener(conf, classOf[IcebergWriteReportListener].getName)
+ }
+ }
+
+ private def registerQueryExecutionListener(conf: SparkConf, listenerClass: String): Unit = {
+ val listenerKey = "spark.sql.queryExecutionListeners"
+ val listeners = conf.get(listenerKey, "")
+ if (listeners.isEmpty) {
+ logInfo(s"Setting $listenerKey=$listenerClass")
+ conf.set(listenerKey, listenerClass)
+ } else {
+ val currentListeners = listeners.split(",").map(_.trim)
+ if (!currentListeners.contains(listenerClass)) {
+ val newValue = s"$listeners,$listenerClass"
+ logInfo(s"Setting $listenerKey=$newValue")
+ conf.set(listenerKey, newValue)
+ }
+ }
+ }
+
def registerCometSessionExtension(conf: SparkConf): Unit = {
val extensionKey = StaticSQLConf.SPARK_SESSION_EXTENSIONS.key
val extensionClass = classOf[CometSparkSessionExtensions].getName
diff --git a/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala b/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala
new file mode 100644
index 0000000000..4d66a37466
--- /dev/null
+++ b/spark/src/main/spark-3.4/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala
@@ -0,0 +1,40 @@
+/*
+ * 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.comet.iceberg
+
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.execution.SparkPlan
+import org.apache.spark.sql.execution.datasources.v2.{AtomicCreateTableAsSelectExec, AtomicReplaceTableAsSelectExec, CreateTableAsSelectExec, ReplaceTableAsSelectExec}
+
+/**
+ * Spark 3.4: CTAS and RTAS write the table from the create or replace exec itself, through
+ * `TableWriteExecHelper.writeWithV2`, so the write never appears as a write node of its own.
+ */
+private[iceberg] object IcebergTableAsSelectShim {
+
+ /** The catalog `plan` writes through, when `plan` is a CTAS or RTAS exec. */
+ def writeCatalog(plan: SparkPlan): Option[TableCatalog] = plan match {
+ case p: CreateTableAsSelectExec => Some(p.catalog)
+ case p: AtomicCreateTableAsSelectExec => Some(p.catalog)
+ case p: ReplaceTableAsSelectExec => Some(p.catalog)
+ case p: AtomicReplaceTableAsSelectExec => Some(p.catalog)
+ case _ => None
+ }
+}
diff --git a/spark/src/main/spark-3.5/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala b/spark/src/main/spark-3.5/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala
new file mode 100644
index 0000000000..ef02700a9c
--- /dev/null
+++ b/spark/src/main/spark-3.5/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala
@@ -0,0 +1,34 @@
+/*
+ * 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.comet.iceberg
+
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.execution.SparkPlan
+
+/**
+ * Spark 3.5+: CTAS and RTAS run their write as a nested `AppendData` or `OverwriteByExpression`
+ * query, which is planned and reported like any other write, so no create or replace exec writes
+ * a table itself.
+ */
+private[iceberg] object IcebergTableAsSelectShim {
+
+ /** The catalog `plan` writes through, when `plan` is a CTAS or RTAS exec. */
+ def writeCatalog(plan: SparkPlan): Option[TableCatalog] = None
+}
diff --git a/spark/src/main/spark-4.x/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala b/spark/src/main/spark-4.x/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala
new file mode 100644
index 0000000000..ef02700a9c
--- /dev/null
+++ b/spark/src/main/spark-4.x/org/apache/comet/iceberg/IcebergTableAsSelectShim.scala
@@ -0,0 +1,34 @@
+/*
+ * 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.comet.iceberg
+
+import org.apache.spark.sql.connector.catalog.TableCatalog
+import org.apache.spark.sql.execution.SparkPlan
+
+/**
+ * Spark 3.5+: CTAS and RTAS run their write as a nested `AppendData` or `OverwriteByExpression`
+ * query, which is planned and reported like any other write, so no create or replace exec writes
+ * a table itself.
+ */
+private[iceberg] object IcebergTableAsSelectShim {
+
+ /** The catalog `plan` writes through, when `plan` is a CTAS or RTAS exec. */
+ def writeCatalog(plan: SparkPlan): Option[TableCatalog] = None
+}
diff --git a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
index 3edb69f3e2..01a8188f29 100644
--- a/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
+++ b/spark/src/test/scala/org/apache/comet/CometIcebergWriteActionSuite.scala
@@ -20,6 +20,7 @@
package org.apache.comet
import java.io.File
+import java.nio.file.Files
import java.sql.Timestamp
import java.util.concurrent.{CountDownLatch, TimeUnit}
@@ -29,7 +30,10 @@ import scala.concurrent.ExecutionContext.Implicits.global
import scala.concurrent.duration.DurationInt
import scala.jdk.CollectionConverters._
-import org.apache.spark.{SparkConf, SparkException, Success}
+import org.json4s.{DefaultFormats, Formats}
+import org.json4s.jackson.JsonMethods.parse
+
+import org.apache.spark.{CometListenerBusUtils, SparkConf, SparkException, Success}
import org.apache.spark.rdd.RDD
import org.apache.spark.scheduler.{SparkListener, SparkListenerTaskEnd}
import org.apache.spark.sql.CometTestBase
@@ -43,13 +47,20 @@ import org.apache.spark.sql.connector.write.{BatchWrite, DataWriterFactory, Phys
import org.apache.spark.sql.execution.{ColumnarToRowTransition, LeafExecNode, SparkPlan}
import org.apache.spark.sql.execution.adaptive.AdaptiveSparkPlanHelper
import org.apache.spark.sql.internal.SQLConf
+import org.apache.spark.sql.streaming.Trigger
import org.apache.spark.sql.types.{DoubleType, IntegerType, StringType, StructField, StructType}
import org.apache.comet.CometSparkSessionExtensions.{isSpark35Plus, isSpark41Plus}
-import org.apache.comet.iceberg.IcebergReflection
+import org.apache.comet.iceberg.{IcebergReflection, IcebergWriteReportListener}
private case class WriteSnapshot(snapshotDelta: Long, plans: Seq[SparkPlan])
+private case class ReportedWrite(
+ writer: String,
+ node: String,
+ reasons: Seq[String],
+ failed: Boolean)
+
class CometIcebergWriteActionSuite
extends CometTestBase
with AdaptiveSparkPlanHelper
@@ -2629,6 +2640,128 @@ class CometIcebergWriteActionSuite
}
}
+ test("write report records which writer ran each Iceberg write") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ createTable(warehouseDir, "report_parquet", partitionSpec = "PARTITIONED BY (region)")
+ createTable(
+ warehouseDir,
+ "report_orc",
+ partitionSpec = "",
+ properties = Some("'write.format.default'='orc'"))
+
+ val writes = reportedWrites {
+ withNativeEnabled {
+ // Collecting the result runs a second query over the command's result; the write
+ // must still be reported once.
+ spark
+ .sql(s"INSERT INTO $catalog.$ns.report_parquet VALUES (1, 'us-east', 1.5)")
+ .collect()
+ spark.sql(s"INSERT INTO $catalog.$ns.report_orc VALUES (2, 'eu', 2.5)")
+ }
+ withSQLConf(CometConf.COMET_ICEBERG_WRITE_SPLIT_OPERATOR_ENABLED.key -> "false") {
+ spark.sql(s"INSERT INTO $catalog.$ns.report_parquet VALUES (3, 'eu', 3.5)")
+ }
+ }
+ assert(writes.map(_.writer) == Seq("native", "jvm", "spark"), writes.mkString("\n"))
+ assert(writes(1).reasons.exists(_.contains("only parquet")))
+ assert(writes(2).node == "AppendData")
+ assert(writes.forall(!_.failed))
+ }
+ }
+
+ test("write report records a CTAS and an RTAS once each") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { _ =>
+ val writes = reportedWrites {
+ withNativeEnabled {
+ spark.sql(s"""
+ CREATE TABLE $catalog.$ns.report_ctas USING iceberg AS
+ SELECT * FROM VALUES (1, 'us', 1.0), (2, 'eu', 2.0) AS t(id, region, amount)
+ """)
+ spark.sql(s"""
+ REPLACE TABLE $catalog.$ns.report_ctas USING iceberg AS
+ SELECT * FROM VALUES (3, 'us', 3.0) AS t(id, region, amount)
+ """)
+ }
+ }
+ // Spark 3.5+ runs each write as a nested append or overwrite, which Comet's split operator
+ // plans. Spark 3.4 writes from the create and replace execs, which it never sees.
+ val expected =
+ if (isSpark35Plus) Seq("native" -> "CometIcebergWrite", "native" -> "CometIcebergWrite")
+ else Seq("spark" -> "AtomicCreateTableAsSelect", "spark" -> "AtomicReplaceTableAsSelect")
+ assert(writes.map(w => w.writer -> w.node) == expected, writes.mkString("\n"))
+ assert(writes.forall(!_.failed))
+ assertRows("report_ctas", Seq(3))
+ }
+ }
+
+ test("write report records a streaming micro-batch write") {
+ assumeNativeAcceleration()
+ withIcebergCatalog { warehouseDir =>
+ createTable(warehouseDir, "report_stream", partitionSpec = "")
+ withTempIcebergDir { dir =>
+ val source = new File(dir, "source").getAbsolutePath
+ val session = spark
+ import session.implicits._
+ Seq((1, "us", 1.0), (2, "eu", 2.0)).toDF("id", "region", "amount").write.parquet(source)
+ val schema = spark.table(s"$catalog.$ns.report_stream").schema
+
+ val writes = reportedWrites {
+ withNativeEnabled {
+ spark.readStream
+ .schema(schema)
+ .parquet(source)
+ .writeStream
+ .format("iceberg")
+ .outputMode("append")
+ .trigger(Trigger.AvailableNow())
+ .option("checkpointLocation", new File(dir, "checkpoint").getAbsolutePath)
+ .toTable(s"$catalog.$ns.report_stream")
+ .awaitTermination()
+ }
+ }
+ assert(
+ writes.map(w => w.writer -> w.node) == Seq("spark" -> "WriteToDataSourceV2"),
+ writes.mkString("\n"))
+ assert(writes.forall(!_.failed))
+ assertRows("report_stream", Seq(1, 2))
+ }
+ }
+ }
+
+ /** The Iceberg writes `IcebergWriteReportListener` records while `action` runs, in order. */
+ private def reportedWrites(action: => Unit): Seq[ReportedWrite] = {
+ var writes = Seq.empty[ReportedWrite]
+ withTempIcebergDir { reportDir =>
+ val listener = new IcebergWriteReportListener(
+ new SparkConf()
+ .set(CometConf.COMET_ICEBERG_WRITE_REPORT_DIR.key, reportDir.getAbsolutePath))
+ spark.listenerManager.register(listener)
+ try {
+ action
+ CometListenerBusUtils.waitUntilEmpty(spark.sparkContext)
+ } finally {
+ spark.listenerManager.unregister(listener)
+ }
+
+ implicit val formats: Formats = DefaultFormats
+ writes = reportDir
+ .listFiles()
+ .toSeq
+ .flatMap(f => Files.readAllLines(f.toPath).asScala)
+ .map { line =>
+ val w = parse(line)
+ ReportedWrite(
+ (w \ "writer").extract[String],
+ (w \ "node").extract[String],
+ (w \ "reasons").extract[Seq[String]],
+ (w \ "failed").extract[Boolean])
+ }
+ }
+ writes
+ }
+
private def assertNativeWriteEngages(tableName: String, expectedIds: Seq[Int])(
action: => Unit): Unit = {
val snapshot = withNativeEnabled { captureWrite(tableName)(action) }
diff --git a/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala b/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala
index 4207bd6621..82ea0a1889 100644
--- a/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala
+++ b/spark/src/test/scala/org/apache/spark/CometPluginsSuite.scala
@@ -88,6 +88,22 @@ class CometPluginsSuite extends CometTestBase {
}
}
+ test("Iceberg write report listener is registered only when a report directory is set") {
+ val listenerKey = "spark.sql.queryExecutionListeners"
+ val listenerClass = "org.apache.comet.iceberg.IcebergWriteReportListener"
+
+ val unset = new SparkConf()
+ CometDriverPlugin.registerIcebergWriteReport(unset)
+ assert(!unset.contains(listenerKey))
+
+ val set = new SparkConf()
+ .set(CometConf.COMET_ICEBERG_WRITE_REPORT_DIR.key, "/tmp/report")
+ .set(listenerKey, "foo")
+ CometDriverPlugin.registerIcebergWriteReport(set)
+ CometDriverPlugin.registerIcebergWriteReport(set)
+ assert(set.get(listenerKey) == s"foo,$listenerClass")
+ }
+
test("Comet version is exposed as a Spark config") {
// The driver plugin sets spark.comet.version, which is then visible both on the SparkContext
// conf and through the session runtime config (SET / spark.conf.get).
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org