feat: add ALTER TABLE ... OPTIMIZE INDEX for incremental index maintenance

docs/src/operations/ddl/create-index.md

@@ -263,9 +263,34 @@ to scanning the data until it is populated. There are two ways to populate it:
     ALTER TABLE lance.db.users CREATE INDEX idx_id USING zonemap (id);
     ```
 
-- **Incremental build through the SDK:** when only some fragments are unindexed (for example after
-  appending data to an already-built index), `Dataset.optimizeIndices` indexes just the unindexed
-  fragments. This currently runs on a single node:
+- **Incremental build with `OPTIMIZE INDEX`:** when only some fragments are unindexed (for example
+  after appending data to an already-built index), `ALTER TABLE ... OPTIMIZE INDEX` merges just the
+  unindexed fragments into existing indexes. Unlike `CREATE INDEX`, it does not rebuild coverage
+  already committed. This currently runs on a single node (the driver):
+
+    === "SQL"
+        ```sql
+        -- optimize a single index
+        ALTER TABLE lance.db.users OPTIMIZE INDEX idx_id;
+
+        -- optimize all indexes on the table
+        ALTER TABLE lance.db.users OPTIMIZE INDEX;
+
+        -- bound the number of delta indices merged per index
+        ALTER TABLE lance.db.users OPTIMIZE INDEX idx_id WITH (num_indices_to_merge = 2);
+        ```
+
+    A named index must already exist and must be a user index (Lance system indexes such as
+    `__lance_frag_reuse` cannot be optimized); optimizing a missing or system index raises an error
+    rather than silently succeeding. Supported `WITH` options (names are case-insensitive):
+
+    | Option | Type | Description |
+    |--------|------|-------------|
+    | `num_indices_to_merge` | Integer >= 0 (default core-defined) | Number of delta indices to merge per index; `0` creates a new delta index instead of merging into the base. |
+
+    Retraining an index from source data is a vector-index operation in lance-core; it is not
+    exposed through this SQL command. Use `Dataset.optimizeIndices` in the SDK for that. The same
+    incremental operation is also available via the SDK:
 
     ```java
     dataset.optimizeIndices(OptimizeOptions.builder().build());
@@ -277,7 +302,7 @@ that populates the index instead.
 
 Creating a scalar index on an empty table also registers an empty index with zero fragment
 coverage. The index is immediately visible through `SHOW INDEXES`. After data is appended, populate
-it by re-running `CREATE INDEX` or calling `Dataset.optimizeIndices`.
+it by re-running `CREATE INDEX` or running `ALTER TABLE ... OPTIMIZE INDEX`.
 
 ## Output
 
@@ -309,4 +334,4 @@ The `CREATE INDEX` command operates as follows:
 - **Index Methods**: The `zonemap`, `bitmap`, `label_list`, `ngram`, `bloomfilter`, `rtree`, `btree`, and `fts` (or `inverted`) methods are supported for index creation.
 - **Indexed Column Count**: `zonemap`, `bitmap`, `label_list`, `ngram`, `bloomfilter`, `rtree`, and `fts` (or `inverted`) currently support exactly one indexed column.
 - **Index Replacement**: If you create an index with the same name as an existing one, the old index will be replaced by the new one.
-- **Deferred Training**: With `train = false` the index is registered empty and is populated later, either by re-running `CREATE INDEX` (a full distributed build that replaces the empty index) or, for incremental coverage of newly appended fragments, by `Dataset.optimizeIndices` in the SDK. The SQL `OPTIMIZE` command compacts fragments and does not train deferred indexes.
+- **Deferred Training**: With `train = false` the index is registered empty and is populated later, either by re-running `CREATE INDEX` (a full distributed build that replaces the empty index) or, for incremental coverage of newly appended fragments, by `ALTER TABLE ... OPTIMIZE INDEX` (equivalently `Dataset.optimizeIndices` in the SDK). The SQL `OPTIMIZE` command compacts fragments and does not train deferred indexes; use `OPTIMIZE INDEX` for that.

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

@@ -16,7 +16,7 @@ package org.apache.spark.sql.catalyst.parser.extensions
 import org.antlr.v4.runtime.ParserRuleContext
 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, OptimizeIndex, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.collection.JavaConverters._
@@ -116,6 +116,19 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     LanceDropIndex(table, indexName)
   }
 
+  override def visitOptimizeIndex(ctx: LanceSqlExtensionsParser.OptimizeIndexContext)
+      : OptimizeIndex = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    val indexName = Option(ctx.indexName).map(name => cleanIdentifier(name.getText))
+    val args = ctx.namedArgument().asScala.map(a =>
+      LanceNamedArgument(
+        cleanIdentifier(a.identifier().getText),
+        a.constant().accept(this)))
+      .toSeq
+
+    OptimizeIndex(table, indexName, args)
+  }
+
   override def visitCreateBranchRefMain(ctx: LanceSqlExtensionsParser.CreateBranchRefMainContext)
       : LanceCreateBranch = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))

lance-spark-3.4_2.12/src/test/java/org/lance/spark/update/LanceSqlExtensionsAstBuilderTest.java

@@ -23,6 +23,7 @@
 import org.apache.spark.sql.catalyst.plans.logical.AddColumnsBackfill;
 import org.apache.spark.sql.catalyst.plans.logical.AddIndex;
 import org.apache.spark.sql.catalyst.plans.logical.Optimize;
+import org.apache.spark.sql.catalyst.plans.logical.OptimizeIndex;
 import org.apache.spark.sql.catalyst.plans.logical.ShowIndexes;
 import org.apache.spark.sql.catalyst.plans.logical.UpdateColumnsBackfill;
 import org.apache.spark.sql.catalyst.plans.logical.Vacuum;
@@ -152,6 +153,46 @@ public void testOptimizeWithBacktickedTableName() {
         List.of("my-catalog", "my-table"), JavaConverters.seqAsJavaList(table.nameParts()));
   }
 
+  @Test
+  public void testOptimizeIndexWithIndexName() {
+    LanceSqlExtensionsParser parser =
+        createParser("ALTER TABLE `my-catalog`.`my-table` OPTIMIZE INDEX `my-idx`");
+    OptimizeIndex plan = (OptimizeIndex) astBuilder.visitSingleStatement(parser.singleStatement());
+
+    UnresolvedIdentifier table = (UnresolvedIdentifier) plan.table();
+    assertEquals(
+        List.of("my-catalog", "my-table"), JavaConverters.seqAsJavaList(table.nameParts()));
+    assertEquals("my-idx", plan.indexName().get());
+    assertTrue(plan.args().isEmpty());
+  }
+
+  @Test
+  public void testOptimizeIndexAllIndexes() {
+    LanceSqlExtensionsParser parser = createParser("ALTER TABLE CATALOG.TBL OPTIMIZE INDEX");
+    OptimizeIndex plan = (OptimizeIndex) astBuilder.visitSingleStatement(parser.singleStatement());
+
+    UnresolvedIdentifier table = (UnresolvedIdentifier) plan.table();
+    assertEquals(List.of("CATALOG", "TBL"), JavaConverters.seqAsJavaList(table.nameParts()));
+    assertTrue(plan.indexName().isEmpty());
+    assertTrue(plan.args().isEmpty());
+  }
+
+  @Test
+  public void testOptimizeIndexWithArgs() {
+    // The grammar accepts arbitrary named arguments; option validation happens in the executor.
+    // The unit-test lexer sees raw input (no UpperCaseCharStream), so its IDENTIFIER rule only
+    // matches [A-Z]; backtick-quote the lowercase argument name so it tokenizes here.
+    LanceSqlExtensionsParser parser =
+        createParser(
+            "ALTER TABLE CATALOG.TBL OPTIMIZE INDEX MY_IDX WITH (`num_indices_to_merge` = 2)");
+    OptimizeIndex plan = (OptimizeIndex) astBuilder.visitSingleStatement(parser.singleStatement());
+
+    assertEquals("MY_IDX", plan.indexName().get());
+    assertEquals(1, plan.args().size());
+    assertEquals("num_indices_to_merge", plan.args().apply(0).name());
+    assertEquals(2L, plan.args().apply(0).value());
+  }
+
   @Test
   public void testShowIndexesWithBacktickedTableName() {
     LanceSqlExtensionsParser parser = createParser("SHOW INDEXES FROM `my-catalog`.`my-table`");

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

@@ -16,7 +16,7 @@ package org.apache.spark.sql.catalyst.parser.extensions
 import org.antlr.v4.runtime.ParserRuleContext
 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, OptimizeIndex, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.collection.JavaConverters._
@@ -116,6 +116,19 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     LanceDropIndex(table, indexName)
   }
 
+  override def visitOptimizeIndex(ctx: LanceSqlExtensionsParser.OptimizeIndexContext)
+      : OptimizeIndex = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    val indexName = Option(ctx.indexName).map(name => cleanIdentifier(name.getText))
+    val args = ctx.namedArgument().asScala.map(a =>
+      LanceNamedArgument(
+        cleanIdentifier(a.identifier().getText),
+        a.constant().accept(this)))
+      .toSeq
+
+    OptimizeIndex(table, indexName, args)
+  }
+
   override def visitCreateBranchRefMain(ctx: LanceSqlExtensionsParser.CreateBranchRefMainContext)
       : LanceCreateBranch = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))

lance-spark-3.5_2.12/src/test/java/org/lance/spark/update/LanceSqlExtensionsAstBuilderTest.java

@@ -26,6 +26,7 @@
 import org.apache.spark.sql.catalyst.plans.logical.LanceDropBranch;
 import org.apache.spark.sql.catalyst.plans.logical.LanceShowBranches;
 import org.apache.spark.sql.catalyst.plans.logical.Optimize;
+import org.apache.spark.sql.catalyst.plans.logical.OptimizeIndex;
 import org.apache.spark.sql.catalyst.plans.logical.ShowIndexes;
 import org.apache.spark.sql.catalyst.plans.logical.UpdateColumnsBackfill;
 import org.apache.spark.sql.catalyst.plans.logical.Vacuum;
@@ -176,6 +177,46 @@ public void testOptimizeWithBacktickedTableName() {
         List.of("my-catalog", "my-table"), JavaConverters.seqAsJavaList(table.nameParts()));
   }
 
+  @Test
+  public void testOptimizeIndexWithIndexName() {
+    LanceSqlExtensionsParser parser =
+        createParser("ALTER TABLE `my-catalog`.`my-table` OPTIMIZE INDEX `my-idx`");
+    OptimizeIndex plan = (OptimizeIndex) astBuilder.visitSingleStatement(parser.singleStatement());
+
+    UnresolvedIdentifier table = (UnresolvedIdentifier) plan.table();
+    assertEquals(
+        List.of("my-catalog", "my-table"), JavaConverters.seqAsJavaList(table.nameParts()));
+    assertEquals("my-idx", plan.indexName().get());
+    assertTrue(plan.args().isEmpty());
+  }
+
+  @Test
+  public void testOptimizeIndexAllIndexes() {
+    LanceSqlExtensionsParser parser = createParser("ALTER TABLE CATALOG.TBL OPTIMIZE INDEX");
+    OptimizeIndex plan = (OptimizeIndex) astBuilder.visitSingleStatement(parser.singleStatement());
+
+    UnresolvedIdentifier table = (UnresolvedIdentifier) plan.table();
+    assertEquals(List.of("CATALOG", "TBL"), JavaConverters.seqAsJavaList(table.nameParts()));
+    assertTrue(plan.indexName().isEmpty());
+    assertTrue(plan.args().isEmpty());
+  }
+
+  @Test
+  public void testOptimizeIndexWithArgs() {
+    // The grammar accepts arbitrary named arguments; option validation happens in the executor.
+    // The unit-test lexer sees raw input (no UpperCaseCharStream), so its IDENTIFIER rule only
+    // matches [A-Z]; backtick-quote the lowercase argument name so it tokenizes here.
+    LanceSqlExtensionsParser parser =
+        createParser(
+            "ALTER TABLE CATALOG.TBL OPTIMIZE INDEX MY_IDX WITH (`num_indices_to_merge` = 2)");
+    OptimizeIndex plan = (OptimizeIndex) astBuilder.visitSingleStatement(parser.singleStatement());
+
+    assertEquals("MY_IDX", plan.indexName().get());
+    assertEquals(1, plan.args().size());
+    assertEquals("num_indices_to_merge", plan.args().apply(0).name());
+    assertEquals(2L, plan.args().apply(0).value());
+  }
+
   @Test
   public void testShowIndexesWithBacktickedTableName() {
     LanceSqlExtensionsParser parser = createParser("SHOW INDEXES FROM `my-catalog`.`my-table`");

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

@@ -16,7 +16,7 @@ package org.apache.spark.sql.catalyst.parser.extensions
 import org.antlr.v4.runtime.ParserRuleContext
 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, OptimizeIndex, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.jdk.CollectionConverters._
@@ -116,6 +116,19 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     LanceDropIndex(table, indexName)
   }
 
+  override def visitOptimizeIndex(ctx: LanceSqlExtensionsParser.OptimizeIndexContext)
+      : OptimizeIndex = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    val indexName = Option(ctx.indexName).map(name => cleanIdentifier(name.getText))
+    val args = ctx.namedArgument().asScala.map(a =>
+      LanceNamedArgument(
+        cleanIdentifier(a.identifier().getText),
+        a.constant().accept(this)))
+      .toSeq
+
+    OptimizeIndex(table, indexName, args)
+  }
+
   override def visitCreateBranchRefMain(ctx: LanceSqlExtensionsParser.CreateBranchRefMainContext)
       : LanceCreateBranch = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))

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

@@ -16,7 +16,7 @@ package org.apache.spark.sql.catalyst.parser.extensions
 import org.antlr.v4.runtime.ParserRuleContext
 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, OptimizeIndex, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.jdk.CollectionConverters._
@@ -116,6 +116,19 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     LanceDropIndex(table, indexName)
   }
 
+  override def visitOptimizeIndex(ctx: LanceSqlExtensionsParser.OptimizeIndexContext)
+      : OptimizeIndex = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    val indexName = Option(ctx.indexName).map(name => cleanIdentifier(name.getText))
+    val args = ctx.namedArgument().asScala.map(a =>
+      LanceNamedArgument(
+        cleanIdentifier(a.identifier().getText),
+        a.constant().accept(this)))
+      .toSeq
+
+    OptimizeIndex(table, indexName, args)
+  }
+
   override def visitCreateBranchRefMain(ctx: LanceSqlExtensionsParser.CreateBranchRefMainContext)
       : LanceCreateBranch = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))

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

@@ -16,7 +16,7 @@ package org.apache.spark.sql.catalyst.parser.extensions
 import org.antlr.v4.runtime.ParserRuleContext
 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, OptimizeIndex, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import scala.jdk.CollectionConverters._
@@ -116,6 +116,19 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     LanceDropIndex(table, indexName)
   }
 
+  override def visitOptimizeIndex(ctx: LanceSqlExtensionsParser.OptimizeIndexContext)
+      : OptimizeIndex = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    val indexName = Option(ctx.indexName).map(name => cleanIdentifier(name.getText))
+    val args = ctx.namedArgument().asScala.map(a =>
+      LanceNamedArgument(
+        cleanIdentifier(a.identifier().getText),
+        a.constant().accept(this)))
+      .toSeq
+
+    OptimizeIndex(table, indexName, args)
+  }
+
   override def visitCreateBranchRefMain(ctx: LanceSqlExtensionsParser.CreateBranchRefMainContext)
       : LanceCreateBranch = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))

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

@@ -23,6 +23,7 @@ statement
     | ALTER TABLE multipartIdentifier UPDATE COLUMNS columnList FROM identifier                 #updateColumnsBackfill
     | ALTER TABLE multipartIdentifier CREATE INDEX indexName=identifier USING method=identifier '(' fieldPathList ')' (WITH '(' (namedArgument (',' namedArgument)*)? ')')? #createIndex
     | ALTER TABLE multipartIdentifier DROP INDEX indexName=identifier                           #dropIndex
+    | ALTER TABLE multipartIdentifier OPTIMIZE INDEX (indexName=identifier)? (WITH '(' (namedArgument (',' namedArgument)*)? ')')? #optimizeIndex
     | ALTER TABLE multipartIdentifier CREATE BRANCH (IF NOT EXISTS)? branchName=identifier
         (AS OF VERSION refMainVersion=versionNumber)?                                           #createBranchRefMain
     | ALTER TABLE multipartIdentifier CREATE BRANCH (IF NOT EXISTS)? branchName=identifier