@@ -29,7 +29,15 @@ val scalaVersion =
         "4.0" to "2.13" -> libs.findVersion("scala213Spark40").get().requiredVersion
         else -> throw GradleException("Unsupported Spark/Scala variant: $sparkLine/$scalaBinaryVersion")
     }
-val spark41Version = libs.findVersion("spark41").get().requiredVersion
+// Later Spark 4 releases run the Spark 4.0 module's tests against its jar to catch binary breaks.
+val sparkCompatVersions =
+    if (sparkLine == "4.0") {
+        listOf("4.1" to "spark41", "4.2" to "spark42").map { (line, alias) ->
+            line to libs.findVersion(alias).get().requiredVersion
+        }
+    } else {
+        emptyList()
+    }
 
 layout.buildDirectory = layout.projectDirectory.dir("build/${project.name}")
 
@@ -139,8 +147,8 @@ testing {
             targets.all { testTask.configure { enabled = false } }
         }
 
-        if (sparkLine == "4.0") {
-            register<JvmTestSuite>("spark41CompatTest") {
+        sparkCompatVersions.forEach { (line, version) ->
+            register<JvmTestSuite>("spark${line.replace(".", "")}CompatTest") {
                 useJUnitJupiter()
                 sources {
                     java.setSrcDirs(
@@ -159,8 +167,8 @@ testing {
                 dependencies {
                     implementation(project())
                     implementation(libs.findLibrary("junit-jupiter").get())
-                    implementation("org.apache.spark:spark-core_2.13:$spark41Version")
-                    implementation("org.apache.spark:spark-sql_2.13:$spark41Version")
+                    implementation("org.apache.spark:spark-core_2.13:$version")
+                    implementation("org.apache.spark:spark-sql_2.13:$version")
                     implementation(libs.findLibrary("s3mock-testcontainers").get())
                     implementation(libs.findLibrary("testcontainers-juputer").get())
                     runtimeOnly(libs.findLibrary("slf4j-simple").get())
@@ -170,9 +178,9 @@ testing {
     }
 }
 
-if (sparkLine == "4.0") {
-    tasks.named("check") {
-        dependsOn("spark41CompatTest")
+tasks.named("check") {
+    sparkCompatVersions.forEach { (line, _) ->
+        dependsOn("spark${line.replace(".", "")}CompatTest")
     }
 }
 
 
@@ -20,6 +20,7 @@ scala213Spark41 = "2.13.17"
 spark35 = "3.5.9"
 spark40 = "4.0.4"
 spark41 = "4.1.2"
+spark42 = "4.2.0"
 s3mock = "5.0.0"
 testcontainers-jupiter = "1.21.4"
 
 
@@ -265,11 +265,10 @@ private static Optional<Expression> literal(Object value, DataType dataType) {
         if (dataType instanceof LongType) {
             return numericLiteral(value, Expression.DType.I64, number -> Expression.literal(number.longValue()));
         }
-        if (dataType instanceof FloatType) {
-            return numericLiteral(value, Expression.DType.F32, number -> Expression.literal(number.floatValue()));
-        }
-        if (dataType instanceof DoubleType) {
-            return numericLiteral(value, Expression.DType.F64, number -> Expression.literal(number.doubleValue()));
+        if (dataType instanceof FloatType || dataType instanceof DoubleType) {
+            // Vortex compares floats in total order, so -0.0 and 0.0 differ, while Spark treats them as equal.
+            // Spark cannot restore rows that Vortex drops, so keep these comparisons in Spark (see #10064).
+            return Optional.empty();
         }
         if (dataType instanceof StringType) {
             if (value == null) {
 
@@ -107,7 +107,8 @@ final class VortexScan(
     }
   }
 
-  /** The listed files that belong to the dataset. The readers skip the others. */
+  /** The listed files that belong to the dataset. The readers skip the others.
+    */
   private def vortexFiles =
     partitions
       .flatMap(_.files)
 
@@ -82,8 +82,7 @@ void bucketByWritesSparkBuckets() throws IOException {
         assertFalse(buckets.isEmpty(), "bucketed table should hold bucket files");
         assertTrue(buckets.stream().allMatch(id -> id >= 0 && id < 4), "bucket ids: " + buckets);
 
-        Dataset<Row> joined = spark.sql(
-                "SELECT l.id FROM bucketed_left l JOIN bucketed_right r ON l.id = r.id");
+        Dataset<Row> joined = spark.sql("SELECT l.id FROM bucketed_left l JOIN bucketed_right r ON l.id = r.id");
         String plan = joined.queryExecution().executedPlan().toString();
         assertFalse(plan.contains("Exchange"), "a bucketed join should not shuffle:\n" + plan);
         assertEquals(50, joined.count());
@@ -95,7 +94,8 @@ void bucketTransformCreatesBucketedTable() throws IOException {
         spark.sql("CREATE TABLE bucketed_sql (id INT, name STRING) USING vortex PARTITIONED BY (bucket(4, id))");
         spark.sql("INSERT INTO bucketed_sql SELECT CAST(id AS INT), CAST(id AS STRING) FROM range(40)");
 
-        List<Row> rows = spark.sql("SELECT count(*), count(DISTINCT id) FROM bucketed_sql").collectAsList();
+        List<Row> rows = spark.sql("SELECT count(*), count(DISTINCT id) FROM bucketed_sql")
+                .collectAsList();
         assertEquals(40L, rows.get(0).getLong(0));
         assertEquals(40L, rows.get(0).getLong(1));
 
 
@@ -27,8 +27,8 @@
  * Only files that end with {@code .vortex} belong to a Vortex dataset.
  *
  * <p>Spark's file index keeps {@code _metadata} and {@code _common_metadata}, and it keeps every extension it does not
- * recognise, so the connector is the only thing that can decide what belongs to the dataset. It skips every other
- * file, as the connector before the file-source rewrite did. These tests hold the paths that see the listing — schema
+ * recognise, so the connector is the only thing that can decide what belongs to the dataset. It skips every other file,
+ * as the connector before the file-source rewrite did. These tests hold the paths that see the listing — schema
  * inference, scan statistics, {@code COUNT(*)} pushdown, and the V1 and V2 scans — to the same answer.
  */
 @TestInstance(TestInstance.Lifecycle.PER_CLASS)
 
@@ -27,6 +27,8 @@
 final class SparkFilterToVortexExpressionTest {
     private static final StructType SCHEMA = new StructType()
             .add("id", DataTypes.IntegerType)
+            .add("float", DataTypes.FloatType)
+            .add("double", DataTypes.DoubleType)
             .add("date", DataTypes.DateType)
             .add("timestamp", DataTypes.TimestampType)
             .add("decimal", DataTypes.createDecimalType(10, 2))
@@ -73,6 +75,15 @@ void rejectsDecimalLiteralsThatCannotHoldTheirScale() {
         assertPushable(new EqualTo("decimal", new BigDecimal("1.00")));
     }
 
+    @Test
+    void rejectsFloatingPointComparisons() {
+        // Vortex orders -0.0 below 0.0, but Spark treats them as equal.
+        assertFalse(SparkFilterToVortexExpression.isPushable(new EqualTo("float", 0.0f), SCHEMA));
+        assertFalse(SparkFilterToVortexExpression.isPushable(new GreaterThan("double", 0.0), SCHEMA));
+        assertFalse(SparkFilterToVortexExpression.isPushable(new In("double", new Object[] {0.0, 1.0}), SCHEMA));
+        assertPushable(new IsNull("double"));
+    }
+
     @Test
     void rejectsAPartlyConvertibleConjunctionAndDisjunction() {
         assertFalse(SparkFilterToVortexExpression.isPushable(
Original file line number	Diff line number	Diff line change
`@@ -107,7 +107,8 @@ final class VortexScan(`
`107`	`107`	`}`
`108`	`108`	`}`
`109`	`109`
`110`		`- /** The listed files that belong to the dataset. The readers skip the others. */`
	`110`	`+ /** The listed files that belong to the dataset. The readers skip the others.`
	`111`	`+ */`
`111`	`112`	`private def vortexFiles =`
`112`	`113`	`partitions`
`113`	`114`	`.flatMap(_.files)`