diff --git a/docs/src/operations/ddl/create-index.md b/docs/src/operations/ddl/create-index.md index b19b5886d..86d140d51 100755 --- a/docs/src/operations/ddl/create-index.md +++ b/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. diff --git a/lance-spark-3.4_2.12/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala b/lance-spark-3.4_2.12/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala index 1ca2fb957..0a54a207c 100644 --- a/lance-spark-3.4_2.12/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala +++ b/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())) diff --git a/lance-spark-3.4_2.12/src/test/java/org/lance/spark/update/LanceSqlExtensionsAstBuilderTest.java b/lance-spark-3.4_2.12/src/test/java/org/lance/spark/update/LanceSqlExtensionsAstBuilderTest.java index 691284eca..fdd388601 100644 --- a/lance-spark-3.4_2.12/src/test/java/org/lance/spark/update/LanceSqlExtensionsAstBuilderTest.java +++ b/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`"); diff --git a/lance-spark-3.5_2.12/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala b/lance-spark-3.5_2.12/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala index 40a228e97..0e4a7434c 100644 --- a/lance-spark-3.5_2.12/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala +++ b/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())) diff --git a/lance-spark-3.5_2.12/src/test/java/org/lance/spark/update/LanceSqlExtensionsAstBuilderTest.java b/lance-spark-3.5_2.12/src/test/java/org/lance/spark/update/LanceSqlExtensionsAstBuilderTest.java index d387087e9..31e73a4bb 100644 --- a/lance-spark-3.5_2.12/src/test/java/org/lance/spark/update/LanceSqlExtensionsAstBuilderTest.java +++ b/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`"); diff --git a/lance-spark-4.0_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala b/lance-spark-4.0_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala index 86068fd67..ab131454d 100644 --- a/lance-spark-4.0_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala +++ b/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())) diff --git a/lance-spark-4.1_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala b/lance-spark-4.1_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala index 86068fd67..ab131454d 100644 --- a/lance-spark-4.1_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala +++ b/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())) diff --git a/lance-spark-4.2_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala b/lance-spark-4.2_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala index 86068fd67..ab131454d 100644 --- a/lance-spark-4.2_2.13/src/main/scala/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensionsAstBuilder.scala +++ b/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())) diff --git a/lance-spark-base_2.12/src/main/antlr4/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensions.g4 b/lance-spark-base_2.12/src/main/antlr4/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensions.g4 index 788525249..d6b59d6a3 100644 --- a/lance-spark-base_2.12/src/main/antlr4/org/apache/spark/sql/catalyst/parser/extensions/LanceSqlExtensions.g4 +++ b/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 diff --git a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/OptimizeIndex.scala b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/OptimizeIndex.scala new file mode 100644 index 000000000..4d9daa4d4 --- /dev/null +++ b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/catalyst/plans/logical/OptimizeIndex.scala @@ -0,0 +1,51 @@ +/* + * 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.apache.spark.sql.catalyst.plans.logical + +import org.apache.spark.sql.catalyst.expressions.{Attribute, AttributeReference} +import org.apache.spark.sql.types.{DataTypes, StructField, StructType} + +/** + * OptimizeIndex logical plan representing incremental index maintenance on a Lance dataset. + * + * Unlike CREATE INDEX (a full distributed rebuild across all fragments), this merges only the + * unindexed (newly-appended) fragments into existing indexes via lance-core's optimizeIndices + * API. When indexName is empty, all indexes are optimized. + */ +case class OptimizeIndex( + table: LogicalPlan, + indexName: Option[String], + args: Seq[LanceNamedArgument]) extends Command { + + override def children: Seq[LogicalPlan] = Seq(table) + + override def output: Seq[Attribute] = OptimizeIndexOutputType.SCHEMA + + override def simpleString(maxFields: Int): String = { + s"OptimizeIndex(${indexName.getOrElse("")})" + } + + override protected def withNewChildrenInternal(newChildren: IndexedSeq[LogicalPlan]) + : OptimizeIndex = { + copy(newChildren(0), this.indexName, this.args) + } +} + +object OptimizeIndexOutputType { + val SCHEMA: Seq[Attribute] = StructType( + Array( + StructField("index_name", DataTypes.StringType, nullable = true), + StructField("status", DataTypes.StringType, nullable = true))) + .map(field => AttributeReference(field.name, field.dataType, field.nullable, field.metadata)()) +} diff --git a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/LanceDataSourceV2Strategy.scala b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/LanceDataSourceV2Strategy.scala index f17e0d52c..0300b7c2a 100644 --- a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/LanceDataSourceV2Strategy.scala +++ b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/LanceDataSourceV2Strategy.scala @@ -23,6 +23,10 @@ import org.apache.spark.sql.execution.{SparkPlan, SparkStrategy} case class LanceDataSourceV2Strategy(session: SparkSession) extends SparkStrategy with PredicateHelper { + // Index names are lower-cased for a case-insensitive SQL contract, using Locale.ROOT so + // normalization is independent of the JVM default locale (see LanceSystemIndex.normalizeName). + private def normalizeIndexName(name: String): String = LanceSystemIndex.normalizeName(name) + override def apply(plan: LogicalPlan): Seq[SparkPlan] = plan match { case AddColumnsBackfill(ResolvedIdentifier(catalog, ident), columnNames, source) => AddColumnsBackfillExec(asTableCatalog(catalog), ident, columnNames, source) :: Nil @@ -40,7 +44,7 @@ case class LanceDataSourceV2Strategy(session: SparkSession) extends SparkStrateg AddIndexExec( asTableCatalog(catalog), ident, - indexName.toLowerCase, + normalizeIndexName(indexName), method, columns, args) :: Nil @@ -49,7 +53,14 @@ case class LanceDataSourceV2Strategy(session: SparkSession) extends SparkStrateg ShowIndexesExec(asTableCatalog(catalog), ident) :: Nil case LanceDropIndex(ResolvedIdentifier(catalog, ident), indexName) => - LanceDropIndexExec(asTableCatalog(catalog), ident, indexName.toLowerCase) :: Nil + LanceDropIndexExec(asTableCatalog(catalog), ident, normalizeIndexName(indexName)) :: Nil + + case OptimizeIndex(ResolvedIdentifier(catalog, ident), indexName, args) => + OptimizeIndexExec( + asTableCatalog(catalog), + ident, + indexName.map(normalizeIndexName), + args) :: Nil case LanceCreateBranch(ResolvedIdentifier(catalog, ident), branchName, ref, ifNotExists) => LanceCreateBranchExec(asTableCatalog(catalog), ident, branchName, ref, ifNotExists) :: Nil diff --git a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/LanceSystemIndex.scala b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/LanceSystemIndex.scala new file mode 100644 index 000000000..1a1b8ae3f --- /dev/null +++ b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/LanceSystemIndex.scala @@ -0,0 +1,40 @@ +/* + * 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.apache.spark.sql.execution.datasources.v2 + +import java.util.Locale + +/** + * Helpers for Lance index names shared across the index DDL commands. + * + * Holds the set of system (internal) index names that are not user-manageable — maintained by + * lance-core itself and filtered out of user-facing index operations such as SHOW INDEXES and + * OPTIMIZE INDEX — and the locale-independent name normalization used to give the SQL commands a + * consistent, case-insensitive contract. + */ +object LanceSystemIndex { + val SystemIndexNames: Set[String] = Set("__lance_frag_reuse", "__lance_mem_wal") + + def isSystemIndex(indexName: String): Boolean = + indexName != null && SystemIndexNames.exists(_.equalsIgnoreCase(indexName)) + + /** + * Lower-cases an index name for the case-insensitive SQL contract using [[Locale.ROOT]], so + * normalization does not depend on the JVM default locale (e.g. under tr-TR a naive + * {@code toLowerCase} maps 'I' to a dotless 'ı', which would make CREATE and OPTIMIZE/DROP INDEX + * disagree on the same name). + */ + def normalizeName(indexName: String): String = + if (indexName == null) null else indexName.toLowerCase(Locale.ROOT) +} diff --git a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/OptimizeIndexExec.scala b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/OptimizeIndexExec.scala new file mode 100644 index 000000000..a533bb982 --- /dev/null +++ b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/OptimizeIndexExec.scala @@ -0,0 +1,139 @@ +/* + * 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.apache.spark.sql.execution.datasources.v2 + +import org.apache.spark.sql.catalyst.InternalRow +import org.apache.spark.sql.catalyst.expressions.{Attribute, GenericInternalRow} +import org.apache.spark.sql.catalyst.plans.logical.{LanceNamedArgument, OptimizeIndexOutputType} +import org.apache.spark.sql.connector.catalog.{Identifier, TableCatalog} +import org.apache.spark.unsafe.types.UTF8String +import org.lance.index.OptimizeOptions +import org.lance.spark.LanceDataset +import org.lance.spark.utils.Utils + +import java.util.Locale + +import scala.collection.JavaConverters._ + +/** + * Physical execution of ALTER TABLE ... OPTIMIZE INDEX for Lance datasets. + * + * Incrementally merges unindexed (newly-appended) fragments into existing indexes via lance-core's + * optimizeIndices API. This is distinct from CREATE INDEX, which performs a full distributed + * rebuild over all fragments. When indexName is empty, all user indexes are optimized. This runs on + * the driver (single node). + * + * Supported WITH options (option names are case-insensitive): + * + * + *

The {@code retrain} option is intentionally not exposed here: in lance-core it is a + * v3-vector-index rebuild and has no effect for the scalar indexes this incremental command + * targets. Use {@code Dataset.optimizeIndices} in the SDK to retrain vector indexes. + */ +case class OptimizeIndexExec( + catalog: TableCatalog, + ident: Identifier, + indexName: Option[String], + args: Seq[LanceNamedArgument]) extends LeafV2CommandExec { + + override def output: Seq[Attribute] = OptimizeIndexOutputType.SCHEMA + + private val NUM_INDICES_TO_MERGE = "num_indices_to_merge" + private val ALLOWED_OPTIONS = Set(NUM_INDICES_TO_MERGE) + + /** + * Validates the WITH options and builds the core OptimizeOptions. Option names are normalized + * with {@code Locale.ROOT} (locale-independent) and matched case-insensitively; unknown names, + * duplicates, wrong value types, and out-of-range integers are rejected rather than silently + * ignored or reinterpreted. + */ + private def buildOptions(): OptimizeOptions = { + val normalized = args.map(arg => (arg.name.toLowerCase(Locale.ROOT), arg)) + + val unknown = normalized.map(_._1).filterNot(ALLOWED_OPTIONS.contains) + if (unknown.nonEmpty) { + throw new IllegalArgumentException( + s"Unsupported OPTIMIZE INDEX option(s): ${unknown.distinct.mkString(", ")}. " + + s"Supported options: ${ALLOWED_OPTIONS.toSeq.sorted.mkString(", ")}") + } + + val byName = normalized.groupBy(_._1) + byName.collectFirst { + case (name, occurrences) if occurrences.size > 1 => + throw new IllegalArgumentException(s"Duplicate OPTIMIZE INDEX option: $name") + } + + val builder = OptimizeOptions.builder() + indexName.foreach(name => builder.indexNames(List(name).asJava)) + + byName.get(NUM_INDICES_TO_MERGE).foreach { occurrences => + occurrences.head._2.value match { + case n: java.lang.Long => + if (n < 0 || n > Int.MaxValue) { + throw new IllegalArgumentException( + s"OPTIMIZE INDEX option '$NUM_INDICES_TO_MERGE' must be between 0 and " + + s"${Int.MaxValue}, got: $n") + } + builder.numIndicesToMerge(n.intValue()) + case other => + throw new IllegalArgumentException( + s"OPTIMIZE INDEX option '$NUM_INDICES_TO_MERGE' expects an integer, got: $other") + } + } + + builder.build() + } + + override protected def run(): Seq[InternalRow] = { + val lanceDataset = catalog.loadTable(ident) match { + case ds: LanceDataset => ds + case _ => + throw new UnsupportedOperationException("OptimizeIndex only supports LanceDataset") + } + + val options = buildOptions() + val readOptions = lanceDataset.readOptions() + + val dataset = Utils.openDatasetBuilder(readOptions).build() + try { + // lance-core exact-matches indexNames and treats an empty match set as a successful no-op, + // so a typo (or a system-index name that core filters out) would otherwise be reported as + // completed maintenance. Validate up front against the user-optimizable index set (the same + // system-index filtering SHOW INDEXES uses), and reject system indexes explicitly. + indexName.foreach { name => + if (LanceSystemIndex.isSystemIndex(name)) { + throw new IllegalArgumentException( + s"Index '$name' is a Lance system index and cannot be optimized.") + } + val userIndexes = + dataset.listIndexes().asScala.filterNot(LanceSystemIndex.isSystemIndex) + if (!userIndexes.contains(name)) { + throw new IllegalArgumentException( + s"Index '$name' does not exist on table ${ident.name()}. " + + s"Existing indexes: ${userIndexes.toSeq.sorted.mkString(", ")}") + } + } + dataset.optimizeIndices(options) + } finally { + dataset.close() + } + + Seq(new GenericInternalRow(Array[Any]( + UTF8String.fromString(indexName.getOrElse("")), + UTF8String.fromString("optimized")))) + } +} diff --git a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala index 997778458..23bb974df 100755 --- a/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala +++ b/lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/execution/datasources/v2/ShowIndexesExec.scala @@ -24,13 +24,6 @@ import org.lance.spark.utils.{FieldPathUtils, Utils} import scala.collection.JavaConverters._ -object ShowIndexesExec { - private val SystemIndexNames = Set("__lance_frag_reuse", "__lance_mem_wal") - - private def isSystemIndex(indexName: String): Boolean = - indexName != null && SystemIndexNames.exists(_.equalsIgnoreCase(indexName)) -} - /** * Physical execution of SHOW INDEXES for Lance datasets. * @@ -54,7 +47,7 @@ case class ShowIndexesExec( val dataset = Utils.openDatasetBuilder(readOptions).build() try { val indexes = dataset.getIndexes.asScala.toSeq - .filterNot(idx => ShowIndexesExec.isSystemIndex(idx.name())) + .filterNot(idx => LanceSystemIndex.isSystemIndex(idx.name())) .groupBy(_.name()) .toSeq .sortBy(_._1) diff --git a/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java b/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java index b99b99d37..6e8368e85 100755 --- a/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java +++ b/lance-spark-base_2.12/src/test/java/org/lance/spark/update/BaseAddIndexTest.java @@ -716,6 +716,186 @@ public void testCreateZonemapIndexDeferred() { Assertions.assertEquals("text_15", afterOptimize.collectAsList().get(0).getString(1)); } + /** + * A deferred ZONEMAP can be populated incrementally through the SQL {@code OPTIMIZE INDEX} + * command, which merges the unindexed fragments into the existing index. + */ + @Test + public void testDeferredZonemapPopulatedByOptimizeIndexSql() { + prepareDataset(); + + spark.sql( + String.format( + "alter table %s create index idx_oi using zonemap (id) with (train=false)", fullTable)); + + // Incrementally populate the deferred index via SQL (not the SDK). + Dataset result = + spark.sql(String.format("alter table %s optimize index idx_oi", fullTable)); + Row row = result.collectAsList().get(0); + Assertions.assertEquals("idx_oi", row.getString(0)); + Assertions.assertEquals("optimized", row.getString(1)); + + org.lance.Dataset lanceDataset = org.lance.Dataset.open().uri(tableDir).build(); + try { + int fragmentCount = lanceDataset.getFragments().size(); + Assertions.assertTrue(fragmentCount >= 2, "Expected multiple fragments"); + int coveredFragments = + lanceDataset.getIndexes().stream() + .filter(index -> "idx_oi".equals(index.name())) + .map(index -> index.fragments().orElse(Collections.emptyList()).size()) + .mapToInt(Integer::intValue) + .sum(); + Assertions.assertEquals( + fragmentCount, + coveredFragments, + "Expected OPTIMIZE INDEX to populate the deferred zonemap over all fragments"); + } finally { + lanceDataset.close(); + } + + // Query remains correct after the index is populated. + Dataset afterOptimize = + spark.sql(String.format("select * from %s where id=15", fullTable)); + Assertions.assertEquals(1L, afterOptimize.count()); + Assertions.assertEquals("text_15", afterOptimize.collectAsList().get(0).getString(1)); + } + + /** + * OPTIMIZE INDEX on a non-existent index name fails instead of silently reporting success (which + * would make a typo indistinguishable from completed maintenance). + */ + @Test + public void testOptimizeIndexMissingNameFails() { + prepareDataset(); + + Assertions.assertThrows( + Exception.class, + () -> + spark + .sql(String.format("alter table %s optimize index missing_idx", fullTable)) + .collect()); + } + + /** + * OPTIMIZE INDEX option names are normalized locale-independently and matched case-insensitively: + * an unquoted (upper-cased) option name is recognized rather than rejected as unknown. + */ + @Test + public void testOptimizeIndexOptionNameCaseInsensitive() { + prepareDataset(); + + spark.sql( + String.format( + "alter table %s create index idx_ci using zonemap (id) with (train=false)", fullTable)); + + // NUM_INDICES_TO_MERGE (unquoted -> upper-cased by the parser) must be recognized, not + // rejected. + Dataset result = + spark.sql( + String.format( + "alter table %s optimize index idx_ci with (NUM_INDICES_TO_MERGE = 1)", fullTable)); + Assertions.assertEquals("optimized", result.collectAsList().get(0).getString(1)); + + org.lance.Dataset lanceDataset = org.lance.Dataset.open().uri(tableDir).build(); + try { + int fragmentCount = lanceDataset.getFragments().size(); + int coveredFragments = + lanceDataset.getIndexes().stream() + .filter(index -> "idx_ci".equals(index.name())) + .map(index -> index.fragments().orElse(Collections.emptyList()).size()) + .mapToInt(Integer::intValue) + .sum(); + Assertions.assertEquals( + fragmentCount, coveredFragments, "Expected OPTIMIZE INDEX to cover all fragments"); + } finally { + lanceDataset.close(); + } + } + + /** OPTIMIZE INDEX rejects unknown WITH options rather than silently ignoring them. */ + @Test + public void testOptimizeIndexUnknownOptionFails() { + prepareDataset(); + + spark.sql( + String.format( + "alter table %s create index idx_uk using zonemap (id) with (train=false)", fullTable)); + + Assertions.assertThrows( + Exception.class, + () -> + spark + .sql( + String.format( + "alter table %s optimize index idx_uk with (bogus = true)", fullTable)) + .collect()); + } + + /** + * OPTIMIZE INDEX rejects the {@code retrain} option: it is a vector-index rebuild in lance-core + * and is not exposed through this incremental SQL command. + */ + @Test + public void testOptimizeIndexRetrainOptionRejected() { + prepareDataset(); + + spark.sql( + String.format( + "alter table %s create index idx_rt using zonemap (id) with (train=false)", fullTable)); + + Assertions.assertThrows( + Exception.class, + () -> + spark + .sql( + String.format( + "alter table %s optimize index idx_rt with (retrain = true)", fullTable)) + .collect()); + } + + /** + * OPTIMIZE INDEX on a Lance system index (e.g. {@code __lance_frag_reuse}) fails instead of + * silently reporting success. lance-core filters system indexes before optimizing, so without + * this guard the executor would emit {@code optimized} for a no-op. + */ + @Test + public void testOptimizeIndexSystemIndexRejected() { + prepareDataset(); + + // Create one full-coverage segment, then OPTIMIZE with deferred remap to produce the + // __lance_frag_reuse system index (mirrors testShowIndexesFiltersFragmentReuseIndex). + org.lance.index.IndexParams indexParams = + org.lance.index.IndexParams.builder() + .setScalarIndexParams(org.lance.index.scalar.ScalarIndexParams.create("BTREE")) + .build(); + try (org.lance.Dataset dataset = org.lance.Dataset.open().uri(tableDir).build()) { + dataset.createIndex( + org.lance.index.IndexOptions.builder(List.of("id"), IndexType.BTREE, indexParams) + .replace(true) + .train(true) + .withIndexName("test_index") + .build()); + } + spark.sql( + String.format( + "optimize %s with (target_rows_per_fragment=20000, defer_index_remap=true)", + fullTable)); + + try (org.lance.Dataset dataset = org.lance.Dataset.open().uri(tableDir).build()) { + Assertions.assertTrue( + dataset.getIndexes().stream() + .anyMatch(index -> "__lance_frag_reuse".equalsIgnoreCase(index.name())), + "Expected a fragment-reuse system index to exist for this test"); + } + + Assertions.assertThrows( + Exception.class, + () -> + spark + .sql(String.format("alter table %s optimize index __lance_frag_reuse", fullTable)) + .collect()); + } + /** * A deferred ZONEMAP can be populated by re-running CREATE INDEX (eager): the distributed segment * build replaces the empty index and covers all fragments. diff --git a/lance-spark-base_2.12/src/test/java/org/lance/spark/update/LanceSystemIndexTest.java b/lance-spark-base_2.12/src/test/java/org/lance/spark/update/LanceSystemIndexTest.java new file mode 100644 index 000000000..98c6ed4a3 --- /dev/null +++ b/lance-spark-base_2.12/src/test/java/org/lance/spark/update/LanceSystemIndexTest.java @@ -0,0 +1,57 @@ +/* + * 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.update; + +import org.apache.spark.sql.execution.datasources.v2.LanceSystemIndex; +import org.junit.jupiter.api.Test; + +import java.util.Locale; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** Unit tests for shared Lance index-name helpers. */ +public class LanceSystemIndexTest { + + /** + * Normalization must be locale-independent. Under tr-TR, {@code "IDX".toLowerCase()} yields + * {@code "ıdx"} (dotless i), which would make CREATE and OPTIMIZE/DROP INDEX disagree on the same + * name. {@code Locale.ROOT} keeps it {@code "idx"} regardless of the JVM default locale. + */ + @Test + public void testNormalizeNameIsLocaleIndependent() { + Locale previous = Locale.getDefault(); + try { + Locale.setDefault(new Locale("tr", "TR")); + assertEquals("idx_istanbul", LanceSystemIndex.normalizeName("IDX_ISTANBUL")); + assertEquals("idx", LanceSystemIndex.normalizeName("IDX")); + } finally { + Locale.setDefault(previous); + } + } + + @Test + public void testNormalizeNameNull() { + assertEquals(null, LanceSystemIndex.normalizeName(null)); + } + + @Test + public void testIsSystemIndex() { + assertTrue(LanceSystemIndex.isSystemIndex("__lance_frag_reuse")); + assertTrue(LanceSystemIndex.isSystemIndex("__LANCE_MEM_WAL")); + assertFalse(LanceSystemIndex.isSystemIndex("idx_user")); + assertFalse(LanceSystemIndex.isSystemIndex(null)); + } +}