feat: surface _score metadata column on FTS predicate queries

docs/src/operations/dql/fts.md

@@ -111,9 +111,34 @@ SELECT * FROM lance.db.documents
 WHERE lance_multi_match('machine learning', 'operator=AND', title, body);
 ```
 
+## Relevance Score (`_score`)
+
+The `_score` metadata column exposes the BM25 relevance score computed by the FTS index. It is available only when a full-text search predicate is active.
+
+```sql
+-- Project _score alongside data columns
+SELECT id, body, _score FROM lance.db.documents
+WHERE lance_match(body, 'machine learning');
+
+-- Sort by relevance
+SELECT id, _score FROM lance.db.documents
+WHERE lance_match(body, 'vector database')
+ORDER BY _score DESC;
+
+-- Combine with scalar filters
+SELECT id, _score FROM lance.db.documents
+WHERE lance_match(body, 'deep learning') AND year >= 2024
+ORDER BY _score DESC;
+```
+
+`_score` is a hidden metadata column — it does not appear in `SELECT *` output. Selecting `_score` without an FTS predicate raises an error at planning time.
+
+!!! note "Not a ranked-retrieval push"
+    `ORDER BY _score DESC LIMIT k` sorts in Spark above the scan — it does **not** push a ranked top-k query to the Lance engine. All matching rows are scanned and scored; Spark applies the sort and limit. A pushed ranked-retrieval path is planned as a future extension.
+
 ## Known Limitations
 
-- **No global relevance ordering.** FTS predicates act as WHERE filters and return all matching rows. There is no `lance_score()` function or relevance-based `ORDER BY` — rows are returned in storage order. In particular, `SELECT * FROM t WHERE lance_match(...) LIMIT N` returns N matching rows determined by Spark task scheduling, not by BM25 rank — it is not a "top-N by relevance" query.
+- **No pushed ranked retrieval.** `ORDER BY _score DESC LIMIT k` is correct but not optimized — all matching rows are scanned, scored, and sorted in Spark. A pushed top-k path that leverages Lance's native scoring order is a planned future extension.
 - **Column names must match the schema exactly.** Column references are resolved at planning time; aliases or expressions are not supported.
 - **`lance_match_phrase` requires positional index.** The FTS index must be built with `with_position = true`. Without it, phrase queries will fail.
 - **WHERE-filter uses full BM25 scoring (no WAND early stopping).** Every row matching the query is evaluated — `wand_factor` is not exposed because SQL WHERE semantics require returning all matching rows.

lance-spark-3.4_2.12/src/test/java/org/lance/spark/read/FtsScoreColumnTest.java

@@ -0,0 +1,16 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.lance.spark.read;
+
+public class FtsScoreColumnTest extends BaseFtsScoreColumnTest {}

lance-spark-3.5_2.12/src/test/java/org/lance/spark/read/FtsScoreColumnTest.java

@@ -0,0 +1,16 @@
+/*
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.lance.spark.read;
+
+public class FtsScoreColumnTest extends BaseFtsScoreColumnTest {}

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

@@ -19,8 +19,8 @@ public class LanceConstant {
   public static final String ROW_ADDRESS = "_rowaddr";
 
   /**
-   * Relevance score column. Lance auto-projects it onto the scan output whenever a full-text query
-   * is set, so it is exposed as a metadata column and stripped from the native column projection.
+   * BM25 relevance score column. Explicitly included in the native column projection when a
+   * full-text query is active. Exposed as a metadata column so it is hidden from SELECT *.
    */
   public static final String SCORE = "_score";
 

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

@@ -157,10 +157,10 @@ public DataType dataType() {
       };
 
   /**
-   * Relevance score, auto-projected by Lance only when a full-text query is active on the scan.
+   * BM25 relevance score, explicitly projected when a full-text query is active on the scan.
    * Advertised unconditionally because {@code metadataColumns()} is consulted by the analyzer
-   * before the query is known to contain FTS; selecting it without a full-text search yields an
-   * empty scan result for this column rather than a plan-time error (validation is a follow-up).
+   * before the query is known to contain FTS; selecting it without a full-text search raises {@code
+   * IllegalArgumentException} at scan-build time with a clear diagnostic message.
    */
   public static final MetadataColumn SCORE_COLUMN =
       new MetadataColumn() {

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

@@ -156,6 +156,15 @@ private List<ColumnVector> buildSparkOrderedVectors(
           BlobSizeColumnVector sizeVector = new BlobSizeColumnVector((StructVector) blobVector);
           fieldVectors.add(sizeVector);
         }
+      } else if (fieldName.equals(LanceConstant.SCORE)) {
+        FieldVector vector = actualFields.get(fieldName);
+        if (vector == null) {
+          throw new IllegalStateException(
+              "Lance scan did not return '_score'. This indicates a full-text query was expected "
+                  + "but not applied to the native scanner. Verify that the FTS predicate rule "
+                  + "injected the query into the relation options.");
+        }
+        fieldVectors.add(new LanceArrowColumnVector(vector, false, field));
       } else {
         FieldVector vector = actualFields.get(fieldName);
         if (vector == null) {

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

@@ -110,6 +110,9 @@ public static LanceFragmentScanner create(int fragmentId, LanceInputPartition in
       boolean hasBlobColumns = !blobColumnNames.isEmpty();
 
       List<String> projectedColumns = getColumnNames(scanSchema);
+      if (readOptions.getFullTextQuery() != null && hasField(scanSchema, LanceConstant.SCORE)) {
+        projectedColumns.add(LanceConstant.SCORE);
+      }
       if (projectedColumns.isEmpty() && scanSchema.isEmpty()) {
         scanOptions.withRowId(true);
       }
@@ -131,6 +134,7 @@ public static LanceFragmentScanner create(int fragmentId, LanceInputPartition in
       scanOptions.batchSize(readOptions.getBatchSize());
       if (readOptions.getFullTextQuery() != null) {
         scanOptions.fullTextQuery(readOptions.getFullTextQuery());
+        scanOptions.disableScoringAutoprojection(true);
       }
       scanOptions.useScalarIndex(readOptions.isUseScalarIndex());
       if (inputPartition.getLimit().isPresent()) {

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

@@ -172,6 +172,18 @@ public Scan build() {
         return localScan;
       }
 
+      // Reject _score without an active FTS query. The metadata column is advertised
+      // unconditionally (analyzer resolves it before the FTS rule runs), so validation
+      // must happen here at build time, after the optimizer fixed-point has finalized
+      // the relation options.
+      if (hasScoreColumn(schema) && readOptions.getFullTextQuery() == null) {
+        throw new IllegalArgumentException(
+            "Column '_score' requires a full-text search predicate (lance_match, "
+                + "lance_match_phrase, or lance_multi_match) in the WHERE clause. "
+                + "_score is a relevance score computed by the Lance FTS index and has no "
+                + "value without an active full-text query.");
+      }
+
       // Namespace-configured full-text search executes server-side via queryTable (single
       // partition). A full-text query without a namespace, or against a catalog-only namespace such
       // as Glue that does not implement queryTable, falls through to the local per-fragment scan
@@ -426,12 +438,9 @@ public boolean isPartiallyPushed() {
 
   @Override
   public boolean pushTopN(SortOrder[] orders, int limit) {
-    // The Order by operator will use compute thread in lance.
-    // So it's better to have an option to enable it.
     if (!readOptions.isTopNPushDown() || hasResidualPredicates) {
       return false;
     }
-    this.limit = Optional.of(limit);
     List<ColumnOrdering> topNSortOrders = new ArrayList<>();
     for (SortOrder sortOrder : orders) {
       ColumnOrdering.Builder builder = new ColumnOrdering.Builder();
@@ -441,9 +450,14 @@ public boolean pushTopN(SortOrder[] orders, int limit) {
         return false;
       }
       FieldReference reference = (FieldReference) sortOrder.expression();
-      builder.setColumnName(reference.fieldNames()[0]);
+      String columnName = reference.fieldNames()[0];
+      if (columnName.equals(LanceConstant.SCORE)) {
+        return false;
+      }
+      builder.setColumnName(columnName);
       topNSortOrders.add(builder.build());
     }
+    this.limit = Optional.of(limit);
     this.topNSortOrders = Optional.of(topNSortOrders);
     return true;
   }
@@ -478,6 +492,15 @@ public boolean pushAggregation(Aggregation aggregation) {
     return false;
   }
 
+  private static boolean hasScoreColumn(StructType schema) {
+    for (StructField field : schema.fields()) {
+      if (field.name().equals(LanceConstant.SCORE)) {
+        return true;
+      }
+    }
+    return false;
+  }
+
   private static Optional<Long> getCountFromMetadata(Dataset dataset) {
     try {
       ManifestSummary summary = dataset.getVersion().getManifestSummary();

lance-spark-base_2.12/src/test/java/org/lance/spark/internal/LanceFragmentScannerTest.java

@@ -150,8 +150,8 @@ public void testGetColumnNamesWithAllMetadataColumns() throws Exception {
 
   @Test
   public void testGetColumnNamesExcludesScore() throws Exception {
-    // _score is auto-projected by Lance when a full-text query is set, so it must not be requested
-    // in the native column projection.
+    // getColumnNames() strips _score unconditionally; the caller (create()) conditionally re-adds
+    // it when FTS is active. This test validates the stripping behavior only.
     StructType schema =
         new StructType(
             new StructField[] {