INSERT REPLACE WHERE with less than condition should be evaluated as …

…less than

INSERT REPLACE WHERE with a less than condition was evaluated as a less than or equal condition. This was wrong, and leads to data being deleted that shouldn't be.

GitOrigin-RevId: f188a2be4d62f87a0711a93c4c75168bab98d8d9

core/src/main/scala/org/apache/spark/sql/delta/sources/DeltaSourceUtils.scala

@@ -77,7 +77,7 @@ object DeltaSourceUtils {
       expressions.GreaterThanOrEqual(
         UnresolvedAttribute(attribute), expressions.Literal.create(value))
     case sources.LessThan(attribute, value) =>
-      expressions.LessThanOrEqual(UnresolvedAttribute(attribute), expressions.Literal.create(value))
+      expressions.LessThan(UnresolvedAttribute(attribute), expressions.Literal.create(value))
     case sources.LessThanOrEqual(attribute, value) =>
       expressions.LessThanOrEqual(UnresolvedAttribute(attribute), expressions.Literal.create(value))
     case sources.In(attribute, values) =>

core/src/test/scala/org/apache/spark/sql/delta/DeltaSuite.scala

@@ -40,7 +40,7 @@ import org.apache.spark.sql.catalyst.plans.logical.Filter
 import org.apache.spark.sql.execution.FileSourceScanExec
 import org.apache.spark.sql.execution.datasources.{HadoopFsRelation, LogicalRelation}
 import org.apache.spark.sql.execution.streaming.MemoryStream
-import org.apache.spark.sql.functions.{asc, expr, lit, map_values, struct}
+import org.apache.spark.sql.functions.{asc, col, expr, lit, map_values, struct}
 import org.apache.spark.sql.internal.SQLConf
 import org.apache.spark.sql.streaming.StreamingQuery
 import org.apache.spark.sql.test.{SharedSparkSession, SQLTestUtils}
@@ -3047,4 +3047,17 @@ class DeltaNameColumnMappingSuite extends DeltaSuite
       }
     }
   }
+
+  test("replaceWhere dataframe V2 API with less than predicate") {
+    withTempDir { dir =>
+      val insertedDF = spark.range(10).toDF()
+
+      insertedDF.write.format("delta").save(dir.toString)
+
+      val otherDF = spark.range(start = 0, end = 4).toDF()
+      otherDF.writeTo(s"delta.`${dir.toString}`").overwrite(col("id") < 6)
+      checkAnswer(spark.read.load(dir.toString),
+        insertedDF.filter(col("id") >= 6).union(otherDF))
+    }
+  }
 }