(spark) branch branch-4.x updated: [SPARK-58030][SQL][TEST] Make FileDataSourceV2FallBackSuite robust to extra listener events

[gu...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

HyukjinKwon pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git


The following commit(s) were added to refs/heads/branch-4.x by this push:
     new 68faae75bb71 [SPARK-58030][SQL][TEST] Make FileDataSourceV2FallBackSuite robust to extra listener events
68faae75bb71 is described below

commit 68faae75bb716aef4351659b863391220119c3b0
Author: Hyukjin Kwon <gu...@apache.org>
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 <gu...@apache.org>
    Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
    (cherry picked from commit 8dcb844b986c134132d0c9876e80fb94da932d31)
    Signed-off-by: Hyukjin Kwon <hy...@databricks.com>
---
 .../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 c84d02c447c1..0fbfaedfdcac 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
@@ -179,10 +179,13 @@ class FileDataSourceV2FallBackSuite extends 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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org