diff --git a/CHANGELOG.md b/CHANGELOG.md
index 066ecbd41..97932ce90 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -13,6 +13,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Fixed
 
+- Background table-tab eviction now releases unpinned row data without dropping query, pinned, edited or in-flight results.
+- The result display cache now enforces its memory budget when a cached row's formatted values grow.
+- Undo, redo, a theme change and a display-format change no longer leave the data grid reformatting every cell as you scroll.
+- Autocomplete no longer bulk-loads every column of a schema too big to cache, so on those databases column suggestions arrive per table as you reference one.
+- A background tab whose column metadata arrived after its rows were freed now reloads on return instead of showing an empty grid.
 - A saved connection whose SSH settings were written before the agent socket field existed now loads instead of disappearing.
 - Switch Connection and Open Database now open on a narrow window, and after you remove their toolbar button, instead of doing nothing at all.
 - A large text value in the row inspector now scrolls in a resizable text view instead of being clipped, and stays selectable and copyable when the row is read-only.
diff --git a/TablePro/Core/Autocomplete/SQLSchemaProvider.swift b/TablePro/Core/Autocomplete/SQLSchemaProvider.swift
index 6506fb7a7..dcf5e93c1 100644
--- a/TablePro/Core/Autocomplete/SQLSchemaProvider.swift
+++ b/TablePro/Core/Autocomplete/SQLSchemaProvider.swift
@@ -12,18 +12,25 @@ import TableProPluginKit
 /// Provides cached database schema information for autocomplete
 actor SQLSchemaProvider {
     private static let logger = Logger(subsystem: "com.TablePro", category: "SQLSchemaProvider")
+
+    /// How many tables' columns the cache holds, and therefore how large a schema is worth
+    /// preloading in one bulk fetch. The two are one number on purpose: fetching every column of a
+    /// schema and then keeping a fraction of them is the waste the eager load exists to avoid.
+    /// Columns are small next to row data, so this sits well above the point where a bulk fetch is
+    /// still one cheap query.
+    static let maxCachedTables = 300
     // MARK: - Properties
 
     private var tables: [TableInfo] = []
     private var columnCache: [String: [ColumnInfo]] = [:]
     private var columnAccessOrder: [String] = []
-    private let maxCachedTables = 50
     private var isLoading = false
     private var lastLoadError: Error?
     private var lastRetryAttempt: Date?
     private let retryCooldown: TimeInterval = 30
     private var loadTask: Task<Void, Never>?
     private var eagerColumnTask: Task<Void, Never>?
+    private var eagerLoadSchema: String?
 
     struct ColumnMetadataSource: Sendable {
         let fetchColumns: @Sendable (_ table: String, _ schema: String?) async throws -> [ColumnInfo]
@@ -75,6 +82,7 @@ actor SQLSchemaProvider {
         Self.logger.info("[schema] loadSchema starting new fetch")
         let t0 = Date()
         self.cachedDriver = driver
+        self.eagerLoadSchema = (driver as? SchemaSwitchable)?.currentSchema
         if let connection { self.connectionInfo = connection }
         isLoading = true
         lastLoadError = nil
@@ -146,7 +154,7 @@ actor SQLSchemaProvider {
     }
 
     private func evictIfNeeded() {
-        while columnAccessOrder.count > maxCachedTables {
+        while columnAccessOrder.count > Self.maxCachedTables {
             let evicted = columnAccessOrder.removeFirst()
             columnCache.removeValue(forKey: evicted)
         }
@@ -176,19 +184,21 @@ actor SQLSchemaProvider {
     }
 
     func resetForDatabase(_ database: String?, tables newTables: [TableInfo], driver: DatabaseDriver) {
-        eagerColumnTask?.cancel()
-        eagerColumnTask = nil
         self.tables = newTables
         self.columnCache.removeAll()
         self.columnAccessOrder.removeAll()
         self.fieldPathCache.removeAll()
         self.fieldPathTasks.removeAll()
         self.cachedDriver = driver
+        self.eagerLoadSchema = (driver as? SchemaSwitchable)?.currentSchema
         self.isLoading = false
         self.lastLoadError = nil
         startEagerColumnLoad()
     }
 
+    /// Empties the cache without refilling it. The refresh signal that reaches here also runs a
+    /// schema reload, and that ends in `resetForDatabase`, which starts the preload; restarting it
+    /// here as well sends a second whole-schema column query for every refresh.
     func clearColumnCache() {
         eagerColumnTask?.cancel()
         eagerColumnTask = nil
@@ -196,20 +206,40 @@ actor SQLSchemaProvider {
         columnAccessOrder.removeAll()
         fieldPathCache.removeAll()
         fieldPathTasks.removeAll()
-        if cachedDriver != nil {
-            startEagerColumnLoad()
-        }
     }
 
     // MARK: - Eager Column Loading
 
+    /// How many tables the bulk fetch will actually return.
+    ///
+    /// `tables` is the union of the current schema and every other schema the sidebar has expanded,
+    /// while `fetchAllColumns()` covers one schema. Counting the union turned the preload off for a
+    /// ten-table `public` as soon as eight other schemas were open. A table the list does not
+    /// attribute to any schema counts, which is what a flat engine reports for all of them; zero
+    /// means the list describes other schemas only, and a fetch sized by it would be a guess.
+    private var eagerLoadTableCount: Int {
+        guard let eagerLoadSchema else { return tables.count }
+        return tables.filter { table in
+            guard let tableSchema = table.schema else { return true }
+            return tableSchema.caseInsensitiveCompare(eagerLoadSchema) == .orderedSame
+        }.count
+    }
+
     private func startEagerColumnLoad() {
-        guard !tables.isEmpty else { return }
+        eagerColumnTask?.cancel()
+        eagerColumnTask = nil
+
+        let tableCount = eagerLoadTableCount
+        guard tableCount > 0 else { return }
+        guard tableCount <= Self.maxCachedTables else {
+            Self.logger.info(
+                "[schema] eager column load skipped tableCount=\(tableCount) limit=\(Self.maxCachedTables)"
+            )
+            return
+        }
         let source = metadataSource
         let driver = cachedDriver
         guard source != nil || driver != nil else { return }
-        eagerColumnTask?.cancel()
-        let tableCount = tables.count
         eagerColumnTask = Task(priority: .utility) {
             Self.logger.info("[schema] eager column load starting tableCount=\(tableCount)")
             do {
@@ -231,14 +261,35 @@ actor SQLSchemaProvider {
         }
     }
 
+    /// Fills the cache in the order the schema lists its tables, so which tables survive the cache
+    /// limit is the same on every run. Walking the fetched dictionary took whatever order hashing
+    /// produced, which made the cached set differ between two loads of the same database.
     private func populateColumnCache(_ allColumns: [String: [ColumnInfo]]) {
+        var pending: [String: [ColumnInfo]] = [:]
+        pending.reserveCapacity(allColumns.count)
         for (tableName, columns) in allColumns {
-            let key = tableName.lowercased()
-            guard columnCache[key] == nil else { continue }
-            guard columnAccessOrder.count < maxCachedTables else { break }
-            columnCache[key] = columns
-            columnAccessOrder.append(key)
+            pending[tableName.lowercased()] = columns
+        }
+
+        for table in tables {
+            guard let columns = pending.removeValue(forKey: table.name.lowercased()) else { continue }
+            insertIntoColumnCache(columns, forKey: table.name.lowercased())
         }
+        for key in pending.keys.sorted() {
+            guard let columns = pending[key] else { continue }
+            insertIntoColumnCache(columns, forKey: key)
+        }
+    }
+
+    private func insertIntoColumnCache(_ columns: [ColumnInfo], forKey key: String) {
+        guard columnCache[key] == nil else { return }
+        guard columnAccessOrder.count < Self.maxCachedTables else { return }
+        columnCache[key] = columns
+        columnAccessOrder.append(key)
+    }
+
+    func waitForEagerColumnLoad() async {
+        await eagerColumnTask?.value
     }
 
     /// Find table name from alias
@@ -411,8 +462,13 @@ actor SQLSchemaProvider {
 
     /// Values a column is restricted to, when the database declares them (a PostgreSQL enum type,
     /// a MongoDB `$jsonSchema` enum). Returns nothing for an ordinary column.
-    /// Reads only what the column cache already holds. Completion runs on every keystroke, so it
-    /// must never trigger a schema fetch; the eager column preload is what fills this cache.
+    ///
+    /// Reads only what the column cache already holds, because completion runs on every keystroke
+    /// and this must never add a fetch of its own. The cache is filled by the eager preload on a
+    /// schema small enough to preload, and otherwise by `getColumns`, which column completion on the
+    /// same statement has already called for every table the statement names. A statement whose
+    /// value position is reached before any column completion ran therefore offers nothing here
+    /// until the next request.
     func allowedValues(forColumn column: String, in references: [TableReference]) -> [String] {
         let name = column.lowercased()
         let candidates = references.isEmpty
diff --git a/TablePro/Core/DataGrid/RowDisplayBox.swift b/TablePro/Core/DataGrid/RowDisplayBox.swift
index 511b397d3..b4f475587 100644
--- a/TablePro/Core/DataGrid/RowDisplayBox.swift
+++ b/TablePro/Core/DataGrid/RowDisplayBox.swift
@@ -15,7 +15,16 @@ final class RowDisplayBox {
 
 @MainActor
 final class RowDisplayCache {
-    private var storage: [RowID: RowDisplayBox] = [:]
+    /// A box is a reference and callers mutate one in place before handing it back, so the cost
+    /// recorded at insertion is the only number that still describes what was added. Recomputing it
+    /// from the box on removal subtracts a different figure than was added and drifts `totalCost`
+    /// away from the budget it is there to enforce.
+    private struct Entry {
+        let box: RowDisplayBox
+        let cost: Int
+    }
+
+    private var storage: [RowID: Entry] = [:]
     private var insertionOrder: [RowID] = []
     private var insertionHead: Int = 0
     private var totalCost: Int = 0
@@ -28,16 +37,17 @@ final class RowDisplayCache {
     }
 
     func box(forID id: RowID) -> RowDisplayBox? {
-        storage[id]
+        storage[id]?.box
     }
 
-    func setBox(_ box: RowDisplayBox, forID id: RowID, cost: Int) {
+    func setBox(_ box: RowDisplayBox, forID id: RowID) {
+        let cost = Self.rowCost(box.values)
         if let existing = storage[id] {
-            totalCost -= rowCost(existing.values)
+            totalCost -= existing.cost
         } else {
             insertionOrder.append(id)
         }
-        storage[id] = box
+        storage[id] = Entry(box: box, cost: cost)
         totalCost += cost
         evictIfNeeded()
     }
@@ -54,11 +64,12 @@ final class RowDisplayCache {
     /// whose content changed in place keeps its id and would otherwise be served
     /// its pre-edit text.
     func clearValues(forID id: RowID) {
-        guard let box = storage[id] else { return }
-        totalCost -= rowCost(box.values)
-        for index in box.values.indices {
-            box.values[index] = nil
+        guard let existing = storage[id] else { return }
+        totalCost -= existing.cost
+        for index in existing.box.values.indices {
+            existing.box.values[index] = nil
         }
+        storage[id] = Entry(box: existing.box, cost: 0)
     }
 
     private func evictIfNeeded() {
@@ -67,7 +78,7 @@ final class RowDisplayCache {
             let oldest = insertionOrder[insertionHead]
             insertionHead += 1
             if let removed = storage.removeValue(forKey: oldest) {
-                totalCost -= rowCost(removed.values)
+                totalCost -= removed.cost
             }
         }
         if insertionHead > 10_000 {
@@ -76,10 +87,10 @@ final class RowDisplayCache {
         }
     }
 
-    private func rowCost(_ values: ContiguousArray<String?>) -> Int {
+    private static func rowCost(_ values: ContiguousArray<String?>) -> Int {
         var total = 0
         for value in values {
-            if let s = value { total &+= s.utf8.count }
+            if let value { total &+= value.utf8.count }
         }
         return total
     }
diff --git a/TablePro/Core/Execution/TabExecutionRegistry.swift b/TablePro/Core/Execution/TabExecutionRegistry.swift
index bc578b5bb..fe97f00da 100644
--- a/TablePro/Core/Execution/TabExecutionRegistry.swift
+++ b/TablePro/Core/Execution/TabExecutionRegistry.swift
@@ -176,6 +176,15 @@ internal struct TabExecutionRegistry {
         entries[tabId] != nil
     }
 
+    /// One tab's whole answer to "is anything running here", the per-tab counterpart of
+    /// `isAnyExecuting`. Work that cannot claim the tab still runs on it: Fetch All extends the
+    /// result already on screen, so it registers unclaimed work rather than minting a content epoch
+    /// that would discard its own rows. Anything deciding whether a tab is idle asks this;
+    /// `isExecuting` answers the narrower question of whether a claim is outstanding.
+    internal func isBusy(_ tabId: UUID) -> Bool {
+        entries[tabId] != nil || unclaimedWork[tabId] != nil
+    }
+
     /// The window's whole answer to "is anything running here", and the only one.
     ///
     /// The toolbar's indicator, Stop, `Cmd+.` and the disconnect warning all read this rather than
diff --git a/TablePro/Models/Query/TabSessionRegistry.swift b/TablePro/Models/Query/TabSessionRegistry.swift
index 3aaf042de..f53d57691 100644
--- a/TablePro/Models/Query/TabSessionRegistry.swift
+++ b/TablePro/Models/Query/TabSessionRegistry.swift
@@ -44,12 +44,22 @@ final class TabSessionRegistry {
         session.dataRevision &+= 1
     }
 
+    /// A mutation of what the tab already holds, so it cannot resurrect a tab that holds nothing.
+    ///
+    /// Phase-2 metadata (foreign keys, defaults, enum values) lands at background priority long
+    /// after the load that asked for it, and it is keyed on the result rather than on the rows, so
+    /// it arrives on tabs that were evicted while it was in flight. Clearing `isEvicted` for a
+    /// mutation that leaves the buffer empty leaves a tab with no rows that `canAutoLoadTableTab`
+    /// reads as already loaded, and eviction cannot re-mark it because it has nothing left to lose:
+    /// the grid stays empty until an explicit refresh.
     func updateTableRows(for tabId: UUID, _ mutate: (inout TableRows) -> Void) {
         let session = ensureSession(for: tabId)
         var rows = session.tableRows
         mutate(&rows)
         session.tableRows = rows
-        session.isEvicted = false
+        if !rows.rows.isEmpty {
+            session.isEvicted = false
+        }
         session.dataRevision &+= 1
     }
 
@@ -72,7 +82,7 @@ final class TabSessionRegistry {
     func evict(for tabId: UUID) {
         guard let session = sessions[tabId] else { return }
         guard !session.tableRows.rows.isEmpty else { return }
-        session.tableRows.rows = []
+        session.tableRows.discardRowsKeepingMetadata()
         session.isEvicted = true
         session.dataRevision &+= 1
     }
diff --git a/TablePro/Models/Query/TableRows.swift b/TablePro/Models/Query/TableRows.swift
index 899f4780e..49dc73850 100644
--- a/TablePro/Models/Query/TableRows.swift
+++ b/TablePro/Models/Query/TableRows.swift
@@ -57,6 +57,12 @@ struct TableRows: Sendable {
         return rows[index]
     }
 
+    /// Releases the row payload while keeping the schema needed to render and reload the table.
+    mutating func discardRowsKeepingMetadata() {
+        rows = []
+        indexByID = [:]
+    }
+
     @discardableResult
     mutating func edit(row: Int, column: Int, value: PluginCellValue) -> Delta {
         guard row >= 0, row < rows.count else { return .none }
diff --git a/TablePro/Views/Main/Extensions/MainContentCoordinator+TabSwitch.swift b/TablePro/Views/Main/Extensions/MainContentCoordinator+TabSwitch.swift
index 093de0344..14162c2a5 100644
--- a/TablePro/Views/Main/Extensions/MainContentCoordinator+TabSwitch.swift
+++ b/TablePro/Views/Main/Extensions/MainContentCoordinator+TabSwitch.swift
@@ -126,15 +126,54 @@ extension MainContentCoordinator {
         }
     }
 
-    private func evictInactiveTabs(excluding activeTabIds: Set<UUID>) {
+    /// Whether dropping this tab's rows is safe, which is exactly whether `canAutoLoadTableTab`
+    /// will bring them back. The two answers have to agree: a tab evicted without a route back to
+    /// its rows shows an empty grid until the user refreshes it by hand.
+    ///
+    /// Table tabs qualify because their rows follow from their generated query. Query tabs do not,
+    /// a pinned result shares the buffer it would lose, and a tab with an execution, a load task or
+    /// a page fetch in flight would have the result land on a buffer that moved out from under it.
+    func canEvictReloadableTableRows(_ tab: QueryTab) -> Bool {
+        guard tab.id != tabManager.selectedTabId,
+              tab.tabType == .table,
+              tab.execution.errorMessage == nil,
+              tab.content.query.contains(where: { !$0.isWhitespace }),
+              !tab.pendingChanges.hasChanges,
+              !tab.display.hasPinnedResults,
+              !tab.pagination.isLoading,
+              !tab.pagination.isLoadingMore,
+              !tabExecution.isBusy(tab.id),
+              tableLoadTasks[tab.id] == nil,
+              !tabSessionRegistry.isEvicted(tab.id),
+              let rows = tabSessionRegistry.existingTableRows(for: tab.id),
+              !rows.rows.isEmpty
+        else { return false }
+        return true
+    }
+
+    @discardableResult
+    func evictReloadableTableRows(for tabId: UUID) -> Bool {
+        guard let index = tabManager.tabs.firstIndex(where: { $0.id == tabId }),
+              canEvictReloadableTableRows(tabManager.tabs[index])
+        else { return false }
+
+        tabManager.mutate(at: index) { tab in
+            for resultSet in tab.display.resultSets where !resultSet.isPinned {
+                resultSet.tableRows.discardRowsKeepingMetadata()
+            }
+            tab.loadEpoch &+= 1
+        }
+        tabSessionRegistry.evict(for: tabId)
+        return true
+    }
+
+    func evictInactiveTabs(excluding activeTabIds: Set<UUID>) {
         let start = Date()
         let candidates: [(tab: QueryTab, rows: TableRows)] = tabManager.tabs.compactMap { tab in
             guard !activeTabIds.contains(tab.id),
                   tab.execution.lastExecutedAt != nil,
-                  !tab.pendingChanges.hasChanges,
-                  let rows = tabSessionRegistry.existingTableRows(for: tab.id),
-                  !tabSessionRegistry.isEvicted(tab.id),
-                  !rows.rows.isEmpty
+                  canEvictReloadableTableRows(tab),
+                  let rows = tabSessionRegistry.existingTableRows(for: tab.id)
             else { return nil }
             return (tab, rows)
         }
@@ -163,12 +202,12 @@ extension MainContentCoordinator {
         }
         let toEvict = sorted.dropLast(maxInactiveLoaded)
 
+        var evicted = 0
         for entry in toEvict {
-            tabSessionRegistry.evict(for: entry.tab.id)
-            tabManager.mutate(tabId: entry.tab.id) { $0.loadEpoch &+= 1 }
+            if evictReloadableTableRows(for: entry.tab.id) { evicted += 1 }
         }
         Self.lifecycleLogger.debug(
-            "[switch] evictInactiveTabs evicted=\(toEvict.count) keptInactive=\(maxInactiveLoaded) elapsedMs=\(Int(Date().timeIntervalSince(start) * 1_000))"
+            "[switch] evictInactiveTabs evicted=\(evicted) attempted=\(toEvict.count) candidates=\(sorted.count) keptInactive=\(maxInactiveLoaded) elapsedMs=\(Int(Date().timeIntervalSince(start) * 1_000))"
         )
     }
 }
diff --git a/TablePro/Views/Main/MainContentCoordinator.swift b/TablePro/Views/Main/MainContentCoordinator.swift
index 76a51298b..0e9899d96 100644
--- a/TablePro/Views/Main/MainContentCoordinator.swift
+++ b/TablePro/Views/Main/MainContentCoordinator.swift
@@ -568,17 +568,14 @@ final class MainContentCoordinator {
         }
     }()
 
-    /// Evict row data for background tabs in this coordinator to free memory.
-    /// Called when the coordinator's native window-tab becomes inactive.
-    /// The currently selected tab is kept in memory so the user sees no
-    /// refresh flicker when switching back — matching native macOS behavior.
-    /// Background tabs are re-fetched automatically when selected.
+    /// Frees the row data of every background tab that can fetch it again, called when this
+    /// coordinator's window stops being key. The selected tab keeps its rows so returning to the
+    /// window costs no refresh, and `canEvictReloadableTableRows` decides the rest: a query tab, a
+    /// tab holding a pinned result, a failed tab and a tab with work in flight all stay resident,
+    /// because none of them would come back on their own.
     func evictInactiveRowData() {
-        let selectedId = tabManager.selectedTabId
-        for (index, tab) in tabManager.tabs.enumerated()
-        where tab.id != selectedId && !tab.pendingChanges.hasChanges {
-            tabSessionRegistry.evict(for: tab.id)
-            tabManager.mutate(at: index) { $0.loadEpoch &+= 1 }
+        for tab in tabManager.tabs {
+            evictReloadableTableRows(for: tab.id)
         }
     }
 
diff --git a/TablePro/Views/Results/DataGridCoordinator.swift b/TablePro/Views/Results/DataGridCoordinator.swift
index 23a69c439..740a09453 100644
--- a/TablePro/Views/Results/DataGridCoordinator.swift
+++ b/TablePro/Views/Results/DataGridCoordinator.swift
@@ -438,6 +438,7 @@ final class TableViewCoordinator: NSObject, NSTableViewDelegate, NSTableViewData
                     columnIndexes: IndexSet(integersIn: 0..<tableView.numberOfColumns)
                 )
             }
+            startBackgroundPrewarm()
         }
     }
 
@@ -649,7 +650,7 @@ final class TableViewCoordinator: NSObject, NSTableViewDelegate, NSTableViewData
         if column >= 0, column < box.values.count {
             box.values[column] = formatted
         }
-        displayCache.setBox(box, forID: id, cost: displayCacheCost(box.values))
+        displayCache.setBox(box, forID: id)
         return formatted
     }
 
@@ -887,15 +888,7 @@ final class TableViewCoordinator: NSObject, NSTableViewDelegate, NSTableViewData
             ) ?? row.values[col].asText
         }
         let box = RowDisplayBox(values)
-        displayCache.setBox(box, forID: row.id, cost: displayCacheCost(values))
-    }
-
-    private func displayCacheCost(_ values: ContiguousArray<String?>) -> Int {
-        var total = 0
-        for value in values {
-            if let s = value { total &+= s.utf8.count }
-        }
-        return total
+        displayCache.setBox(box, forID: row.id)
     }
 
     private func invalidateDisplayCache(forDisplayRow displayIndex: Int) {
@@ -908,7 +901,7 @@ final class TableViewCoordinator: NSObject, NSTableViewDelegate, NSTableViewData
         guard let box = displayCache.box(forID: row.id),
               column >= 0, column < box.values.count else { return }
         box.values[column] = nil
-        displayCache.setBox(box, forID: row.id, cost: displayCacheCost(box.values))
+        displayCache.setBox(box, forID: row.id)
     }
 
     func applyDelta(_ delta: Delta) {
@@ -984,6 +977,7 @@ final class TableViewCoordinator: NSObject, NSTableViewDelegate, NSTableViewData
             columnIndexes: IndexSet(integersIn: 0..<tableView.numberOfColumns)
         )
         refreshVisibleRowVisualStates()
+        startBackgroundPrewarm()
     }
 
     /// Single-row equivalent of `reloadVisibleRowsAndStates` for cases where
diff --git a/TableProTests/Core/Autocomplete/SQLSchemaProviderFallbackTests.swift b/TableProTests/Core/Autocomplete/SQLSchemaProviderFallbackTests.swift
index d6cbd0a2d..9dfa0bed2 100644
--- a/TableProTests/Core/Autocomplete/SQLSchemaProviderFallbackTests.swift
+++ b/TableProTests/Core/Autocomplete/SQLSchemaProviderFallbackTests.swift
@@ -422,8 +422,8 @@ struct SQLSchemaProviderFallbackTests {
         #expect(cachedColumns.count == 2)
     }
 
-    @Test("Eager load respects maxCachedTables limit")
-    func eagerLoadRespectsMaxCachedTables() async throws {
+    @Test("Eager load caches a schema that fits the table limit")
+    func eagerLoadCachesASchemaThatFits() async {
         let driver = MockFallbackDriver()
         var tables: [TableInfo] = []
         for i in 0..<60 {
@@ -437,13 +437,32 @@ struct SQLSchemaProviderFallbackTests {
 
         let provider = SQLSchemaProvider()
         await provider.resetForDatabase("testdb", tables: tables, driver: driver)
+        await provider.waitForEagerColumnLoad()
 
-        // Wait for the eager load task
-        try await Task.sleep(nanoseconds: 300_000_000)
+        let items = await provider.allColumnsFromCachedTables()
+        #expect(items.count == 60)
+        #expect(Set(items.map(\.label)) == Set((0..<60).map { "col_\($0)" }))
+    }
+
+    @Test("Eager load caps the cache when the fetch returns more tables than the schema listed")
+    func eagerLoadCapsCacheWhenFetchOverruns() async {
+        let cap = SQLSchemaProvider.maxCachedTables
+        let driver = MockFallbackDriver()
+        for i in 0..<(cap + 25) {
+            driver.columnsPerTable["table_\(i)"] = [
+                TestFixtures.makeColumnInfo(name: "col_\(i)", isPrimaryKey: false)
+            ]
+        }
+        let listed = (0..<10).map { TestFixtures.makeTableInfo(name: "table_\($0)") }
+        driver.tablesToReturn = listed
+
+        let provider = SQLSchemaProvider()
+        await provider.resetForDatabase("testdb", tables: listed, driver: driver)
+        await provider.waitForEagerColumnLoad()
 
-        // allColumnsFromCachedTables should return at most 50 tables worth of columns
         let items = await provider.allColumnsFromCachedTables()
-        #expect(items.count <= 50)
+        #expect(items.count == cap)
+        #expect(Set(items.map(\.label)).isSuperset(of: (0..<10).map { "col_\($0)" }))
     }
 
     @Test("allColumnsFromCachedTables uses canonical table names from tables list")
diff --git a/TableProTests/Core/Autocomplete/SQLSchemaProviderTests.swift b/TableProTests/Core/Autocomplete/SQLSchemaProviderTests.swift
index b4ee96e7d..6ba0f6587 100644
--- a/TableProTests/Core/Autocomplete/SQLSchemaProviderTests.swift
+++ b/TableProTests/Core/Autocomplete/SQLSchemaProviderTests.swift
@@ -27,6 +27,8 @@ final class MockDatabaseDriver: DatabaseDriver, SchemaSwitchable, @unchecked Sen
     var fetchTablesCallCount = 0
     var fetchColumnsCallCount = 0
     var fetchColumnsCalls: [String] = []
+    var fetchAllColumnsCallCount = 0
+    var allColumnsToReturn: [String: [ColumnInfo]] = [:]
     var fetchSchemaTablesCalls: [String] = []
     var applyQueryTimeoutValues: [Int] = []
     var cancelQueryCallCount = 0
@@ -111,7 +113,8 @@ final class MockDatabaseDriver: DatabaseDriver, SchemaSwitchable, @unchecked Sen
     }
 
     func fetchAllColumns() async throws -> [String: [ColumnInfo]] {
-        [:]
+        fetchAllColumnsCallCount += 1
+        return allColumnsToReturn
     }
 
     func fetchIndexes(table: String) async throws -> [IndexInfo] { [] }
@@ -224,9 +227,10 @@ struct SQLSchemaProviderTests {
 
     @Test("evicts oldest columns when cache exceeds limit")
     func lruEvictionOnExceedingMax() async {
+        let total = SQLSchemaProvider.maxCachedTables + 2
         let driver = MockDatabaseDriver()
         var allTables: [TableInfo] = []
-        for i in 0..<52 {
+        for i in 0..<total {
             let name = "table_\(i)"
             allTables.append(TestFixtures.makeTableInfo(name: name))
             driver.columnsToReturn[name] = [
@@ -237,22 +241,23 @@ struct SQLSchemaProviderTests {
 
         let provider = SQLSchemaProvider()
         await provider.loadSchema(using: driver, connection: TestFixtures.makeConnection())
+        await provider.waitForEagerColumnLoad()
 
-        for i in 0..<52 {
+        for i in 0..<total {
             _ = await provider.getColumns(for: "table_\(i)")
         }
 
-        #expect(driver.fetchColumnsCallCount == 52)
+        #expect(driver.fetchColumnsCallCount == total)
 
         // table_0 and table_1 should have been evicted (oldest entries)
         // Fetching them again should trigger new driver calls
         _ = await provider.getColumns(for: "table_0")
         _ = await provider.getColumns(for: "table_1")
-        #expect(driver.fetchColumnsCallCount == 54)
+        #expect(driver.fetchColumnsCallCount == total + 2)
 
-        // table_51 should still be cached (no additional call)
+        // The newest table should still be cached (no additional call)
         let countBefore = driver.fetchColumnsCallCount
-        _ = await provider.getColumns(for: "table_51")
+        _ = await provider.getColumns(for: "table_\(total - 1)")
         #expect(driver.fetchColumnsCallCount == countBefore)
     }
 
@@ -265,7 +270,8 @@ struct SQLSchemaProviderTests {
                 TestFixtures.makeColumnInfo(name: "\(name)_col", isPrimaryKey: false)
             ]
         }
-        for i in 0..<49 {
+        let fillCount = SQLSchemaProvider.maxCachedTables - 1
+        for i in 0..<fillCount {
             let name = "fill_\(i)"
             driver.columnsToReturn[name] = [
                 TestFixtures.makeColumnInfo(name: "col", isPrimaryKey: false)
@@ -273,11 +279,12 @@ struct SQLSchemaProviderTests {
         }
 
         var allTables = tableNames.map { TestFixtures.makeTableInfo(name: $0) }
-        allTables += (0..<49).map { TestFixtures.makeTableInfo(name: "fill_\($0)") }
+        allTables += (0..<fillCount).map { TestFixtures.makeTableInfo(name: "fill_\($0)") }
         driver.tablesToReturn = allTables
 
         let provider = SQLSchemaProvider()
         await provider.loadSchema(using: driver, connection: TestFixtures.makeConnection())
+        await provider.waitForEagerColumnLoad()
 
         // Fetch columns for A, B, C (in that order)
         _ = await provider.getColumns(for: "a")
@@ -289,8 +296,8 @@ struct SQLSchemaProviderTests {
         _ = await provider.getColumns(for: "a")
         #expect(driver.fetchColumnsCallCount == 3)
 
-        // Fill cache with 49 more tables (total becomes 52, evicting 2 oldest: b then c)
-        for i in 0..<49 {
+        // Fill the cache to two over capacity, evicting the two oldest: b then c
+        for i in 0..<fillCount {
             _ = await provider.getColumns(for: "fill_\(i)")
         }
 
@@ -419,7 +426,7 @@ struct SQLSchemaProviderTests {
     }
 
     @Test("eager column load uses injected metadata source instead of cached driver")
-    func eagerLoadUsesMetadataSource() async throws {
+    func eagerLoadUsesMetadataSource() async {
         let driver = MockDatabaseDriver()
         driver.columnsToReturn = ["users": [TestFixtures.makeColumnInfo(name: "from_driver")]]
         let source = SQLSchemaProvider.ColumnMetadataSource(
@@ -428,14 +435,136 @@ struct SQLSchemaProviderTests {
         )
         let provider = SQLSchemaProvider(metadataSource: source)
         await provider.resetForDatabase("db", tables: [TestFixtures.makeTableInfo(name: "users")], driver: driver)
-
-        try await Task.sleep(nanoseconds: 300_000_000)
+        await provider.waitForEagerColumnLoad()
 
         let columns = await provider.getColumns(for: "users")
         #expect(columns.first?.name == "eager_source")
         #expect(driver.fetchColumnsCallCount == 0)
     }
 
+    @Test("eager column load runs when the schema fits in the cache")
+    func eagerLoadRunsForBoundedSchema() async {
+        let driver = MockDatabaseDriver()
+        let fits = SQLSchemaProvider.maxCachedTables
+        driver.tablesToReturn = (0..<fits).map { TestFixtures.makeTableInfo(name: "table_\($0)") }
+        driver.allColumnsToReturn = [
+            "table_\(fits - 1)": [TestFixtures.makeColumnInfo(name: "eager_column")]
+        ]
+
+        let provider = SQLSchemaProvider()
+        await provider.loadSchema(using: driver)
+        await provider.waitForEagerColumnLoad()
+
+        let columns = await provider.getColumns(for: "table_\(fits - 1)")
+        #expect(driver.fetchAllColumnsCallCount == 1)
+        #expect(driver.fetchColumnsCallCount == 0)
+        #expect(columns.first?.name == "eager_column")
+    }
+
+    @Test("large schemas skip eager column loading and fetch columns lazily")
+    func largeSchemaSkipsEagerLoad() async {
+        let driver = MockDatabaseDriver()
+        let overflows = SQLSchemaProvider.maxCachedTables + 1
+        driver.tablesToReturn = (0..<overflows).map { TestFixtures.makeTableInfo(name: "table_\($0)") }
+        driver.columnsToReturn = [
+            "table_\(overflows - 1)": [TestFixtures.makeColumnInfo(name: "lazy_column")]
+        ]
+
+        let provider = SQLSchemaProvider()
+        await provider.loadSchema(using: driver)
+        await provider.waitForEagerColumnLoad()
+
+        let columns = await provider.getColumns(for: "table_\(overflows - 1)")
+        #expect(driver.fetchAllColumnsCallCount == 0)
+        #expect(driver.fetchColumnsCalls == ["table_\(overflows - 1)"])
+        #expect(columns.first?.name == "lazy_column")
+    }
+
+    @Test("eager column load counts the schema it fetches, not every expanded schema")
+    func eagerLoadCountsOnlyTheFetchedSchema() async {
+        let driver = MockDatabaseDriver()
+        driver.currentSchema = "public"
+        let ownSchema = (0..<10).map { TestFixtures.makeTableInfo(name: "public_\($0)", schema: "public") }
+        let otherSchemas = (0..<(SQLSchemaProvider.maxCachedTables + 100)).map {
+            TestFixtures.makeTableInfo(name: "other_\($0)", schema: "audit_\($0 % 8)")
+        }
+        driver.allColumnsToReturn = [
+            "public_0": [TestFixtures.makeColumnInfo(name: "eager_column")]
+        ]
+
+        let provider = SQLSchemaProvider()
+        await provider.resetForDatabase("db", tables: ownSchema + otherSchemas, driver: driver)
+        await provider.waitForEagerColumnLoad()
+
+        let columns = await provider.getColumns(for: "public_0")
+        #expect(driver.fetchAllColumnsCallCount == 1)
+        #expect(driver.fetchColumnsCallCount == 0)
+        #expect(columns.first?.name == "eager_column")
+    }
+
+    @Test("eager column load still skips when the fetched schema is itself too large")
+    func eagerLoadSkipsLargeCurrentSchema() async {
+        let driver = MockDatabaseDriver()
+        driver.currentSchema = "public"
+        driver.tablesToReturn = []
+        driver.columnsToReturn = [
+            "public_0": [TestFixtures.makeColumnInfo(name: "lazy_column")]
+        ]
+        let ownSchema = (0..<(SQLSchemaProvider.maxCachedTables + 1)).map {
+            TestFixtures.makeTableInfo(name: "public_\($0)", schema: "public")
+        }
+
+        let provider = SQLSchemaProvider()
+        await provider.resetForDatabase("db", tables: ownSchema, driver: driver)
+        await provider.waitForEagerColumnLoad()
+
+        let columns = await provider.getColumns(for: "public_0")
+        #expect(driver.fetchAllColumnsCallCount == 0)
+        #expect(columns.first?.name == "lazy_column")
+    }
+
+    @Test("clearing the column cache empties it without sending a second bulk fetch")
+    func clearColumnCacheDoesNotRefetch() async {
+        let driver = MockDatabaseDriver()
+        driver.allColumnsToReturn = ["users": [TestFixtures.makeColumnInfo(name: "eager_source")]]
+        let provider = SQLSchemaProvider()
+        await provider.resetForDatabase(
+            "db", tables: [TestFixtures.makeTableInfo(name: "users")], driver: driver
+        )
+        await provider.waitForEagerColumnLoad()
+        #expect(driver.fetchAllColumnsCallCount == 1)
+
+        await provider.clearColumnCache()
+        await provider.waitForEagerColumnLoad()
+
+        #expect(driver.fetchAllColumnsCallCount == 1)
+        let values = await provider.allColumnsFromCachedTables()
+        #expect(values.isEmpty)
+    }
+
+    @Test("eager column load caches every table the fetch returned")
+    func eagerLoadCachesEveryFetchedTable() async {
+        let names = (0..<40).map { "table_\($0)" }
+        let allColumns: [String: [ColumnInfo]] = names.reduce(into: [:]) { result, name in
+            result[name] = [TestFixtures.makeColumnInfo(name: "col_of_\(name)")]
+        }
+        let source = SQLSchemaProvider.ColumnMetadataSource(
+            fetchColumns: { _, _ in [] },
+            fetchAllColumns: { allColumns }
+        )
+        let provider = SQLSchemaProvider(metadataSource: source)
+
+        await provider.resetForDatabase(
+            "db",
+            tables: names.map { TestFixtures.makeTableInfo(name: $0) },
+            driver: MockDatabaseDriver()
+        )
+        await provider.waitForEagerColumnLoad()
+
+        let items = await provider.allColumnsFromCachedTables()
+        #expect(Set(items.map(\.label)) == Set(names.map { "col_of_\($0)" }))
+    }
+
     // MARK: - Namespaces (database/schema segments)
 
     @Test("isKnownSchema and isKnownDatabase match case-insensitively")
diff --git a/TableProTests/Core/DataGrid/RowDisplayCacheTests.swift b/TableProTests/Core/DataGrid/RowDisplayCacheTests.swift
index 5400eb365..fdf38700a 100644
--- a/TableProTests/Core/DataGrid/RowDisplayCacheTests.swift
+++ b/TableProTests/Core/DataGrid/RowDisplayCacheTests.swift
@@ -14,14 +14,6 @@ struct RowDisplayCacheTests {
         RowDisplayBox(ContiguousArray(values))
     }
 
-    private func cost(of values: [String?]) -> Int {
-        var total = 0
-        for v in values {
-            if let s = v { total &+= s.utf8.count }
-        }
-        return total
-    }
-
     @Test("Empty cache returns nil for any lookup")
     func emptyLookup() {
         let cache = RowDisplayCache()
@@ -33,9 +25,8 @@ struct RowDisplayCacheTests {
     func basicSetGet() {
         let cache = RowDisplayCache()
         let id = RowID.existing(42)
-        let values = ["a", "b", "c"]
-        let box = makeBox(values)
-        cache.setBox(box, forID: id, cost: cost(of: values))
+        let box = makeBox(["a", "b", "c"])
+        cache.setBox(box, forID: id)
 
         #expect(cache.box(forID: id) === box)
     }
@@ -44,12 +35,12 @@ struct RowDisplayCacheTests {
     func countLimitEvictsFIFO() {
         let cache = RowDisplayCache(countLimit: 3, costLimit: 1_000_000)
         for index in 1...3 {
-            cache.setBox(makeBox(["row\(index)"]), forID: .existing(index), cost: 4)
+            cache.setBox(makeBox(["row\(index)"]), forID: .existing(index))
         }
         #expect(cache.box(forID: .existing(1)) != nil)
 
         // Fourth insertion should evict the first.
-        cache.setBox(makeBox(["row4"]), forID: .existing(4), cost: 4)
+        cache.setBox(makeBox(["row4"]), forID: .existing(4))
         #expect(cache.box(forID: .existing(1)) == nil)
         #expect(cache.box(forID: .existing(2)) != nil)
         #expect(cache.box(forID: .existing(3)) != nil)
@@ -60,28 +51,43 @@ struct RowDisplayCacheTests {
     func costLimitEvicts() {
         let cache = RowDisplayCache(countLimit: 1_000, costLimit: 10)
         // First insert costs 6; under cap.
-        cache.setBox(makeBox(["abcdef"]), forID: .existing(1), cost: 6)
+        cache.setBox(makeBox(["abcdef"]), forID: .existing(1))
         // Second insert costs 6 more; total 12 > 10, evicts first.
-        cache.setBox(makeBox(["123456"]), forID: .existing(2), cost: 6)
+        cache.setBox(makeBox(["123456"]), forID: .existing(2))
 
         #expect(cache.box(forID: .existing(1)) == nil)
         #expect(cache.box(forID: .existing(2)) != nil)
     }
 
+    @Test("Updating the same mutable box uses its previously recorded cost")
+    func mutableBoxCostUpdateEvicts() {
+        let cache = RowDisplayCache(countLimit: 1_000, costLimit: 10)
+        let firstID = RowID.existing(1)
+        let box = makeBox(["a"])
+        cache.setBox(box, forID: firstID)
+
+        box.values[0] = "123456789"
+        cache.setBox(box, forID: firstID)
+        cache.setBox(makeBox(["xy"]), forID: .existing(2))
+
+        #expect(cache.box(forID: firstID) == nil)
+        #expect(cache.box(forID: .existing(2)) != nil)
+    }
+
     @Test("Replacing an existing key does not consume queue slot")
     func replaceExistingKey() {
         let cache = RowDisplayCache(countLimit: 2, costLimit: 1_000_000)
-        cache.setBox(makeBox(["v1"]), forID: .existing(1), cost: 2)
-        cache.setBox(makeBox(["v2"]), forID: .existing(2), cost: 2)
+        cache.setBox(makeBox(["v1"]), forID: .existing(1))
+        cache.setBox(makeBox(["v2"]), forID: .existing(2))
 
         // Replace id=1 without expanding the cache.
-        cache.setBox(makeBox(["v1-updated"]), forID: .existing(1), cost: 10)
+        cache.setBox(makeBox(["v1-updated"]), forID: .existing(1))
         #expect(cache.box(forID: .existing(1))?.values.first == "v1-updated")
         #expect(cache.box(forID: .existing(2))?.values.first == "v2")
 
         // Adding a new entry now evicts the oldest in insertion order (still id=1
         // because replacing did not re-add it to the order).
-        cache.setBox(makeBox(["v3"]), forID: .existing(3), cost: 2)
+        cache.setBox(makeBox(["v3"]), forID: .existing(3))
         #expect(cache.box(forID: .existing(1)) == nil)
         #expect(cache.box(forID: .existing(2)) != nil)
         #expect(cache.box(forID: .existing(3)) != nil)
@@ -91,7 +97,7 @@ struct RowDisplayCacheTests {
     func removeAllResetsState() {
         let cache = RowDisplayCache()
         for index in 1...10 {
-            cache.setBox(makeBox(["x"]), forID: .existing(index), cost: 1)
+            cache.setBox(makeBox(["x"]), forID: .existing(index))
         }
         cache.removeAll()
         for index in 1...10 {
@@ -99,7 +105,7 @@ struct RowDisplayCacheTests {
         }
 
         // Cache continues to work after removeAll.
-        cache.setBox(makeBox(["fresh"]), forID: .existing(100), cost: 5)
+        cache.setBox(makeBox(["fresh"]), forID: .existing(100))
         #expect(cache.box(forID: .existing(100))?.values.first == "fresh")
     }
 
@@ -108,7 +114,7 @@ struct RowDisplayCacheTests {
         let cache = RowDisplayCache()
         let id = RowID.existing(1)
         let box = makeBox(["old", "VARCHAR(255)", "YES"])
-        cache.setBox(box, forID: id, cost: cost(of: ["old", "VARCHAR(255)", "YES"]))
+        cache.setBox(box, forID: id)
 
         cache.clearValues(forID: id)
 
@@ -121,8 +127,8 @@ struct RowDisplayCacheTests {
     @Test("Clearing one row leaves the others formatted")
     func clearValuesLeavesOtherRows() throws {
         let cache = RowDisplayCache()
-        cache.setBox(makeBox(["a"]), forID: .existing(0), cost: 1)
-        cache.setBox(makeBox(["b"]), forID: .existing(1), cost: 1)
+        cache.setBox(makeBox(["a"]), forID: .existing(0))
+        cache.setBox(makeBox(["b"]), forID: .existing(1))
 
         cache.clearValues(forID: .existing(1))
 
@@ -135,7 +141,7 @@ struct RowDisplayCacheTests {
     @Test("Clearing an uncached row does nothing")
     func clearValuesForUnknownRow() {
         let cache = RowDisplayCache()
-        cache.setBox(makeBox(["a"]), forID: .existing(0), cost: 1)
+        cache.setBox(makeBox(["a"]), forID: .existing(0))
 
         cache.clearValues(forID: .existing(9))
 
@@ -148,22 +154,50 @@ struct RowDisplayCacheTests {
         let cache = RowDisplayCache()
         let id = RowID.existing(2)
         let box = makeBox(["old"])
-        cache.setBox(box, forID: id, cost: 3)
+        cache.setBox(box, forID: id)
         cache.clearValues(forID: id)
 
         box.values[0] = "new"
-        cache.setBox(box, forID: id, cost: 3)
+        cache.setBox(box, forID: id)
 
         #expect(cache.box(forID: id)?.values.first == "new")
     }
 
+    @Test("Refilling a cleared mutable box restores its recorded cost")
+    func clearedRowRefillRestoresCost() {
+        let cache = RowDisplayCache(countLimit: 1_000, costLimit: 10)
+        let firstID = RowID.existing(1)
+        let box = makeBox(["123456789"])
+        cache.setBox(box, forID: firstID)
+        cache.clearValues(forID: firstID)
+        cache.setBox(makeBox(["abcdefghij"]), forID: .existing(2))
+
+        box.values[0] = "123456789"
+        cache.setBox(box, forID: firstID)
+
+        #expect(cache.box(forID: firstID) == nil)
+        #expect(cache.box(forID: .existing(2)) != nil)
+    }
+
+    @Test("Cost comes from the box, so a grown row is charged what it now holds")
+    func costIsDerivedFromTheBox() {
+        let cache = RowDisplayCache(countLimit: 1_000, costLimit: 10)
+        let grown = makeBox(["a"])
+        cache.setBox(grown, forID: .existing(1))
+
+        grown.values[0] = "12345678901234567890"
+        cache.setBox(grown, forID: .existing(1))
+
+        #expect(cache.box(forID: .existing(1)) == nil)
+    }
+
     @Test("Inserted row IDs of both kinds round-trip")
     func mixedRowIDKinds() {
         let cache = RowDisplayCache()
         let existingID = RowID.existing(5)
         let insertedID = RowID.inserted(UUID())
-        cache.setBox(makeBox(["existing"]), forID: existingID, cost: 8)
-        cache.setBox(makeBox(["inserted"]), forID: insertedID, cost: 8)
+        cache.setBox(makeBox(["existing"]), forID: existingID)
+        cache.setBox(makeBox(["inserted"]), forID: insertedID)
         #expect(cache.box(forID: existingID)?.values.first == "existing")
         #expect(cache.box(forID: insertedID)?.values.first == "inserted")
     }
diff --git a/TableProTests/Helpers/TestFixtures.swift b/TableProTests/Helpers/TestFixtures.swift
index d79d8ec05..88b64edb1 100644
--- a/TableProTests/Helpers/TestFixtures.swift
+++ b/TableProTests/Helpers/TestFixtures.swift
@@ -101,12 +101,14 @@ enum TestFixtures {
 
     static func makeTableInfo(
         name: String = "test_table",
-        type: TableInfo.TableType = .table
+        type: TableInfo.TableType = .table,
+        schema: String? = nil
     ) -> TableInfo {
         return TableInfo(
             name: name,
             type: type,
-            rowCount: 0
+            rowCount: 0,
+            schema: schema
         )
     }
 
diff --git a/TableProTests/Models/Query/TabSessionRegistryTests.swift b/TableProTests/Models/Query/TabSessionRegistryTests.swift
index c631e3adf..5c3182e46 100644
--- a/TableProTests/Models/Query/TabSessionRegistryTests.swift
+++ b/TableProTests/Models/Query/TabSessionRegistryTests.swift
@@ -152,5 +152,6 @@ struct TabSessionRegistryTests {
 
         #expect(afterEvict > before)
         #expect(session.dataRevision == afterEvict)
+        #expect(session.tableRows.index(of: .existing(0)) == nil)
     }
 }
diff --git a/TableProTests/Models/Query/TableRowsTests.swift b/TableProTests/Models/Query/TableRowsTests.swift
index 45a16850b..fc41a8a65 100644
--- a/TableProTests/Models/Query/TableRowsTests.swift
+++ b/TableProTests/Models/Query/TableRowsTests.swift
@@ -4,8 +4,8 @@
 //
 
 import Foundation
-import TableProPluginKit
 @testable import TablePro
+import TableProPluginKit
 import Testing
 
 @Suite("TableRows - construction")
@@ -176,6 +176,23 @@ struct TableRowsIDLookupTests {
         let insertedID = table.rows[0].id
         #expect(table.row(withID: insertedID)?.values == ["v"])
     }
+
+    @Test("discardRowsKeepingMetadata releases rows and their ID index")
+    func discardRowsKeepingMetadataClearsIndex() {
+        var table = TableRows.from(
+            queryRows: [["a"]],
+            columns: ["c1"],
+            columnTypes: [.text(rawType: nil)]
+        )
+
+        table.discardRowsKeepingMetadata()
+
+        #expect(table.rows.isEmpty)
+        #expect(table.index(of: .existing(0)) == nil)
+        #expect(table.row(withID: .existing(0)) == nil)
+        #expect(table.columns == ["c1"])
+        #expect(table.columnTypes == [.text(rawType: nil)])
+    }
 }
 
 @Suite("TableRows - edit")
diff --git a/TableProTests/Views/Main/EvictionTests.swift b/TableProTests/Views/Main/EvictionTests.swift
index 65f11b09e..521bdf014 100644
--- a/TableProTests/Views/Main/EvictionTests.swift
+++ b/TableProTests/Views/Main/EvictionTests.swift
@@ -6,8 +6,8 @@
 //
 
 import Foundation
-import TableProPluginKit
 @testable import TablePro
+import TableProPluginKit
 import Testing
 
 @Suite("Cross-Window Tab Eviction")
@@ -40,7 +40,12 @@ struct EvictionTests {
         let columnTypes: [ColumnType] = Array(repeating: .text(rawType: nil), count: columns.count)
         let tableRows = TableRows.from(queryRows: rows.map { row in row.map(PluginCellValue.fromOptional) }, columns: columns, columnTypes: columnTypes)
         coordinator.setActiveTableRows(tableRows, for: tabId)
-        tabManager.tabs[index].execution.lastExecutedAt = Date()
+        let resultSet = ResultSet(label: tableName, tableRows: tableRows)
+        tabManager.mutate(at: index) { tab in
+            tab.display.resultSets = [resultSet]
+            tab.display.activeResultSetId = resultSet.id
+            tab.execution.lastExecutedAt = Date()
+        }
     }
 
     @Test("evictInactiveRowData evicts background tabs without pending changes")
@@ -48,15 +53,37 @@ struct EvictionTests {
         let (coordinator, tabManager) = makeCoordinator()
         try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
         let backgroundTabId = tabManager.tabs[0].id
+        let backgroundResult = try #require(tabManager.tabs[0].display.activeResultSet)
         try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
 
         #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.count == 10)
+        #expect(backgroundResult.tableRows.rows.count == 10)
         #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == false)
 
         coordinator.evictInactiveRowData()
 
         #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == true)
         #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.isEmpty)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).index(of: .existing(0)) == nil)
+        #expect(backgroundResult.tableRows.rows.isEmpty)
+        #expect(backgroundResult.tableRows.index(of: .existing(0)) == nil)
+    }
+
+    @Test("eviction helper never evicts the selected tab")
+    func preservesSelectedTab() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let selectedTabId = try #require(tabManager.selectedTabId)
+        let selectedResult = try #require(tabManager.selectedTab?.display.activeResultSet)
+        let loadEpoch = try #require(tabManager.selectedTab?.loadEpoch)
+
+        let didEvict = coordinator.evictReloadableTableRows(for: selectedTabId)
+
+        #expect(didEvict == false)
+        #expect(coordinator.tabSessionRegistry.isEvicted(selectedTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: selectedTabId).rows.count == 10)
+        #expect(selectedResult.tableRows.rows.count == 10)
+        #expect(tabManager.selectedTab?.loadEpoch == loadEpoch)
     }
 
     @Test("evictInactiveRowData skips tabs with pending changes")
@@ -65,12 +92,139 @@ struct EvictionTests {
         try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
 
         tabManager.tabs[0].pendingChanges.deletedRowIndices = [0]
+        let loadEpoch = tabManager.tabs[0].loadEpoch
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
 
         coordinator.evictInactiveRowData()
 
         let tabId = tabManager.tabs[0].id
         #expect(coordinator.tabSessionRegistry.isEvicted(tabId) == false)
         #expect(coordinator.tabSessionRegistry.tableRows(for: tabId).rows.count == 10)
+        #expect(tabManager.tabs[0].loadEpoch == loadEpoch)
+    }
+
+    @Test("evictInactiveRowData skips tabs with pinned results")
+    func preservesPinnedResults() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let backgroundTabId = tabManager.tabs[0].id
+        let pinnedResult = try #require(tabManager.tabs[0].display.activeResultSet)
+        pinnedResult.isPinned = true
+        let loadEpoch = tabManager.tabs[0].loadEpoch
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.count == 10)
+        #expect(pinnedResult.tableRows.rows.count == 10)
+        #expect(pinnedResult.tableRows.index(of: .existing(0)) == 0)
+        #expect(tabManager.tabs[0].loadEpoch == loadEpoch)
+    }
+
+    @Test("evictInactiveRowData skips table tabs that cannot auto-reload")
+    func skipsNonReloadableTableTabs() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let backgroundTabId = tabManager.tabs[0].id
+        tabManager.tabs[0].execution.errorMessage = "connection lost"
+        let loadEpoch = tabManager.tabs[0].loadEpoch
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.count == 10)
+        #expect(tabManager.tabs[0].loadEpoch == loadEpoch)
+    }
+
+    @Test("evictInactiveRowData skips an executing table tab")
+    func skipsExecutingTableTab() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let backgroundTabId = tabManager.tabs[0].id
+        let loadEpoch = tabManager.tabs[0].loadEpoch
+        let claim = coordinator.tabExecution.claim(backgroundTabId)
+        defer { _ = coordinator.tabExecution.settle(claim) }
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.count == 10)
+        #expect(tabManager.tabs[0].loadEpoch == loadEpoch)
+    }
+
+    @Test("evictInactiveRowData skips a table tab with an active load")
+    func skipsTableTabWithActiveLoad() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let backgroundTabId = tabManager.tabs[0].id
+        let loadEpoch = tabManager.tabs[0].loadEpoch
+        let inFlight = Task<Void, Never> { _ = try? await Task.sleep(for: .seconds(60)) }
+        coordinator.tableLoadTasks[backgroundTabId] = (UUID(), inFlight)
+        defer {
+            inFlight.cancel()
+            coordinator.tableLoadTasks[backgroundTabId] = nil
+        }
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.count == 10)
+        #expect(tabManager.tabs[0].loadEpoch == loadEpoch)
+    }
+
+    @Test("evictInactiveRowData skips tables loading rows", arguments: [false, true])
+    func skipsTableTabLoadingRows(isLoadingMore: Bool) throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let backgroundTabId = tabManager.tabs[0].id
+        let loadEpoch = tabManager.tabs[0].loadEpoch
+        tabManager.mutate(at: 0) { tab in
+            if isLoadingMore {
+                tab.pagination.isLoadingMore = true
+            } else {
+                tab.pagination.isLoading = true
+            }
+        }
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.count == 10)
+        #expect(tabManager.tabs[0].loadEpoch == loadEpoch)
+    }
+
+    @Test("evictInactiveRowData does not evict query results")
+    func preservesQueryResults() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        tabManager.addTab(initialQuery: "SELECT 1")
+        let queryIndex = try #require(tabManager.selectedTabIndex)
+        let queryTabId = tabManager.tabs[queryIndex].id
+        let queryRows = TableRows.from(
+            queryRows: [[.text("1")]],
+            columns: ["value"],
+            columnTypes: [.integer(rawType: nil)]
+        )
+        coordinator.setActiveTableRows(queryRows, for: queryTabId)
+        let queryResult = ResultSet(label: "SELECT 1", tableRows: queryRows)
+        tabManager.mutate(at: queryIndex) { tab in
+            tab.display.resultSets = [queryResult]
+            tab.display.activeResultSetId = queryResult.id
+            tab.execution.lastExecutedAt = Date()
+        }
+        let loadEpoch = tabManager.tabs[queryIndex].loadEpoch
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(queryTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: queryTabId).rows.count == 1)
+        #expect(queryResult.tableRows.rows.count == 1)
+        #expect(tabManager.tabs[queryIndex].loadEpoch == loadEpoch)
     }
 
     @Test("evictInactiveRowData preserves column metadata after eviction")
@@ -92,4 +246,119 @@ struct EvictionTests {
         let (coordinator, _) = makeCoordinator()
         coordinator.evictInactiveRowData()
     }
+
+    @Test("eviction skips a table tab whose query is blank")
+    func skipsTableTabWithBlankQuery() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let backgroundTabId = tabManager.tabs[0].id
+        tabManager.mutate(at: 0) { $0.content.query = "   \n\t " }
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.count == 10)
+    }
+
+    @Test("eviction skips a table tab running work that took no execution claim")
+    func skipsTableTabWithUnclaimedWork() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let backgroundTabId = tabManager.tabs[0].id
+        let token = coordinator.tabExecution.beginUnclaimedWork(for: backgroundTabId)
+        defer { coordinator.tabExecution.endUnclaimedWork(token, for: backgroundTabId) }
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.count == 10)
+    }
+
+    @Test("metadata landing after eviction does not resurrect an empty tab")
+    func lateMetadataKeepsTabEvicted() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let backgroundTabId = tabManager.tabs[0].id
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == true)
+
+        coordinator.mutateActiveTableRows(for: backgroundTabId) { rows in
+            rows.columnEnumValues["status"] = ["active", "archived"]
+            return .columnsReplaced
+        }
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == true)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.isEmpty)
+    }
+
+    @Test("rows arriving after eviction clear the evicted flag")
+    func reloadedRowsClearTheEvictedFlag() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "users")
+        let backgroundTabId = tabManager.tabs[0].id
+        try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "orders")
+
+        coordinator.evictInactiveRowData()
+        let reloaded = TableRows.from(
+            queryRows: [[.text("1")]],
+            columns: ["id"],
+            columnTypes: [.integer(rawType: nil)]
+        )
+        coordinator.setActiveTableRows(reloaded, for: backgroundTabId)
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(backgroundTabId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: backgroundTabId).rows.count == 1)
+    }
+
+    @Test("evictInactiveTabs keeps the newest tabs within the memory budget")
+    func budgetKeepsNewestTabs() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        let budget = MemoryPressureAdvisor.budgetForInactiveTabs()
+        let total = budget + 3
+        var executedAt: [UUID: Date] = [:]
+        for index in 0..<total {
+            try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "table_\(index)")
+            let tabIndex = try #require(tabManager.selectedTabIndex)
+            let stamp = Date(timeIntervalSince1970: TimeInterval(1_000 + index))
+            tabManager.mutate(at: tabIndex) { $0.execution.lastExecutedAt = stamp }
+            executedAt[tabManager.tabs[tabIndex].id] = stamp
+        }
+        let selectedId = try #require(tabManager.selectedTabId)
+
+        coordinator.evictInactiveTabs(excluding: [selectedId])
+
+        let evictable = tabManager.tabs
+            .filter { $0.id != selectedId }
+            .sorted { executedAt[$0.id] ?? .distantPast < executedAt[$1.id] ?? .distantPast }
+        let expectedEvicted = evictable.count - budget
+        #expect(expectedEvicted > 0)
+        for (position, tab) in evictable.enumerated() {
+            #expect(coordinator.tabSessionRegistry.isEvicted(tab.id) == (position < expectedEvicted))
+        }
+        #expect(coordinator.tabSessionRegistry.isEvicted(selectedId) == false)
+    }
+
+    @Test("evictInactiveTabs never evicts a tab it was told is active")
+    func budgetSkipsActiveTabs() throws {
+        let (coordinator, tabManager) = makeCoordinator()
+        let budget = MemoryPressureAdvisor.budgetForInactiveTabs()
+        for index in 0..<(budget + 3) {
+            try addLoadedTab(to: coordinator, tabManager: tabManager, tableName: "table_\(index)")
+            let tabIndex = try #require(tabManager.selectedTabIndex)
+            tabManager.mutate(at: tabIndex) {
+                $0.execution.lastExecutedAt = Date(timeIntervalSince1970: TimeInterval(1_000 + index))
+            }
+        }
+        let selectedId = try #require(tabManager.selectedTabId)
+        let oldestId = tabManager.tabs[0].id
+
+        coordinator.evictInactiveTabs(excluding: [selectedId, oldestId])
+
+        #expect(coordinator.tabSessionRegistry.isEvicted(oldestId) == false)
+        #expect(coordinator.tabSessionRegistry.tableRows(for: oldestId).rows.count == 10)
+    }
 }