feat: add custom Lance metrics to trace read-path scan performance

docs/src/performance.md

@@ -157,6 +157,43 @@ For large scans, increasing this to match your CPU core count enables more concu
 export LANCE_IO_THREADS=128
 ```
 
+### Custom Read Metrics
+
+Lance Spark reports per-task custom metrics on the Spark UI Scan node, viewable in the SQL tab's
+physical plan. These are useful for diagnosing read-path performance — pruning effectiveness, JNI
+overhead, and where time is being spent.
+
+Naming conventions:
+
+- Counters use the `num*` prefix.
+- Durations use the `*TimeNs` suffix and are displayed in the UI as formatted strings
+  (e.g. `1.2 s`, `350 ms`, `47 us`) rather than raw nanoseconds.
+
+| Metric | Type | Description |
+|---|---|---|
+| `numFragmentsScanned` | counter | Lance fragments actually opened by this task. Compare against the table fragment count to verify pruning is working. |
+| `numBatchesLoaded` | counter | Arrow batches returned from the JNI scanner. |
+| `numRowsScanned` | counter | Rows read from storage before filter evaluation. Pair with Spark's built-in `numOutputRows` to compute filter selectivity (`numOutputRows / numRowsScanned`). |
+| `datasetOpenTimeNs` | duration | Time spent in `Dataset.open(...)` — manifest load, namespace lookup, credential fetch. High values indicate catalog or metadata cache misses. |
+| `scannerCreateTimeNs` | duration | Time spent in `fragment.newScan(...)` — scan planning, predicate compilation, index lookup setup. |
+| `batchLoadTimeNs` | duration | Wall-clock time inside `loadNextBatch` — JNI crossing, IO, and Arrow IPC deserialization. Divide by `numBatchesLoaded` to get per-batch cost. |
+
+How to read these together:
+
+- **Pushdown verification**: a low `numRowsScanned / numOutputRows` ratio means filters are being
+  pushed down effectively. A 1:1 ratio means every row is being scanned and filtered in Spark —
+  check whether your predicate is supported for pushdown.
+- **Per-batch JNI cost**: `batchLoadTimeNs / numBatchesLoaded` gives the average cost of one batch
+  crossing JNI. If this is high relative to total query time, consider increasing
+  `LanceSparkReadOptions.batchSize` to amortize JNI overhead over more rows.
+- **Catalog overhead**: `datasetOpenTimeNs` accumulates per fragment opened. If many fragments are
+  opened per task, this can dominate; metadata cache size and namespace caching matter most here.
+
+!!!note
+      Additional metrics covering bytes read, fragments pruned, index lookups, and IO/decode
+      time breakdown require new APIs in `lance-jni` and will be added once that surface is
+      available upstream.
+
 ## Caching
 
 Lance Spark uses a multi-level caching strategy to minimize redundant I/O and improve query performance.

lance-spark-3.4_2.12/src/test/java/org/lance/spark/read/metric/LanceCustomMetricsTest.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.metric;
+
+public class LanceCustomMetricsTest extends BaseLanceCustomMetricsTest {}

lance-spark-3.4_2.12/src/test/java/org/lance/spark/read/metric/LanceReadMetricsIntegrationTest.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.metric;
+
+public class LanceReadMetricsIntegrationTest extends BaseLanceReadMetricsIntegrationTest {}

lance-spark-3.5_2.12/src/test/java/org/lance/spark/read/metric/LanceCustomMetricsTest.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.metric;
+
+public class LanceCustomMetricsTest extends BaseLanceCustomMetricsTest {}

lance-spark-3.5_2.12/src/test/java/org/lance/spark/read/metric/LanceReadMetricsIntegrationTest.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.metric;
+
+public class LanceReadMetricsIntegrationTest extends BaseLanceReadMetricsIntegrationTest {}

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

@@ -41,6 +41,7 @@ public class LanceFragmentColumnarBatchScanner implements AutoCloseable {
   private final LanceFragmentScanner fragmentScanner;
   private final ArrowReader arrowReader;
   private ColumnarBatch currentColumnarBatch;
+  private long lastBatchLoadTimeNs;
 
   public LanceFragmentColumnarBatchScanner(
       LanceFragmentScanner fragmentScanner, ArrowReader arrowReader) {
@@ -55,8 +56,13 @@ public static LanceFragmentColumnarBatchScanner create(
   }
 
   public boolean loadNextBatch() throws IOException {
-    if (arrowReader.loadNextBatch()) {
+    long start = System.nanoTime();
+    boolean hasNext = arrowReader.loadNextBatch();
+    lastBatchLoadTimeNs = System.nanoTime() - start;
+
+    if (hasNext) {
       VectorSchemaRoot root = arrowReader.getVectorSchemaRoot();
+
       List<ColumnVector> fieldVectors =
           root.getFieldVectors().stream()
               .map(LanceArrowColumnVector::new)
@@ -86,6 +92,18 @@ public ColumnarBatch getCurrentBatch() {
     return currentColumnarBatch;
   }
 
+  public long getLastBatchLoadTimeNs() {
+    return lastBatchLoadTimeNs;
+  }
+
+  public long getDatasetOpenTimeNs() {
+    return fragmentScanner.getDatasetOpenTimeNs();
+  }
+
+  public long getScannerCreateTimeNs() {
+    return fragmentScanner.getScannerCreateTimeNs();
+  }
+
   @Override
   public void close() throws IOException {
     try {

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

@@ -38,33 +38,42 @@ public class LanceFragmentScanner implements AutoCloseable {
   private final int fragmentId;
   private final boolean withFragemtId;
   private final LanceInputPartition inputPartition;
+  private final long datasetOpenTimeNs;
+  private final long scannerCreateTimeNs;
 
   private LanceFragmentScanner(
       Dataset dataset,
       LanceScanner scanner,
       int fragmentId,
       boolean withFragmentId,
-      LanceInputPartition inputPartition) {
+      LanceInputPartition inputPartition,
+      long datasetOpenTimeNs,
+      long scannerCreateTimeNs) {
     this.dataset = dataset;
     this.scanner = scanner;
     this.fragmentId = fragmentId;
     this.withFragemtId = withFragmentId;
     this.inputPartition = inputPartition;
+    this.datasetOpenTimeNs = datasetOpenTimeNs;
+    this.scannerCreateTimeNs = scannerCreateTimeNs;
   }
 
   public static LanceFragmentScanner create(int fragmentId, LanceInputPartition inputPartition) {
     Dataset dataset = null;
+    LanceScanner lanceScanner = null;
     try {
       LanceSparkReadOptions readOptions = inputPartition.getReadOptions();
       if (inputPartition.getNamespaceImpl() != null) {
         readOptions.setNamespace(
             LanceRuntime.getOrCreateNamespace(
                 inputPartition.getNamespaceImpl(), inputPartition.getNamespaceProperties()));
       }
+      long dsOpenStart = System.nanoTime();
       dataset =
           Utils.openDatasetBuilder(readOptions)
               .initialStorageOptions(inputPartition.getInitialStorageOptions())
               .build();
+      long dsOpenTimeNs = System.nanoTime() - dsOpenStart;
       Fragment fragment = dataset.getFragment(fragmentId);
       if (fragment == null) {
         throw new IllegalStateException(
@@ -109,13 +118,25 @@ public static LanceFragmentScanner create(int fragmentId, LanceInputPartition in
       }
       boolean withFragmentId =
           inputPartition.getSchema().getFieldIndex(LanceConstant.FRAGMENT_ID).nonEmpty();
+      long scanCreateStart = System.nanoTime();
+      lanceScanner = fragment.newScan(scanOptions.build());
+      long scanCreateTimeNs = System.nanoTime() - scanCreateStart;
       return new LanceFragmentScanner(
           dataset,
-          fragment.newScan(scanOptions.build()),
+          lanceScanner,
           fragmentId,
           withFragmentId,
-          inputPartition);
+          inputPartition,
+          dsOpenTimeNs,
+          scanCreateTimeNs);
     } catch (Throwable throwable) {
+      if (lanceScanner != null) {
+        try {
+          lanceScanner.close();
+        } catch (Throwable closeError) {
+          throwable.addSuppressed(closeError);
+        }
+      }
       if (dataset != null) {
         try {
           dataset.close();
@@ -181,6 +202,14 @@ public LanceInputPartition getInputPartition() {
     return inputPartition;
   }
 
+  public long getDatasetOpenTimeNs() {
+    return datasetOpenTimeNs;
+  }
+
+  public long getScannerCreateTimeNs() {
+    return scannerCreateTimeNs;
+  }
+
   /**
    * Builds the projection column list for the scanner. Regular data columns come first, followed by
    * special metadata columns in the order matching {@link

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

@@ -14,7 +14,9 @@
 package org.lance.spark.read;
 
 import org.lance.spark.internal.LanceFragmentColumnarBatchScanner;
+import org.lance.spark.read.metric.LanceReadMetricsTracker;
 
+import org.apache.spark.sql.connector.metric.CustomTaskMetric;
 import org.apache.spark.sql.connector.read.PartitionReader;
 import org.apache.spark.sql.vectorized.ColumnarBatch;
 
@@ -25,6 +27,7 @@ public class LanceColumnarPartitionReader implements PartitionReader<ColumnarBat
   private int fragmentIndex;
   private LanceFragmentColumnarBatchScanner fragmentReader;
   private ColumnarBatch currentBatch;
+  private final LanceReadMetricsTracker metricsTracker = new LanceReadMetricsTracker();
 
   public LanceColumnarPartitionReader(LanceInputPartition inputPartition) {
     this.inputPartition = inputPartition;
@@ -44,6 +47,9 @@ public boolean next() throws IOException {
           LanceFragmentColumnarBatchScanner.create(
               inputPartition.getLanceSplit().getFragments().get(fragmentIndex), inputPartition);
       fragmentIndex++;
+      metricsTracker.addNumFragmentsScanned(1);
+      metricsTracker.addDatasetOpenTimeNs(fragmentReader.getDatasetOpenTimeNs());
+      metricsTracker.addScannerCreateTimeNs(fragmentReader.getScannerCreateTimeNs());
       if (loadNextBatchFromCurrentReader()) {
         return true;
       }
@@ -52,8 +58,14 @@ public boolean next() throws IOException {
   }
 
   private boolean loadNextBatchFromCurrentReader() throws IOException {
-    if (fragmentReader != null && fragmentReader.loadNextBatch()) {
+    if (fragmentReader == null) {
+      return false;
+    }
+    if (fragmentReader.loadNextBatch()) {
       currentBatch = fragmentReader.getCurrentBatch();
+      metricsTracker.addNumBatchesLoaded(1);
+      metricsTracker.addNumRowsScanned(currentBatch.numRows());
+      metricsTracker.addBatchLoadTimeNs(fragmentReader.getLastBatchLoadTimeNs());
       return true;
     }
     return false;
@@ -64,6 +76,11 @@ public ColumnarBatch get() {
     return currentBatch;
   }
 
+  @Override
+  public CustomTaskMetric[] currentMetricsValues() {
+    return metricsTracker.currentMetricsValues();
+  }
+
   @Override
   public void close() throws IOException {
     if (fragmentReader != null) {

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

@@ -18,6 +18,7 @@
 import org.lance.ipc.ScanOptions;
 import org.lance.spark.LanceRuntime;
 import org.lance.spark.LanceSparkReadOptions;
+import org.lance.spark.read.metric.LanceReadMetricsTracker;
 import org.lance.spark.utils.Utils;
 import org.lance.spark.vectorized.LanceArrowColumnVector;
 
@@ -26,6 +27,7 @@
 import org.apache.arrow.vector.BigIntVector;
 import org.apache.arrow.vector.VectorSchemaRoot;
 import org.apache.arrow.vector.ipc.ArrowReader;
+import org.apache.spark.sql.connector.metric.CustomTaskMetric;
 import org.apache.spark.sql.connector.read.PartitionReader;
 import org.apache.spark.sql.types.StructType;
 import org.apache.spark.sql.util.LanceArrowUtils;
@@ -43,6 +45,7 @@ public class LanceCountStarPartitionReader implements PartitionReader<ColumnarBa
   private final BufferAllocator allocator;
   private boolean finished = false;
   private ColumnarBatch currentBatch;
+  private final LanceReadMetricsTracker metricsTracker = new LanceReadMetricsTracker();
 
   public LanceCountStarPartitionReader(LanceInputPartition inputPartition) {
     this.inputPartition = inputPartition;
@@ -64,14 +67,18 @@ private long computeCount() {
     LanceSparkReadOptions readOptions = inputPartition.getReadOptions();
     long totalCount = 0;
 
+    long dsOpenStart = System.nanoTime();
     try (Dataset dataset =
         Utils.openDatasetBuilder(readOptions)
             .initialStorageOptions(inputPartition.getInitialStorageOptions())
             .build()) {
+      metricsTracker.addDatasetOpenTimeNs(System.nanoTime() - dsOpenStart);
+
       List<Integer> fragmentIds = inputPartition.getLanceSplit().getFragments();
       if (fragmentIds.isEmpty()) {
         return 0;
       }
+      metricsTracker.addNumFragmentsScanned(fragmentIds.size());
 
       ScanOptions.Builder scanOptionsBuilder = new ScanOptions.Builder();
       if (inputPartition.getWhereCondition().isPresent()) {
@@ -80,10 +87,23 @@ private long computeCount() {
       scanOptionsBuilder.withRowId(true);
       scanOptionsBuilder.columns(Lists.newArrayList());
       scanOptionsBuilder.fragmentIds(fragmentIds);
+
+      long scanCreateStart = System.nanoTime();
       try (LanceScanner scanner = dataset.newScan(scanOptionsBuilder.build())) {
+        metricsTracker.addScannerCreateTimeNs(System.nanoTime() - scanCreateStart);
         try (ArrowReader reader = scanner.scanBatches()) {
-          while (reader.loadNextBatch()) {
-            totalCount += reader.getVectorSchemaRoot().getRowCount();
+          while (true) {
+            long batchStart = System.nanoTime();
+            boolean hasNext = reader.loadNextBatch();
+            long batchTimeNs = System.nanoTime() - batchStart;
+            if (!hasNext) {
+              break;
+            }
+            metricsTracker.addBatchLoadTimeNs(batchTimeNs);
+            long rowCount = reader.getVectorSchemaRoot().getRowCount();
+            totalCount += rowCount;
+            metricsTracker.addNumBatchesLoaded(1);
+            metricsTracker.addNumRowsScanned(rowCount);
           }
         }
       } catch (Exception e) {
@@ -118,13 +138,20 @@ private ColumnarBatch createCountResultBatch(long count, StructType resultSchema
 
   @Override
   public ColumnarBatch get() {
-    long rowCount = computeCount();
-    StructType countSchema =
-        new StructType().add("count", org.apache.spark.sql.types.DataTypes.LongType);
-    currentBatch = createCountResultBatch(rowCount, countSchema);
+    if (currentBatch == null) {
+      long rowCount = computeCount();
+      StructType countSchema =
+          new StructType().add("count", org.apache.spark.sql.types.DataTypes.LongType);
+      currentBatch = createCountResultBatch(rowCount, countSchema);
+    }
     return currentBatch;
   }
 
+  @Override
+  public CustomTaskMetric[] currentMetricsValues() {
+    return metricsTracker.currentMetricsValues();
+  }
+
   @Override
   public void close() throws IOException {
     if (currentBatch != null) {