This is an automated email from the ASF dual-hosted git repository.
HyukjinKwon pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 062fc8e3985c [SPARK-58030][SQL][TEST] Make
FileDataSourceV2FallBackSuite robust to extra listener events
062fc8e3985c is described below
commit 062fc8e3985c2015d33133c7120d5c7a260aa387
Author: Hyukjin Kwon <[email protected]>
AuthorDate: Wed Jul 8 14:55:42 2026 +0900
[SPARK-58030][SQL][TEST] Make FileDataSourceV2FallBackSuite robust to extra
listener events
### What changes were proposed in this pull request?
Fixes a flaky failure in scheduled CI on `master`
(`build_maven_java21_macos26`, `sql#core - other tests`).
`FileDataSourceV2FallBackSuite` "Fallback Parquet V2 to V1" asserted
`commands.length == 1` on a `QueryExecutionListener`. An extra eager `collect`
callback on the analyzed `UnresolvedDataSource` can race onto the listener bus,
making the count `2` and failing the test non-deterministically. Fixed by
filtering the recorded events to the write `"command"` entry and asserting on
that, instead of requiring the bus to contain exactly one event.
### Why are the changes needed?
The flake intermittently reds scheduled builds without indicating a real
product regression.
### Does this PR introduce _any_ user-facing change?
No. Test-only change.
### How was this patch tested?
- Passes locally (`Tests: succeeded 5, failed 0`, including "Fallback
Parquet V2 to V1").
- Fork CI (`build_maven_java21_macos26`, `sql#core - other tests`): ✅
**PASS** — https://github.com/HyukjinKwon/spark/actions/runs/28906196443
`Tests: succeeded 19657, failed 0` — "All tests passed", with "Fallback
Parquet V2 to V1" running clean. (The run's overall status shows a failure only
from an unrelated `TransformWithStateAvroSuite` state-store schema-evolution
flake in the separate `sql#core - slow tests` shard, which shares no code path
with this test-only change.)
JIRA: https://issues.apache.org/jira/browse/SPARK-58030
### Related-but-not-fixed (documented for reviewers)
- **`DynamicPartitionPruningHiveScanSuiteAEOff` "broadcast a single key in
a HashedRelation"** (`sql#hive - slow tests` on the same workflow): an
embedded-Derby internal `ClassCastException` (`ReferencedColumnsDescriptorImpl`
→ `ExecRowBuilder`) in the DataNucleus metastore layer — Derby-internal state
corruption under the slow macOS-26 JDK21 run, with no Spark product line to
fix. It passed on this run, consistent with a non-deterministic flake. Not
addressed here.
Closes #57097 from HyukjinKwon/DO-NOT-MERGE/deflake-master.
Authored-by: Hyukjin Kwon <[email protected]>
Signed-off-by: Hyukjin Kwon <[email protected]>
(cherry picked from commit 8dcb844b986c134132d0c9876e80fb94da932d31)
Signed-off-by: Hyukjin Kwon <[email protected]>
---
.../spark/sql/connector/FileDataSourceV2FallBackSuite.scala | 11 +++++++----
1 file changed, 7 insertions(+), 4 deletions(-)
diff --git
a/sql/core/src/test/scala/org/apache/spark/sql/connector/FileDataSourceV2FallBackSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/connector/FileDataSourceV2FallBackSuite.scala
index 2a0ab21ddb09..370cd9318093 100644
---
a/sql/core/src/test/scala/org/apache/spark/sql/connector/FileDataSourceV2FallBackSuite.scala
+++
b/sql/core/src/test/scala/org/apache/spark/sql/connector/FileDataSourceV2FallBackSuite.scala
@@ -180,10 +180,13 @@ class FileDataSourceV2FallBackSuite extends QueryTest
with SharedSparkSession {
val inputData = spark.range(10)
inputData.write.format(format).save(path.getCanonicalPath)
sparkContext.listenerBus.waitUntilEmpty()
- assert(commands.length == 1)
- assert(commands.head._1 == "command")
-
assert(commands.head._2.isInstanceOf[InsertIntoHadoopFsRelationCommand])
-
assert(commands.head._2.asInstanceOf[InsertIntoHadoopFsRelationCommand]
+ // The write should be executed as a single "command" callback.
Other callbacks
+ // (e.g. an eager "collect" on the analyzed plan) may be observed
on the listener
+ // bus, so filter to the write command rather than asserting on
the total count.
+ val writeCommands = commands.filter(_._1 == "command")
+ assert(writeCommands.length == 1)
+
assert(writeCommands.head._2.isInstanceOf[InsertIntoHadoopFsRelationCommand])
+
assert(writeCommands.head._2.asInstanceOf[InsertIntoHadoopFsRelationCommand]
.fileFormat.isInstanceOf[ParquetFileFormat])
val df = spark.read.format(format).load(path.getCanonicalPath)
checkAnswer(df, inputData.toDF())
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
