diff --git a/lance-spark-base_2.12/src/main/java/org/lance/spark/internal/LanceFragmentScanner.java b/lance-spark-base_2.12/src/main/java/org/lance/spark/internal/LanceFragmentScanner.java
index a235a1171..025b63c19 100644
--- a/lance-spark-base_2.12/src/main/java/org/lance/spark/internal/LanceFragmentScanner.java
+++ b/lance-spark-base_2.12/src/main/java/org/lance/spark/internal/LanceFragmentScanner.java
@@ -19,7 +19,6 @@
 import org.lance.ipc.ScanOptions;
 import org.lance.ipc.ScanStats;
 import org.lance.spark.LanceConstant;
-import org.lance.spark.LanceRuntime;
 import org.lance.spark.LanceSparkReadOptions;
 import org.lance.spark.read.LanceInputPartition;
 import org.lance.spark.utils.BlobUtils;
@@ -81,15 +80,6 @@ public static LanceFragmentScanner create(int fragmentId, LanceInputPartition in
     LanceScanner lanceScanner = null;
     try {
       LanceSparkReadOptions readOptions = inputPartition.getReadOptions();
-      if (inputPartition.getNamespaceImpl() != null && readOptions.isExecutorCredentialRefresh()) {
-        if (LanceRuntime.useNamespaceOnWorkers(inputPartition.getNamespaceImpl())) {
-          readOptions.setNamespace(
-              LanceRuntime.getOrCreateNamespace(
-                  inputPartition.getNamespaceImpl(), inputPartition.getNamespaceProperties()));
-        } else {
-          readOptions.setNamespace(null);
-        }
-      }
       long dsOpenStart = System.nanoTime();
       dataset =
           Utils.openDatasetBuilder(readOptions)
diff --git a/lance-spark-base_2.12/src/main/java/org/lance/spark/read/LanceColumnarPartitionReader.java b/lance-spark-base_2.12/src/main/java/org/lance/spark/read/LanceColumnarPartitionReader.java
index 5bf36f1f8..64537db40 100644
--- a/lance-spark-base_2.12/src/main/java/org/lance/spark/read/LanceColumnarPartitionReader.java
+++ b/lance-spark-base_2.12/src/main/java/org/lance/spark/read/LanceColumnarPartitionReader.java
@@ -13,6 +13,9 @@
  */
 package org.lance.spark.read;
 
+import org.lance.namespace.LanceNamespace;
+import org.lance.spark.LanceRuntime;
+import org.lance.spark.LanceSparkReadOptions;
 import org.lance.spark.internal.LanceFragmentColumnarBatchScanner;
 import org.lance.spark.read.metric.LanceReadMetricsTracker;
 
@@ -29,6 +32,8 @@ public class LanceColumnarPartitionReader implements PartitionReader<ColumnarBat
   private ColumnarBatch currentBatch;
   private final LanceReadMetricsTracker metricsTracker = new LanceReadMetricsTracker();
   private boolean currentScanStatsAdded = false;
+  private boolean executorNamespaceInitialized = false;
+  private LanceNamespace executorNamespace;
 
   public LanceColumnarPartitionReader(LanceInputPartition inputPartition) {
     this.inputPartition = inputPartition;
@@ -37,6 +42,14 @@ public LanceColumnarPartitionReader(LanceInputPartition inputPartition) {
 
   @Override
   public boolean next() throws IOException {
+    try {
+      return nextInternal();
+    } catch (Throwable t) {
+      throw asIOException(closeResources(t));
+    }
+  }
+
+  private boolean nextInternal() throws IOException {
     if (loadNextBatchFromCurrentReader()) {
       return true;
     }
@@ -49,6 +62,7 @@ public boolean next() throws IOException {
         fragmentReader = null;
         toClose.close();
       }
+      initializeExecutorNamespace();
       fragmentReader =
           LanceFragmentColumnarBatchScanner.create(
               inputPartition.getLanceSplit().getFragments().get(fragmentIndex), inputPartition);
@@ -66,6 +80,26 @@ public boolean next() throws IOException {
     return false;
   }
 
+  private void initializeExecutorNamespace() {
+    if (executorNamespaceInitialized) {
+      return;
+    }
+    executorNamespaceInitialized = true;
+
+    LanceSparkReadOptions readOptions = inputPartition.getReadOptions();
+    String namespaceImpl = inputPartition.getNamespaceImpl();
+    if (namespaceImpl == null || !readOptions.isExecutorCredentialRefresh()) {
+      return;
+    }
+    if (LanceRuntime.useNamespaceOnWorkers(namespaceImpl)) {
+      executorNamespace =
+          LanceRuntime.getOrCreateNamespace(namespaceImpl, inputPartition.getNamespaceProperties());
+      readOptions.setNamespace(executorNamespace);
+    } else {
+      readOptions.setNamespace(null);
+    }
+  }
+
   private boolean loadNextBatchFromCurrentReader() throws IOException {
     if (fragmentReader == null) {
       return false;
@@ -98,22 +132,64 @@ public CustomTaskMetric[] currentMetricsValues() {
 
   @Override
   public void close() throws IOException {
-    if (fragmentReader == null) {
-      return;
+    Throwable failure = null;
+    if (fragmentReader != null && !currentScanStatsAdded) {
+      try {
+        metricsTracker.addScanStats(fragmentReader.getScanStats());
+        currentScanStatsAdded = true;
+      } catch (Throwable t) {
+        failure = t;
+      }
     }
-    if (!currentScanStatsAdded) {
-      metricsTracker.addScanStats(fragmentReader.getScanStats());
-      currentScanStatsAdded = true;
+    failure = closeResources(failure);
+    if (failure != null) {
+      throw asIOException(failure);
     }
-    // Null-first so close() is idempotent (PartitionReader extends Closeable, whose contract
-    // requires it): a repeat call short-circuits rather than raising
-    // `ArrowArrayStream is already closed` from a second ArrowArrayStream.release().
-    LanceFragmentColumnarBatchScanner toClose = fragmentReader;
+  }
+
+  private Throwable closeResources(Throwable primary) {
+    LanceFragmentColumnarBatchScanner scannerToClose = fragmentReader;
     fragmentReader = null;
+    LanceNamespace namespaceToClose = executorNamespace;
+    executorNamespace = null;
+
+    if (namespaceToClose != null
+        && inputPartition.getReadOptions().getNamespace() == namespaceToClose) {
+      inputPartition.getReadOptions().setNamespace(null);
+    }
+
+    primary = closeResource(scannerToClose, primary);
+    if (namespaceToClose instanceof AutoCloseable) {
+      primary = closeResource((AutoCloseable) namespaceToClose, primary);
+    }
+    return primary;
+  }
+
+  private static Throwable closeResource(AutoCloseable resource, Throwable primary) {
+    if (resource == null) {
+      return primary;
+    }
     try {
-      toClose.close();
-    } catch (Exception e) {
-      throw new IOException(e);
+      resource.close();
+    } catch (Throwable closeError) {
+      if (primary == null) {
+        return closeError;
+      }
+      primary.addSuppressed(closeError);
+    }
+    return primary;
+  }
+
+  private static IOException asIOException(Throwable failure) {
+    if (failure instanceof IOException) {
+      return (IOException) failure;
+    }
+    if (failure instanceof RuntimeException) {
+      throw (RuntimeException) failure;
+    }
+    if (failure instanceof Error) {
+      throw (Error) failure;
     }
+    return new IOException(failure);
   }
 }
diff --git a/lance-spark-base_2.12/src/test/java/org/lance/spark/internal/LanceFragmentScannerTest.java b/lance-spark-base_2.12/src/test/java/org/lance/spark/internal/LanceFragmentScannerTest.java
index e0230f994..ffa06c7e8 100644
--- a/lance-spark-base_2.12/src/test/java/org/lance/spark/internal/LanceFragmentScannerTest.java
+++ b/lance-spark-base_2.12/src/test/java/org/lance/spark/internal/LanceFragmentScannerTest.java
@@ -14,9 +14,14 @@
 package org.lance.spark.internal;
 
 import org.lance.namespace.LanceNamespace;
+import org.lance.namespace.model.DescribeTableRequest;
+import org.lance.namespace.model.DescribeTableResponse;
 import org.lance.spark.LanceConstant;
 import org.lance.spark.LanceSparkReadOptions;
+import org.lance.spark.TestUtils;
+import org.lance.spark.read.LanceColumnarPartitionReader;
 import org.lance.spark.read.LanceInputPartition;
+import org.lance.spark.read.LanceSplit;
 import org.lance.spark.utils.BlobUtils;
 import org.lance.spark.utils.Optional;
 
@@ -25,6 +30,7 @@
 import org.apache.spark.sql.types.MetadataBuilder;
 import org.apache.spark.sql.types.StructField;
 import org.apache.spark.sql.types.StructType;
+import org.apache.spark.sql.vectorized.ColumnarBatch;
 import org.junit.jupiter.api.Test;
 
 import java.lang.reflect.InvocationTargetException;
@@ -36,6 +42,7 @@
 import java.util.concurrent.atomic.AtomicInteger;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
 import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 
@@ -224,49 +231,29 @@ public void testGetColumnNamesWithFragmentId() throws Exception {
 
   /**
    * Locks down the executor-branch contract for {@code executor_credential_refresh=false}: when an
-   * executor opens a fragment for a namespace-backed table with the flag disabled, {@link
-   * LanceFragmentScanner#create} must <i>not</i> reconstruct the namespace client. Without this
-   * gate, executors of Kerberized HMS catalogs hit {@code GSS initiate failed} because they lack a
-   * TGT for the eager {@code describeTable()} RPC.
-   *
-   * <p>Strategy: hand a real, loadable {@link LanceNamespace} impl to the partition. If the gate
-   * regresses (rebuild not skipped), {@code LanceNamespace.connect} would succeed via {@code
-   * Class.forName}, {@link RecordingNamespace#initialize} would run, and {@code
-   * readOptions.setNamespace(...)} would fire — all observable here. The bogus dataset URI lets the
-   * outer {@code Utils.openDatasetBuilder().build()} call fail predictably, since the gate runs
-   * <i>before</i> the dataset is opened. No real Lance dataset is required.
+   * executor opens a namespace-backed table with the flag disabled, the partition reader must
+   * <i>not</i> reconstruct the namespace client. Without this gate, executors of Kerberized HMS
+   * catalogs hit {@code GSS initiate failed} because they lack a TGT for the eager {@code
+   * describeTable()} RPC.
    */
   @Test
-  public void testCreateSkipsNamespaceRebuildWhenExecutorCredentialRefreshDisabled() {
-    RecordingNamespace.INITIALIZE_CALLS.set(0);
-
-    LanceSparkReadOptions readOptions =
-        LanceSparkReadOptions.builder()
-            .datasetUri("file:///tmp/__lance_nonexistent_for_executor_gate_test__")
-            .executorCredentialRefresh(false)
-            .build();
-
+  public void testPartitionReaderSkipsNamespaceWhenExecutorCredentialRefreshDisabled()
+      throws Exception {
+    RecordingNamespace.reset();
     LanceInputPartition partition =
-        new LanceInputPartition(
-            new StructType(),
-            0,
-            null,
-            readOptions,
-            Optional.empty(),
-            Optional.empty(),
-            Optional.empty(),
-            Optional.empty(),
-            Optional.empty(),
-            "test-scan",
-            Collections.emptyMap(),
-            RecordingNamespace.class.getName(),
-            Collections.emptyMap(),
-            null);
-
-    assertThrows(RuntimeException.class, () -> LanceFragmentScanner.create(0, partition));
+        namespacePartition(
+            TestUtils.TestTable1Config.datasetUri,
+            Collections.singletonList(0),
+            "testPartitionReaderSkipsNamespaceWhenExecutorCredentialRefreshDisabled",
+            false);
+    try (LanceColumnarPartitionReader reader = new LanceColumnarPartitionReader(partition)) {
+      while (reader.next()) {
+        reader.get().close();
+      }
+    }
 
     assertNull(
-        readOptions.getNamespace(),
+        partition.getReadOptions().getNamespace(),
         "executor_credential_refresh=false must skip namespace rebuild on the executor");
     assertEquals(
         0,
@@ -274,6 +261,92 @@ public void testCreateSkipsNamespaceRebuildWhenExecutorCredentialRefreshDisabled
         "executor_credential_refresh=false must not load or initialize the namespace impl");
   }
 
+  @Test
+  public void testPartitionReaderReusesAndClosesExecutorNamespace() throws Exception {
+    RecordingNamespace.reset();
+    LanceInputPartition partition =
+        namespacePartition(
+            TestUtils.TestTable1Config.datasetUri,
+            Arrays.asList(0, 1),
+            "testPartitionReaderReusesAndClosesExecutorNamespace",
+            true);
+
+    LanceColumnarPartitionReader reader = new LanceColumnarPartitionReader(partition);
+    int rowsRead = 0;
+    try {
+      while (reader.next()) {
+        ColumnarBatch batch = reader.get();
+        assertNotNull(batch);
+        rowsRead += batch.numRows();
+        batch.close();
+      }
+    } finally {
+      reader.close();
+    }
+    reader.close();
+
+    assertEquals(TestUtils.TestTable1Config.expectedValues.size(), rowsRead);
+    assertEquals(
+        1,
+        RecordingNamespace.INITIALIZE_CALLS.get(),
+        "all fragments in one Spark task must reuse one namespace client");
+    assertEquals(
+        1,
+        RecordingNamespace.CLOSE_CALLS.get(),
+        "the task-owned namespace client must be closed exactly once");
+  }
+
+  @Test
+  public void testPartitionReaderClosesExecutorNamespaceWhenFragmentOpenFails() throws Exception {
+    RecordingNamespace.reset();
+    LanceInputPartition partition =
+        namespacePartition(
+            TestUtils.TestTable1Config.datasetUri,
+            Collections.singletonList(999999),
+            "testPartitionReaderClosesExecutorNamespaceWhenFragmentOpenFails",
+            true);
+    LanceColumnarPartitionReader reader = new LanceColumnarPartitionReader(partition);
+
+    RuntimeException failure = assertThrows(RuntimeException.class, reader::next);
+
+    assertNotNull(failure.getCause());
+    assertEquals(1, RecordingNamespace.INITIALIZE_CALLS.get());
+    assertEquals(
+        1,
+        RecordingNamespace.CLOSE_CALLS.get(),
+        "namespace initialization must be rolled back when fragment open fails");
+    reader.close();
+    assertEquals(
+        1,
+        RecordingNamespace.CLOSE_CALLS.get(),
+        "close after failed initialization cleanup must be idempotent");
+  }
+
+  private LanceInputPartition namespacePartition(
+      String location, List<Integer> fragments, String scanId, boolean executorCredentialRefresh) {
+    LanceSparkReadOptions readOptions =
+        LanceSparkReadOptions.builder()
+            .datasetUri(location)
+            .tableId(Collections.singletonList(TestUtils.TestTable1Config.datasetName))
+            .executorCredentialRefresh(executorCredentialRefresh)
+            .build();
+    return new LanceInputPartition(
+        TestUtils.TestTable1Config.schema,
+        0,
+        new LanceSplit(fragments),
+        readOptions,
+        Optional.empty(),
+        Optional.empty(),
+        Optional.empty(),
+        Optional.empty(),
+        Optional.empty(),
+        scanId,
+        Collections.emptyMap(),
+        RecordingNamespace.class.getName(),
+        Collections.singletonMap("location", location),
+        null);
+  }
+
   @Test
   public void getBlobColumnNamesIncludesBlobV2ReadColumns() throws Exception {
     Method method =
@@ -302,21 +375,40 @@ public void getBlobColumnNamesIncludesBlobV2ReadColumns() throws Exception {
   /**
    * Public, top-level-by-FQCN, no-arg {@link LanceNamespace} so that {@link
    * LanceNamespace#connect(String, Map, BufferAllocator)} can resolve it via {@code Class.forName}
-   * if the executor branch is (incorrectly) taken.
+   * during executor-side namespace initialization.
    */
-  public static class RecordingNamespace implements LanceNamespace {
+  public static class RecordingNamespace implements LanceNamespace, AutoCloseable {
     static final AtomicInteger INITIALIZE_CALLS = new AtomicInteger();
+    static final AtomicInteger CLOSE_CALLS = new AtomicInteger();
+
+    private String location;
 
     public RecordingNamespace() {}
 
+    static void reset() {
+      INITIALIZE_CALLS.set(0);
+      CLOSE_CALLS.set(0);
+    }
+
     @Override
     public void initialize(Map<String, String> properties, BufferAllocator allocator) {
       INITIALIZE_CALLS.incrementAndGet();
+      location = properties.get("location");
     }
 
     @Override
     public String namespaceId() {
       return "recording";
     }
+
+    @Override
+    public DescribeTableResponse describeTable(DescribeTableRequest request) {
+      return new DescribeTableResponse().location(location);
+    }
+
+    @Override
+    public void close() {
+      CLOSE_CALLS.incrementAndGet();
+    }
   }
 }