fix: cache executor namespace metadata across fragments

[charleshuang119]

Summary

Reuse Lance Namespace clients and describeTable metadata across fragment tasks belonging to the same Spark scan and executor JVM.

This PR is intentionally limited to the catalog layer. It no longer retains a Dataset as an object-store anchor. Azure object-store, HTTP-client, and token-provider reuse require a separate fix in lance-format/lance.

Production impact and reproduction

Lance Spark creates approximately one input task per Lance fragment. A controlled Gravitino-backed read with 368 fragments and one 4-core executor generated 11,042 DNS queries in 29 seconds, almost exactly 30 per fragment:

• ~10 queries to Gravitino
• ~10 queries to login.microsoftonline.com
• ~10 queries to Azure Blob Storage

At production fragment counts, even a small number of Spark pods can overload cluster DNS. Changing the same workload to direct ABFSS removed the storm.

The three groups have different causes. This PR addresses the Gravitino group: before this change, every fragment task rebuilt the worker-side namespace client and called describeTable. Azure identity and Blob amplification happen below Spark in Lance object-store construction and lifetime management and are explicitly out of scope here.

Root cause

LanceFragmentScanner.create opens a Dataset for each fragment. When executor_credential_refresh=true, it also previously rebuilt the namespace client for each fragment. Opening the namespace-backed Dataset calls describeTable, so namespace initialization, Gravitino requests, and associated DNS traffic scaled linearly with fragment count.

Filtered COUNT(*) uses LanceCountStarPartitionReader instead of LanceFragmentScanner and had the same worker-side namespace behavior, so it must participate in the same cache.

Solution

Add a bounded, scan-scoped executor namespace cache:

• Cache namespace clients by (namespace implementation, namespace properties, scanId) in each executor JVM.
• Share the namespace and describeTable response across both regular fragment readers and filtered COUNT(*) readers.
• Coalesce concurrent identical describeTable calls with a FutureTask single-flight.
• Keep separate entries for separate Spark scans so locations and credentials cannot leak into later queries that reuse the executor JVM.
• Give each scan-scoped wrapper a distinct namespaceId, preventing Lance from reusing an earlier scan’s object-store provider after that scan’s namespace is evicted and closed.
• Refresh temporary namespace-vended credentials before expires_at_millis using a 10% safety window bounded between 1 and 60 seconds.
• Cache responses without storage options for five minutes. Use a conservative one-minute TTL when non-empty storage options omit a valid expiry.
• Bound the executor cache to 1,000 scans and each description cache to 1,000 requests; expire idle entries after one hour.
• Use reference-counted leases so eviction never closes a namespace still used by an active task.
• Delegate listTableVersions and describeTableVersion to preserve managed-versioning reads.
• Leave executor_credential_refresh=false behavior unchanged.

Memory and lifecycle safety

An earlier revision pinned one full Dataset per executor and scan to keep the Lance object store alive. That approach was removed after review because a Dataset retains manifest and fragment metadata, making an hour-long cache with up to 1,000 scans unsafe for large tables. It also created ambiguous namespace/credential-provider ownership across scans.

The final implementation caches only namespace clients and small metadata responses. Scan-scoped provider identity also ensures that object stores cannot outlive and call a namespace delegate owned by another scan. It does not cache Datasets, fragment readers, object stores, or Azure credential providers, and it does not enumerate fragments into an executor-side map.

Expected effect and explicit non-goals

Expected for regular fragment scans and filtered COUNT(*):

• Gravitino namespace initialization and describeTable: from O(fragments) toward O(executor JVMs × scans × metadata refresh windows).

Not solved by this PR:

• login.microsoftonline.com token-acquisition DNS amplification.
• Azure Blob DNS amplification caused by repeatedly creating object stores/HTTP clients.

Those require bounded strong object-store reuse and creation single-flight in lance-format/lance. The original 368-fragment DNS capture should be rerun after both changes; this PR alone should only be credited for reducing the Gravitino portion.

Test plan

Targeted tests:

./mvnw test -pl lance-spark-3.5_2.13 \
  -Dtest=ExecutorNamespaceCacheTest,LanceFragmentScannerTest,LanceCountStarPartitionReaderTest

Result: 24 tests run, 0 failures, 0 errors.

Coverage verifies:

• Separately constructed, value-equivalent DescribeTableRequest instances share one cached response.
• Unexpected namespace operations retain the interface unsupported defaults and never reach the delegate.
• Different scans expose different provider identities, preventing cross-scan reuse of an evictable namespace client.
• Eight concurrent describeTable calls collapse to one backing request.
• Credentials refresh before expires_at_millis.
• Failed refreshes remain retryable.
• Expired credentials do not create an internal spin loop.
• Fragment tasks in one scan initialize and describe once.
• Different scan IDs do not share table descriptions.
• Filtered COUNT(*) tasks reuse the same namespace metadata cache.
• Managed-versioning read APIs are delegated.
• Eviction waits for active leases before closing the namespace.
• executor_credential_refresh=false remains unchanged.

A full local Spark 3.5 / Scala 2.13 run executed 1,240 tests. The changed and targeted tests passed; two unrelated index tests hit the local Lance DataFusion 150 MB external-sort memory limit. The previous full run before adding the isolated filtered-count coverage passed, and GitHub CI remains the authoritative clean-environment run.

References

• Dataset cache removal: refactor: remove Java-side dataset cache, rely on Rust-side Session #353
• Executor credential refresh option: fix(spark): gss initiate failed on hms executors; spark.sql.catalog read options not applied #476
• Lance Namespace expiry contract: https://github.com/lance-format/lance-namespace/blob/main/docs/src/spec.yaml