diff --git a/src/include/storage/table/arrow_node_table.h b/src/include/storage/table/arrow_node_table.h
index febe70920..fed706897 100644
--- a/src/include/storage/table/arrow_node_table.h
+++ b/src/include/storage/table/arrow_node_table.h
@@ -19,10 +19,10 @@ struct ArrowNodeTableScanState final : ColumnarNodeTableScanState {
     size_t currentMorselStartOffset = 0;
     size_t currentMorselEndOffset = 0;
 
-    ArrowNodeTableScanState(MemoryManager& mm, common::ValueVector* nodeIDVector,
+    ArrowNodeTableScanState(common::ValueVector* nodeIDVector,
         std::vector<common::ValueVector*> outputVectors,
         std::shared_ptr<common::DataChunkState> outChunkState)
-        : ColumnarNodeTableScanState{mm, nodeIDVector, std::move(outputVectors),
+        : ColumnarNodeTableScanState{nodeIDVector, std::move(outputVectors),
               std::move(outChunkState)} {}
 };
 
@@ -96,13 +96,14 @@ class ArrowNodeTable final : public ColumnarNodeTableBase {
     const ArrowSchemaWrapper& getArrowSchema() const { return schema; }
     const std::vector<ArrowArrayWrapper>& getArrowArrays() const { return arrays; }
 
-    common::node_group_idx_t getNumBatches(
+    common::node_group_idx_t getNumScanMorsels(
         const transaction::Transaction* transaction) const override;
 
-    size_t getNumScanMorsels(const transaction::Transaction* transaction) const;
-
     const catalog::NodeTableCatalogEntry* getCatalogEntry() const { return nodeTableCatalogEntry; }
 
+    std::unique_ptr<TableScanState> createScanState(common::ValueVector* nodeIDVector,
+        const std::vector<common::ValueVector*>& outVectors) const override;
+
 protected:
     std::string getColumnarFormatName() const override { return "Arrow"; }
     common::row_idx_t getTotalRowCount(const transaction::Transaction* transaction) const override;
diff --git a/src/include/storage/table/arrow_rel_table.h b/src/include/storage/table/arrow_rel_table.h
index 7f6d8c928..6fe3cb531 100644
--- a/src/include/storage/table/arrow_rel_table.h
+++ b/src/include/storage/table/arrow_rel_table.h
@@ -41,6 +41,9 @@ class ArrowRelTable final : public ColumnarRelTableBase {
         bool resetCachedBoundNodeSelVec = true) const override;
 
     bool scanInternal(transaction::Transaction* transaction, TableScanState& scanState) override;
+    std::unique_ptr<TableScanState> createScanState(common::ValueVector* nodeIDVector,
+        const std::vector<common::ValueVector*>& outVectors,
+        std::shared_ptr<common::DataChunkState> outChunkState) const override;
 
 protected:
     std::string getColumnarFormatName() const override { return "Arrow"; }
diff --git a/src/include/storage/table/columnar_node_table_base.h b/src/include/storage/table/columnar_node_table_base.h
index a719aa840..47759dce5 100644
--- a/src/include/storage/table/columnar_node_table_base.h
+++ b/src/include/storage/table/columnar_node_table_base.h
@@ -18,8 +18,8 @@ struct ColumnarNodeTableScanState : NodeTableScanState {
     size_t totalRows = 0;
     bool scanCompleted = false; // Track if this scan state has finished reading
 
-    ColumnarNodeTableScanState([[maybe_unused]] MemoryManager& mm,
-        common::ValueVector* nodeIDVector, std::vector<common::ValueVector*> outputVectors,
+    ColumnarNodeTableScanState(common::ValueVector* nodeIDVector,
+        std::vector<common::ValueVector*> outputVectors,
         std::shared_ptr<common::DataChunkState> outChunkState)
         : NodeTableScanState{nodeIDVector, std::move(outputVectors), std::move(outChunkState)} {}
 };
@@ -72,8 +72,6 @@ class ColumnarNodeTableBase : public NodeTable {
 
     // Template method pattern: subclasses implement format-specific operations
     virtual std::string getColumnarFormatName() const = 0;
-    virtual common::node_group_idx_t getNumBatches(
-        const transaction::Transaction* transaction) const = 0;
     virtual common::row_idx_t getTotalRowCount(
         const transaction::Transaction* transaction) const = 0;
 
@@ -81,6 +79,10 @@ class ColumnarNodeTableBase : public NodeTable {
     ColumnarNodeTableScanSharedState* getTableScanSharedState() const {
         return tableScanSharedState.get();
     }
+    virtual std::unique_ptr<TableScanState> createScanState(common::ValueVector* nodeIDVector,
+        const std::vector<common::ValueVector*>& outVectors) const = 0;
+    virtual common::node_group_idx_t getNumScanMorsels(
+        const transaction::Transaction* transaction) const = 0;
 };
 
 } // namespace storage
diff --git a/src/include/storage/table/columnar_rel_table_base.h b/src/include/storage/table/columnar_rel_table_base.h
index 0cbc6eb2f..8d41170d8 100644
--- a/src/include/storage/table/columnar_rel_table_base.h
+++ b/src/include/storage/table/columnar_rel_table_base.h
@@ -51,6 +51,9 @@ class ColumnarRelTableBase : public RelTable {
     std::vector<std::pair<common::offset_t, common::row_idx_t>> getTopKDegrees(
         const transaction::Transaction* transaction, common::RelDataDirection direction,
         common::idx_t k) override;
+    virtual std::unique_ptr<TableScanState> createScanState(common::ValueVector* nodeIDVector,
+        const std::vector<common::ValueVector*>& outVectors,
+        std::shared_ptr<common::DataChunkState> outChunkState) const = 0;
 
 protected:
     catalog::RelGroupCatalogEntry* relGroupEntry;
diff --git a/src/include/storage/table/ice_disk_node_table.h b/src/include/storage/table/ice_disk_node_table.h
index 2a6dc251f..bffa42ae7 100644
--- a/src/include/storage/table/ice_disk_node_table.h
+++ b/src/include/storage/table/ice_disk_node_table.h
@@ -21,10 +21,10 @@ struct IceDiskNodeTableScanState final : ColumnarNodeTableScanState {
     std::unique_ptr<processor::ParquetReaderScanState> parquetScanState;
     uint64_t lastQueryId = 0; // Track the last query ID to detect new queries
 
-    IceDiskNodeTableScanState([[maybe_unused]] MemoryManager& mm, common::ValueVector* nodeIDVector,
+    IceDiskNodeTableScanState(common::ValueVector* nodeIDVector,
         std::vector<common::ValueVector*> outputVectors,
         std::shared_ptr<common::DataChunkState> outChunkState)
-        : ColumnarNodeTableScanState{mm, nodeIDVector, std::move(outputVectors),
+        : ColumnarNodeTableScanState{nodeIDVector, std::move(outputVectors),
               std::move(outChunkState)} {
         parquetScanState = std::make_unique<processor::ParquetReaderScanState>();
     }
@@ -76,21 +76,26 @@ class IceDiskNodeTable final : public ColumnarNodeTableBase {
         common::offset_t offset) const override;
 
     const std::string& getParquetFilePath() const { return parquetFilePath; }
+    std::unique_ptr<TableScanState> createScanState(common::ValueVector* nodeIDVector,
+        const std::vector<common::ValueVector*>& outVectors) const override;
+    common::node_group_idx_t getNumScanMorsels(
+        const transaction::Transaction* transaction) const override;
 
 protected:
     // Implement ColumnarNodeTableBase interface
     std::string getColumnarFormatName() const override { return "icebug-disk"; }
-    common::node_group_idx_t getNumBatches(
-        const transaction::Transaction* transaction) const override;
     common::row_idx_t getTotalRowCount(const transaction::Transaction* transaction) const override;
 
 private:
     std::string parquetFilePath;
     mutable std::atomic<common::row_idx_t> cachedRowCount{common::INVALID_ROW_IDX};
+    mutable std::atomic<common::node_group_idx_t> cachedNumRowGroups{
+        common::INVALID_NODE_GROUP_IDX};
 
     void initializeParquetReader(transaction::Transaction* transaction) const;
     void initParquetScanForRowGroup(transaction::Transaction* transaction,
         IceDiskNodeTableScanState& iceDiskScanState) const;
+    common::node_group_idx_t getNumRowGroups(const transaction::Transaction* transaction) const;
 };
 
 } // namespace storage
diff --git a/src/include/storage/table/ice_disk_rel_table.h b/src/include/storage/table/ice_disk_rel_table.h
index e04bc75dc..a6d4cf68e 100644
--- a/src/include/storage/table/ice_disk_rel_table.h
+++ b/src/include/storage/table/ice_disk_rel_table.h
@@ -65,6 +65,9 @@ class IceDiskRelTable final : public ColumnarRelTableBase {
         bool resetCachedBoundNodeSelVec = true) const override;
 
     bool scanInternal(transaction::Transaction* transaction, TableScanState& scanState) override;
+    std::unique_ptr<TableScanState> createScanState(common::ValueVector* nodeIDVector,
+        const std::vector<common::ValueVector*>& outVectors,
+        std::shared_ptr<common::DataChunkState> outChunkState) const override;
 
 protected:
     // Implement ColumnarRelTableBase interface
diff --git a/src/processor/operator/scan/scan_multi_rel_tables.cpp b/src/processor/operator/scan/scan_multi_rel_tables.cpp
index f24eae254..ff9d57013 100644
--- a/src/processor/operator/scan/scan_multi_rel_tables.cpp
+++ b/src/processor/operator/scan/scan_multi_rel_tables.cpp
@@ -67,34 +67,27 @@ void ScanMultiRelTable::initLocalStateInternal(ResultSet* resultSet, ExecutionCo
     auto nbrNodeIDVector = outVectors[0];
 
     // Check if any table in any scanner is an external rel table with a custom scan state.
-    bool hasArrowTable = false;
-    bool hasIceDiskTable = false;
+    storage::ColumnarRelTableBase* columnarTable = nullptr;
     for (auto& [_, scanner] : scanners) {
         for (auto& relInfo : scanner.relInfos) {
-            if (dynamic_cast<storage::ArrowRelTable*>(relInfo.table) != nullptr) {
-                hasArrowTable = true;
-                break;
-            }
-            if (dynamic_cast<storage::IceDiskRelTable*>(relInfo.table) != nullptr) {
-                hasIceDiskTable = true;
+            if (dynamic_cast<storage::ColumnarRelTableBase*>(relInfo.table) != nullptr) {
+                columnarTable = dynamic_cast<storage::ColumnarRelTableBase*>(relInfo.table);
                 break;
             }
         }
-        if (hasArrowTable || hasIceDiskTable) {
+
+        if (columnarTable != nullptr) {
             break;
         }
     }
 
     // IceDisk scan state extends the common rel scan state and Arrow stores its per-table state
     // there, so one scan state can now cover IceDisk, Arrow, and native rel tables.
-    if (hasIceDiskTable) {
-        scanState =
-            std::make_unique<storage::IceDiskRelTableScanState>(*MemoryManager::Get(*clientContext),
-                boundNodeIDVector, outVectors, nbrNodeIDVector->state);
-    } else if (hasArrowTable) {
-        scanState =
-            std::make_unique<storage::ArrowRelTableScanState>(*MemoryManager::Get(*clientContext),
-                boundNodeIDVector, outVectors, nbrNodeIDVector->state);
+    if (columnarTable != nullptr) {
+        auto tableScanState =
+            columnarTable->createScanState(boundNodeIDVector, outVectors, nbrNodeIDVector->state);
+        scanState = std::unique_ptr<RelTableScanState>(
+            dynamic_cast<RelTableScanState*>(tableScanState.release()));
     } else {
         scanState = std::make_unique<RelTableScanState>(*MemoryManager::Get(*clientContext),
             boundNodeIDVector, outVectors, nbrNodeIDVector->state);
diff --git a/src/processor/operator/scan/scan_node_table.cpp b/src/processor/operator/scan/scan_node_table.cpp
index e9c952485..7d89aa9cb 100644
--- a/src/processor/operator/scan/scan_node_table.cpp
+++ b/src/processor/operator/scan/scan_node_table.cpp
@@ -15,16 +15,11 @@ using namespace lbug::storage;
 namespace lbug {
 namespace processor {
 static std::unique_ptr<TableScanState> createNodeTableScanState(NodeTable* table,
-    ValueVector* nodeIDVector, const std::vector<ValueVector*>& outVectors,
-    MemoryManager* memoryManager) {
-    if (dynamic_cast<IceDiskNodeTable*>(table) != nullptr) {
-        return std::make_unique<IceDiskNodeTableScanState>(*memoryManager, nodeIDVector, outVectors,
-            nodeIDVector->state);
-    }
-    if (dynamic_cast<ArrowNodeTable*>(table) != nullptr) {
-        return std::make_unique<ArrowNodeTableScanState>(*memoryManager, nodeIDVector, outVectors,
-            nodeIDVector->state);
+    ValueVector* nodeIDVector, const std::vector<ValueVector*>& outVectors) {
+    if (dynamic_cast<ColumnarNodeTableBase*>(table) != nullptr) {
+        return table->cast<ColumnarNodeTableBase>().createScanState(nodeIDVector, outVectors);
     }
+
     return std::make_unique<NodeTableScanState>(nodeIDVector, outVectors, nodeIDVector->state);
 }
 
@@ -56,23 +51,8 @@ void ScanNodeTableSharedState::initialize(const transaction::Transaction* transa
     // Initialize table-specific scan coordination (e.g., for IceDiskNodeTable)
     table->initializeScanCoordination(transaction);
 
-    if (const auto iceDiskTable = dynamic_cast<IceDiskNodeTable*>(table)) {
-        // For ice-disk tables, set numCommittedNodeGroups to number of row groups
-        std::vector<bool> columnSkips;
-        try {
-            auto context = transaction->getClientContext();
-            auto resolvedPath =
-                common::VirtualFileSystem::resolvePath(context, iceDiskTable->getParquetFilePath());
-            auto tempReader =
-                std::make_unique<processor::ParquetReader>(resolvedPath, columnSkips, context);
-            this->numCommittedNodeGroups = tempReader->getNumRowGroups();
-        } catch (const std::exception& e) {
-            this->numCommittedNodeGroups = 1;
-        }
-    } else if (const auto arrowTable = dynamic_cast<ArrowNodeTable*>(table)) {
-        // For Arrow tables, set numCommittedNodeGroups to number of morsels
-        this->numCommittedNodeGroups =
-            static_cast<common::node_group_idx_t>(arrowTable->getNumScanMorsels(transaction));
+    if (const auto columnarNodeTable = dynamic_cast<ColumnarNodeTableBase*>(table)) {
+        this->numCommittedNodeGroups = columnarNodeTable->getNumScanMorsels(transaction);
     } else {
         this->numCommittedNodeGroups = table->getNumCommittedNodeGroups();
     }
@@ -145,13 +125,12 @@ void ScanNodeTable::initLocalStateInternal(ResultSet* resultSet, ExecutionContex
 
 void ScanNodeTable::initCurrentTable(ExecutionContext* context) {
     auto& currentInfo = tableInfos[currentTableIdx];
-    scanState = createNodeTableScanState(currentInfo.table->ptrCast<NodeTable>(), nodeIDVector,
-        outVectors, MemoryManager::Get(*context->clientContext));
+    scanState =
+        createNodeTableScanState(currentInfo.table->ptrCast<NodeTable>(), nodeIDVector, outVectors);
     currentInfo.initScanState(*scanState, outVectors, context->clientContext);
     scanState->semiMask = sharedStates[currentTableIdx]->getSemiMask();
-    // Call table->initScanState for IceDiskNodeTable or ArrowNodeTable
-    if (dynamic_cast<IceDiskNodeTable*>(tableInfos[currentTableIdx].table) ||
-        dynamic_cast<ArrowNodeTable*>(tableInfos[currentTableIdx].table)) {
+    // Call table->initScanState for ColumnarNodeTables
+    if (dynamic_cast<ColumnarNodeTableBase*>(tableInfos[currentTableIdx].table)) {
         auto transaction = transaction::Transaction::Get(*context->clientContext);
         tableInfos[currentTableIdx].table->initScanState(transaction, *scanState);
     }
diff --git a/src/processor/operator/scan/scan_rel_table.cpp b/src/processor/operator/scan/scan_rel_table.cpp
index 8e7b5a7d1..d1b04be93 100644
--- a/src/processor/operator/scan/scan_rel_table.cpp
+++ b/src/processor/operator/scan/scan_rel_table.cpp
@@ -21,15 +21,9 @@ namespace lbug {
 namespace processor {
 
 static std::unique_ptr<TableScanState> createSourceNodeTableScanState(NodeTable* table,
-    ValueVector* nodeIDVector, const std::vector<ValueVector*>& outVectors,
-    MemoryManager* memoryManager) {
-    if (dynamic_cast<IceDiskNodeTable*>(table) != nullptr) {
-        return std::make_unique<IceDiskNodeTableScanState>(*memoryManager, nodeIDVector, outVectors,
-            nodeIDVector->state);
-    }
-    if (dynamic_cast<ArrowNodeTable*>(table) != nullptr) {
-        return std::make_unique<ArrowNodeTableScanState>(*memoryManager, nodeIDVector, outVectors,
-            nodeIDVector->state);
+    ValueVector* nodeIDVector, const std::vector<ValueVector*>& outVectors) {
+    if (dynamic_cast<ColumnarNodeTableBase*>(table) != nullptr) {
+        return table->cast<ColumnarNodeTableBase>().createScanState(nodeIDVector, outVectors);
     }
     return std::make_unique<NodeTableScanState>(nodeIDVector, outVectors, nodeIDVector->state);
 }
@@ -91,17 +85,13 @@ void ScanRelTable::initLocalStateInternal(ResultSet* resultSet, ExecutionContext
     auto boundNodeIDVector = resultSet->getValueVector(opInfo.nodeIDPos).get();
     auto nbrNodeIDVector = outVectors[0];
     // Check if this is an external rel table and create the corresponding scan state.
-    auto* arrowTable = dynamic_cast<storage::ArrowRelTable*>(tableInfo.table);
-    auto* iceDiskTable = dynamic_cast<storage::IceDiskRelTable*>(tableInfo.table);
+    auto* columnarTable = dynamic_cast<storage::ColumnarRelTableBase*>(tableInfo.table);
     auto* foreignTable = dynamic_cast<storage::ForeignRelTable*>(tableInfo.table);
-    if (arrowTable) {
-        scanState =
-            std::make_unique<storage::ArrowRelTableScanState>(*MemoryManager::Get(*clientContext),
-                boundNodeIDVector, outVectors, nbrNodeIDVector->state);
-    } else if (iceDiskTable) {
-        scanState =
-            std::make_unique<storage::IceDiskRelTableScanState>(*MemoryManager::Get(*clientContext),
-                boundNodeIDVector, outVectors, nbrNodeIDVector->state);
+    if (columnarTable) {
+        auto tableScanState =
+            columnarTable->createScanState(boundNodeIDVector, outVectors, nbrNodeIDVector->state);
+        scanState = std::unique_ptr<RelTableScanState>(
+            dynamic_cast<RelTableScanState*>(tableScanState.release()));
     } else if (foreignTable) {
         scanState =
             std::make_unique<storage::ForeignRelTableScanState>(*MemoryManager::Get(*clientContext),
@@ -140,10 +130,9 @@ static void initSourceNodeScanState(ScanNodeTableInfo& sourceInfo,
     std::unique_ptr<TableScanState>& sourceScanState, ValueVector* boundNodeIDVector,
     const std::vector<ValueVector*>& sourceNodeOutVectors, main::ClientContext* context) {
     sourceScanState = createSourceNodeTableScanState(sourceInfo.table->ptrCast<NodeTable>(),
-        boundNodeIDVector, sourceNodeOutVectors, MemoryManager::Get(*context));
+        boundNodeIDVector, sourceNodeOutVectors);
     sourceInfo.initScanState(*sourceScanState, sourceNodeOutVectors, context);
-    if (dynamic_cast<IceDiskNodeTable*>(sourceInfo.table) ||
-        dynamic_cast<ArrowNodeTable*>(sourceInfo.table)) {
+    if (dynamic_cast<ColumnarNodeTableBase*>(sourceInfo.table)) {
         sourceInfo.table->initScanState(transaction::Transaction::Get(*context), *sourceScanState);
     }
 }
diff --git a/src/storage/table/arrow_node_table.cpp b/src/storage/table/arrow_node_table.cpp
index a139e1464..bcbd3c93e 100644
--- a/src/storage/table/arrow_node_table.cpp
+++ b/src/storage/table/arrow_node_table.cpp
@@ -50,6 +50,11 @@ ArrowNodeTable::~ArrowNodeTable() {
     }
 }
 
+std::unique_ptr<TableScanState> ArrowNodeTable::createScanState(common::ValueVector* nodeIDVector,
+    const std::vector<common::ValueVector*>& outVectors) const {
+    return std::make_unique<ArrowNodeTableScanState>(nodeIDVector, outVectors, nodeIDVector->state);
+}
+
 void ArrowNodeTable::initializeScanCoordination(const transaction::Transaction* transaction) {
     auto arrowScanSharedState =
         static_cast<ArrowNodeTableScanSharedState*>(tableScanSharedState.get());
@@ -135,11 +140,6 @@ bool ArrowNodeTable::scanInternal([[maybe_unused]] transaction::Transaction* tra
     return true;
 }
 
-common::node_group_idx_t ArrowNodeTable::getNumBatches(
-    [[maybe_unused]] const transaction::Transaction* transaction) const {
-    return arrays.size();
-}
-
 common::row_idx_t ArrowNodeTable::getTotalRowCount(
     [[maybe_unused]] const transaction::Transaction* transaction) const {
     return totalRows;
@@ -156,9 +156,9 @@ std::vector<size_t> ArrowNodeTable::getBatchSizes(
     return batchSizes;
 }
 
-size_t ArrowNodeTable::getNumScanMorsels(
+common::node_group_idx_t ArrowNodeTable::getNumScanMorsels(
     [[maybe_unused]] const transaction::Transaction* transaction) const {
-    size_t numMorsels = 0;
+    common::node_group_idx_t numMorsels = 0;
     for (const auto& array : arrays) {
         auto batchLength = getArrowBatchLength(array);
         numMorsels += (batchLength + scanMorselSize - 1) / scanMorselSize;
diff --git a/src/storage/table/arrow_rel_table.cpp b/src/storage/table/arrow_rel_table.cpp
index 327443eaf..a532b397e 100644
--- a/src/storage/table/arrow_rel_table.cpp
+++ b/src/storage/table/arrow_rel_table.cpp
@@ -38,6 +38,13 @@ static int64_t findColumnIdx(const ArrowSchemaWrapper& schema, const std::string
     return -1;
 }
 
+std::unique_ptr<TableScanState> ArrowRelTable::createScanState(common::ValueVector* nodeIDVector,
+    const std::vector<common::ValueVector*>& outVectors,
+    std::shared_ptr<common::DataChunkState> outChunkState) const {
+    return std::make_unique<storage::ArrowRelTableScanState>(*(this->memoryManager), nodeIDVector,
+        outVectors, outChunkState);
+}
+
 void ArrowRelTableScanState::setToTable(const transaction::Transaction* transaction, Table* table_,
     std::vector<column_id_t> columnIDs_, std::vector<ColumnPredicateSet> columnPredicateSets_,
     RelDataDirection direction_) {
diff --git a/src/storage/table/ice_disk_node_table.cpp b/src/storage/table/ice_disk_node_table.cpp
index d61b02947..fe1c33382 100644
--- a/src/storage/table/ice_disk_node_table.cpp
+++ b/src/storage/table/ice_disk_node_table.cpp
@@ -40,11 +40,17 @@ IceDiskNodeTable::IceDiskNodeTable(const StorageManager* storageManager,
     parquetFilePath = resolvedPath;
 }
 
+std::unique_ptr<TableScanState> IceDiskNodeTable::createScanState(common::ValueVector* nodeIDVector,
+    const std::vector<common::ValueVector*>& outVectors) const {
+    return std::make_unique<IceDiskNodeTableScanState>(nodeIDVector, outVectors,
+        nodeIDVector->state);
+}
+
 void IceDiskNodeTable::initializeScanCoordination(const transaction::Transaction* transaction) {
     auto iceDiskScanSharedState =
         static_cast<IceDiskNodeTableScanSharedState*>(tableScanSharedState.get());
-    auto numBatches = getNumBatches(transaction);
-    iceDiskScanSharedState->reset(numBatches);
+    auto numMorsels = getNumRowGroups(transaction);
+    iceDiskScanSharedState->reset(numMorsels);
 }
 
 void IceDiskNodeTable::initScanState(Transaction* transaction, TableScanState& scanState,
@@ -85,7 +91,13 @@ void IceDiskNodeTable::initScanState(Transaction* transaction, TableScanState& s
     initParquetScanForRowGroup(transaction, iceDiskScanState);
 }
 
-common::node_group_idx_t IceDiskNodeTable::getNumBatches(const Transaction* transaction) const {
+common::node_group_idx_t IceDiskNodeTable::getNumRowGroups(
+    const transaction::Transaction* transaction) const {
+    const auto cached = cachedNumRowGroups.load(std::memory_order_relaxed);
+    if (cached != INVALID_NODE_GROUP_IDX) {
+        return cached;
+    }
+
     auto context = transaction->getClientContext();
     if (!context) {
         return 1;
@@ -94,12 +106,20 @@ common::node_group_idx_t IceDiskNodeTable::getNumBatches(const Transaction* tran
     std::vector<bool> columnSkips;
     try {
         auto tempReader = std::make_unique<ParquetReader>(parquetFilePath, columnSkips, context);
-        return tempReader->getNumRowGroups();
+        const auto numRowGroups =
+            static_cast<common::node_group_idx_t>(tempReader->getNumRowGroups());
+        cachedNumRowGroups.store(numRowGroups, std::memory_order_relaxed);
+        return numRowGroups;
     } catch (const std::exception& e) {
         return 1; // Fallback
     }
 }
 
+common::node_group_idx_t IceDiskNodeTable::getNumScanMorsels(
+    const transaction::Transaction* transaction) const {
+    return getNumRowGroups(transaction);
+}
+
 void IceDiskNodeTable::initParquetScanForRowGroup(Transaction* transaction,
     IceDiskNodeTableScanState& iceDiskScanState) const {
     auto context = transaction->getClientContext();
diff --git a/src/storage/table/ice_disk_rel_table.cpp b/src/storage/table/ice_disk_rel_table.cpp
index e662b7162..ca6d32b77 100644
--- a/src/storage/table/ice_disk_rel_table.cpp
+++ b/src/storage/table/ice_disk_rel_table.cpp
@@ -22,6 +22,13 @@ using namespace lbug::transaction;
 namespace lbug {
 namespace storage {
 
+std::unique_ptr<TableScanState> IceDiskRelTable::createScanState(common::ValueVector* nodeIDVector,
+    const std::vector<common::ValueVector*>& outVectors,
+    std::shared_ptr<common::DataChunkState> outChunkState) const {
+    return std::make_unique<storage::IceDiskRelTableScanState>(*(this->memoryManager), nodeIDVector,
+        outVectors, outChunkState);
+}
+
 void IceDiskRelTableScanState::setToTable(const Transaction* transaction, Table* table_,
     std::vector<column_id_t> columnIDs_, std::vector<ColumnPredicateSet> columnPredicateSets_,
     RelDataDirection direction_) {