feat: support OPTIMIZE INDEX SQL

docs/src/config.md

@@ -56,6 +56,7 @@ The following features require the Lance Spark SQL extension to be enabled:
 - [ADD COLUMNS with backfill](operations/dml/add-columns.md) - Add new columns and backfill existing rows with data
 - [UPDATE COLUMNS with backfill](operations/dml/update-columns.md) - Update existing columns using data from a source
 - [OPTIMIZE](operations/ddl/optimize.md) - Compact table fragments for improved query performance
+- [OPTIMIZE INDEX](operations/ddl/optimize-index.md) - Incrementally maintain a named index
 - [VACUUM](operations/ddl/vacuum.md) - Remove old versions and reclaim storage space
 
 ## Basic Setup

docs/src/operations/ddl/.pages

@@ -12,6 +12,7 @@ nav:
   - drop-table.md
   - create-index.md
   - show-indexes.md
+  - optimize-index.md
   - create-branch.md
   - drop-branch.md
   - show-branches.md

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

@@ -266,12 +266,12 @@ 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 maintenance:** when only some fragments are unindexed (for example after appending
+  data to an already-built index), [`OPTIMIZE INDEX`](optimize-index.md) indexes the uncovered
+  fragments. This currently runs on the Spark driver:
 
-    ```java
-    dataset.optimizeIndices(OptimizeOptions.builder().build());
+    ```sql
+    ALTER TABLE lance.db.users OPTIMIZE INDEX idx_id;
     ```
 
 `train = false` is supported for all index methods. Because deferred index creation does not build
@@ -312,4 +312,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**: All supported index methods 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 through `ALTER TABLE ... OPTIMIZE INDEX` for driver-side incremental maintenance. The table-level `OPTIMIZE` command compacts fragments and does not train deferred indexes.

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

@@ -0,0 +1,61 @@
+# OPTIMIZE INDEX
+
+Incrementally maintains an existing named Lance index.
+
+!!! warning "Spark Extension Required"
+    This feature requires the Lance Spark SQL extension to be enabled. See [Spark SQL Extensions](../../config.md#spark-sql-extensions) for configuration details.
+
+## Syntax
+
+```sql
+ALTER TABLE table_name OPTIMIZE INDEX index_name
+[WITH (
+    num_indices_to_merge = non_negative_integer
+)];
+```
+
+The index must already exist. Lance builds index data for fragments not currently covered by the
+named index and may merge existing index segments according to the supplied options.
+
+## Options
+
+| Option | Type | Description |
+|--------|------|-------------|
+| `num_indices_to_merge` | Integer | Number of existing index segments Lance should merge during maintenance. When omitted, Lance chooses its default. Set to `0` to add coverage without requesting a merge of existing segments. |
+
+The option is passed directly to Lance's `OptimizeOptions`. The target index name is passed as
+the sole entry in `indexNames`, so other indexes on the table are not maintained by this command.
+
+## Examples
+
+Maintain a scalar index after new data is appended:
+
+```sql
+ALTER TABLE lance.db.users OPTIMIZE INDEX idx_user_id;
+```
+
+Build coverage for new fragments without requesting a merge of existing segments:
+
+```sql
+ALTER TABLE lance.db.users OPTIMIZE INDEX idx_user_id WITH (
+    num_indices_to_merge = 0
+);
+```
+
+## Output
+
+| Column | Type | Description |
+|--------|------|-------------|
+| `index_name` | String | Name of the maintained index. |
+| `fragments_indexed` | Long | Number of previously uncovered live fragments indexed by the operation. |
+| `segments_before` | Long | Number of physical segments for the named index before maintenance. |
+| `segments_after` | Long | Number of physical segments for the named index after maintenance. |
+
+## Execution
+
+This command currently invokes Lance index maintenance on the Spark driver. Its SQL contract is
+independent of execution strategy, so a future distributed implementation can retain the same
+syntax and options.
+
+`ALTER TABLE ... OPTIMIZE INDEX` maintains index coverage. The table-level [`OPTIMIZE`](optimize.md)
+command compacts data fragments and is a separate operation.

integration-tests/test_lance_spark.py

@@ -915,6 +915,39 @@ def test_create_distributed_bitmap_index(self, spark):
         )
         assert spark.sql("SELECT * FROM default.test_table").count() == 4
 
+    def test_optimize_index(self, spark):
+        """Test incremental index maintenance through Spark SQL."""
+        spark.sql("CREATE TABLE default.test_table (id INT, name STRING)")
+        spark.sql("INSERT INTO default.test_table VALUES (1, 'one'), (2, 'two')")
+        spark.sql("""
+            ALTER TABLE default.test_table
+            CREATE INDEX idx_id USING zonemap (id)
+        """)
+        spark.sql("INSERT INTO default.test_table VALUES (3, 'three')")
+
+        before = next(
+            row
+            for row in spark.sql("SHOW INDEXES IN default.test_table").collect()
+            if row.name == "idx_id"
+        )
+        assert before.num_unindexed_fragments > 0
+
+        result = spark.sql("""
+            ALTER TABLE default.test_table OPTIMIZE INDEX idx_id
+            WITH (num_indices_to_merge = 0)
+        """).first()
+
+        assert result.index_name == "idx_id"
+        assert result.fragments_indexed == before.num_unindexed_fragments
+        assert result.segments_after >= result.segments_before
+
+        after = next(
+            row
+            for row in spark.sql("SHOW INDEXES IN default.test_table").collect()
+            if row.name == "idx_id"
+        )
+        assert after.num_unindexed_fragments == 0
+
     def test_create_btree_index_on_nested_literal_dot_field(self, spark):
         """Test CREATE INDEX on nested struct fields, including literal dots."""
         spark.sql("""

lance-spark-3.4_2.12/src/main/java/org/lance/spark/write/SparkPositionDeltaWrite.java

@@ -61,6 +61,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.concurrent.FutureTask;
 import java.util.stream.Collectors;
@@ -202,6 +203,7 @@ public void commit(WriterCommitMessage[] messages) {
                 .removedFragmentIds(removedFragmentIds)
                 .updatedFragments(updatedFragments)
                 .newFragments(newFragments)
+                .updateMode(Optional.of(Update.UpdateMode.RewriteRows))
                 .build();
 
         CommitBuilder commitBuilder =

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, LanceOptimizeIndex, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import java.util.Locale
@@ -93,6 +93,19 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     Optimize(table, args)
   }
 
+  override def visitOptimizeIndex(ctx: LanceSqlExtensionsParser.OptimizeIndexContext)
+      : LanceOptimizeIndex = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    val indexName = cleanIdentifier(ctx.indexName.getText)
+    val args = ctx.namedArgument().asScala.map(a =>
+      LanceNamedArgument(
+        normalizedOptionName(a.identifier().getText),
+        a.constant().accept(this)))
+      .toSeq
+
+    LanceOptimizeIndex(table, indexName, args)
+  }
+
   override def visitVacuum(ctx: LanceSqlExtensionsParser.VacuumContext): Vacuum = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
     val args = ctx.namedArgument().asScala.map(a =>

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

@@ -22,6 +22,8 @@
 import org.apache.spark.sql.catalyst.parser.extensions.LanceSqlExtensionsParser;
 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.LanceNamedArgument;
+import org.apache.spark.sql.catalyst.plans.logical.LanceOptimizeIndex;
 import org.apache.spark.sql.catalyst.plans.logical.Optimize;
 import org.apache.spark.sql.catalyst.plans.logical.ShowIndexes;
 import org.apache.spark.sql.catalyst.plans.logical.UpdateColumnsBackfill;
@@ -182,6 +184,26 @@ public void testOptimizeNormalizesOptionNames() {
     assertEquals("target_rows_per_fragment", plan.args().apply(0).name());
   }
 
+  @Test
+  public void testOptimizeIndexWithOptions() {
+    LanceSqlExtensionsParser parser =
+        createParser(
+            "ALTER TABLE `my-catalog`.`my-table` OPTIMIZE INDEX `my-idx` "
+                + "WITH (NUM_INDICES_TO_MERGE = 2)");
+    LanceOptimizeIndex plan =
+        (LanceOptimizeIndex) 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());
+
+    List<LanceNamedArgument> args = JavaConverters.seqAsJavaList(plan.args());
+    assertEquals(1, args.size());
+    assertEquals("num_indices_to_merge", args.get(0).name());
+    assertEquals(2L, args.get(0).value());
+  }
+
   @Test
   public void testShowIndexesWithBacktickedTableName() {
     LanceSqlExtensionsParser parser = createParser("SHOW INDEXES FROM `my-catalog`.`my-table`");

lance-spark-3.5_2.12/src/main/java/org/lance/spark/write/SparkPositionDeltaWrite.java

@@ -66,6 +66,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Optional;
 import java.util.concurrent.Callable;
 import java.util.concurrent.FutureTask;
 import java.util.stream.Collectors;
@@ -224,6 +225,7 @@ public void commit(WriterCommitMessage[] messages) {
                 .removedFragmentIds(removedFragmentIds)
                 .updatedFragments(updatedFragments)
                 .newFragments(newFragments)
+                .updateMode(Optional.of(Update.UpdateMode.RewriteRows))
                 .build();
 
         CommitBuilder commitBuilder =

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, LanceOptimizeIndex, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import java.util.Locale
@@ -93,6 +93,19 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     Optimize(table, args)
   }
 
+  override def visitOptimizeIndex(ctx: LanceSqlExtensionsParser.OptimizeIndexContext)
+      : LanceOptimizeIndex = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    val indexName = cleanIdentifier(ctx.indexName.getText)
+    val args = ctx.namedArgument().asScala.map(a =>
+      LanceNamedArgument(
+        normalizedOptionName(a.identifier().getText),
+        a.constant().accept(this)))
+      .toSeq
+
+    LanceOptimizeIndex(table, indexName, args)
+  }
+
   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/update/LanceSqlExtensionsAstBuilderTest.java

@@ -24,6 +24,8 @@
 import org.apache.spark.sql.catalyst.plans.logical.AddIndex;
 import org.apache.spark.sql.catalyst.plans.logical.LanceCreateBranch;
 import org.apache.spark.sql.catalyst.plans.logical.LanceDropBranch;
+import org.apache.spark.sql.catalyst.plans.logical.LanceNamedArgument;
+import org.apache.spark.sql.catalyst.plans.logical.LanceOptimizeIndex;
 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.ShowIndexes;
@@ -206,6 +208,26 @@ public void testOptimizeNormalizesOptionNames() {
     assertEquals("target_rows_per_fragment", plan.args().apply(0).name());
   }
 
+  @Test
+  public void testOptimizeIndexWithOptions() {
+    LanceSqlExtensionsParser parser =
+        createParser(
+            "ALTER TABLE `my-catalog`.`my-table` OPTIMIZE INDEX `my-idx` "
+                + "WITH (NUM_INDICES_TO_MERGE = 2)");
+    LanceOptimizeIndex plan =
+        (LanceOptimizeIndex) 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());
+
+    List<LanceNamedArgument> args = JavaConverters.seqAsJavaList(plan.args());
+    assertEquals(1, args.size());
+    assertEquals("num_indices_to_merge", args.get(0).name());
+    assertEquals(2L, args.get(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, LanceOptimizeIndex, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import java.util.Locale
@@ -93,6 +93,19 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     Optimize(table, args)
   }
 
+  override def visitOptimizeIndex(ctx: LanceSqlExtensionsParser.OptimizeIndexContext)
+      : LanceOptimizeIndex = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    val indexName = cleanIdentifier(ctx.indexName.getText)
+    val args = ctx.namedArgument().asScala.map(a =>
+      LanceNamedArgument(
+        normalizedOptionName(a.identifier().getText),
+        a.constant().accept(this)))
+      .toSeq
+
+    LanceOptimizeIndex(table, indexName, args)
+  }
+
   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

@@ -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, LanceOptimizeIndex, LanceShowBranches, LanceShowTags, LogicalPlan, Optimize, SetUnenforcedPrimaryKey, ShowIndexes, UpdateColumnsBackfill, Vacuum}
 import org.lance.spark.utils.{FieldPathUtils, ParserUtils}
 
 import java.util.Locale
@@ -93,6 +93,19 @@ class LanceSqlExtensionsAstBuilder(delegate: ParserInterface)
     Optimize(table, args)
   }
 
+  override def visitOptimizeIndex(ctx: LanceSqlExtensionsParser.OptimizeIndexContext)
+      : LanceOptimizeIndex = {
+    val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
+    val indexName = cleanIdentifier(ctx.indexName.getText)
+    val args = ctx.namedArgument().asScala.map(a =>
+      LanceNamedArgument(
+        normalizedOptionName(a.identifier().getText),
+        a.constant().accept(this)))
+      .toSeq
+
+    LanceOptimizeIndex(table, indexName, args)
+  }
+
   override def visitVacuum(ctx: LanceSqlExtensionsParser.VacuumContext): Vacuum = {
     val table = UnresolvedIdentifier(visitMultipartIdentifier(ctx.multipartIdentifier()))
     val args = ctx.namedArgument().asScala.map(a =>