feat: add SQL vector_search table-valued function

docs/src/config.md

@@ -53,6 +53,7 @@ The following features require the Lance Spark SQL extension to be enabled:
 - [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
 - [VACUUM](operations/ddl/vacuum.md) - Remove old versions and reclaim storage space
+- [SQL vector search](operations/dql/select.md#vector-search) - Search vector columns with `vector_search(...)`
 
 ## Basic Setup
 

docs/src/operations/dql/select.md

@@ -53,6 +53,53 @@ Query data from Lance tables using SQL or DataFrames.
     SELECT * FROM users WHERE age > 25;
     ```
 
+## Vector Search
+
+!!! note
+    This feature requires the Lance Spark SQL extension to be enabled. See [Spark SQL Extensions](../../config.md#spark-sql-extensions) for configuration details.
+
+Use the `vector_search` table-valued function to search a vector column from SQL:
+
+```sql
+vector_search(table, nearest)
+```
+
+Arguments:
+
+- `table` - A Spark multi-part identifier (catalog-qualified table name), passed as a
+  single-quoted SQL string literal. Bare filesystem paths are **not** accepted - they
+  are rejected when the identifier is parsed. To target a Lance dataset by path,
+  register a `LanceNamespaceSparkCatalog` (e.g. as `lance_default`) and quote the path
+  part with backticks inside the outer single-quoted string.
+- `nearest` - A Lance nearest-neighbor query JSON string. Include `column`, `key`,
+  and `k`; optional fields such as `distanceType` and `useIndex` are passed through
+  to Lance. The `k` value is also used as Spark's final global TopK limit.
+
+Path-based catalog example (mirrors the test fixtures):
+
+```sql
+SELECT id, title
+FROM vector_search(
+  '`lance_default`.`file:/data/lance/documents.lance`',
+  '{"column":"embedding","key":[0.12,0.34,0.56],"k":10,"distanceType":"Cosine","useIndex":true}'
+);
+```
+
+Namespace-backed catalog example:
+
+```sql
+SELECT id, title
+FROM vector_search(
+  'lance_prod.reports.documents',
+  '{"column":"embedding","key":[0.12,0.34,0.56],"k":10}'
+);
+```
+
+`vector_search` uses Lance's native nearest-neighbor scan on each fragment, keeps the normal
+fragment scan partitions, and applies the final global TopK ordering in Spark using Lance's
+native `_distance` value. The `_distance` column is internal to the SQL plan and is not returned
+by `SELECT *`.
+
 ## Aggregate Queries
 
 === "SQL"

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

@@ -85,4 +85,19 @@ public RowLevelOperationBuilder newRowLevelOperationBuilder(
         getFileFormatVersion(),
         properties());
   }
+
+  @Override
+  public LancePositionDeltaDataset withSchema(StructType newSchema) {
+    return new LancePositionDeltaDataset(
+        readOptions(),
+        newSchema,
+        getInitialStorageOptions(),
+        getNamespaceImpl(),
+        getNamespaceProperties(),
+        getManagedVersioning(),
+        getStagedCommit(),
+        getFileFormatVersion(),
+        getTableProperties(),
+        getShardingSpec());
+  }
 }

lance-spark-3.4_2.12/src/main/scala/org/lance/spark/extensions/LanceSparkSessionExtensions.scala

@@ -14,8 +14,10 @@
 package org.lance.spark.extensions
 
 import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.spark.sql.catalyst.analysis.LanceTableValuedFunctionResolution
 import org.apache.spark.sql.catalyst.optimizer.{LanceBlobSourceContextRule, LanceFragmentAwareJoinRule}
 import org.apache.spark.sql.catalyst.parser.extensions.LanceSparkSqlExtensionsParser
+import org.apache.spark.sql.catalyst.plans.logical.LanceTableValuedFunctions
 import org.apache.spark.sql.execution.datasources.v2.LanceDataSourceV2Strategy
 
 class LanceSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
@@ -24,6 +26,15 @@ class LanceSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
     // parser extensions
     extensions.injectParser { case (_, parser) => new LanceSparkSqlExtensionsParser(parser) }
 
+    // table-valued function resolution
+    extensions.injectResolutionRule(spark => LanceTableValuedFunctionResolution(spark))
+
+    // table function extensions
+    LanceTableValuedFunctions.supportedFnNames.foreach { fnName =>
+      extensions.injectTableFunction(
+        LanceTableValuedFunctions.getTableValueFunctionInjection(fnName))
+    }
+
     // optimizer rules for fragment-aware joins
     extensions.injectOptimizerRule(_ => LanceFragmentAwareJoinRule())
 

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

@@ -85,4 +85,19 @@ public RowLevelOperationBuilder newRowLevelOperationBuilder(
         getFileFormatVersion(),
         properties());
   }
+
+  @Override
+  public LancePositionDeltaDataset withSchema(StructType newSchema) {
+    return new LancePositionDeltaDataset(
+        readOptions(),
+        newSchema,
+        getInitialStorageOptions(),
+        getNamespaceImpl(),
+        getNamespaceProperties(),
+        getManagedVersioning(),
+        getStagedCommit(),
+        getFileFormatVersion(),
+        getTableProperties(),
+        getShardingSpec());
+  }
 }

lance-spark-3.5_2.12/src/main/scala/org/lance/spark/extensions/LanceSparkSessionExtensions.scala

@@ -14,8 +14,10 @@
 package org.lance.spark.extensions
 
 import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.spark.sql.catalyst.analysis.LanceTableValuedFunctionResolution
 import org.apache.spark.sql.catalyst.optimizer.{LanceBlobSourceContextRule, LanceFragmentAwareJoinRule}
 import org.apache.spark.sql.catalyst.parser.extensions.LanceSparkSqlExtensionsParser
+import org.apache.spark.sql.catalyst.plans.logical.LanceTableValuedFunctions
 import org.apache.spark.sql.execution.datasources.v2.LanceDataSourceV2Strategy
 
 class LanceSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
@@ -24,6 +26,15 @@ class LanceSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
     // parser extensions
     extensions.injectParser { case (_, parser) => new LanceSparkSqlExtensionsParser(parser) }
 
+    // table-valued function resolution
+    extensions.injectResolutionRule(spark => LanceTableValuedFunctionResolution(spark))
+
+    // table function extensions
+    LanceTableValuedFunctions.supportedFnNames.foreach { fnName =>
+      extensions.injectTableFunction(
+        LanceTableValuedFunctions.getTableValueFunctionInjection(fnName))
+    }
+
     // optimizer rules for fragment-aware joins
     extensions.injectOptimizerRule(_ => LanceFragmentAwareJoinRule())
 

lance-spark-4.0_2.13/src/main/scala/org/lance/spark/extensions/LanceSparkSessionExtensions.scala

@@ -14,8 +14,10 @@
 package org.lance.spark.extensions
 
 import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.spark.sql.catalyst.analysis.LanceTableValuedFunctionResolution
 import org.apache.spark.sql.catalyst.optimizer.{LanceBlobSourceContextRule, LanceFragmentAwareJoinRule}
 import org.apache.spark.sql.catalyst.parser.extensions.LanceSparkSqlExtensionsParser
+import org.apache.spark.sql.catalyst.plans.logical.LanceTableValuedFunctions
 import org.apache.spark.sql.execution.datasources.v2.LanceDataSourceV2Strategy
 
 class LanceSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
@@ -24,6 +26,15 @@ class LanceSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
     // parser extensions
     extensions.injectParser { case (_, parser) => new LanceSparkSqlExtensionsParser(parser) }
 
+    // table-valued function resolution
+    extensions.injectResolutionRule(spark => LanceTableValuedFunctionResolution(spark))
+
+    // table function extensions
+    LanceTableValuedFunctions.supportedFnNames.foreach { fnName =>
+      extensions.injectTableFunction(
+        LanceTableValuedFunctions.getTableValueFunctionInjection(fnName))
+    }
+
     // optimizer rules for fragment-aware joins
     extensions.injectOptimizerRule(_ => LanceFragmentAwareJoinRule())
 

lance-spark-4.1_2.13/src/main/scala/org/lance/spark/extensions/LanceSparkSessionExtensions.scala

@@ -14,8 +14,10 @@
 package org.lance.spark.extensions
 
 import org.apache.spark.sql.SparkSessionExtensions
+import org.apache.spark.sql.catalyst.analysis.LanceTableValuedFunctionResolution
 import org.apache.spark.sql.catalyst.optimizer.{LanceBlobSourceContextRule, LanceFragmentAwareJoinRule}
 import org.apache.spark.sql.catalyst.parser.extensions.LanceSparkSqlExtensionsParser
+import org.apache.spark.sql.catalyst.plans.logical.LanceTableValuedFunctions
 import org.apache.spark.sql.execution.datasources.v2.LanceDataSourceV2Strategy
 
 class LanceSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
@@ -24,6 +26,15 @@ class LanceSparkSessionExtensions extends (SparkSessionExtensions => Unit) {
     // parser extensions
     extensions.injectParser { case (_, parser) => new LanceSparkSqlExtensionsParser(parser) }
 
+    // table-valued function resolution
+    extensions.injectResolutionRule(spark => LanceTableValuedFunctionResolution(spark))
+
+    // table function extensions
+    LanceTableValuedFunctions.supportedFnNames.foreach { fnName =>
+      extensions.injectTableFunction(
+        LanceTableValuedFunctions.getTableValueFunctionInjection(fnName))
+    }
+
     // optimizer rules for fragment-aware joins
     extensions.injectOptimizerRule(_ => LanceFragmentAwareJoinRule())
 

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

@@ -126,7 +126,7 @@ private static boolean isPathBasedIdentifier(Identifier ident) {
           || firstPart.startsWith("gs://")
           || firstPart.startsWith("az://")
           || firstPart.startsWith("abfss://")
-          || firstPart.startsWith("file://")
+          || firstPart.startsWith("file:/")
           || firstPart.startsWith("hdfs://")) {
         return true;
       }
@@ -139,7 +139,7 @@ private static boolean isPathBasedIdentifier(Identifier ident) {
         || name.startsWith("gs://")
         || name.startsWith("az://")
         || name.startsWith("abfss://")
-        || name.startsWith("file://")
+        || name.startsWith("file:/")
         || name.startsWith("hdfs://");
   }
 

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

@@ -18,6 +18,9 @@ public class LanceConstant {
   public static final String ROW_ID = "_rowid";
   public static final String ROW_ADDRESS = "_rowaddr";
 
+  // Internal distance column emitted by Lance nearest-neighbor scans.
+  public static final String VECTOR_DISTANCE = "_distance";
+
   // CDF (Change Data Feed) version tracking columns
   public static final String ROW_CREATED_AT_VERSION = "_row_created_at_version";
   public static final String ROW_LAST_UPDATED_AT_VERSION = "_row_last_updated_at_version";

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

@@ -272,6 +272,42 @@ public String getFileFormatVersion() {
     return fileFormatVersion;
   }
 
+  public StagedCommit getStagedCommit() {
+    return stagedCommit;
+  }
+
+  public Map<String, String> getTableProperties() {
+    return tableProperties;
+  }
+
+  public ShardingSpec getShardingSpec() {
+    return shardingSpec;
+  }
+
+  /**
+   * Returns a copy of this dataset with the given Spark schema, preserving every other piece of
+   * state (read options, namespace credentials, staged commit, file format version, table
+   * properties, sharding spec).
+   *
+   * <p>Used by callers that need to extend the projected schema without losing dataset
+   * configuration -- for example, vector_search adds a synthetic {@code _distance} column.
+   *
+   * <p>Subclasses should override to preserve their runtime type and any extra contracts.
+   */
+  public LanceDataset withSchema(StructType newSchema) {
+    return new LanceDataset(
+        readOptions,
+        newSchema,
+        initialStorageOptions,
+        namespaceImpl,
+        namespaceProperties,
+        managedVersioning,
+        stagedCommit,
+        fileFormatVersion,
+        tableProperties,
+        shardingSpec);
+  }
+
   @Override
   public ScanBuilder newScanBuilder(CaseInsensitiveStringMap caseInsensitiveStringMap) {
     // Merge scan-time options with the existing read options

lance-spark-base_2.12/src/main/java/org/lance/spark/internal/LanceFragmentScanner.java

@@ -29,6 +29,7 @@
 import org.apache.spark.sql.types.StructType;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashSet;
 import java.util.List;
@@ -107,7 +108,12 @@ public static LanceFragmentScanner create(int fragmentId, LanceInputPartition in
       Set<String> blobColumnNames = getBlobColumnNames(inputPartition.getSchema());
       boolean hasBlobColumns = !blobColumnNames.isEmpty();
 
-      List<String> projectedColumns = getColumnNames(inputPartition.getSchema());
+      List<String> projectedColumns = new ArrayList<>(getColumnNames(inputPartition.getSchema()));
+      if (readOptions.getNearest() != null
+          && hasField(inputPartition.getSchema(), LanceConstant.VECTOR_DISTANCE)
+          && !projectedColumns.contains(LanceConstant.VECTOR_DISTANCE)) {
+        projectedColumns.add(LanceConstant.VECTOR_DISTANCE);
+      }
       if (projectedColumns.isEmpty() && inputPartition.getSchema().isEmpty()) {
         scanOptions.withRowId(true);
       }
@@ -276,6 +282,7 @@ private static List<String> getColumnNames(StructType schema) {
                     !name.equals(LanceConstant.FRAGMENT_ID)
                         && !name.equals(LanceConstant.ROW_ID)
                         && !name.equals(LanceConstant.ROW_ADDRESS)
+                        && !name.equals(LanceConstant.VECTOR_DISTANCE)
                         && !name.equals(LanceConstant.ROW_CREATED_AT_VERSION)
                         && !name.equals(LanceConstant.ROW_LAST_UPDATED_AT_VERSION)
                         && !name.endsWith(LanceConstant.BLOB_POSITION_SUFFIX)

lance-spark-base_2.12/src/main/java/org/lance/spark/read/LanceScan.java

@@ -147,6 +147,11 @@ public LanceScan(
     this.namespaceProperties = namespaceProperties;
   }
 
+  /** Visible for tests in the same package; not part of the public API. */
+  LanceSparkReadOptions getReadOptions() {
+    return readOptions;
+  }
+
   @Override
   public Batch toBatch() {
     return this;

lance-spark-base_2.12/src/main/java/org/lance/spark/read/LanceScanBuilder.java

@@ -304,6 +304,10 @@ public boolean isPartiallyPushed() {
 
   @Override
   public boolean pushTopN(SortOrder[] orders, int limit) {
+    if (readOptions.getNearest() != null) {
+      return false;
+    }
+
     // The Order by operator will use compute thread in lance.
     // So it's better to have an option to enable it.
     if (!readOptions.isTopNPushDown()) {

lance-spark-base_2.12/src/main/scala/org/apache/spark/sql/catalyst/analysis/LanceTableValuedFunctionResolution.scala

@@ -0,0 +1,26 @@
+/*
+ * 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.analysis
+
+import org.apache.spark.sql.SparkSession
+import org.apache.spark.sql.catalyst.plans.logical.{LanceTableValuedFunctions, LanceTableValueFunction, LogicalPlan}
+import org.apache.spark.sql.catalyst.rules.Rule
+
+case class LanceTableValuedFunctionResolution(session: SparkSession) extends Rule[LogicalPlan] {
+
+  override def apply(plan: LogicalPlan): LogicalPlan = plan.resolveOperatorsDown {
+    case func: LanceTableValueFunction if func.args.forall(_.resolved) =>
+      LanceTableValuedFunctions.resolveLanceTableValuedFunction(session, func)
+  }
+}