diff --git a/dataset b/dataset
index 1eb7993b8c..8eb1aa703c 160000
--- a/dataset
+++ b/dataset
@@ -1 +1 @@
-Subproject commit 1eb7993b8caeed3b75cb931698522640b9557481
+Subproject commit 8eb1aa703c7b4a34a2e7500b29b032316283ff28
diff --git a/docs/icebug-disk.md b/docs/icebug-disk.md
index 3f48fb6494..5976fb5749 100644
--- a/docs/icebug-disk.md
+++ b/docs/icebug-disk.md
@@ -25,7 +25,7 @@ CREATE REL TABLE livesin(FROM user TO city) WITH (storage = '<path-to-dir>', for
 
 File paths can be relative or absolute and are resolved as `<path-to-dir>/nodes_{tableName}.parquet` for node tables, and `<path-to-dir>/indices_{tableName}.parquet` and `<path-to-dir>/indptr_{tableName}.parquet` for relationship tables.
 
-Object-store URIs (e.g. `s3://bucket/path`, `https://host/path`) are supported as `storage` values and are passed through unchanged.
+Object-store URIs (e.g. `s3://bucket/path`, `https://host/path`) are also supported as `storage` values.
 
 If `storage` is omitted when `format = 'icebug-disk'` is set, files are resolved relative to the current working directory.
 
diff --git a/src/binder/bind/bind_ddl.cpp b/src/binder/bind/bind_ddl.cpp
index b71327c9b2..d2b5487c35 100644
--- a/src/binder/bind/bind_ddl.cpp
+++ b/src/binder/bind/bind_ddl.cpp
@@ -15,6 +15,7 @@
 #include "catalog/catalog_entry/sequence_catalog_entry.h"
 #include "common/constants.h"
 #include "common/enums/extend_direction_util.h"
+#include "common/enums/storage_format.h"
 #include "common/exception/binder.h"
 #include "common/exception/message.h"
 #include "common/string_utils.h"
@@ -203,12 +204,12 @@ static ExtendDirection getStorageDirection(const case_insensitive_map_t<Value>&
     return DEFAULT_EXTEND_DIRECTION;
 }
 
-static std::string getStorageFormat(const case_insensitive_map_t<Value>& options) {
+static StorageFormat getStorageFormat(const case_insensitive_map_t<Value>& options) {
     if (options.contains(TableOptionConstants::STORAGE_FORMAT_OPTION)) {
-        return options.at(TableOptionConstants::STORAGE_FORMAT_OPTION).toString();
+        return StorageFormatUtils::fromString(
+            options.at(TableOptionConstants::STORAGE_FORMAT_OPTION).toString());
     }
-
-    return "";
+    return StorageFormat::NONE;
 }
 
 BoundCreateTableInfo Binder::bindCreateNodeTableInfo(const CreateTableInfo* info) {
@@ -252,8 +253,7 @@ BoundCreateTableInfo Binder::bindCreateRelTableGroupInfo(const CreateTableInfo*
         // Handle special case where icebug-disk storage could contain a dot
         // Otherwise, treat as file path (e.g., "dataset/demo-db/icebug-disk/demo" or
         // "data.parquet")
-        if (!TableOptionConstants::isIceBugDiskFormat(storageFormat) &&
-            dotPos != std::string::npos) {
+        if (storageFormat != StorageFormat::ICEBUG_DISK && dotPos != std::string::npos) {
             std::string dbName = storage.substr(0, dotPos);
             std::string tableName = storage.substr(dotPos + 1);
             if (!dbName.empty()) {
@@ -344,17 +344,15 @@ BoundCreateTableInfo Binder::bindCreateRelTableGroupInfo(const CreateTableInfo*
             }
         }
 
-        bool isSrcIcebugDisk =
-            srcEntry->getType() == CatalogEntryType::NODE_TABLE_ENTRY ?
-                TableOptionConstants::isIceBugDiskFormat(
-                    srcEntry->ptrCast<NodeTableCatalogEntry>()->getStorageFormat()) :
-                false;
-        bool isDstIcebugDisk =
-            dstEntry->getType() == CatalogEntryType::NODE_TABLE_ENTRY ?
-                TableOptionConstants::isIceBugDiskFormat(
-                    dstEntry->ptrCast<NodeTableCatalogEntry>()->getStorageFormat()) :
-                false;
-        bool isRelIcebugDisk = TableOptionConstants::isIceBugDiskFormat(storageFormat);
+        bool isSrcIcebugDisk = srcEntry->getType() == CatalogEntryType::NODE_TABLE_ENTRY ?
+                                   srcEntry->ptrCast<NodeTableCatalogEntry>()->getStorageFormat() ==
+                                       StorageFormat::ICEBUG_DISK :
+                                   false;
+        bool isDstIcebugDisk = dstEntry->getType() == CatalogEntryType::NODE_TABLE_ENTRY ?
+                                   dstEntry->ptrCast<NodeTableCatalogEntry>()->getStorageFormat() ==
+                                       StorageFormat::ICEBUG_DISK :
+                                   false;
+        bool isRelIcebugDisk = (storageFormat == StorageFormat::ICEBUG_DISK);
 
         // We don't allow mixing icebug-disk tables with non-icebug-disk tables
         // We only allow icebug-disk rel tables to connect icebug-disk node tables
@@ -609,7 +607,7 @@ static void validateNotIceDiskTable(main::ClientContext* clientContext,
     }
 
     auto tableEntry = catalog->getTableCatalogEntry(transaction, tableName);
-    std::string storageFormat;
+    StorageFormat storageFormat = StorageFormat::NONE;
 
     if (tableEntry->getTableType() == common::TableType::NODE) {
         storageFormat = tableEntry->ptrCast<NodeTableCatalogEntry>()->getStorageFormat();
@@ -617,7 +615,7 @@ static void validateNotIceDiskTable(main::ClientContext* clientContext,
         storageFormat = tableEntry->ptrCast<RelGroupCatalogEntry>()->getStorageFormat();
     }
 
-    if (TableOptionConstants::isIceBugDiskFormat(storageFormat)) {
+    if (storageFormat == StorageFormat::ICEBUG_DISK) {
         throw BinderException(
             std::format("Cannot alter table {}: icebug-disk tables are immutable.", tableName));
     }
diff --git a/src/catalog/catalog_entry/node_table_catalog_entry.cpp b/src/catalog/catalog_entry/node_table_catalog_entry.cpp
index fc9279db35..0c1411be6e 100644
--- a/src/catalog/catalog_entry/node_table_catalog_entry.cpp
+++ b/src/catalog/catalog_entry/node_table_catalog_entry.cpp
@@ -2,50 +2,26 @@
 
 #include "binder/ddl/bound_create_table_info.h"
 #include "common/constants.h"
-#include "common/serializer/buffered_file.h"
+#include "common/enums/storage_format.h"
 #include "common/serializer/deserializer.h"
 #include "common/string_utils.h"
 #include "storage/storage_version_info.h"
 #include <format>
 
 using namespace lbug::binder;
+using namespace lbug::common;
 
 namespace lbug {
 namespace catalog {
 
-static void upgradeLegacyStorageFormat(const std::string& storage, std::string& storageFormat) {
+static void upgradeLegacyStorageFormat(const std::string& storage,
+    common::StorageFormat& storageFormat) {
     const auto lowerStorage = common::StringUtils::getLower(storage);
     if (lowerStorage.ends_with("parquet")) {
-        storageFormat = std::string(common::TableOptionConstants::ICEBUG_DISK_FORMAT);
+        storageFormat = common::StorageFormat::ICEBUG_DISK;
     }
 }
 
-static bool tryDeserializeStorageFormat(common::Deserializer& deserializer,
-    std::string& storageFormat) {
-    auto* reader = dynamic_cast<common::BufferedFileReader*>(deserializer.getReader());
-    if (reader == nullptr) {
-        deserializer.deserializeValue(storageFormat);
-        return true;
-    }
-    const auto readOffset = reader->getReadOffset();
-    uint64_t valueLength = 0;
-    deserializer.deserializeValue(valueLength);
-    constexpr uint64_t MAX_STORAGE_FORMAT_LENGTH = 1024;
-    if (valueLength > MAX_STORAGE_FORMAT_LENGTH) {
-        reader->resetReadOffset(readOffset);
-        return false;
-    }
-    storageFormat.resize(valueLength);
-    deserializer.read(reinterpret_cast<uint8_t*>(storageFormat.data()), valueLength);
-    if (!storageFormat.empty() &&
-        !common::TableOptionConstants::isIceBugDiskFormat(storageFormat)) {
-        reader->resetReadOffset(readOffset);
-        storageFormat.clear();
-        return false;
-    }
-    return true;
-}
-
 void NodeTableCatalogEntry::renameProperty(const std::string& propertyName,
     const std::string& newName) {
     TableCatalogEntry::renameProperty(propertyName, newName);
@@ -61,7 +37,7 @@ void NodeTableCatalogEntry::serialize(common::Serializer& serializer) const {
     serializer.writeDebuggingInfo("storage");
     serializer.write(storage);
     serializer.writeDebuggingInfo("storageFormat");
-    serializer.write(storageFormat);
+    serializer.serializeValue(storageFormat);
 }
 
 std::unique_ptr<NodeTableCatalogEntry> NodeTableCatalogEntry::deserialize(
@@ -69,7 +45,7 @@ std::unique_ptr<NodeTableCatalogEntry> NodeTableCatalogEntry::deserialize(
     std::string debuggingInfo;
     std::string primaryKeyName;
     std::string storage;
-    std::string storageFormat;
+    auto storageFormat = StorageFormat::NONE;
     deserializer.validateDebuggingInfo(debuggingInfo, "primaryKeyName");
     deserializer.deserializeValue(primaryKeyName);
     deserializer.validateDebuggingInfo(debuggingInfo, "storage");
@@ -77,9 +53,7 @@ std::unique_ptr<NodeTableCatalogEntry> NodeTableCatalogEntry::deserialize(
     if (deserializer.getStorageVersion() >=
         ::lbug::storage::StorageVersionInfo::STORAGE_VERSION_41) {
         deserializer.validateDebuggingInfo(debuggingInfo, "storageFormat");
-        if (!tryDeserializeStorageFormat(deserializer, storageFormat)) {
-            upgradeLegacyStorageFormat(storage, storageFormat);
-        }
+        deserializer.deserializeValue(storageFormat);
     } else {
         upgradeLegacyStorageFormat(storage, storageFormat);
     }
diff --git a/src/catalog/catalog_entry/rel_group_catalog_entry.cpp b/src/catalog/catalog_entry/rel_group_catalog_entry.cpp
index f54fad4cd1..ee0e17222d 100644
--- a/src/catalog/catalog_entry/rel_group_catalog_entry.cpp
+++ b/src/catalog/catalog_entry/rel_group_catalog_entry.cpp
@@ -5,7 +5,7 @@
 #include "binder/ddl/bound_create_table_info.h"
 #include "catalog/catalog.h"
 #include "common/constants.h"
-#include "common/serializer/buffered_file.h"
+#include "common/enums/storage_format.h"
 #include "common/serializer/deserializer.h"
 #include "common/string_utils.h"
 #include "storage/storage_version_info.h"
@@ -18,38 +18,14 @@ using namespace lbug::main;
 namespace lbug {
 namespace catalog {
 
-static void upgradeLegacyStorageFormat(const std::string& storage, std::string& storageFormat) {
+static void upgradeLegacyStorageFormat(const std::string& storage,
+    common::StorageFormat& storageFormat) {
     const auto lowerStorage = common::StringUtils::getLower(storage);
     if (lowerStorage.ends_with("parquet")) {
-        storageFormat = std::string(common::TableOptionConstants::ICEBUG_DISK_FORMAT);
+        storageFormat = common::StorageFormat::ICEBUG_DISK;
     }
 }
 
-static bool tryDeserializeStorageFormat(Deserializer& deserializer, std::string& storageFormat) {
-    auto* reader = dynamic_cast<common::BufferedFileReader*>(deserializer.getReader());
-    if (reader == nullptr) {
-        deserializer.deserializeValue(storageFormat);
-        return true;
-    }
-    const auto readOffset = reader->getReadOffset();
-    uint64_t valueLength = 0;
-    deserializer.deserializeValue(valueLength);
-    constexpr uint64_t MAX_STORAGE_FORMAT_LENGTH = 1024;
-    if (valueLength > MAX_STORAGE_FORMAT_LENGTH) {
-        reader->resetReadOffset(readOffset);
-        return false;
-    }
-    storageFormat.resize(valueLength);
-    deserializer.read(reinterpret_cast<uint8_t*>(storageFormat.data()), valueLength);
-    if (!storageFormat.empty() &&
-        !common::TableOptionConstants::isIceBugDiskFormat(storageFormat)) {
-        reader->resetReadOffset(readOffset);
-        storageFormat.clear();
-        return false;
-    }
-    return true;
-}
-
 void RelGroupCatalogEntry::addFromToConnection(table_id_t srcTableID, table_id_t dstTableID,
     oid_t oid) {
     relTableInfos.emplace_back(NodeTableIDPair{srcTableID, dstTableID}, oid);
@@ -152,7 +128,7 @@ std::unique_ptr<RelGroupCatalogEntry> RelGroupCatalogEntry::deserialize(
     auto dstMultiplicity = RelMultiplicity::MANY;
     auto storageDirection = ExtendDirection::BOTH;
     std::string storage;
-    std::string storageFormat;
+    auto storageFormat = StorageFormat::NONE;
     std::vector<RelTableCatalogInfo> relTableInfos;
     deserializer.validateDebuggingInfo(debuggingInfo, "srcMultiplicity");
     deserializer.deserializeValue(srcMultiplicity);
@@ -174,9 +150,7 @@ std::unique_ptr<RelGroupCatalogEntry> RelGroupCatalogEntry::deserialize(
     if (deserializer.getStorageVersion() >=
         ::lbug::storage::StorageVersionInfo::STORAGE_VERSION_41) {
         deserializer.validateDebuggingInfo(debuggingInfo, "storageFormat");
-        if (!tryDeserializeStorageFormat(deserializer, storageFormat)) {
-            upgradeLegacyStorageFormat(storage, storageFormat);
-        }
+        deserializer.deserializeValue(storageFormat);
     } else {
         upgradeLegacyStorageFormat(storage, storageFormat);
     }
diff --git a/src/common/enums/CMakeLists.txt b/src/common/enums/CMakeLists.txt
index f789e52a0d..1205d9fe0d 100644
--- a/src/common/enums/CMakeLists.txt
+++ b/src/common/enums/CMakeLists.txt
@@ -1,6 +1,7 @@
 add_library(lbug_common_enums
         OBJECT
         accumulate_type.cpp
+        storage_format.cpp
         path_semantic.cpp
         query_rel_type.cpp
         rel_direction.cpp
diff --git a/src/common/enums/storage_format.cpp b/src/common/enums/storage_format.cpp
new file mode 100644
index 0000000000..b1da19c8f1
--- /dev/null
+++ b/src/common/enums/storage_format.cpp
@@ -0,0 +1,18 @@
+#include "common/enums/storage_format.h"
+
+#include "common/exception/binder.h"
+#include <format>
+
+namespace lbug {
+namespace common {
+
+StorageFormat StorageFormatUtils::fromString(const std::string& str) {
+    if (str == "icebug-disk") {
+        return StorageFormat::ICEBUG_DISK;
+    }
+    throw BinderException(
+        std::format("Unsupported storage format '{}'. Valid options are: icebug-disk.", str));
+}
+
+} // namespace common
+} // namespace lbug
diff --git a/src/include/binder/ddl/bound_create_table_info.h b/src/include/binder/ddl/bound_create_table_info.h
index 2466a0dead..94d2f1096a 100644
--- a/src/include/binder/ddl/bound_create_table_info.h
+++ b/src/include/binder/ddl/bound_create_table_info.h
@@ -7,6 +7,7 @@
 #include "common/enums/conflict_action.h"
 #include "common/enums/extend_direction.h"
 #include "common/enums/rel_multiplicity.h"
+#include "common/enums/storage_format.h"
 #include "function/table/bind_data.h"
 #include "function/table/table_function.h"
 #include "property_definition.h"
@@ -76,14 +77,14 @@ struct LBUG_API BoundExtraCreateTableInfo : BoundExtraCreateCatalogEntryInfo {
 struct BoundExtraCreateNodeTableInfo final : BoundExtraCreateTableInfo {
     std::string primaryKeyName;
     std::string storage;
-    std::string storageFormat;
+    common::StorageFormat storageFormat = common::StorageFormat::NONE;
 
     BoundExtraCreateNodeTableInfo(std::string primaryKeyName,
         std::vector<PropertyDefinition> definitions, std::string storage = "",
-        std::string storageFormat = "")
+        common::StorageFormat storageFormat = common::StorageFormat::NONE)
         : BoundExtraCreateTableInfo{std::move(definitions)},
           primaryKeyName{std::move(primaryKeyName)}, storage{std::move(storage)},
-          storageFormat{std::move(storageFormat)} {}
+          storageFormat{storageFormat} {}
     BoundExtraCreateNodeTableInfo(const BoundExtraCreateNodeTableInfo& other)
         : BoundExtraCreateTableInfo{copyVector(other.propertyDefinitions)},
           primaryKeyName{other.primaryKeyName}, storage{other.storage},
@@ -100,7 +101,7 @@ struct BoundExtraCreateRelTableGroupInfo final : BoundExtraCreateTableInfo {
     common::ExtendDirection storageDirection;
     std::vector<catalog::NodeTableIDPair> nodePairs;
     std::string storage;
-    std::string storageFormat;
+    common::StorageFormat storageFormat = common::StorageFormat::NONE;
     std::optional<function::TableFunction> scanFunction;
     std::optional<std::shared_ptr<function::TableFuncBindData>> scanBindData;
     std::string foreignDatabaseName;
@@ -108,14 +109,14 @@ struct BoundExtraCreateRelTableGroupInfo final : BoundExtraCreateTableInfo {
     explicit BoundExtraCreateRelTableGroupInfo(std::vector<PropertyDefinition> definitions,
         common::RelMultiplicity srcMultiplicity, common::RelMultiplicity dstMultiplicity,
         common::ExtendDirection storageDirection, std::vector<catalog::NodeTableIDPair> nodePairs,
-        std::string storage = "", std::string storageFormat = "",
+        std::string storage = "", common::StorageFormat storageFormat = common::StorageFormat::NONE,
         std::optional<function::TableFunction> scanFunction = std::nullopt,
         std::optional<std::shared_ptr<function::TableFuncBindData>> scanBindData = std::nullopt,
         std::string foreignDatabaseName = "")
         : BoundExtraCreateTableInfo{std::move(definitions)}, srcMultiplicity{srcMultiplicity},
           dstMultiplicity{dstMultiplicity}, storageDirection{storageDirection},
           nodePairs{std::move(nodePairs)}, storage{std::move(storage)},
-          storageFormat{std::move(storageFormat)}, scanFunction{std::move(scanFunction)},
+          storageFormat{storageFormat}, scanFunction{std::move(scanFunction)},
           scanBindData{std::move(scanBindData)},
           foreignDatabaseName{std::move(foreignDatabaseName)} {}
 
diff --git a/src/include/catalog/catalog_entry/node_table_catalog_entry.h b/src/include/catalog/catalog_entry/node_table_catalog_entry.h
index fe466c222a..15773b80b9 100644
--- a/src/include/catalog/catalog_entry/node_table_catalog_entry.h
+++ b/src/include/catalog/catalog_entry/node_table_catalog_entry.h
@@ -3,6 +3,7 @@
 #include <functional>
 #include <optional>
 
+#include "common/enums/storage_format.h"
 #include "function/table/table_function.h"
 #include "table_catalog_entry.h"
 
@@ -28,9 +29,9 @@ class LBUG_API NodeTableCatalogEntry final : public TableCatalogEntry {
 public:
     NodeTableCatalogEntry() = default;
     NodeTableCatalogEntry(std::string name, std::string primaryKeyName, std::string storage = "",
-        std::string storageFormat = "")
+        common::StorageFormat storageFormat = common::StorageFormat::NONE)
         : TableCatalogEntry{entryType_, std::move(name)}, primaryKeyName{std::move(primaryKeyName)},
-          storage{std::move(storage)}, storageFormat{std::move(storageFormat)} {}
+          storage{std::move(storage)}, storageFormat{storageFormat} {}
 
     // Constructor for foreign-backed tables
     NodeTableCatalogEntry(std::string name, std::string primaryKeyName,
@@ -57,7 +58,7 @@ class LBUG_API NodeTableCatalogEntry final : public TableCatalogEntry {
         return getProperty(primaryKeyName);
     }
     const std::string& getStorage() const { return storage; }
-    const std::string& getStorageFormat() const { return storageFormat; }
+    common::StorageFormat getStorageFormat() const { return storageFormat; }
     std::optional<function::TableFunction> getScanFunction() const override;
     const CreateBindDataFunc& getCreateBindDataFunc() const { return createBindDataFunc; }
     const std::string& getForeignDatabaseName() const { return foreignDatabaseName; }
@@ -84,7 +85,7 @@ class LBUG_API NodeTableCatalogEntry final : public TableCatalogEntry {
 private:
     std::string primaryKeyName;
     std::string storage;
-    std::string storageFormat;
+    common::StorageFormat storageFormat = common::StorageFormat::NONE;
     std::optional<function::TableFunction> scanFunction;
     CreateBindDataFunc createBindDataFunc; // Callback to create bind data
     std::string foreignDatabaseName;
diff --git a/src/include/catalog/catalog_entry/rel_group_catalog_entry.h b/src/include/catalog/catalog_entry/rel_group_catalog_entry.h
index 7d6e792bf1..c2a0e4ec92 100644
--- a/src/include/catalog/catalog_entry/rel_group_catalog_entry.h
+++ b/src/include/catalog/catalog_entry/rel_group_catalog_entry.h
@@ -6,6 +6,7 @@
 #include "common/enums/extend_direction.h"
 #include "common/enums/rel_direction.h"
 #include "common/enums/rel_multiplicity.h"
+#include "common/enums/storage_format.h"
 #include "function/table/bind_data.h"
 #include "function/table/table_function.h"
 #include "node_table_id_pair.h"
@@ -39,14 +40,14 @@ class LBUG_API RelGroupCatalogEntry final : public TableCatalogEntry {
     RelGroupCatalogEntry(std::string tableName, common::RelMultiplicity srcMultiplicity,
         common::RelMultiplicity dstMultiplicity, common::ExtendDirection storageDirection,
         std::vector<RelTableCatalogInfo> relTableInfos, std::string storage = "",
-        std::string storageFormat = "",
+        common::StorageFormat storageFormat = common::StorageFormat::NONE,
         std::optional<function::TableFunction> scanFunction = std::nullopt,
         std::optional<std::shared_ptr<function::TableFuncBindData>> scanBindData = std::nullopt,
         std::string foreignDatabaseName = "")
         : TableCatalogEntry{type_, std::move(tableName)}, srcMultiplicity{srcMultiplicity},
           dstMultiplicity{dstMultiplicity}, storageDirection{storageDirection},
           relTableInfos{std::move(relTableInfos)}, storage{std::move(storage)},
-          storageFormat{std::move(storageFormat)}, scanFunction{std::move(scanFunction)},
+          storageFormat{storageFormat}, scanFunction{std::move(scanFunction)},
           scanBindData{std::move(scanBindData)},
           foreignDatabaseName{std::move(foreignDatabaseName)} {
         propertyCollection =
@@ -65,7 +66,7 @@ class LBUG_API RelGroupCatalogEntry final : public TableCatalogEntry {
 
     common::ExtendDirection getStorageDirection() const { return storageDirection; }
     const std::string& getStorage() const { return storage; }
-    const std::string& getStorageFormat() const { return storageFormat; }
+    common::StorageFormat getStorageFormat() const { return storageFormat; }
     std::optional<function::TableFunction> getScanFunction() const override { return scanFunction; }
     const std::optional<std::shared_ptr<function::TableFuncBindData>>& getScanBindData() const {
         return scanBindData;
@@ -116,7 +117,7 @@ class LBUG_API RelGroupCatalogEntry final : public TableCatalogEntry {
     common::ExtendDirection storageDirection = common::ExtendDirection::BOTH;
     std::vector<RelTableCatalogInfo> relTableInfos;
     std::string storage;
-    std::string storageFormat;
+    common::StorageFormat storageFormat = common::StorageFormat::NONE;
     std::optional<function::TableFunction> scanFunction;
     std::optional<std::shared_ptr<function::TableFuncBindData>> scanBindData;
     std::string foreignDatabaseName; // Database name for foreign-backed rel tables
diff --git a/src/include/common/constants.h b/src/include/common/constants.h
index bf4c38ba1d..16e018d55b 100644
--- a/src/include/common/constants.h
+++ b/src/include/common/constants.h
@@ -87,12 +87,6 @@ struct TableOptionConstants {
     static constexpr char REL_STORAGE_DIRECTION_OPTION[] = "STORAGE_DIRECTION";
     static constexpr char REL_STORAGE_OPTION[] = "STORAGE";
     static constexpr char STORAGE_FORMAT_OPTION[] = "FORMAT";
-
-    static constexpr std::string_view ICEBUG_DISK_FORMAT = "icebug-disk";
-
-    static bool isIceBugDiskFormat(const std::string& format) {
-        return format.find(ICEBUG_DISK_FORMAT) != std::string::npos;
-    }
 };
 
 // Hash Index Configurations
diff --git a/src/include/common/enums/storage_format.h b/src/include/common/enums/storage_format.h
new file mode 100644
index 0000000000..58b39f14e1
--- /dev/null
+++ b/src/include/common/enums/storage_format.h
@@ -0,0 +1,16 @@
+#pragma once
+
+#include <cstdint>
+#include <string>
+
+namespace lbug {
+namespace common {
+
+enum class StorageFormat : uint8_t { NONE, ICEBUG_DISK };
+
+struct StorageFormatUtils {
+    static StorageFormat fromString(const std::string& str);
+};
+
+} // namespace common
+} // namespace lbug
diff --git a/src/include/storage/storage_version_info.h b/src/include/storage/storage_version_info.h
index 0e6a54fb9e..577c87f82b 100644
--- a/src/include/storage/storage_version_info.h
+++ b/src/include/storage/storage_version_info.h
@@ -15,7 +15,7 @@ struct StorageVersionInfo {
     // Storage version 40 spans the releases after 0.11.0 where the on-disk catalog/data format did
     // not change.
     static constexpr storage_version_t STORAGE_VERSION_40 = 40;
-    // Storage version 41 adds the table storage FORMAT field to catalog entries.
+    // Storage version 41 adds the table storage FORMAT field to catalog entries (enum encoding).
     static constexpr storage_version_t STORAGE_VERSION_41 = 41;
 
     static std::unordered_map<std::string, storage_version_t> getStorageVersionInfo() {
diff --git a/src/include/storage/table/ice_disk_rel_table.h b/src/include/storage/table/ice_disk_rel_table.h
index 975424d6db..6c5b2d947c 100644
--- a/src/include/storage/table/ice_disk_rel_table.h
+++ b/src/include/storage/table/ice_disk_rel_table.h
@@ -16,9 +16,14 @@ namespace storage {
 struct IceDiskRelTableScanState final : RelTableScanState {
     std::unique_ptr<processor::ParquetReaderScanState> parquetScanState;
 
-    // Row group range for morsel-driven parallelism
-    uint64_t currentRowGroup = 0;
-    uint64_t endRowGroup = 0;
+    // cached data for the current batch in current row group
+    std::unique_ptr<common::DataChunk> cachedBatchData;
+    common::offset_t currentBatchStartOffset =
+        0; // Global row index of the start of the current batch of the current row group
+    common::offset_t currentLocalRowIdx =
+        0; // Row index within the current batch of the current row group
+    std::unordered_map<common::offset_t, common::sel_t>
+        boundNodeOffsets; // Map from bound node offset to selection vector index
 
     // Per-scan-state readers for thread safety
     std::unique_ptr<processor::ParquetReader> indicesReader;
@@ -35,6 +40,15 @@ struct IceDiskRelTableScanState final : RelTableScanState {
         std::vector<common::column_id_t> columnIDs_,
         std::vector<ColumnPredicateSet> columnPredicateSets_,
         common::RelDataDirection direction_) override;
+
+    void reset(std::unordered_map<common::offset_t, common::sel_t> boundNodeOffsets_) {
+        cachedBatchData = nullptr;
+        currentBatchStartOffset = 0;
+        currentLocalRowIdx = 0;
+        boundNodeOffsets = std::move(boundNodeOffsets_);
+    }
+
+    void reloadCachedBatchData(transaction::Transaction* transaction);
 };
 
 class IceDiskRelTable final : public ColumnarRelTableBase {
@@ -65,11 +79,6 @@ class IceDiskRelTable final : public ColumnarRelTableBase {
     void initializeParquetReaders(transaction::Transaction* transaction) const;
     void initializeIndptrReader(transaction::Transaction* transaction) const;
     void loadIndptrData(transaction::Transaction* transaction) const;
-    bool scanInternalByRowGroups(transaction::Transaction* transaction,
-        IceDiskRelTableScanState& iceDiskScanState);
-    bool scanRowGroupForBoundNodes(transaction::Transaction* transaction,
-        IceDiskRelTableScanState& iceDiskScanState, const std::vector<uint64_t>& rowGroupsToProcess,
-        const std::unordered_map<common::offset_t, common::sel_t>& boundNodeOffsets);
     common::offset_t findSourceNodeForRow(common::offset_t globalRowIdx) const;
 };
 
diff --git a/src/include/storage/table/ice_disk_utils.h b/src/include/storage/table/ice_disk_utils.h
index 07b5942036..03101adbae 100644
--- a/src/include/storage/table/ice_disk_utils.h
+++ b/src/include/storage/table/ice_disk_utils.h
@@ -25,6 +25,7 @@ class IceDiskUtils {
 public:
     // Joins a base path with a filename. When base is empty the filename is returned
     // as-is (i.e. relative to the current working directory)
+    // public for testing
     static std::string joinPath(const std::string& base, const std::string& part) {
         if (base.empty()) {
             return part;
diff --git a/src/storage/storage_manager.cpp b/src/storage/storage_manager.cpp
index c232dedc86..6b54d0d2e4 100644
--- a/src/storage/storage_manager.cpp
+++ b/src/storage/storage_manager.cpp
@@ -5,6 +5,7 @@
 #include "catalog/catalog_entry/rel_group_catalog_entry.h"
 #include "common/arrow/arrow.h"
 #include "common/constants.h"
+#include "common/enums/storage_format.h"
 #include "common/file_system/virtual_file_system.h"
 #include "common/random_engine.h"
 #include "common/serializer/in_mem_file_writer.h"
@@ -102,14 +103,15 @@ void StorageManager::recover(main::ClientContext& clientContext, bool throwOnWal
 void StorageManager::createNodeTable(NodeTableCatalogEntry* entry, main::ClientContext* context) {
     tableNameCache[entry->getTableID()] = entry->getName();
 
-    if (!entry->getStorageFormat().empty()) {
-        if (TableOptionConstants::isIceBugDiskFormat(entry->getStorageFormat())) {
+    if (entry->getStorageFormat() != StorageFormat::NONE) {
+        if (entry->getStorageFormat() == StorageFormat::ICEBUG_DISK) {
             // Create icebug-disk-backed node table
             tables[entry->getTableID()] =
                 std::make_unique<IceDiskNodeTable>(this, entry, &memoryManager, context);
         } else {
             throw common::RuntimeException(
-                "Unsupported storage format option for node table: " + entry->getStorageFormat());
+                "Unsupported storage format option for node table: " +
+                std::to_string(static_cast<int>(entry->getStorageFormat())));
         }
     } else if (!entry->getStorage().empty()) {
         // Check if storage is Arrow backed
@@ -155,14 +157,15 @@ void StorageManager::addRelTable(RelGroupCatalogEntry* entry, const RelTableCata
         tables[info.oid] = std::make_unique<ForeignRelTable>(entry, info.nodePair.srcTableID,
             info.nodePair.dstTableID, this, &memoryManager, *entry->getScanFunction(),
             std::move(entry->getScanBindData().value()));
-    } else if (!entry->getStorageFormat().empty()) {
-        if (TableOptionConstants::isIceBugDiskFormat(entry->getStorageFormat())) {
+    } else if (entry->getStorageFormat() != StorageFormat::NONE) {
+        if (entry->getStorageFormat() == StorageFormat::ICEBUG_DISK) {
             // Create icebug-disk-backed rel table
             tables[info.oid] = std::make_unique<IceDiskRelTable>(entry, info.nodePair.srcTableID,
                 info.nodePair.dstTableID, this, &memoryManager, context);
         } else {
             throw common::RuntimeException(
-                "Unsupported storage format option for rel table: " + entry->getStorageFormat());
+                "Unsupported storage format option for rel table: " +
+                std::to_string(static_cast<int>(entry->getStorageFormat())));
         }
     } else if (!entry->getStorage().empty()) {
         if (entry->getStorage().substr(0, 8) == "arrow://") {
@@ -459,7 +462,7 @@ void StorageManager::deserialize(main::ClientContext* context, const Catalog* ca
         auto tableEntry = catalog->getTableCatalogEntry(&DUMMY_TRANSACTION, tableID)
                               ->ptrCast<NodeTableCatalogEntry>();
         tableNameCache[tableID] = tableEntry->getName();
-        if (TableOptionConstants::isIceBugDiskFormat(tableEntry->getStorageFormat())) {
+        if (tableEntry->getStorageFormat() == StorageFormat::ICEBUG_DISK) {
             // Create icebug-disk-backed node table
             tables[tableID] =
                 std::make_unique<IceDiskNodeTable>(this, tableEntry, &memoryManager, context);
@@ -488,7 +491,7 @@ void StorageManager::deserialize(main::ClientContext* context, const Catalog* ca
         for (auto k = 0u; k < numInnerRelTables; k++) {
             RelTableCatalogInfo info = RelTableCatalogInfo::deserialize(deSer);
             DASSERT(!tables.contains(info.oid));
-            if (TableOptionConstants::isIceBugDiskFormat(relGroupEntry->getStorageFormat())) {
+            if (relGroupEntry->getStorageFormat() == StorageFormat::ICEBUG_DISK) {
                 // Create icebug-disk-backed rel table
                 tables[info.oid] =
                     std::make_unique<IceDiskRelTable>(relGroupEntry, info.nodePair.srcTableID,
diff --git a/src/storage/table/ice_disk_node_table.cpp b/src/storage/table/ice_disk_node_table.cpp
index 19fb02e5b4..682c40fd8a 100644
--- a/src/storage/table/ice_disk_node_table.cpp
+++ b/src/storage/table/ice_disk_node_table.cpp
@@ -307,11 +307,21 @@ bool IceDiskNodeTable::scanInternal(Transaction* transaction, TableScanState& sc
         }
     }
 
+    // calc current global row index based on assigned row group and local row index within that
+    // group
+    auto metadata = iceDiskScanState.parquetReader->getMetadata();
+    offset_t startOffset = 0;
+
+    for (common::node_group_idx_t rg = 0;
+         rg < iceDiskScanState.nodeGroupIdx && rg < metadata->row_groups.size(); ++rg) {
+        startOffset += metadata->row_groups[rg].num_rows;
+    }
+
     // Set node ID for this row
     auto tableID = this->getTableID();
     auto& nodeID = scanState.nodeIDVector->getValue<nodeID_t>(0);
     nodeID.tableID = tableID;
-    nodeID.offset = rowIndex; // Use the actual row index from parquet
+    nodeID.offset = startOffset + rowIndex; // Use the actual row index from parquet
 
     scanState.outState->getSelVectorUnsafe().setSelSize(1); // Return exactly one row
     return true;
diff --git a/src/storage/table/ice_disk_rel_table.cpp b/src/storage/table/ice_disk_rel_table.cpp
index a87c68eac7..3ed1004896 100644
--- a/src/storage/table/ice_disk_rel_table.cpp
+++ b/src/storage/table/ice_disk_rel_table.cpp
@@ -42,6 +42,25 @@ void IceDiskRelTableScanState::setToTable(const Transaction* transaction, Table*
     // IceDiskRelTable does not support local storage, so we skip the local table initialization
 }
 
+void IceDiskRelTableScanState::reloadCachedBatchData(Transaction* transaction) {
+    auto context = transaction->getClientContext();
+
+    // Create DataChunk matching the indices parquet file schema
+    auto numIndicesColumns = indicesReader->getNumColumns();
+    cachedBatchData = std::make_unique<DataChunk>(numIndicesColumns);
+
+    // Insert value vectors for all columns in the parquet file
+    auto memoryManager = MemoryManager::Get(*context);
+    for (uint32_t colIdx = 0; colIdx < numIndicesColumns; ++colIdx) {
+        const auto& columnTypeRef = indicesReader->getColumnType(colIdx);
+        auto columnType = columnTypeRef.copy();
+        auto vector = std::make_shared<ValueVector>(std::move(columnType), memoryManager);
+        cachedBatchData->insert(colIdx, vector);
+    }
+
+    indicesReader->scan(*parquetScanState, *cachedBatchData);
+}
+
 IceDiskRelTable::IceDiskRelTable(RelGroupCatalogEntry* relGroupEntry, table_id_t fromTableID,
     table_id_t toTableID, const StorageManager* storageManager, MemoryManager* memoryManager,
     main::ClientContext* context)
@@ -66,28 +85,6 @@ void IceDiskRelTable::initScanState(Transaction* transaction, TableScanState& sc
     relScanState.nodeGroup = nullptr;
     relScanState.nodeGroupIdx = INVALID_NODE_GROUP_IDX;
 
-    // Initialize ParquetReaders for this scan state (per-thread)
-    auto& iceDiskScanState = static_cast<IceDiskRelTableScanState&>(relScanState);
-
-    // Initialize readers if not already done for this scan state
-    if (!iceDiskScanState.indicesReader) {
-        std::vector<bool> columnSkips; // Read all columns
-        auto context = transaction->getClientContext();
-        iceDiskScanState.indicesReader =
-            std::make_unique<ParquetReader>(indicesFilePath, columnSkips, context);
-    }
-    if (!indptrFilePath.empty() && !iceDiskScanState.indptrReader) {
-        std::vector<bool> columnSkips; // Read all columns
-        auto context = transaction->getClientContext();
-        iceDiskScanState.indptrReader =
-            std::make_unique<ParquetReader>(indptrFilePath, columnSkips, context);
-    }
-
-    // Load shared indptr data - thread-safe to read
-    if (!indptrFilePath.empty()) {
-        loadIndptrData(transaction);
-    }
-
     // For morsel-driven parallelism, each scan state maintains its own bound node processing state
     // No shared state needed between threads
     if (resetCachedBoundNodeSelVec) {
@@ -104,11 +101,44 @@ void IceDiskRelTable::initScanState(Transaction* transaction, TableScanState& sc
             relScanState.nodeIDVector->state->getSelVector().getSelSize());
     }
 
-    // Initialize row group ranges for morsel-driven parallelism
-    // For now, assign all row groups to this scan state (will be partitioned by the scan operator)
-    iceDiskScanState.currentRowGroup = 0;
-    iceDiskScanState.endRowGroup =
-        iceDiskScanState.indicesReader ? iceDiskScanState.indicesReader->getNumRowGroups() : 0;
+    // Initialize ParquetReaders for this scan state (per-thread)
+    auto context = transaction->getClientContext();
+    auto vfs = VirtualFileSystem::GetUnsafe(*context);
+    auto& iceDiskScanState = static_cast<IceDiskRelTableScanState&>(relScanState);
+
+    // Initialize readers if not already done for this scan state
+    if (!iceDiskScanState.indicesReader) {
+        iceDiskScanState.indicesReader =
+            std::make_unique<ParquetReader>(indicesFilePath, std::vector<bool>{}, context);
+    }
+
+    if (!iceDiskScanState.indptrReader) {
+        iceDiskScanState.indptrReader =
+            std::make_unique<ParquetReader>(indptrFilePath, std::vector<bool>{}, context);
+    }
+
+    // Load shared indptr data - thread-safe to read
+    loadIndptrData(transaction);
+
+    auto numRowGroups = iceDiskScanState.indicesReader->getNumRowGroups();
+
+    // Initialize parquet reader scan state once per morsel
+    std::vector<uint64_t> rowGroupsToProcess;
+    for (uint64_t i = 0; i < numRowGroups; ++i) {
+        rowGroupsToProcess.push_back(i);
+    }
+
+    // Create a set of bound node IDs for fast lookup
+    std::unordered_map<common::offset_t, common::sel_t> boundNodeOffsets;
+    for (size_t i = 0; i < iceDiskScanState.cachedBoundNodeSelVector.getSelSize(); ++i) {
+        common::sel_t boundNodeIdx = iceDiskScanState.cachedBoundNodeSelVector[i];
+        const auto boundNodeID = iceDiskScanState.nodeIDVector->getValue<nodeID_t>(boundNodeIdx);
+        boundNodeOffsets.insert({boundNodeID.offset, boundNodeIdx});
+    }
+
+    iceDiskScanState.reset(std::move(boundNodeOffsets));
+    iceDiskScanState.indicesReader->initializeScan(*iceDiskScanState.parquetScanState,
+        rowGroupsToProcess, vfs);
 }
 
 void IceDiskRelTable::initializeParquetReaders(Transaction* transaction) const {
@@ -187,133 +217,76 @@ void IceDiskRelTable::loadIndptrData(Transaction* transaction) const {
 }
 
 bool IceDiskRelTable::scanInternal(Transaction* transaction, TableScanState& scanState) {
-    auto& relScanState = scanState.cast<RelTableScanState>();
-
-    // Get the IceDiskRelTableScanState
-    auto& iceDiskScanState = static_cast<IceDiskRelTableScanState&>(relScanState);
-
-    // Load shared indptr data - thread-safe to read
-    if (!indptrFilePath.empty()) {
-        loadIndptrData(transaction);
-    }
-
-    // True morsel-driven parallelism: each scan state processes its assigned row groups
-    // Process all row groups assigned to this scan state, collecting relationships for bound nodes
-    return scanInternalByRowGroups(transaction, iceDiskScanState);
-}
+    auto& iceDiskScanState = static_cast<IceDiskRelTableScanState&>(scanState);
 
-bool IceDiskRelTable::scanInternalByRowGroups(Transaction* transaction,
-    IceDiskRelTableScanState& iceDiskScanState) {
-    // True morsel-driven parallelism: process assigned row groups and collect relationships for
-    // bound nodes
+    scanState.resetOutVectors();
 
-    // Check if we have any row groups left to process
-    if (iceDiskScanState.currentRowGroup >= iceDiskScanState.endRowGroup) {
-        // No more row groups to process
+    if (iceDiskScanState.boundNodeOffsets.empty()) {
+        // No bound nodes, return empty result
         iceDiskScanState.outState->getSelVectorUnsafe().setToFiltered(0);
         return false;
     }
 
-    // Process the current row group
-    std::vector<uint64_t> rowGroupsToProcess = {iceDiskScanState.currentRowGroup};
-
-    // Create a set of bound node IDs for fast lookup
-    std::unordered_map<common::offset_t, common::sel_t> boundNodeOffsets;
-    for (size_t i = 0; i < iceDiskScanState.cachedBoundNodeSelVector.getSelSize(); ++i) {
-        common::sel_t boundNodeIdx = iceDiskScanState.cachedBoundNodeSelVector[i];
-        const auto boundNodeID = iceDiskScanState.nodeIDVector->getValue<nodeID_t>(boundNodeIdx);
-        boundNodeOffsets.insert({boundNodeID.offset, boundNodeIdx});
-    }
-
-    // Scan the current row group and collect relationships for bound nodes
-    bool hasData = scanRowGroupForBoundNodes(transaction, iceDiskScanState, rowGroupsToProcess,
-        boundNodeOffsets);
-
-    // Move to next row group for next call
-    iceDiskScanState.currentRowGroup++;
-
-    return hasData;
-}
-
-common::offset_t IceDiskRelTable::findSourceNodeForRow(common::offset_t globalRowIdx) const {
-    // Use base class helper for binary search
-    return findSourceNodeForRowInternal(globalRowIdx, indptrData);
-}
-
-bool IceDiskRelTable::scanRowGroupForBoundNodes(Transaction* transaction,
-    IceDiskRelTableScanState& iceDiskScanState, const std::vector<uint64_t>& rowGroupsToProcess,
-    const std::unordered_map<common::offset_t, common::sel_t>& boundNodeOffsets) {
-
-    // Initialize readers if needed
-    initializeParquetReaders(transaction);
-
-    // Initialize scan state for the assigned row groups
-    auto context = transaction->getClientContext();
-    auto vfs = VirtualFileSystem::GetUnsafe(*context);
-    iceDiskScanState.indicesReader->initializeScan(*iceDiskScanState.parquetScanState,
-        rowGroupsToProcess, vfs);
-
-    // Create DataChunk matching the indices parquet file schema
-    auto numIndicesColumns = iceDiskScanState.indicesReader->getNumColumns();
-    DataChunk indicesChunk(numIndicesColumns);
-
-    // Insert value vectors for all columns in the parquet file
-    auto memoryManager = MemoryManager::Get(*context);
-    for (uint32_t colIdx = 0; colIdx < numIndicesColumns; ++colIdx) {
-        const auto& columnTypeRef = iceDiskScanState.indicesReader->getColumnType(colIdx);
-        auto columnType = columnTypeRef.copy();
-        auto vector = std::make_shared<ValueVector>(std::move(columnType), memoryManager);
-        indicesChunk.insert(colIdx, vector);
-    }
+    // Load shared indptr data - thread-safe to read
+    loadIndptrData(transaction);
 
+    // start local scan
     // Scan the row groups and collect relationships for bound nodes.
     const auto isFwd = iceDiskScanState.direction != RelDataDirection::BWD;
     uint64_t totalRowsCollected = 0;
     const uint64_t maxRowsPerCall = DEFAULT_VECTOR_CAPACITY;
-    uint64_t currentGlobalRowIdx = 0;
     auto activeBoundSelPos = INVALID_SEL;
     auto activeBoundOffset = INVALID_OFFSET;
     auto hasActiveBound = false;
-
-    // Calculate the starting global row index for the first row group
-    if (!rowGroupsToProcess.empty()) {
-        auto metadata = iceDiskScanState.indicesReader->getMetadata();
-        for (uint64_t rgIdx = 0; rgIdx < rowGroupsToProcess[0]; ++rgIdx) {
-            currentGlobalRowIdx += metadata->row_groups[rgIdx].num_rows;
+    auto differentBoundNodeEncountered = false;
+
+    while (totalRowsCollected < maxRowsPerCall) {
+        if (!iceDiskScanState.cachedBatchData ||
+            iceDiskScanState.currentLocalRowIdx ==
+                iceDiskScanState.cachedBatchData->state->getSelVector().getSelSize()) {
+            // This means we are at the start of a new batch, so we need to reset the local row
+            // index and update the batch start offset
+            iceDiskScanState.currentBatchStartOffset += iceDiskScanState.currentLocalRowIdx;
+            iceDiskScanState.currentLocalRowIdx = 0;
+            iceDiskScanState.reloadCachedBatchData(transaction);
         }
-    }
 
-    while (totalRowsCollected < maxRowsPerCall &&
-           iceDiskScanState.indicesReader->scanInternal(*iceDiskScanState.parquetScanState,
-               indicesChunk)) {
+        auto selSize = iceDiskScanState.cachedBatchData->state->getSelVector().getSelSize();
 
-        auto selSize = indicesChunk.state->getSelVector().getSelSize();
+        if (selSize == 0) {
+            break; // No more data to read
+        }
 
-        for (size_t i = 0; i < selSize && totalRowsCollected < maxRowsPerCall;
-             ++i, ++currentGlobalRowIdx) {
+        for (; iceDiskScanState.currentLocalRowIdx < selSize && totalRowsCollected < maxRowsPerCall;
+             ++iceDiskScanState.currentLocalRowIdx) {
             // Find which source node this row belongs to.
+            const auto currentGlobalRowIdx =
+                iceDiskScanState.currentBatchStartOffset + iceDiskScanState.currentLocalRowIdx;
             const auto sourceNodeOffset = findSourceNodeForRow(currentGlobalRowIdx);
             if (sourceNodeOffset == common::INVALID_OFFSET) {
                 continue; // Invalid row
             }
 
             // Column 0 in indices file is the destination node offset.
-            const auto dstOffset = indicesChunk.getValueVector(0).getValue<common::offset_t>(i);
+            const auto dstOffset =
+                iceDiskScanState.cachedBatchData->getValueVector(0).getValue<common::offset_t>(
+                    iceDiskScanState.currentLocalRowIdx);
             const auto boundOffset = isFwd ? sourceNodeOffset : dstOffset;
-            if (boundNodeOffsets.find(boundOffset) == boundNodeOffsets.end()) {
+            if (iceDiskScanState.boundNodeOffsets.find(boundOffset) ==
+                iceDiskScanState.boundNodeOffsets.end()) {
                 continue; // Not a bound node, skip
             }
 
             if (!hasActiveBound) {
                 hasActiveBound = true;
                 activeBoundOffset = boundOffset;
-                activeBoundSelPos = boundNodeOffsets.at(boundOffset);
+                activeBoundSelPos = iceDiskScanState.boundNodeOffsets.at(boundOffset);
             } else if (boundOffset != activeBoundOffset) {
+                differentBoundNodeEncountered = true;
                 break;
             }
 
             // This row belongs to a bound node, collect the relationship
-
             const auto nbrOffset = isFwd ? dstOffset : sourceNodeOffset;
             const auto nbrTableID = isFwd ? getToNodeTableID() : getFromNodeTableID();
             auto nbrNodeID = internalID_t(nbrOffset, nbrTableID);
@@ -344,20 +317,22 @@ bool IceDiskRelTable::scanRowGroupForBoundNodes(Transaction* transaction,
                 }
 
                 iceDiskScanState.outputVectors[outCol]->copyFromVectorData(totalRowsCollected,
-                    &indicesChunk.getValueVector(colID - 1), i);
+                    &iceDiskScanState.cachedBatchData->getValueVector(colID - 1),
+                    iceDiskScanState.currentLocalRowIdx);
             }
 
             totalRowsCollected++;
         }
+
+        if (differentBoundNodeEncountered) {
+            break;
+        }
     }
 
     // Set up the output state
     if (totalRowsCollected > 0) {
         auto& selVector = iceDiskScanState.outState->getSelVectorUnsafe();
-        selVector.setToFiltered(totalRowsCollected);
-        for (uint64_t i = 0; i < totalRowsCollected; ++i) {
-            selVector[i] = i;
-        }
+        selVector.setToUnfiltered(totalRowsCollected);
         iceDiskScanState.setNodeIDVectorToFlat(activeBoundSelPos);
 
         return true;
@@ -368,6 +343,11 @@ bool IceDiskRelTable::scanRowGroupForBoundNodes(Transaction* transaction,
     }
 }
 
+common::offset_t IceDiskRelTable::findSourceNodeForRow(common::offset_t globalRowIdx) const {
+    // Use base class helper for binary search
+    return findSourceNodeForRowInternal(globalRowIdx, indptrData);
+}
+
 row_idx_t IceDiskRelTable::getTotalRowCount(const Transaction* transaction) const {
     initializeParquetReaders(const_cast<Transaction*>(transaction));
     if (!indicesReader) {
diff --git a/test/storage/ice_disk_utils_test.cpp b/test/storage/ice_disk_utils_test.cpp
index 83693bd6d3..4179a3a652 100644
--- a/test/storage/ice_disk_utils_test.cpp
+++ b/test/storage/ice_disk_utils_test.cpp
@@ -18,10 +18,7 @@ using namespace lbug::main;
 using namespace lbug::storage;
 using namespace lbug::testing;
 
-static const std::string FIXTURES_DIR =
-    TestHelper::appendLbugRootPath("dataset/ice-disk-test/fixtures");
-static const std::string DEMO_DB_ICEBUG_DISK =
-    TestHelper::appendLbugRootPath("dataset/demo-db/icebug-disk");
+static const std::string DATASET_DIR = TestHelper::appendLbugRootPath("dataset/ice-disk-test/");
 
 // ─────────────────────────────────────────────────────────────
 // joinPath
@@ -102,30 +99,29 @@ class IceDiskCheckVersionTest : public EmptyDBTest {
     }
 
     ClientContext* context = nullptr;
-    const std::string dbDir = FIXTURES_DIR;
 };
 
 TEST_F(IceDiskCheckVersionTest, NullContext) {
-    EXPECT_THROW(IceDiskUtils::checkVersionCompatibility(nullptr,
-                     DEMO_DB_ICEBUG_DISK + "/nodes_person.parquet"),
+    EXPECT_THROW(
+        IceDiskUtils::checkVersionCompatibility(nullptr, DATASET_DIR + "/nodes_user.parquet"),
         RuntimeException);
 }
 
 TEST_F(IceDiskCheckVersionTest, FileDoesNotExist) {
     EXPECT_THROW(IceDiskUtils::checkVersionCompatibility(context,
-                     FIXTURES_DIR + "/nodes_nonexistent.parquet"),
+                     DATASET_DIR + "/nodes_nonexistent.parquet"),
         IOException);
 }
 
 TEST_F(IceDiskCheckVersionTest, NotAParquetFile) {
-    EXPECT_THROW(IceDiskUtils::checkVersionCompatibility(context,
-                     FIXTURES_DIR + "/nodes_notparquet.parquet"),
+    EXPECT_THROW(
+        IceDiskUtils::checkVersionCompatibility(context, DATASET_DIR + "/nodes_notparquet.parquet"),
         CopyException);
 }
 
 TEST_F(IceDiskCheckVersionTest, MissingVersionKey) {
     try {
-        IceDiskUtils::checkVersionCompatibility(context, FIXTURES_DIR + "/nodes_noversion.parquet");
+        IceDiskUtils::checkVersionCompatibility(context, DATASET_DIR + "/nodes_noversion.parquet");
         FAIL() << "Expected RuntimeException for missing version key";
     } catch (const RuntimeException& e) {
         EXPECT_TRUE(std::string(e.what()).find("missing icebug_disk_version") != std::string::npos);
@@ -135,7 +131,7 @@ TEST_F(IceDiskCheckVersionTest, MissingVersionKey) {
 TEST_F(IceDiskCheckVersionTest, WrongVersionValue) {
     try {
         IceDiskUtils::checkVersionCompatibility(context,
-            FIXTURES_DIR + "/nodes_wrongversion.parquet");
+            DATASET_DIR + "/nodes_wrongversion.parquet");
         FAIL() << "Expected RuntimeException for wrong version";
     } catch (const RuntimeException& e) {
         EXPECT_TRUE(std::string(e.what()).find("does not support icebug_disk_version: v99") !=
@@ -146,10 +142,10 @@ TEST_F(IceDiskCheckVersionTest, WrongVersionValue) {
 TEST_F(IceDiskCheckVersionTest, UppercaseVersionSucceeds) {
     // "V1" should match "v1" case-insensitively
     EXPECT_NO_THROW(IceDiskUtils::checkVersionCompatibility(context,
-        FIXTURES_DIR + "/nodes_upperversion.parquet"));
+        DATASET_DIR + "/nodes_upperversion.parquet"));
 }
 
 TEST_F(IceDiskCheckVersionTest, ValidV1Succeeds) {
-    EXPECT_NO_THROW(IceDiskUtils::checkVersionCompatibility(context,
-        DEMO_DB_ICEBUG_DISK + "/nodes_user.parquet"));
+    EXPECT_NO_THROW(
+        IceDiskUtils::checkVersionCompatibility(context, DATASET_DIR + "/nodes_user.parquet"));
 }
diff --git a/test/test_files/demo_db/demo_db_icebug_disk.test b/test/test_files/demo_db/demo_db_icebug_disk.test
index 18462de648..47bc0b8ddb 100644
--- a/test/test_files/demo_db/demo_db_icebug_disk.test
+++ b/test/test_files/demo_db/demo_db_icebug_disk.test
@@ -133,57 +133,3 @@ Adam|Zhang
 Adam|45.000000
 Karissa|50.000000
 Zhang|25.000000
-
--LOG TwoHopCrossRel
--STATEMENT MATCH (a:user {id: 100})-[:follows]->(b)-[:livesin]->(c:city) RETURN b.name, c.name ORDER BY b.name;
----- 2
-Karissa|Waterloo
-Zhang|Kitchener
-
--LOG BackwardInNeighbors
--STATEMENT MATCH (u)<-[:follows]-(v) WHERE u.id = 300 RETURN v.name ORDER BY v.name;
----- 2
-Adam
-Karissa
-
--LOG UndirectedLivesIn
--STATEMENT MATCH (a:user)-[:livesin]-(c:city) RETURN a.name, c.name ORDER BY a.name;
----- 4
-Adam|Waterloo
-Karissa|Waterloo
-Noura|Guelph
-Zhang|Kitchener
-
--LOG CyclicTriangle
--STATEMENT MATCH (a:user)-[:follows]->(b:user)-[:follows]->(c:user), (a)-[:follows]->(c) RETURN a.name, b.name, c.name;
----- 1
-Adam|Karissa|Zhang
-
--LOG SemiMaskerVarLen
--STATEMENT MATCH (a:user {id: 100})-[:follows*1..2 (r, n | WHERE n.age > 40)]->(b:user) RETURN b.name ORDER BY b.name;
----- 3
-Karissa
-Zhang
-Noura
-
--LOG VarLenThreeHop
--STATEMENT MATCH (a:user)-[:follows*3..3]->(b:user) RETURN a.name, b.name ORDER BY a.name, b.name;
----- 1
-Adam|Noura
-
--LOG FlattenMultiPartMatch
--STATEMENT MATCH (a:user)-[:follows]->(b:user) WITH a, b MATCH (b)-[:livesin]->(c:city) RETURN a.name, b.name, c.name ORDER BY a.name, b.name;
----- 4
-Adam|Karissa|Waterloo
-Adam|Zhang|Kitchener
-Karissa|Zhang|Kitchener
-Zhang|Noura|Guelph
-
--LOG HashJoinSharedFollowee
--STATEMENT MATCH (a:user)-[:follows]->(b:user), (c:user)-[:follows]->(b) WHERE a.id < c.id RETURN a.name, b.name, c.name;
----- 1
-Adam|Zhang|Karissa
-
--LOG ProfileQuery
--STATEMENT PROFILE MATCH (u:user)-[:follows]->(v:user) RETURN u.name, v.name;
----- ok
diff --git a/test/test_files/ice_disk/ice_disk_complex_queries.test b/test/test_files/ice_disk/ice_disk_complex_queries.test
new file mode 100644
index 0000000000..fd9e3c667d
--- /dev/null
+++ b/test/test_files/ice_disk/ice_disk_complex_queries.test
@@ -0,0 +1,102 @@
+-DATASET ICEBUG-DISK ice-disk-test
+--
+
+-CASE IceDiskComplexQueries
+
+-LOG TwoHopCrossRel
+-STATEMENT MATCH (a:user {id: 100})-[:follows]->(b)-[:livesin]->(c:city) RETURN b.name, c.name ORDER BY b.name;
+---- 3
+Adam|Waterloo
+Karissa|Waterloo
+Zhang|Kitchener
+
+-LOG BackwardInNeighbors
+-STATEMENT MATCH (u)<-[:follows]-(v) WHERE u.id = 300 RETURN v.name ORDER BY v.name;
+---- 2
+Adam
+Karissa
+
+-LOG UndirectedLivesIn
+-STATEMENT MATCH (a:user)-[:livesin]-(c:city) RETURN a.name, c.name ORDER BY a.name;
+---- 4
+Adam|Waterloo
+Karissa|Waterloo
+Noura|Guelph
+Zhang|Kitchener
+
+-LOG CyclicTriangle
+-STATEMENT MATCH (a:user)-[:follows]->(b:user)-[:follows]->(c:user), (a)-[:follows]->(c) WHERE a.id <> b.id AND b.id <> c.id AND a.id <> c.id RETURN a.name, b.name, c.name ORDER BY a.name, b.name;
+---- 2
+Adam|Karissa|Zhang
+Karissa|Adam|Zhang
+
+-LOG SemiMaskerVarLen
+-STATEMENT MATCH (a:user {id: 100})-[:follows*1..2 (r, n | WHERE n.age > 40)]->(b:user) RETURN b.name ORDER BY b.name;
+---- 4
+Adam
+Karissa
+Noura
+Zhang
+
+-LOG VarLenThreeHop
+-STATEMENT MATCH (a:user {id: 100})-[:follows*3..3]->(b:user) RETURN DISTINCT b.name ORDER BY b.name;
+---- 4
+Adam
+Karissa
+Noura
+Zhang
+
+-LOG FlattenMultiPartMatch
+-STATEMENT MATCH (a:user)-[:follows]->(b:user) WITH a, b MATCH (b)-[:livesin]->(c:city) RETURN a.name, b.name, c.name ORDER BY a.name, b.name;
+---- 7
+Adam|Adam|Waterloo
+Adam|Karissa|Waterloo
+Adam|Zhang|Kitchener
+Karissa|Adam|Waterloo
+Karissa|Zhang|Kitchener
+Noura|Adam|Waterloo
+Zhang|Noura|Guelph
+
+-LOG HashJoinSharedFollowee
+-STATEMENT MATCH (a:user)-[:follows]->(b:user), (c:user)-[:follows]->(b) WHERE a.id < c.id RETURN a.name, b.name, c.name ORDER BY a.name, b.name, c.name;
+---- 4
+Adam|Adam|Karissa
+Adam|Zhang|Karissa
+Noura|Adam|Adam
+Noura|Adam|Karissa
+
+-LOG ProfileQuery
+-STATEMENT PROFILE MATCH (u:user)-[:follows]->(v:user) RETURN u.name, v.name;
+---- ok
+
+-LOG SelfLoopFollows
+-STATEMENT MATCH (a:user)-[:follows]->(a) RETURN a.name;
+---- 1
+Adam
+
+-LOG SelfLoopExcluded
+-STATEMENT MATCH (a:user)-[:follows]->(b:user) WHERE a.id <> b.id RETURN COUNT(*);
+---- 1
+6
+
+-LOG BackwardMultiHopCityUserUser
+-STATEMENT MATCH (c:city)<-[:livesin]-(u:user)<-[:follows]-(f:user) RETURN f.name, u.name, c.name ORDER BY f.name, u.name;
+---- 7
+Adam|Adam|Waterloo
+Adam|Karissa|Waterloo
+Adam|Zhang|Kitchener
+Karissa|Adam|Waterloo
+Karissa|Zhang|Kitchener
+Noura|Adam|Waterloo
+Zhang|Noura|Guelph
+
+-LOG CrossRelCityFollowsCity
+-STATEMENT MATCH (c1:city)<-[:livesin]-(u:user)-[:follows]->(v:user)-[:livesin]->(c2:city) RETURN c1.name, u.name, v.name, c2.name ORDER BY u.name, v.name;
+---- 7
+Waterloo|Adam|Adam|Waterloo
+Waterloo|Adam|Karissa|Waterloo
+Waterloo|Adam|Zhang|Kitchener
+Waterloo|Karissa|Adam|Waterloo
+Waterloo|Karissa|Zhang|Kitchener
+Guelph|Noura|Adam|Waterloo
+Kitchener|Zhang|Noura|Guelph
diff --git a/test/test_files/ice_disk/ice_disk_invalid_storage.test b/test/test_files/ice_disk/ice_disk_invalid_storage.test
index 255a290d19..4434c30f34 100644
--- a/test/test_files/ice_disk/ice_disk_invalid_storage.test
+++ b/test/test_files/ice_disk/ice_disk_invalid_storage.test
@@ -25,7 +25,7 @@
 -CASE IceDiskNoStorageRelFails
 
 -LOG IceDiskNoStorageRelMissingParquet
--STATEMENT CREATE NODE TABLE upperversion(id INT64 PRIMARY KEY) WITH (storage = '${LBUG_ROOT_DIRECTORY}/dataset/ice-disk-test/fixtures', format = 'icebug-disk')
+-STATEMENT CREATE NODE TABLE upperversion(id INT64 PRIMARY KEY) WITH (storage = '${LBUG_ROOT_DIRECTORY}/dataset/ice-disk-test/', format = 'icebug-disk')
 ---- ok
 -STATEMENT CREATE REL TABLE rel(FROM upperversion TO upperversion) WITH (format = 'icebug-disk')
 ---- error(regex)
diff --git a/test/test_files/ice_disk/ice_disk_mix_tables.test b/test/test_files/ice_disk/ice_disk_mix_tables.test
index dd88abbc63..c1215b22b4 100644
--- a/test/test_files/ice_disk/ice_disk_mix_tables.test
+++ b/test/test_files/ice_disk/ice_disk_mix_tables.test
@@ -8,14 +8,14 @@
 ---- ok
 
 -LOG IceDiskRelWithRegularNodeFails
--STATEMENT CREATE REL TABLE rel(FROM person TO person) WITH (storage = '${LBUG_ROOT_DIRECTORY}/dataset/ice-disk-test/fixtures', format = 'icebug-disk')
+-STATEMENT CREATE REL TABLE rel(FROM person TO person) WITH (storage = '${LBUG_ROOT_DIRECTORY}/dataset/ice-disk-test/', format = 'icebug-disk')
 ---- error
 Binder exception: Cannot mix icebug-disk tables with non-icebug-disk tables in CREATE REL TABLE.
 
 -CASE NodeIceDiskRelRegular
 
 -LOG CreateIceDiskNodeTable
--STATEMENT CREATE NODE TABLE upperversion(id INT64 PRIMARY KEY) WITH (storage = '${LBUG_ROOT_DIRECTORY}/dataset/ice-disk-test/fixtures', format = 'icebug-disk')
+-STATEMENT CREATE NODE TABLE upperversion(id INT64 PRIMARY KEY) WITH (storage = '${LBUG_ROOT_DIRECTORY}/dataset/ice-disk-test/', format = 'icebug-disk')
 ---- ok
 
 -LOG CreateRegularNodeTable