feat(sql): add REPLACE ... WHERE for atomic predicate-scoped overwrite

docs/src/operations/dml/.pages

@@ -1,6 +1,8 @@
 title: DML
 nav:
   - insert-into.md
+  - insert-overwrite.md
+  - replace.md
   - update.md
   - delete.md
   - add-columns.md

docs/src/operations/dml/replace.md

@@ -0,0 +1,60 @@
+# REPLACE ... WHERE
+
+`REPLACE` atomically replaces the rows of a table that match a predicate with the result of a
+query, in a single table version. It is the predicate-scoped analogue of `INSERT OVERWRITE`: rather
+than replacing the whole table, it replaces only the rows selected by `WHERE`.
+
+The delete of the matching rows and the append of the new rows are committed as one atomic Lance
+`Update` operation, so readers never observe a state where the old rows are gone but the new rows
+are not yet present, and a failure cannot leave the region half-written.
+
+## Syntax
+
+```sql
+REPLACE <table> WHERE <predicate> AS <query>
+```
+
+- `<predicate>` is any SQL boolean expression over the table's columns. It selects the existing rows
+  to delete.
+- `<query>` is any `SELECT` producing rows with the table's schema. Its result becomes the new rows.
+
+## Examples
+
+=== "Spark SQL"
+    ```sql
+    -- Replace one day's data with freshly computed rows
+    REPLACE lance.db.events
+      WHERE dt = '2026-08-01'
+      AS SELECT id, dt, value FROM staging_events WHERE dt = '2026-08-01';
+
+    -- Predicates may span a range or combine conditions
+    REPLACE lance.db.events
+      WHERE dt >= '2026-08-01' AND dt < '2026-08-08'
+      AS SELECT id, dt, value FROM staging_events;
+    ```
+
+=== "PySpark"
+    ```python
+    spark.sql(
+        "REPLACE lance.db.events "
+        "WHERE dt = '2026-08-01' "
+        "AS SELECT id, dt, value FROM staging_events WHERE dt = '2026-08-01'"
+    )
+    ```
+
+## Notes
+
+- The `spark.sql.extensions` entry `org.lance.spark.extensions.LanceSparkSessionExtensions` must be
+  configured, as `REPLACE` is a Lance SQL extension.
+- A row filter that matches no existing rows makes `REPLACE` behave as a plain append of the query
+  result.
+- When a fragment contains both matching and non-matching rows, only the matching rows are removed
+  (via a deletion vector); the rest are preserved. A fragment whose rows all match is dropped
+  outright.
+- **Performance:** for an equality predicate (a conjunction of `column = value`) on columns that
+  have a zonemap index, `REPLACE` uses the zonemap to drop fully-covered fragments by id without
+  scanning their rows — so replacing a partition that occupies its own fragments is metadata-only.
+  A predicate with no zonemap, or any non-equality shape (ranges, `OR`, functions), still works but
+  falls back to scanning the affected fragments to find the rows to delete. Multiple single-column
+  zonemaps combine for a multi-column predicate (e.g. a zonemap on `dt` and one on `hr` for
+  `WHERE dt = ... AND hr = ...`).

lance-spark-3.4_2.12/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala

@@ -14,9 +14,10 @@
 package org.apache.spark.sql.catalyst.parser.extensions
 
 import org.antlr.v4.runtime.ParserRuleContext
+import org.antlr.v4.runtime.misc.Interval
 import org.apache.spark.sql.catalyst.analysis.{UnresolvedIdentifier, UnresolvedRelation}
 import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
-import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
+import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, ReplaceWhere, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.collection.JavaConverters._
@@ -80,6 +81,23 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     Optimize(table, args)
   }
 
+  override def visitReplaceWhere(ctx: LanceSqlExtensionsParser.ReplaceWhereContext)
+      : ReplaceWhere = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    // Recover the raw text following WHERE by character interval (the grammar captures it as an
+    // opaque token run, so slicing the original stream preserves operators, quoting, and spacing),
+    // then split it at the first top-level AS into predicate and query. The query is parsed by
+    // Spark's own parser (delegate); the predicate is handed to Lance at execution time.
+    val parts = ParserUtils.splitReplaceBody(originalText(ctx.body))
+    val query = delegate.parsePlan(parts(1))
+    ReplaceWhere(table, parts(0), query)
+  }
+
+  private def originalText(ctx: ParserRuleContext): String = {
+    val stream = ctx.getStart.getInputStream
+    stream.getText(Interval.of(ctx.getStart.getStartIndex, ctx.getStop.getStopIndex)).trim
+  }
+
   override def visitVacuum(ctx: LanceSqlExtensionsParser.VacuumContext): Vacuum = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
     val args = ctx.namedArgument().asScala.map(a =>

lance-spark-3.5_2.12/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala

@@ -14,9 +14,10 @@
 package org.apache.spark.sql.catalyst.parser.extensions
 
 import org.antlr.v4.runtime.ParserRuleContext
+import org.antlr.v4.runtime.misc.Interval
 import org.apache.spark.sql.catalyst.analysis.{UnresolvedIdentifier, UnresolvedRelation}
 import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
-import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
+import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, ReplaceWhere, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.collection.JavaConverters._
@@ -80,6 +81,23 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     Optimize(table, args)
   }
 
+  override def visitReplaceWhere(ctx: LanceSqlExtensionsParser.ReplaceWhereContext)
+      : ReplaceWhere = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    // Recover the raw text following WHERE by character interval (the grammar captures it as an
+    // opaque token run, so slicing the original stream preserves operators, quoting, and spacing),
+    // then split it at the first top-level AS into predicate and query. The query is parsed by
+    // Spark's own parser (delegate); the predicate is handed to Lance at execution time.
+    val parts = ParserUtils.splitReplaceBody(originalText(ctx.body))
+    val query = delegate.parsePlan(parts(1))
+    ReplaceWhere(table, parts(0), query)
+  }
+
+  private def originalText(ctx: ParserRuleContext): String = {
+    val stream = ctx.getStart.getInputStream
+    stream.getText(Interval.of(ctx.getStart.getStartIndex, ctx.getStop.getStopIndex)).trim
+  }
+
   override def visitVacuum(ctx: LanceSqlExtensionsParser.VacuumContext): Vacuum = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
     val args = ctx.namedArgument().asScala.map(a =>

lance-spark-3.5_2.12/src/test/java/org/lance/spark/write/ReplaceWhereTest.java

@@ -0,0 +1,16 @@
+/*
+ * Licensed 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.lance.spark.write;
+
+public class ReplaceWhereTest extends BaseReplaceWhereTest {}

lance-spark-4.0_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala

@@ -14,9 +14,10 @@
 package org.apache.spark.sql.catalyst.parser.extensions
 
 import org.antlr.v4.runtime.ParserRuleContext
+import org.antlr.v4.runtime.misc.Interval
 import org.apache.spark.sql.catalyst.analysis.{UnresolvedIdentifier, UnresolvedRelation}
 import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
-import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
+import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, ReplaceWhere, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.jdk.CollectionConverters._
@@ -80,6 +81,23 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     Optimize(table, args)
   }
 
+  override def visitReplaceWhere(ctx: LanceSqlExtensionsParser.ReplaceWhereContext)
+      : ReplaceWhere = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    // Recover the raw text following WHERE by character interval (the grammar captures it as an
+    // opaque token run, so slicing the original stream preserves operators, quoting, and spacing),
+    // then split it at the first top-level AS into predicate and query. The query is parsed by
+    // Spark's own parser (delegate); the predicate is handed to Lance at execution time.
+    val parts = ParserUtils.splitReplaceBody(originalText(ctx.body))
+    val query = delegate.parsePlan(parts(1))
+    ReplaceWhere(table, parts(0), query)
+  }
+
+  private def originalText(ctx: ParserRuleContext): String = {
+    val stream = ctx.getStart.getInputStream
+    stream.getText(Interval.of(ctx.getStart.getStartIndex, ctx.getStop.getStopIndex)).trim
+  }
+
   override def visitVacuum(ctx: LanceSqlExtensionsParser.VacuumContext): Vacuum = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
     val args = ctx.namedArgument().asScala.map(a =>

lance-spark-4.1_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala

@@ -14,9 +14,10 @@
 package org.apache.spark.sql.catalyst.parser.extensions
 
 import org.antlr.v4.runtime.ParserRuleContext
+import org.antlr.v4.runtime.misc.Interval
 import org.apache.spark.sql.catalyst.analysis.{UnresolvedIdentifier, UnresolvedRelation}
 import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
-import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
+import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, ReplaceWhere, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.jdk.CollectionConverters._
@@ -80,6 +81,23 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     Optimize(table, args)
   }
 
+  override def visitReplaceWhere(ctx: LanceSqlExtensionsParser.ReplaceWhereContext)
+      : ReplaceWhere = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    // Recover the raw text following WHERE by character interval (the grammar captures it as an
+    // opaque token run, so slicing the original stream preserves operators, quoting, and spacing),
+    // then split it at the first top-level AS into predicate and query. The query is parsed by
+    // Spark's own parser (delegate); the predicate is handed to Lance at execution time.
+    val parts = ParserUtils.splitReplaceBody(originalText(ctx.body))
+    val query = delegate.parsePlan(parts(1))
+    ReplaceWhere(table, parts(0), query)
+  }
+
+  private def originalText(ctx: ParserRuleContext): String = {
+    val stream = ctx.getStart.getInputStream
+    stream.getText(Interval.of(ctx.getStart.getStartIndex, ctx.getStop.getStopIndex)).trim
+  }
+
   override def visitVacuum(ctx: LanceSqlExtensionsParser.VacuumContext): Vacuum = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
     val args = ctx.namedArgument().asScala.map(a =>

lance-spark-4.2_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala

@@ -14,9 +14,10 @@
 package org.apache.spark.sql.catalyst.parser.extensions
 
 import org.antlr.v4.runtime.ParserRuleContext
+import org.antlr.v4.runtime.misc.Interval
 import org.apache.spark.sql.catalyst.analysis.{UnresolvedIdentifier, UnresolvedRelation}
 import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
-import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
+import org.apache.spark.sql.catalyst.plans.logical.{AddColumnsBackfill, AddIndex, LanceCreateBranch, LanceCreateTag, LanceDropBranch, LanceDropIndex, LanceDropTag, LanceNamedArgument, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, ReplaceWhere, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.jdk.CollectionConverters._
@@ -80,6 +81,23 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     Optimize(table, args)
   }
 
+  override def visitReplaceWhere(ctx: LanceSqlExtensionsParser.ReplaceWhereContext)
+      : ReplaceWhere = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    // Recover the raw text following WHERE by character interval (the grammar captures it as an
+    // opaque token run, so slicing the original stream preserves operators, quoting, and spacing),
+    // then split it at the first top-level AS into predicate and query. The query is parsed by
+    // Spark's own parser (delegate); the predicate is handed to Lance at execution time.
+    val parts = ParserUtils.splitReplaceBody(originalText(ctx.body))
+    val query = delegate.parsePlan(parts(1))
+    ReplaceWhere(table, parts(0), query)
+  }
+
+  private def originalText(ctx: ParserRuleContext): String = {
+    val stream = ctx.getStart.getInputStream
+    stream.getText(Interval.of(ctx.getStart.getStartIndex, ctx.getStop.getStopIndex)).trim
+  }
+
   override def visitVacuum(ctx: LanceSqlExtensionsParser.VacuumContext): Vacuum = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
     val args = ctx.namedArgument().asScala.map(a =>

lance-spark-base_2.12/src/main/antlr4/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensions.g4

@@ -43,6 +43,17 @@ statement
     | OPTIMIZE multipartIdentifier (WITH '(' (namedArgument (',' namedArgument)*)? ')')?        #optimize
     | VACUUM multipartIdentifier (WITH '(' (namedArgument (',' namedArgument)*)? ')')?          #vacuum
     | ALTER TABLE multipartIdentifier SET UNENFORCED PRIMARY KEY '(' columnList ')'             #setUnenforcedPrimaryKey
+    | REPLACE multipartIdentifier WHERE body=replaceBody                                       #replaceWhere
+    ;
+
+// Everything after WHERE (the predicate, the AS separator, and the query) is captured verbatim as
+// one raw region and re-parsed with Spark's own parser, so the Lance grammar itself does not need
+// to model SQL expressions or SELECT statements. The AST builder splits this text at the first
+// top-level `AS` (honoring parentheses, string literals, and comments), which lets predicates
+// contain `AS` themselves (e.g. `CAST(x AS STRING)`) and lets the query contain column aliases.
+// Recovering by character interval preserves operators and whitespace regardless of tokenization.
+replaceBody
+    : .+
     ;
 
 multipartIdentifier
@@ -113,6 +124,7 @@ NOT: 'NOT';
 OF: 'OF';
 OPTIMIZE: 'OPTIMIZE';
 PRIMARY: 'PRIMARY';
+REPLACE: 'REPLACE';
 SET: 'SET';
 SHOW: 'SHOW';
 TABLE: 'TABLE';
@@ -123,6 +135,7 @@ UPDATE: 'UPDATE';
 USING: 'USING';
 VACUUM: 'VACUUM';
 VERSION: 'VERSION';
+WHERE: 'WHERE';
 WITH: 'WITH';
 
 TRUE: 'TRUE';
@@ -173,3 +186,15 @@ fragment LETTER
     : [A-Z]
     ;
 
+WS
+    : [ \t\r\n]+ -> skip
+    ;
+
+// Catch-all so any character not matched by a specific token above (SQL operators like '>', '<',
+// '*', etc.) still produces a token instead of a silently-dropped lexer error. This lets the raw
+// predicate/query regions of REPLACE tokenize completely; their text is recovered by character
+// interval, so the specific token kind does not matter.
+ANY
+    : .
+    ;
+

lance-spark-base_2.12/src/main/java/org/lance/spark/LanceConstant.java

@@ -35,6 +35,23 @@ public class LanceConstant {
   public static final String BACKFILL_COLUMNS_KEY = "backfill_columns";
   public static final String UPDATE_COLUMNS_KEY = "update_columns";
 
+  /**
+   * Internal write option carrying the row filter for a {@code REPLACE ... WHERE ... AS ...}
+   * command. When present, the batch write commits a single atomic {@code Update} that deletes the
+   * existing rows matching this predicate and appends the newly written fragments, instead of a
+   * plain append. Set on the driver by {@code ReplaceWhereExec}; not a user-facing option.
+   */
+  public static final String REPLACE_WHERE_KEY = "__lance_replace_where";
+
+  /**
+   * Internal write option carrying the JSON-encoded equality terms of a {@code REPLACE ... WHERE}
+   * predicate (a pure conjunction of {@code column = literal}), when the predicate has that shape.
+   * The commit uses these terms plus zonemap statistics to drop fully-covered fragments by id
+   * without scanning their rows; when absent, commit falls back to the exact scan-based deletion.
+   * Set on the driver by {@code ReplaceWhereExec}; not a user-facing option.
+   */
+  public static final String REPLACE_WHERE_EQUALITY_KEY = "__lance_replace_where_equality";
+
   /**
    * Internal write option carrying the encoded blob source credential/open contexts for an INSERT
    * whose query reads blob columns. Set on the driver by {@code LanceBlobSourceContextRule} and