diff --git a/dataset b/dataset
index 55531118c5..1eb7993b8c 160000
--- a/dataset
+++ b/dataset
@@ -1 +1 @@
-Subproject commit 55531118c5e0c683fc3a3d806b7abd0b09a31ff8
+Subproject commit 1eb7993b8caeed3b75cb931698522640b9557481
diff --git a/docs/icebug-disk.md b/docs/icebug-disk.md
new file mode 100644
index 0000000000..3f48fb6494
--- /dev/null
+++ b/docs/icebug-disk.md
@@ -0,0 +1,54 @@
+# Icebug-Disk Storage Format
+
+## Overview
+
+This is LadybugDB's implementation of [Icebug-Disk](https://github.com/Ladybug-Memory/icebug-format), a read-only graph storage format based on Parquet files. It is designed for efficient analytical queries on large graphs.
+
+## V1
+
+Implements Icebug-Disk v1.
+
+### Version
+
+Version is stored in each file's metadata footer as a key-value pair: `icebug_disk_version = v1`.
+
+### schema.cypher
+
+The graph schema is declared in a `schema.cypher`, which can be loaded using `lbug -i schema.cypher` to create tables in Ladybug.
+
+```cypher
+CREATE NODE TABLE city(id INT32, name STRING, population INT64, PRIMARY KEY(id)) WITH (storage = '<path-to-dir>', format = 'icebug-disk');
+CREATE NODE TABLE user(id INT32, name STRING, age INT64, PRIMARY KEY(id)) WITH (storage = '<path-to-dir>', format = 'icebug-disk');
+CREATE REL TABLE follows(FROM user TO user, since INT32) WITH (storage = '<path-to-dir>', format = 'icebug-disk');
+CREATE REL TABLE livesin(FROM user TO city) WITH (storage = '<path-to-dir>', format = 'icebug-disk');
+```
+
+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.
+
+If `storage` is omitted when `format = 'icebug-disk'` is set, files are resolved relative to the current working directory.
+
+Tables can also be created by manually running the above queries in the Ladybug CLI.
+
+If the directory is moved, the affected tables must be dropped and re-created with the updated path. This updates the file pointers in the catalog. A fresh db instance can also be used to run the `CREATE TABLE` queries with the new paths.
+
+Mixed tables are not supported — `CREATE REL TABLE` queries involving both `icebug-disk` and non-`icebug-disk` tables will throw a `BinderException`.
+
+Note: Icebug-disk tables are **immutable**: `ALTER TABLE` operations, Data insertions, updates, and deletions are not supported.
+
+### Node tables
+
+For each node table, there is a corresponding Parquet file named `nodes_{tableName}.parquet` containing a primary key column and one column per property as declared in the schema.
+
+### Indices
+
+Each relationship table has a corresponding `indices_{tableName}.parquet` file containing one row per edge. The first column is always `target` (the destination node offset), followed by zero or more edge property columns as declared in the schema.
+
+### Indptr
+
+Each relationship table has a corresponding `indptr_{tableName}.parquet` file containing the CSR row pointers. It has a single integer column with `N+1` entries, where `N` is the number of source nodes.
+
+## Convert from other formats
+
+You can convert from other graph formats (e.g. duckdb, parquet tables) to Icebug-Disk using the script at https://github.com/Ladybug-Memory/icebug-format
diff --git a/docs/morsel_parallelism.md b/docs/morsel_parallelism.md
index ee7f20f935..6108ee0602 100644
--- a/docs/morsel_parallelism.md
+++ b/docs/morsel_parallelism.md
@@ -2,7 +2,7 @@
 
 ## Overview
 
-Morsel-driven parallelism is ladybug's approach to parallel query execution where work is divided into small, independently processable chunks called "morsels." This document explains how it works for native node tables versus columnar formats (Arrow, Parquet).
+Morsel-driven parallelism is ladybug's approach to parallel query execution where work is divided into small, independently processable chunks called "morsels." This document explains how it works for native node tables versus columnar formats (Arrow, Icebug Disk).
 
 ## Architecture
 
diff --git a/src/binder/bind/bind_ddl.cpp b/src/binder/bind/bind_ddl.cpp
index 9c4c702791..fe2acbdd5d 100644
--- a/src/binder/bind/bind_ddl.cpp
+++ b/src/binder/bind/bind_ddl.cpp
@@ -10,7 +10,9 @@
 #include "binder/expression_visitor.h"
 #include "catalog/catalog.h"
 #include "catalog/catalog_entry/node_table_catalog_entry.h"
+#include "catalog/catalog_entry/rel_group_catalog_entry.h"
 #include "catalog/catalog_entry/sequence_catalog_entry.h"
+#include "common/constants.h"
 #include "common/enums/extend_direction_util.h"
 #include "common/exception/binder.h"
 #include "common/exception/message.h"
@@ -190,14 +192,23 @@ static ExtendDirection getStorageDirection(const case_insensitive_map_t<Value>&
     return DEFAULT_EXTEND_DIRECTION;
 }
 
+static std::string getStorageFormat(const case_insensitive_map_t<Value>& options) {
+    if (options.contains(TableOptionConstants::STORAGE_FORMAT_OPTION)) {
+        return options.at(TableOptionConstants::STORAGE_FORMAT_OPTION).toString();
+    }
+
+    return "";
+}
+
 BoundCreateTableInfo Binder::bindCreateNodeTableInfo(const CreateTableInfo* info) {
     auto propertyDefinitions = bindPropertyDefinitions(info->propertyDefinitions, info->tableName);
     auto& extraInfo = info->extraInfo->constCast<ExtraCreateNodeTableInfo>();
     validatePrimaryKey(extraInfo.pKName, propertyDefinitions);
     auto boundOptions = bindParsingOptions(extraInfo.options);
     auto storage = getStorage(boundOptions);
+    auto storageFormat = getStorageFormat(boundOptions);
     auto boundExtraInfo = std::make_unique<BoundExtraCreateNodeTableInfo>(extraInfo.pKName,
-        std::move(propertyDefinitions), std::move(storage));
+        std::move(propertyDefinitions), std::move(storage), std::move(storageFormat));
     return BoundCreateTableInfo(CatalogEntryType::NODE_TABLE_ENTRY, info->tableName,
         info->onConflict, std::move(boundExtraInfo), clientContext->useInternalCatalogEntry());
 }
@@ -220,15 +231,18 @@ BoundCreateTableInfo Binder::bindCreateRelTableGroupInfo(const CreateTableInfo*
     auto boundOptions = bindParsingOptions(extraInfo.options);
     auto storageDirection = getStorageDirection(boundOptions);
     auto storage = getStorage(boundOptions);
+    auto storageFormat = getStorageFormat(boundOptions);
     std::optional<function::TableFunction> scanFunction = std::nullopt;
     std::optional<std::unique_ptr<function::TableFuncBindData>> scanBindData = std::nullopt;
     std::string foreignDatabaseName;
     if (!storage.empty()) {
         auto dotPos = storage.find('.');
         // Check if storage is database.table format by verifying the attached database exists
+        // 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 (dotPos != std::string::npos) {
+        if (!TableOptionConstants::isIceBugDiskFormat(storageFormat) &&
+            dotPos != std::string::npos) {
             std::string dbName = storage.substr(0, dotPos);
             std::string tableName = storage.substr(dotPos + 1);
             if (!dbName.empty()) {
@@ -319,6 +333,26 @@ 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);
+
+        // 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
+        if ((!isRelIcebugDisk && (isSrcIcebugDisk || isDstIcebugDisk)) ||
+            (isRelIcebugDisk && (!isSrcIcebugDisk || !isDstIcebugDisk))) {
+            throw BinderException(
+                "Cannot mix icebug-disk tables with non-icebug-disk tables in CREATE REL TABLE.");
+        }
+
         // Use the actual shadow table IDs, not FOREIGN_TABLE_ID
         // The shadow tables allow the query planner to distinguish between different node tables
         auto srcTableID = srcEntry->getTableID();
@@ -333,8 +367,8 @@ BoundCreateTableInfo Binder::bindCreateRelTableGroupInfo(const CreateTableInfo*
     }
     auto boundExtraInfo = std::make_unique<BoundExtraCreateRelTableGroupInfo>(
         std::move(propertyDefinitions), srcMultiplicity, dstMultiplicity, storageDirection,
-        std::move(nodePairs), std::move(storage), std::move(scanFunction), std::move(scanBindData),
-        std::move(foreignDatabaseName));
+        std::move(nodePairs), std::move(storage), std::move(storageFormat), std::move(scanFunction),
+        std::move(scanBindData), std::move(foreignDatabaseName));
     return BoundCreateTableInfo(CatalogEntryType::REL_GROUP_ENTRY, info->tableName,
         info->onConflict, std::move(boundExtraInfo), clientContext->useInternalCatalogEntry());
 }
@@ -534,8 +568,36 @@ std::unique_ptr<BoundStatement> Binder::bindDrop(const Statement& statement) {
     return std::make_unique<BoundDrop>(drop.getDropInfo());
 }
 
+static void validateNotIceDiskTable(main::ClientContext* clientContext,
+    const std::string& tableName) {
+    auto catalog = Catalog::Get(*clientContext);
+    auto transaction = transaction::Transaction::Get(*clientContext);
+
+    if (!catalog->containsTable(transaction, tableName)) {
+        return;
+    }
+
+    auto tableEntry = catalog->getTableCatalogEntry(transaction, tableName);
+    std::string storageFormat;
+
+    if (tableEntry->getTableType() == common::TableType::NODE) {
+        storageFormat = tableEntry->ptrCast<NodeTableCatalogEntry>()->getStorageFormat();
+    } else if (tableEntry->getTableType() == common::TableType::REL) {
+        storageFormat = tableEntry->ptrCast<RelGroupCatalogEntry>()->getStorageFormat();
+    }
+
+    if (TableOptionConstants::isIceBugDiskFormat(storageFormat)) {
+        throw BinderException(
+            std::format("Cannot alter table {}: icebug-disk tables are immutable.", tableName));
+    }
+}
+
 std::unique_ptr<BoundStatement> Binder::bindAlter(const Statement& statement) {
     auto& alter = statement.constCast<Alter>();
+
+    // we don't support alter operations on icebug-disk tables
+    validateNotIceDiskTable(clientContext, alter.getInfo()->tableName);
+
     switch (alter.getInfo()->type) {
     case AlterType::RENAME: {
         return bindRenameTable(statement);
diff --git a/src/catalog/catalog.cpp b/src/catalog/catalog.cpp
index 930dbc2e68..a565b7285c 100644
--- a/src/catalog/catalog.cpp
+++ b/src/catalog/catalog.cpp
@@ -203,10 +203,11 @@ CatalogEntry* Catalog::createRelGroupEntry(Transaction* transaction,
     for (auto& nodePair : extraInfo->nodePairs) {
         relTableInfos.emplace_back(nodePair, tables->getNextOID());
     }
-    auto relGroupEntry = std::make_unique<RelGroupCatalogEntry>(info.tableName,
-        extraInfo->srcMultiplicity, extraInfo->dstMultiplicity, extraInfo->storageDirection,
-        std::move(relTableInfos), extraInfo->storage, extraInfo->scanFunction,
-        std::move(extraInfo->scanBindData), extraInfo->foreignDatabaseName);
+    auto relGroupEntry =
+        std::make_unique<RelGroupCatalogEntry>(info.tableName, extraInfo->srcMultiplicity,
+            extraInfo->dstMultiplicity, extraInfo->storageDirection, std::move(relTableInfos),
+            extraInfo->storage, extraInfo->storageFormat, extraInfo->scanFunction,
+            std::move(extraInfo->scanBindData), extraInfo->foreignDatabaseName);
     for (auto& definition : extraInfo->propertyDefinitions) {
         relGroupEntry->addProperty(definition);
     }
@@ -561,7 +562,7 @@ CatalogEntry* Catalog::createNodeTableEntry(Transaction* transaction,
     const BoundCreateTableInfo& info) {
     const auto extraInfo = info.extraInfo->constPtrCast<BoundExtraCreateNodeTableInfo>();
     auto entry = std::make_unique<NodeTableCatalogEntry>(info.tableName, extraInfo->primaryKeyName,
-        extraInfo->storage);
+        extraInfo->storage, extraInfo->storageFormat);
     for (auto& definition : extraInfo->propertyDefinitions) {
         entry->addProperty(definition);
     }
diff --git a/src/catalog/catalog_entry/node_table_catalog_entry.cpp b/src/catalog/catalog_entry/node_table_catalog_entry.cpp
index ee7170fc13..350fcf725f 100644
--- a/src/catalog/catalog_entry/node_table_catalog_entry.cpp
+++ b/src/catalog/catalog_entry/node_table_catalog_entry.cpp
@@ -24,6 +24,8 @@ void NodeTableCatalogEntry::serialize(common::Serializer& serializer) const {
     serializer.write(primaryKeyName);
     serializer.writeDebuggingInfo("storage");
     serializer.write(storage);
+    serializer.writeDebuggingInfo("storageFormat");
+    serializer.write(storageFormat);
 }
 
 std::unique_ptr<NodeTableCatalogEntry> NodeTableCatalogEntry::deserialize(
@@ -31,13 +33,17 @@ std::unique_ptr<NodeTableCatalogEntry> NodeTableCatalogEntry::deserialize(
     std::string debuggingInfo;
     std::string primaryKeyName;
     std::string storage;
+    std::string storageFormat;
     deserializer.validateDebuggingInfo(debuggingInfo, "primaryKeyName");
     deserializer.deserializeValue(primaryKeyName);
     deserializer.validateDebuggingInfo(debuggingInfo, "storage");
     deserializer.deserializeValue(storage);
+    deserializer.validateDebuggingInfo(debuggingInfo, "storageFormat");
+    deserializer.deserializeValue(storageFormat);
     auto nodeTableEntry = std::make_unique<NodeTableCatalogEntry>();
     nodeTableEntry->primaryKeyName = primaryKeyName;
     nodeTableEntry->storage = storage;
+    nodeTableEntry->storageFormat = storageFormat;
     return nodeTableEntry;
 }
 
@@ -66,6 +72,7 @@ std::unique_ptr<TableCatalogEntry> NodeTableCatalogEntry::copy() const {
     auto other = std::make_unique<NodeTableCatalogEntry>();
     other->primaryKeyName = primaryKeyName;
     other->storage = storage;
+    other->storageFormat = storageFormat;
     other->scanFunction = scanFunction;
     other->createBindDataFunc = createBindDataFunc;
     other->foreignDatabaseName = foreignDatabaseName;
@@ -76,7 +83,7 @@ std::unique_ptr<TableCatalogEntry> NodeTableCatalogEntry::copy() const {
 std::unique_ptr<BoundExtraCreateCatalogEntryInfo> NodeTableCatalogEntry::getBoundExtraCreateInfo(
     transaction::Transaction*) const {
     return std::make_unique<BoundExtraCreateNodeTableInfo>(primaryKeyName,
-        copyVector(getProperties()), storage);
+        copyVector(getProperties()), storage, storageFormat);
 }
 
 } // namespace catalog
diff --git a/src/catalog/catalog_entry/rel_group_catalog_entry.cpp b/src/catalog/catalog_entry/rel_group_catalog_entry.cpp
index 46da2a4fbe..86acd13dcd 100644
--- a/src/catalog/catalog_entry/rel_group_catalog_entry.cpp
+++ b/src/catalog/catalog_entry/rel_group_catalog_entry.cpp
@@ -105,6 +105,8 @@ void RelGroupCatalogEntry::serialize(Serializer& serializer) const {
     }
     serializer.writeDebuggingInfo("relTableInfos");
     serializer.serializeVector(relTableInfos);
+    serializer.writeDebuggingInfo("storageFormat");
+    serializer.serializeValue(storageFormat);
 }
 
 std::unique_ptr<RelGroupCatalogEntry> RelGroupCatalogEntry::deserialize(
@@ -114,6 +116,7 @@ std::unique_ptr<RelGroupCatalogEntry> RelGroupCatalogEntry::deserialize(
     auto dstMultiplicity = RelMultiplicity::MANY;
     auto storageDirection = ExtendDirection::BOTH;
     std::string storage;
+    std::string storageFormat;
     std::vector<RelTableCatalogInfo> relTableInfos;
     deserializer.validateDebuggingInfo(debuggingInfo, "srcMultiplicity");
     deserializer.deserializeValue(srcMultiplicity);
@@ -132,11 +135,14 @@ std::unique_ptr<RelGroupCatalogEntry> RelGroupCatalogEntry::deserialize(
     }
     deserializer.validateDebuggingInfo(debuggingInfo, "relTableInfos");
     deserializer.deserializeVector(relTableInfos);
+    deserializer.validateDebuggingInfo(debuggingInfo, "storageFormat");
+    deserializer.deserializeValue(storageFormat);
     auto relGroupEntry = std::make_unique<RelGroupCatalogEntry>();
     relGroupEntry->srcMultiplicity = srcMultiplicity;
     relGroupEntry->dstMultiplicity = dstMultiplicity;
     relGroupEntry->storageDirection = storageDirection;
     relGroupEntry->storage = storage;
+    relGroupEntry->storageFormat = storageFormat;
     relGroupEntry->scanFunction = scanFunction;
     relGroupEntry->relTableInfos = relTableInfos;
     return relGroupEntry;
@@ -198,6 +204,7 @@ std::unique_ptr<TableCatalogEntry> RelGroupCatalogEntry::copy() const {
     other->dstMultiplicity = dstMultiplicity;
     other->storageDirection = storageDirection;
     other->storage = storage;
+    other->storageFormat = storageFormat;
     other->scanFunction = scanFunction;
     other->scanBindData = std::nullopt; // TODO: implement copy for bindData if needed
     other->foreignDatabaseName = foreignDatabaseName;
@@ -214,7 +221,7 @@ RelGroupCatalogEntry::getBoundExtraCreateInfo(transaction::Transaction*) const {
     }
     return std::make_unique<binder::BoundExtraCreateRelTableGroupInfo>(
         copyVector(propertyCollection.getDefinitions()), srcMultiplicity, dstMultiplicity,
-        storageDirection, std::move(nodePairs));
+        storageDirection, std::move(nodePairs), storage, storageFormat);
 }
 
 } // namespace catalog
diff --git a/src/graph/on_disk_graph.cpp b/src/graph/on_disk_graph.cpp
index 6f4dd077f4..5ebf9acc21 100644
--- a/src/graph/on_disk_graph.cpp
+++ b/src/graph/on_disk_graph.cpp
@@ -18,8 +18,8 @@
 #include "storage/local_storage/local_storage.h"
 #include "storage/storage_manager.h"
 #include "storage/storage_utils.h"
+#include "storage/table/ice_disk_rel_table.h"
 #include "storage/table/node_table.h"
-#include "storage/table/parquet_rel_table.h"
 #include "storage/table/rel_table.h"
 
 using namespace lbug::catalog;
@@ -129,8 +129,8 @@ OnDiskGraphNbrScanState::OnDiskGraphNbrScanState(ClientContext* context,
         }
 
         std::unique_ptr<RelTableScanState> scanState;
-        if (dynamic_cast<ParquetRelTable*>(table) != nullptr) {
-            scanState = std::make_unique<ParquetRelTableScanState>(*mm, srcNodeIDVector.get(),
+        if (dynamic_cast<IceDiskRelTable*>(table) != nullptr) {
+            scanState = std::make_unique<IceDiskRelTableScanState>(*mm, srcNodeIDVector.get(),
                 outVectors, state);
         } else {
             scanState = std::make_unique<RelTableScanState>(*MemoryManager::Get(*context),
diff --git a/src/include/binder/ddl/bound_create_table_info.h b/src/include/binder/ddl/bound_create_table_info.h
index 2db5ac5d39..2466a0dead 100644
--- a/src/include/binder/ddl/bound_create_table_info.h
+++ b/src/include/binder/ddl/bound_create_table_info.h
@@ -76,14 +76,18 @@ struct LBUG_API BoundExtraCreateTableInfo : BoundExtraCreateCatalogEntryInfo {
 struct BoundExtraCreateNodeTableInfo final : BoundExtraCreateTableInfo {
     std::string primaryKeyName;
     std::string storage;
+    std::string storageFormat;
 
     BoundExtraCreateNodeTableInfo(std::string primaryKeyName,
-        std::vector<PropertyDefinition> definitions, std::string storage = "")
+        std::vector<PropertyDefinition> definitions, std::string storage = "",
+        std::string storageFormat = "")
         : BoundExtraCreateTableInfo{std::move(definitions)},
-          primaryKeyName{std::move(primaryKeyName)}, storage{std::move(storage)} {}
+          primaryKeyName{std::move(primaryKeyName)}, storage{std::move(storage)},
+          storageFormat{std::move(storageFormat)} {}
     BoundExtraCreateNodeTableInfo(const BoundExtraCreateNodeTableInfo& other)
         : BoundExtraCreateTableInfo{copyVector(other.propertyDefinitions)},
-          primaryKeyName{other.primaryKeyName}, storage{other.storage} {}
+          primaryKeyName{other.primaryKeyName}, storage{other.storage},
+          storageFormat{other.storageFormat} {}
 
     std::unique_ptr<BoundExtraCreateCatalogEntryInfo> copy() const override {
         return std::make_unique<BoundExtraCreateNodeTableInfo>(*this);
@@ -96,6 +100,7 @@ struct BoundExtraCreateRelTableGroupInfo final : BoundExtraCreateTableInfo {
     common::ExtendDirection storageDirection;
     std::vector<catalog::NodeTableIDPair> nodePairs;
     std::string storage;
+    std::string storageFormat;
     std::optional<function::TableFunction> scanFunction;
     std::optional<std::shared_ptr<function::TableFuncBindData>> scanBindData;
     std::string foreignDatabaseName;
@@ -103,22 +108,24 @@ 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 storage = "", std::string storageFormat = "",
         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)},
-          scanFunction{std::move(scanFunction)}, scanBindData{std::move(scanBindData)},
+          storageFormat{std::move(storageFormat)}, scanFunction{std::move(scanFunction)},
+          scanBindData{std::move(scanBindData)},
           foreignDatabaseName{std::move(foreignDatabaseName)} {}
 
     BoundExtraCreateRelTableGroupInfo(const BoundExtraCreateRelTableGroupInfo& other)
         : BoundExtraCreateTableInfo{copyVector(other.propertyDefinitions)},
           srcMultiplicity{other.srcMultiplicity}, dstMultiplicity{other.dstMultiplicity},
           storageDirection{other.storageDirection}, nodePairs{other.nodePairs},
-          storage{other.storage}, scanFunction{other.scanFunction},
-          scanBindData{other.scanBindData}, foreignDatabaseName{other.foreignDatabaseName} {}
+          storage{other.storage}, storageFormat{other.storageFormat},
+          scanFunction{other.scanFunction}, scanBindData{other.scanBindData},
+          foreignDatabaseName{other.foreignDatabaseName} {}
 
     std::unique_ptr<BoundExtraCreateCatalogEntryInfo> copy() const override {
         return std::make_unique<BoundExtraCreateRelTableGroupInfo>(*this);
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 32dfd20a82..fe466c222a 100644
--- a/src/include/catalog/catalog_entry/node_table_catalog_entry.h
+++ b/src/include/catalog/catalog_entry/node_table_catalog_entry.h
@@ -27,9 +27,10 @@ class LBUG_API NodeTableCatalogEntry final : public TableCatalogEntry {
 
 public:
     NodeTableCatalogEntry() = default;
-    NodeTableCatalogEntry(std::string name, std::string primaryKeyName, std::string storage = "")
+    NodeTableCatalogEntry(std::string name, std::string primaryKeyName, std::string storage = "",
+        std::string storageFormat = "")
         : TableCatalogEntry{entryType_, std::move(name)}, primaryKeyName{std::move(primaryKeyName)},
-          storage{std::move(storage)} {}
+          storage{std::move(storage)}, storageFormat{std::move(storageFormat)} {}
 
     // Constructor for foreign-backed tables
     NodeTableCatalogEntry(std::string name, std::string primaryKeyName,
@@ -56,6 +57,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; }
     std::optional<function::TableFunction> getScanFunction() const override;
     const CreateBindDataFunc& getCreateBindDataFunc() const { return createBindDataFunc; }
     const std::string& getForeignDatabaseName() const { return foreignDatabaseName; }
@@ -82,6 +84,7 @@ class LBUG_API NodeTableCatalogEntry final : public TableCatalogEntry {
 private:
     std::string primaryKeyName;
     std::string storage;
+    std::string storageFormat;
     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 807dc8f002..7d6e792bf1 100644
--- a/src/include/catalog/catalog_entry/rel_group_catalog_entry.h
+++ b/src/include/catalog/catalog_entry/rel_group_catalog_entry.h
@@ -39,13 +39,15 @@ 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 = "",
         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)},
-          scanFunction{std::move(scanFunction)}, scanBindData{std::move(scanBindData)},
+          storageFormat{std::move(storageFormat)}, scanFunction{std::move(scanFunction)},
+          scanBindData{std::move(scanBindData)},
           foreignDatabaseName{std::move(foreignDatabaseName)} {
         propertyCollection =
             PropertyDefinitionCollection{1}; // Skip NBR_NODE_ID column as the first one.
@@ -63,6 +65,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; }
     std::optional<function::TableFunction> getScanFunction() const override { return scanFunction; }
     const std::optional<std::shared_ptr<function::TableFuncBindData>>& getScanBindData() const {
         return scanBindData;
@@ -113,6 +116,7 @@ class LBUG_API RelGroupCatalogEntry final : public TableCatalogEntry {
     common::ExtendDirection storageDirection = common::ExtendDirection::BOTH;
     std::vector<RelTableCatalogInfo> relTableInfos;
     std::string storage;
+    std::string storageFormat;
     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 dcbbb43b53..bf4c38ba1d 100644
--- a/src/include/common/constants.h
+++ b/src/include/common/constants.h
@@ -86,6 +86,13 @@ struct StorageConstants {
 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/storage/storage_manager.h b/src/include/storage/storage_manager.h
index 3268e0ad3e..8f95064338 100644
--- a/src/include/storage/storage_manager.h
+++ b/src/include/storage/storage_manager.h
@@ -12,6 +12,10 @@ namespace main {
 class Database;
 } // namespace main
 
+namespace common {
+class VirtualFileSystem;
+} // namespace common
+
 namespace catalog {
 class CatalogEntry;
 class Catalog;
@@ -40,9 +44,9 @@ class LBUG_API StorageManager {
     static void recover(main::ClientContext& clientContext, bool throwOnWalReplayFailure,
         bool enableChecksums);
 
-    void createTable(catalog::TableCatalogEntry* entry);
-    void addRelTable(catalog::RelGroupCatalogEntry* entry,
-        const catalog::RelTableCatalogInfo& info);
+    void createTable(catalog::TableCatalogEntry* entry, main::ClientContext* context = nullptr);
+    void addRelTable(catalog::RelGroupCatalogEntry* entry, const catalog::RelTableCatalogInfo& info,
+        main::ClientContext* context = nullptr);
 
     bool checkpoint(main::ClientContext* context, PageAllocator& pageAllocator);
     bool checkpoint(main::ClientContext* context, const transaction::Transaction& snapshotTxn,
@@ -65,6 +69,8 @@ class LBUG_API StorageManager {
     bool defaultHashIndexEnabled() const { return enableDefaultHashIndex; }
     void setDefaultHashIndexEnabled(bool enabled) { enableDefaultHashIndex = enabled; }
 
+    common::VirtualFileSystem* getVFS() const { return vfs_; }
+
     void registerIndexType(IndexType indexType) {
         registeredIndexTypes.push_back(std::move(indexType));
     }
@@ -94,9 +100,11 @@ class LBUG_API StorageManager {
     static StorageManager* Get(const main::ClientContext& context);
 
 private:
-    void createNodeTable(catalog::NodeTableCatalogEntry* entry);
+    void createNodeTable(catalog::NodeTableCatalogEntry* entry,
+        main::ClientContext* context = nullptr);
 
-    void createRelTableGroup(catalog::RelGroupCatalogEntry* entry);
+    void createRelTableGroup(catalog::RelGroupCatalogEntry* entry,
+        main::ClientContext* context = nullptr);
 
     void reclaimDroppedTables(const catalog::Catalog& catalog);
 
@@ -115,6 +123,7 @@ class LBUG_API StorageManager {
     bool inMemory;
     std::vector<IndexType> registeredIndexTypes;
     std::unordered_map<common::table_id_t, std::string> tableNameCache;
+    common::VirtualFileSystem* vfs_; // non-owning, owned by Database
 };
 
 } // namespace storage
diff --git a/src/include/storage/table/columnar_node_table_base.h b/src/include/storage/table/columnar_node_table_base.h
index fed8e34492..a719aa8402 100644
--- a/src/include/storage/table/columnar_node_table_base.h
+++ b/src/include/storage/table/columnar_node_table_base.h
@@ -32,7 +32,7 @@ struct ColumnarNodeTableScanSharedState {
     virtual bool getNextMorsel(ColumnarNodeTableScanState* scanState) = 0;
 };
 
-// Abstract base class for columnar-format node tables (Parquet, Arrow, etc.)
+// Abstract base class for columnar-format node tables (Icebug-Disk, Arrow, etc.)
 class ColumnarNodeTableBase : public NodeTable {
 public:
     ColumnarNodeTableBase(const StorageManager* storageManager,
@@ -77,12 +77,6 @@ class ColumnarNodeTableBase : public NodeTable {
     virtual common::row_idx_t getTotalRowCount(
         const transaction::Transaction* transaction) const = 0;
 
-    // Helper for constructing storage paths
-    std::string constructStoragePath(const std::string& prefix, const std::string& suffix) const {
-        std::string tableName = nodeTableCatalogEntry->getName();
-        return prefix + "_nodes_" + tableName + suffix;
-    }
-
 public:
     ColumnarNodeTableScanSharedState* getTableScanSharedState() const {
         return tableScanSharedState.get();
diff --git a/src/include/storage/table/columnar_rel_table_base.h b/src/include/storage/table/columnar_rel_table_base.h
index 5f04510616..fad0595d83 100644
--- a/src/include/storage/table/columnar_rel_table_base.h
+++ b/src/include/storage/table/columnar_rel_table_base.h
@@ -12,7 +12,7 @@
 namespace lbug {
 namespace storage {
 
-// Abstract base class for columnar-format relationship tables (Parquet, Arrow, etc.)
+// Abstract base class for columnar-format relationship tables (Icebug-Disk, Arrow, etc.)
 class ColumnarRelTableBase : public RelTable {
 public:
     ColumnarRelTableBase(catalog::RelGroupCatalogEntry* relGroupEntry,
@@ -54,19 +54,6 @@ class ColumnarRelTableBase : public RelTable {
     virtual common::row_idx_t getTotalRowCount(
         const transaction::Transaction* transaction) const = 0;
 
-    // Helper for constructing storage paths for CSR format
-    struct CSRFilePaths {
-        std::string indices;
-        std::string indptr;
-        std::string metadata;
-    };
-
-    CSRFilePaths constructCSRPaths(const std::string& prefix, const std::string& suffix) const {
-        std::string relName = relGroupEntry->getName();
-        return {prefix + "_indices_" + relName + suffix, prefix + "_indptr_" + relName + suffix,
-            prefix + "_metadata_" + relName + suffix};
-    }
-
     // Helper for finding source node in CSR format
     // Subclasses should cache indptr data and provide it via this interface
     virtual common::offset_t findSourceNodeForRowInternal(common::offset_t globalRowIdx,
diff --git a/src/include/storage/table/ice_disk_constants.h b/src/include/storage/table/ice_disk_constants.h
new file mode 100644
index 0000000000..7e6a9b5d3c
--- /dev/null
+++ b/src/include/storage/table/ice_disk_constants.h
@@ -0,0 +1,18 @@
+#pragma once
+
+#include <string>
+
+#include "common/constants.h"
+
+namespace lbug {
+namespace storage {
+
+struct IceDiskConstants {
+    static constexpr std::string_view V1 = "v1";
+
+    static constexpr std::string_view CURRENT_VERSION = V1;
+
+    static constexpr std::string_view VERSION_METADATA_KEY = "icebug_disk_version";
+};
+} // namespace storage
+} // namespace lbug
diff --git a/src/include/storage/table/parquet_node_table.h b/src/include/storage/table/ice_disk_node_table.h
similarity index 86%
rename from src/include/storage/table/parquet_node_table.h
rename to src/include/storage/table/ice_disk_node_table.h
index dcf681931e..12a085c64d 100644
--- a/src/include/storage/table/parquet_node_table.h
+++ b/src/include/storage/table/ice_disk_node_table.h
@@ -12,9 +12,12 @@
 #include "storage/table/columnar_node_table_base.h"
 
 namespace lbug {
+namespace main {
+class ClientContext;
+} // namespace main
 namespace storage {
 
-struct ParquetNodeTableScanState final : ColumnarNodeTableScanState {
+struct IceDiskNodeTableScanState final : ColumnarNodeTableScanState {
     std::unique_ptr<processor::ParquetReader> parquetReader;
     std::unique_ptr<processor::ParquetReaderScanState> parquetScanState;
     bool dataRead = false;
@@ -22,7 +25,7 @@ struct ParquetNodeTableScanState final : ColumnarNodeTableScanState {
     size_t nextRowToDistribute = 0;
     uint64_t lastQueryId = 0; // Track the last query ID to detect new queries
 
-    ParquetNodeTableScanState([[maybe_unused]] MemoryManager& mm, common::ValueVector* nodeIDVector,
+    IceDiskNodeTableScanState([[maybe_unused]] MemoryManager& mm, common::ValueVector* nodeIDVector,
         std::vector<common::ValueVector*> outputVectors,
         std::shared_ptr<common::DataChunkState> outChunkState)
         : ColumnarNodeTableScanState{mm, nodeIDVector, std::move(outputVectors),
@@ -31,7 +34,7 @@ struct ParquetNodeTableScanState final : ColumnarNodeTableScanState {
     }
 };
 
-struct ParquetNodeTableScanSharedState final : ColumnarNodeTableScanSharedState {
+struct IceDiskNodeTableScanSharedState final : ColumnarNodeTableScanSharedState {
 private:
     std::mutex mtx;
     common::node_group_idx_t currentBatchIdx = 0;
@@ -58,10 +61,11 @@ struct ParquetNodeTableScanSharedState final : ColumnarNodeTableScanSharedState
     bool getNextMorsel(ColumnarNodeTableScanState*) override { return false; }
 };
 
-class ParquetNodeTable final : public ColumnarNodeTableBase {
+class IceDiskNodeTable final : public ColumnarNodeTableBase {
 public:
-    ParquetNodeTable(const StorageManager* storageManager,
-        const catalog::NodeTableCatalogEntry* nodeTableEntry, MemoryManager* memoryManager);
+    IceDiskNodeTable(const StorageManager* storageManager,
+        const catalog::NodeTableCatalogEntry* nodeTableEntry, MemoryManager* memoryManager,
+        main::ClientContext* context = nullptr);
 
     void initializeScanCoordination(const transaction::Transaction* transaction) override;
 
@@ -79,7 +83,7 @@ class ParquetNodeTable final : public ColumnarNodeTableBase {
 
 protected:
     // Implement ColumnarNodeTableBase interface
-    std::string getColumnarFormatName() const override { return "Parquet"; }
+    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;
@@ -90,7 +94,7 @@ class ParquetNodeTable final : public ColumnarNodeTableBase {
 
     void initializeParquetReader(transaction::Transaction* transaction) const;
     void initParquetScanForRowGroup(transaction::Transaction* transaction,
-        ParquetNodeTableScanState& scanState) const;
+        IceDiskNodeTableScanState& iceDiskScanState) const;
 };
 
 } // namespace storage
diff --git a/src/include/storage/table/parquet_rel_table.h b/src/include/storage/table/ice_disk_rel_table.h
similarity index 83%
rename from src/include/storage/table/parquet_rel_table.h
rename to src/include/storage/table/ice_disk_rel_table.h
index c63a373a7f..975424d6db 100644
--- a/src/include/storage/table/parquet_rel_table.h
+++ b/src/include/storage/table/ice_disk_rel_table.h
@@ -8,9 +8,12 @@
 #include "transaction/transaction.h"
 
 namespace lbug {
+namespace main {
+class ClientContext;
+} // namespace main
 namespace storage {
 
-struct ParquetRelTableScanState final : RelTableScanState {
+struct IceDiskRelTableScanState final : RelTableScanState {
     std::unique_ptr<processor::ParquetReaderScanState> parquetScanState;
 
     // Row group range for morsel-driven parallelism
@@ -21,7 +24,7 @@ struct ParquetRelTableScanState final : RelTableScanState {
     std::unique_ptr<processor::ParquetReader> indicesReader;
     std::unique_ptr<processor::ParquetReader> indptrReader;
 
-    ParquetRelTableScanState(MemoryManager& mm, common::ValueVector* nodeIDVector,
+    IceDiskRelTableScanState(MemoryManager& mm, common::ValueVector* nodeIDVector,
         std::vector<common::ValueVector*> outputVectors,
         std::shared_ptr<common::DataChunkState> outChunkState)
         : RelTableScanState{mm, nodeIDVector, std::move(outputVectors), std::move(outChunkState)} {
@@ -34,11 +37,11 @@ struct ParquetRelTableScanState final : RelTableScanState {
         common::RelDataDirection direction_) override;
 };
 
-class ParquetRelTable final : public ColumnarRelTableBase {
+class IceDiskRelTable final : public ColumnarRelTableBase {
 public:
-    ParquetRelTable(catalog::RelGroupCatalogEntry* relGroupEntry, common::table_id_t fromTableID,
+    IceDiskRelTable(catalog::RelGroupCatalogEntry* relGroupEntry, common::table_id_t fromTableID,
         common::table_id_t toTableID, const StorageManager* storageManager,
-        MemoryManager* memoryManager);
+        MemoryManager* memoryManager, main::ClientContext* context = nullptr);
 
     void initScanState(transaction::Transaction* transaction, TableScanState& scanState,
         bool resetCachedBoundNodeSelVec = true) const override;
@@ -47,7 +50,7 @@ class ParquetRelTable final : public ColumnarRelTableBase {
 
 protected:
     // Implement ColumnarRelTableBase interface
-    std::string getColumnarFormatName() const override { return "Parquet"; }
+    std::string getColumnarFormatName() const override { return "icebug-disk"; }
     common::row_idx_t getTotalRowCount(const transaction::Transaction* transaction) const override;
 
 private:
@@ -63,10 +66,9 @@ class ParquetRelTable final : public ColumnarRelTableBase {
     void initializeIndptrReader(transaction::Transaction* transaction) const;
     void loadIndptrData(transaction::Transaction* transaction) const;
     bool scanInternalByRowGroups(transaction::Transaction* transaction,
-        ParquetRelTableScanState& parquetRelScanState);
+        IceDiskRelTableScanState& iceDiskScanState);
     bool scanRowGroupForBoundNodes(transaction::Transaction* transaction,
-        ParquetRelTableScanState& parquetRelScanState,
-        const std::vector<uint64_t>& rowGroupsToProcess,
+        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
new file mode 100644
index 0000000000..07b5942036
--- /dev/null
+++ b/src/include/storage/table/ice_disk_utils.h
@@ -0,0 +1,106 @@
+#pragma once
+
+#include <algorithm>
+#include <cctype>
+#include <filesystem>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include "common/constants.h"
+#include "common/exception/runtime.h"
+#include "common/file_system/virtual_file_system.h"
+#include "processor/operator/persistent/reader/parquet/parquet_reader.h"
+#include "storage/table/ice_disk_constants.h"
+
+namespace lbug {
+namespace storage {
+
+struct CSRFilePaths {
+    std::string indices;
+    std::string indptr;
+};
+
+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)
+    static std::string joinPath(const std::string& base, const std::string& part) {
+        if (base.empty()) {
+            return part;
+        }
+        const char last = base.back();
+        if (last == '/' || last == '\\') {
+            return base + part;
+        }
+
+        return base + "/" + part;
+    }
+
+    // Get the file path for a given node table's parquet file
+    static std::string constructNodeTablePath(const std::string& dir, const std::string& name,
+        const std::string& suffix) {
+        return IceDiskUtils::joinPath(dir, "nodes_" + name + suffix);
+    }
+
+    // Get the file paths for a given rel table's CSR files
+    static CSRFilePaths constructCSRPaths(const std::string& dir, const std::string& name,
+        const std::string& suffix) {
+        return {IceDiskUtils::joinPath(dir, "indices_" + name + suffix),
+            IceDiskUtils::joinPath(dir, "indptr_" + name + suffix)};
+    }
+
+    // Validates that the parquet file at `path` carries the expected icebug_disk_version metadata.
+    // Note: path is already resolved by VFS
+    static void checkVersionCompatibility(main::ClientContext* context, const std::string& path) {
+        if (!context) {
+            throw common::RuntimeException(
+                path + ": failed to read parquet metadata for version check");
+        }
+
+        auto tempReader =
+            std::make_unique<processor::ParquetReader>(path, std::vector<bool>{}, context);
+        if (!tempReader) {
+            throw common::RuntimeException(
+                path + ": failed to read parquet metadata for version check");
+        }
+
+        auto metadata = tempReader->getMetadata();
+        if (!metadata) {
+            throw common::RuntimeException(
+                path + ": failed to read parquet metadata for version check");
+        }
+
+        // key_value_metadata is std::vector<KeyValue>
+        bool found = false;
+        std::string versionValue;
+        for (const auto& kv : metadata->key_value_metadata) {
+            if (kv.key == IceDiskConstants::VERSION_METADATA_KEY) {
+                versionValue = kv.value;
+                found = true;
+                break;
+            }
+        }
+
+        if (!found) {
+            throw common::RuntimeException(
+                path + ": parquet file is missing icebug_disk_version metadata");
+        }
+
+        const auto& expected = IceDiskConstants::CURRENT_VERSION;
+        const bool versionMatches = versionValue.size() == expected.size() &&
+                                    std::equal(versionValue.begin(), versionValue.end(),
+                                        expected.begin(), [](unsigned char a, unsigned char b) {
+                                            return std::tolower(a) == std::tolower(b);
+                                        });
+
+        if (!versionMatches) {
+            throw common::RuntimeException(
+                path +
+                ": current ladybug version does not support icebug_disk_version: " + versionValue);
+        }
+    }
+};
+
+} // namespace storage
+} // namespace lbug
diff --git a/src/include/storage/table/rel_table.h b/src/include/storage/table/rel_table.h
index 696985463a..5bee7d6223 100644
--- a/src/include/storage/table/rel_table.h
+++ b/src/include/storage/table/rel_table.h
@@ -30,7 +30,7 @@ struct RelTableScanState : TableScanState {
     std::unique_ptr<LocalRelTableScanState> localTableScanState;
 
     // Optional state used by Arrow-backed relationship tables. Keep it on the common scan state so
-    // a single multi-rel scan state can scan native, Parquet-backed, and Arrow-backed tables.
+    // a single multi-rel scan state can scan native, icebug-disk-backed, and Arrow-backed tables.
     size_t arrowCurrentBatchIdx = 0;
     size_t arrowCurrentBatchOffset = 0;
     std::unordered_map<common::offset_t, common::sel_t> arrowBoundNodeOffsetToSelPos;
diff --git a/src/processor/operator/ddl/alter.cpp b/src/processor/operator/ddl/alter.cpp
index 6e99af8b70..72da92f3f3 100644
--- a/src/processor/operator/ddl/alter.cpp
+++ b/src/processor/operator/ddl/alter.cpp
@@ -346,7 +346,7 @@ void Alter::alterTable(main::ClientContext* clientContext, const TableCatalogEnt
         auto connectionInfo = alterInfo.extraInfo->constPtrCast<BoundExtraAlterFromToConnection>();
         auto relEntryInfo =
             relGroupEntry->getRelEntryInfo(connectionInfo->fromTableID, connectionInfo->toTableID);
-        storageManager->addRelTable(relGroupEntry, *relEntryInfo);
+        storageManager->addRelTable(relGroupEntry, *relEntryInfo, clientContext);
     } break;
     default:
         break;
diff --git a/src/processor/operator/ddl/create_table.cpp b/src/processor/operator/ddl/create_table.cpp
index d98e9ee40a..d3479c3b49 100644
--- a/src/processor/operator/ddl/create_table.cpp
+++ b/src/processor/operator/ddl/create_table.cpp
@@ -42,7 +42,8 @@ void CreateTable::executeInternal(ExecutionContext* context) {
     default:
         UNREACHABLE_CODE;
     }
-    storage::StorageManager::Get(*clientContext)->createTable(entry->ptrCast<TableCatalogEntry>());
+    storage::StorageManager::Get(*clientContext)
+        ->createTable(entry->ptrCast<TableCatalogEntry>(), clientContext);
     appendMessage(std::format("Table {} has been created.", info.tableName), memoryManager);
     sharedState->tableCreated = true;
 }
diff --git a/src/processor/operator/scan/scan_multi_rel_tables.cpp b/src/processor/operator/scan/scan_multi_rel_tables.cpp
index 27f5422379..f24eae254b 100644
--- a/src/processor/operator/scan/scan_multi_rel_tables.cpp
+++ b/src/processor/operator/scan/scan_multi_rel_tables.cpp
@@ -3,7 +3,7 @@
 #include "processor/execution_context.h"
 #include "storage/local_storage/local_storage.h"
 #include "storage/table/arrow_rel_table.h"
-#include "storage/table/parquet_rel_table.h"
+#include "storage/table/ice_disk_rel_table.h"
 
 using namespace lbug::common;
 using namespace lbug::storage;
@@ -68,28 +68,28 @@ void ScanMultiRelTable::initLocalStateInternal(ResultSet* resultSet, ExecutionCo
 
     // Check if any table in any scanner is an external rel table with a custom scan state.
     bool hasArrowTable = false;
-    bool hasParquetTable = false;
+    bool hasIceDiskTable = false;
     for (auto& [_, scanner] : scanners) {
         for (auto& relInfo : scanner.relInfos) {
             if (dynamic_cast<storage::ArrowRelTable*>(relInfo.table) != nullptr) {
                 hasArrowTable = true;
                 break;
             }
-            if (dynamic_cast<storage::ParquetRelTable*>(relInfo.table) != nullptr) {
-                hasParquetTable = true;
+            if (dynamic_cast<storage::IceDiskRelTable*>(relInfo.table) != nullptr) {
+                hasIceDiskTable = true;
                 break;
             }
         }
-        if (hasArrowTable || hasParquetTable) {
+        if (hasArrowTable || hasIceDiskTable) {
             break;
         }
     }
 
-    // Parquet scan state extends the common rel scan state and Arrow stores its per-table state
-    // there, so one scan state can now cover Parquet, Arrow, and native rel tables.
-    if (hasParquetTable) {
+    // 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::ParquetRelTableScanState>(*MemoryManager::Get(*clientContext),
+            std::make_unique<storage::IceDiskRelTableScanState>(*MemoryManager::Get(*clientContext),
                 boundNodeIDVector, outVectors, nbrNodeIDVector->state);
     } else if (hasArrowTable) {
         scanState =
diff --git a/src/processor/operator/scan/scan_node_table.cpp b/src/processor/operator/scan/scan_node_table.cpp
index 9496756cf0..e9c9524854 100644
--- a/src/processor/operator/scan/scan_node_table.cpp
+++ b/src/processor/operator/scan/scan_node_table.cpp
@@ -7,7 +7,7 @@
 #include "storage/local_storage/local_node_table.h"
 #include "storage/local_storage/local_storage.h"
 #include "storage/table/arrow_node_table.h"
-#include "storage/table/parquet_node_table.h"
+#include "storage/table/ice_disk_node_table.h"
 
 using namespace lbug::common;
 using namespace lbug::storage;
@@ -17,8 +17,8 @@ namespace processor {
 static std::unique_ptr<TableScanState> createNodeTableScanState(NodeTable* table,
     ValueVector* nodeIDVector, const std::vector<ValueVector*>& outVectors,
     MemoryManager* memoryManager) {
-    if (dynamic_cast<ParquetNodeTable*>(table) != nullptr) {
-        return std::make_unique<ParquetNodeTableScanState>(*memoryManager, nodeIDVector, outVectors,
+    if (dynamic_cast<IceDiskNodeTable*>(table) != nullptr) {
+        return std::make_unique<IceDiskNodeTableScanState>(*memoryManager, nodeIDVector, outVectors,
             nodeIDVector->state);
     }
     if (dynamic_cast<ArrowNodeTable*>(table) != nullptr) {
@@ -53,16 +53,16 @@ void ScanNodeTableSharedState::initialize(const transaction::Transaction* transa
     this->currentCommittedGroupIdx = 0;
     this->currentUnCommittedGroupIdx = 0;
 
-    // Initialize table-specific scan coordination (e.g., for ParquetNodeTable)
+    // Initialize table-specific scan coordination (e.g., for IceDiskNodeTable)
     table->initializeScanCoordination(transaction);
 
-    if (const auto parquetTable = dynamic_cast<ParquetNodeTable*>(table)) {
-        // For parquet tables, set numCommittedNodeGroups to number of row groups
+    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, parquetTable->getParquetFilePath());
+                common::VirtualFileSystem::resolvePath(context, iceDiskTable->getParquetFilePath());
             auto tempReader =
                 std::make_unique<processor::ParquetReader>(resolvedPath, columnSkips, context);
             this->numCommittedNodeGroups = tempReader->getNumRowGroups();
@@ -91,7 +91,7 @@ void ScanNodeTableSharedState::nextMorsel(TableScanState& scanState,
     std::unique_lock lck{mtx};
 
     // ColumnarNodeTables handle morsel assignment internally
-    // TODO: parquet tables https://github.com/LadybugDB/ladybug/issues/245
+    // TODO: icebug-disk tables https://github.com/LadybugDB/ladybug/issues/245
     if (const auto arrowTable = dynamic_cast<ArrowNodeTable*>(this->table)) {
         const auto tableSharedState = arrowTable->getTableScanSharedState();
         if (tableSharedState->getNextMorsel(static_cast<ColumnarNodeTableScanState*>(&scanState))) {
@@ -149,8 +149,8 @@ void ScanNodeTable::initCurrentTable(ExecutionContext* context) {
         outVectors, MemoryManager::Get(*context->clientContext));
     currentInfo.initScanState(*scanState, outVectors, context->clientContext);
     scanState->semiMask = sharedStates[currentTableIdx]->getSemiMask();
-    // Call table->initScanState for ParquetNodeTable or ArrowNodeTable
-    if (dynamic_cast<ParquetNodeTable*>(tableInfos[currentTableIdx].table) ||
+    // Call table->initScanState for IceDiskNodeTable or ArrowNodeTable
+    if (dynamic_cast<IceDiskNodeTable*>(tableInfos[currentTableIdx].table) ||
         dynamic_cast<ArrowNodeTable*>(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 f09b7a9abd..de7c4051df 100644
--- a/src/processor/operator/scan/scan_rel_table.cpp
+++ b/src/processor/operator/scan/scan_rel_table.cpp
@@ -9,8 +9,8 @@
 #include "storage/local_storage/local_rel_table.h"
 #include "storage/table/arrow_rel_table.h"
 #include "storage/table/foreign_rel_table.h"
+#include "storage/table/ice_disk_rel_table.h"
 #include "storage/table/node_table.h"
-#include "storage/table/parquet_rel_table.h"
 
 using namespace lbug::common;
 using namespace lbug::storage;
@@ -76,15 +76,15 @@ void ScanRelTable::initLocalStateInternal(ResultSet* resultSet, ExecutionContext
     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* parquetTable = dynamic_cast<storage::ParquetRelTable*>(tableInfo.table);
+    auto* iceDiskTable = dynamic_cast<storage::IceDiskRelTable*>(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 (parquetTable) {
+    } else if (iceDiskTable) {
         scanState =
-            std::make_unique<storage::ParquetRelTableScanState>(*MemoryManager::Get(*clientContext),
+            std::make_unique<storage::IceDiskRelTableScanState>(*MemoryManager::Get(*clientContext),
                 boundNodeIDVector, outVectors, nbrNodeIDVector->state);
     } else if (foreignTable) {
         scanState =
diff --git a/src/storage/storage_manager.cpp b/src/storage/storage_manager.cpp
index 82a22dd750..6a986fbd75 100644
--- a/src/storage/storage_manager.cpp
+++ b/src/storage/storage_manager.cpp
@@ -4,6 +4,7 @@
 #include "catalog/catalog_entry/node_table_catalog_entry.h"
 #include "catalog/catalog_entry/rel_group_catalog_entry.h"
 #include "common/arrow/arrow.h"
+#include "common/constants.h"
 #include "common/file_system/virtual_file_system.h"
 #include "common/random_engine.h"
 #include "common/serializer/in_mem_file_writer.h"
@@ -19,9 +20,9 @@
 #include "storage/table/arrow_rel_table.h"
 #include "storage/table/arrow_table_support.h"
 #include "storage/table/foreign_rel_table.h"
+#include "storage/table/ice_disk_node_table.h"
+#include "storage/table/ice_disk_rel_table.h"
 #include "storage/table/node_table.h"
-#include "storage/table/parquet_node_table.h"
-#include "storage/table/parquet_rel_table.h"
 #include "storage/table/rel_table.h"
 #include "storage/wal/wal_replayer.h"
 #include "transaction/transaction.h"
@@ -38,7 +39,8 @@ StorageManager::StorageManager(const std::string& databasePath, bool readOnly, b
     MemoryManager& memoryManager, bool enableCompression, bool enableDefaultHashIndex,
     VirtualFileSystem* vfs)
     : databasePath{databasePath}, readOnly{readOnly}, dataFH{nullptr}, memoryManager{memoryManager},
-      enableCompression{enableCompression}, enableDefaultHashIndex{enableDefaultHashIndex} {
+      enableCompression{enableCompression}, enableDefaultHashIndex{enableDefaultHashIndex},
+      vfs_{vfs} {
     wal = std::make_unique<WAL>(databasePath, readOnly, enableChecksums, vfs);
     shadowFile =
         std::make_unique<ShadowFile>(*memoryManager.getBufferManager(), vfs, this->databasePath);
@@ -95,9 +97,19 @@ void StorageManager::recover(main::ClientContext& clientContext, bool throwOnWal
     walReplayer->replay(throwOnWalReplayFailure, enableChecksums);
 }
 
-void StorageManager::createNodeTable(NodeTableCatalogEntry* entry) {
+void StorageManager::createNodeTable(NodeTableCatalogEntry* entry, main::ClientContext* context) {
     tableNameCache[entry->getTableID()] = entry->getName();
-    if (!entry->getStorage().empty()) {
+
+    if (!entry->getStorageFormat().empty()) {
+        if (TableOptionConstants::isIceBugDiskFormat(entry->getStorageFormat())) {
+            // 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());
+        }
+    } else if (!entry->getStorage().empty()) {
         // Check if storage is Arrow backed
         if (entry->getStorage().substr(0, 8) == "arrow://") {
             // Extract Arrow ID from storage string
@@ -122,9 +134,8 @@ void StorageManager::createNodeTable(NodeTableCatalogEntry* entry) {
             tables[entry->getTableID()] = std::make_unique<ArrowNodeTable>(this, entry,
                 &memoryManager, std::move(schemaCopy), std::move(arraysCopy), arrowId);
         } else {
-            // Create parquet-backed node table
-            tables[entry->getTableID()] =
-                std::make_unique<ParquetNodeTable>(this, entry, &memoryManager);
+            throw common::RuntimeException(
+                "Unsupported storage option for node table: " + entry->getStorage());
         }
     } else {
         // Create regular node table
@@ -135,12 +146,22 @@ void StorageManager::createNodeTable(NodeTableCatalogEntry* entry) {
 // TODO(Guodong): This API is added since storageManager doesn't provide an API to add a single
 // rel table. We may have to refactor the existing StorageManager::createTable(TableCatalogEntry*
 // entry).
-void StorageManager::addRelTable(RelGroupCatalogEntry* entry, const RelTableCatalogInfo& info) {
+void StorageManager::addRelTable(RelGroupCatalogEntry* entry, const RelTableCatalogInfo& info,
+    main::ClientContext* context) {
     if (entry->getScanFunction().has_value()) {
         // Create foreign-backed rel table
         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())) {
+            // 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());
+        }
     } else if (!entry->getStorage().empty()) {
         if (entry->getStorage().substr(0, 8) == "arrow://") {
             std::string arrowId = entry->getStorage().substr(8);
@@ -170,9 +191,8 @@ void StorageManager::addRelTable(RelGroupCatalogEntry* entry, const RelTableCata
                 info.nodePair.dstTableID, this, &memoryManager, fromNodeTable, toNodeTable,
                 std::move(schemaCopy), std::move(arraysCopy), arrowId);
         } else {
-            // Create parquet-backed rel table
-            tables[info.oid] = std::make_unique<ParquetRelTable>(entry, info.nodePair.srcTableID,
-                info.nodePair.dstTableID, this, &memoryManager);
+            throw common::RuntimeException(
+                "Unsupported storage option for rel table: " + entry->getStorage());
         }
     } else {
         // Create regular rel table
@@ -181,20 +201,21 @@ void StorageManager::addRelTable(RelGroupCatalogEntry* entry, const RelTableCata
     }
 }
 
-void StorageManager::createRelTableGroup(RelGroupCatalogEntry* entry) {
+void StorageManager::createRelTableGroup(RelGroupCatalogEntry* entry,
+    main::ClientContext* context) {
     for (auto& info : entry->getRelEntryInfos()) {
-        addRelTable(entry, info);
+        addRelTable(entry, info, context);
     }
 }
 
-void StorageManager::createTable(TableCatalogEntry* entry) {
+void StorageManager::createTable(TableCatalogEntry* entry, main::ClientContext* context) {
     std::unique_lock lck{mtx};
     switch (entry->getType()) {
     case CatalogEntryType::NODE_TABLE_ENTRY: {
-        createNodeTable(entry->ptrCast<NodeTableCatalogEntry>());
+        createNodeTable(entry->ptrCast<NodeTableCatalogEntry>(), context);
     } break;
     case CatalogEntryType::REL_GROUP_ENTRY: {
-        createRelTableGroup(entry->ptrCast<RelGroupCatalogEntry>());
+        createRelTableGroup(entry->ptrCast<RelGroupCatalogEntry>(), context);
     } break;
     default: {
         UNREACHABLE_CODE;
@@ -436,9 +457,10 @@ void StorageManager::deserialize(main::ClientContext* context, const Catalog* ca
         auto tableEntry = catalog->getTableCatalogEntry(&DUMMY_TRANSACTION, tableID)
                               ->ptrCast<NodeTableCatalogEntry>();
         tableNameCache[tableID] = tableEntry->getName();
-        if (!tableEntry->getStorage().empty()) {
-            // Create parquet-backed node table
-            tables[tableID] = std::make_unique<ParquetNodeTable>(this, tableEntry, &memoryManager);
+        if (TableOptionConstants::isIceBugDiskFormat(tableEntry->getStorageFormat())) {
+            // Create icebug-disk-backed node table
+            tables[tableID] =
+                std::make_unique<IceDiskNodeTable>(this, tableEntry, &memoryManager, context);
         } else {
             // Create regular node table
             tables[tableID] = std::make_unique<NodeTable>(this, tableEntry, &memoryManager);
@@ -464,10 +486,11 @@ 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 (!relGroupEntry->getStorage().empty()) {
-                // Create parquet-backed rel table
-                tables[info.oid] = std::make_unique<ParquetRelTable>(relGroupEntry,
-                    info.nodePair.srcTableID, info.nodePair.dstTableID, this, &memoryManager);
+            if (TableOptionConstants::isIceBugDiskFormat(relGroupEntry->getStorageFormat())) {
+                // Create icebug-disk-backed rel table
+                tables[info.oid] =
+                    std::make_unique<IceDiskRelTable>(relGroupEntry, info.nodePair.srcTableID,
+                        info.nodePair.dstTableID, this, &memoryManager, context);
             } else {
                 // Create regular rel table
                 tables[info.oid] = std::make_unique<RelTable>(relGroupEntry,
diff --git a/src/storage/table/CMakeLists.txt b/src/storage/table/CMakeLists.txt
index 54a76cca23..121a8e319e 100644
--- a/src/storage/table/CMakeLists.txt
+++ b/src/storage/table/CMakeLists.txt
@@ -28,8 +28,8 @@ add_library(lbug_storage_store
         node_group_collection.cpp
         node_table.cpp
         null_column.cpp
-        parquet_node_table.cpp
-        parquet_rel_table.cpp
+        ice_disk_node_table.cpp
+        ice_disk_rel_table.cpp
         rel_table.cpp
         rel_table_data.cpp
         string_chunk_data.cpp
diff --git a/src/storage/table/arrow_rel_table.cpp b/src/storage/table/arrow_rel_table.cpp
index aa218e87fb..befaa29a20 100644
--- a/src/storage/table/arrow_rel_table.cpp
+++ b/src/storage/table/arrow_rel_table.cpp
@@ -40,7 +40,7 @@ static int64_t findColumnIdx(const ArrowSchemaWrapper& schema, const std::string
 void ArrowRelTableScanState::setToTable(const transaction::Transaction* transaction, Table* table_,
     std::vector<column_id_t> columnIDs_, std::vector<ColumnPredicateSet> columnPredicateSets_,
     RelDataDirection direction_) {
-    // Same behavior as ParquetRelTable: no local table for external data sources.
+    // Same behavior as IceDiskRelTable: no local table for external data sources.
     TableScanState::setToTable(transaction, table_, std::move(columnIDs_),
         std::move(columnPredicateSets_));
     columns.resize(columnIDs.size());
diff --git a/src/storage/table/parquet_node_table.cpp b/src/storage/table/ice_disk_node_table.cpp
similarity index 70%
rename from src/storage/table/parquet_node_table.cpp
rename to src/storage/table/ice_disk_node_table.cpp
index b8178cccd6..0c5ebac24b 100644
--- a/src/storage/table/parquet_node_table.cpp
+++ b/src/storage/table/ice_disk_node_table.cpp
@@ -1,5 +1,6 @@
-#include "storage/table/parquet_node_table.h"
+#include "storage/table/ice_disk_node_table.h"
 
+#include <filesystem>
 #include <mutex>
 
 #include "catalog/catalog_entry/node_table_catalog_entry.h"
@@ -13,6 +14,7 @@
 #include "storage/storage_manager.h"
 #include "storage/storage_utils.h"
 #include "storage/table/column.h"
+#include "storage/table/ice_disk_utils.h"
 #include "transaction/transaction.h"
 
 using namespace lbug::catalog;
@@ -23,27 +25,26 @@ using namespace lbug::transaction;
 namespace lbug {
 namespace storage {
 
-ParquetNodeTable::ParquetNodeTable(const StorageManager* storageManager,
-    const NodeTableCatalogEntry* nodeTableEntry, MemoryManager* memoryManager)
+IceDiskNodeTable::IceDiskNodeTable(const StorageManager* storageManager,
+    const NodeTableCatalogEntry* nodeTableEntry, MemoryManager* memoryManager,
+    main::ClientContext* context)
     : ColumnarNodeTableBase{storageManager, nodeTableEntry, memoryManager,
-          std::make_unique<ParquetNodeTableScanSharedState>()} {
-    std::string prefix = nodeTableEntry->getStorage();
-    if (prefix.empty()) {
-        throw RuntimeException("Parquet file prefix is empty for parquet-backed node table");
-    }
-
-    // Use base class helper to construct storage path
-    parquetFilePath = constructStoragePath(prefix, ".parquet");
+          std::make_unique<IceDiskNodeTableScanSharedState>()} {
+    auto resolvedPath = common::VirtualFileSystem::resolvePath(context,
+        IceDiskUtils::constructNodeTablePath(nodeTableEntry->getStorage(),
+            nodeTableEntry->getName(), ".parquet"));
+    IceDiskUtils::checkVersionCompatibility(context, resolvedPath);
+    parquetFilePath = resolvedPath;
 }
 
-void ParquetNodeTable::initializeScanCoordination(const transaction::Transaction* transaction) {
-    auto parquetScanSharedState =
-        static_cast<ParquetNodeTableScanSharedState*>(tableScanSharedState.get());
+void IceDiskNodeTable::initializeScanCoordination(const transaction::Transaction* transaction) {
+    auto iceDiskScanSharedState =
+        static_cast<IceDiskNodeTableScanSharedState*>(tableScanSharedState.get());
     auto numBatches = getNumBatches(transaction);
-    parquetScanSharedState->reset(numBatches);
+    iceDiskScanSharedState->reset(numBatches);
 }
 
-void ParquetNodeTable::initScanState(Transaction* transaction, TableScanState& scanState,
+void IceDiskNodeTable::initScanState(Transaction* transaction, TableScanState& scanState,
     [[maybe_unused]] bool resetCachedBoundNodeSelVec) const {
     // Set up the scan state similar to how NodeTable does it
     auto& nodeScanState = scanState.cast<NodeTableScanState>();
@@ -51,19 +52,19 @@ void ParquetNodeTable::initScanState(Transaction* transaction, TableScanState& s
 
     // Note: Don't set nodeGroupIdx here - it's set by the morsel-driven parallelism system
 
-    auto& parquetNodeScanState = static_cast<ParquetNodeTableScanState&>(nodeScanState);
+    auto& iceDiskScanState = static_cast<IceDiskNodeTableScanState&>(nodeScanState);
 
     // Reset scan state for each scan to allow multiple scans of the same table in one query
-    parquetNodeScanState.dataRead = false;
-    parquetNodeScanState.allData.clear();
-    parquetNodeScanState.totalRows = 0;
-    parquetNodeScanState.nextRowToDistribute = 0;
+    iceDiskScanState.dataRead = false;
+    iceDiskScanState.allData.clear();
+    iceDiskScanState.totalRows = 0;
+    iceDiskScanState.nextRowToDistribute = 0;
 
     // Reset scan completion flag for this scan state
-    parquetNodeScanState.scanCompleted = false;
+    iceDiskScanState.scanCompleted = false;
 
     // Each scan state gets its own parquet reader for thread safety
-    if (!parquetNodeScanState.initialized) {
+    if (!iceDiskScanState.initialized) {
         auto context = transaction->getClientContext();
         if (!context) {
             throw RuntimeException("Invalid client context for parquet scan state initialization");
@@ -71,10 +72,9 @@ void ParquetNodeTable::initScanState(Transaction* transaction, TableScanState& s
 
         std::vector<bool> columnSkips;
         try {
-            auto resolvedPath = VirtualFileSystem::resolvePath(context, parquetFilePath);
-            parquetNodeScanState.parquetReader =
-                std::make_unique<ParquetReader>(resolvedPath, columnSkips, context);
-            parquetNodeScanState.initialized = true;
+            iceDiskScanState.parquetReader =
+                std::make_unique<ParquetReader>(parquetFilePath, columnSkips, context);
+            iceDiskScanState.initialized = true;
         } catch (const std::exception& e) {
             throw RuntimeException("Failed to initialize parquet reader for file '" +
                                    parquetFilePath + "': " + e.what());
@@ -82,13 +82,13 @@ void ParquetNodeTable::initScanState(Transaction* transaction, TableScanState& s
     }
 
     // Set nodeGroupIdx to invalid initially - will be assigned by getNextBatch
-    parquetNodeScanState.nodeGroupIdx = INVALID_NODE_GROUP_IDX;
+    iceDiskScanState.nodeGroupIdx = INVALID_NODE_GROUP_IDX;
 
     // Initialize scan state for the current row group (assigned via shared state)
-    initParquetScanForRowGroup(transaction, parquetNodeScanState);
+    initParquetScanForRowGroup(transaction, iceDiskScanState);
 }
 
-common::node_group_idx_t ParquetNodeTable::getNumBatches(const Transaction* transaction) const {
+common::node_group_idx_t IceDiskNodeTable::getNumBatches(const Transaction* transaction) const {
     auto context = transaction->getClientContext();
     if (!context) {
         return 1;
@@ -96,16 +96,15 @@ common::node_group_idx_t ParquetNodeTable::getNumBatches(const Transaction* tran
 
     std::vector<bool> columnSkips;
     try {
-        auto resolvedPath = VirtualFileSystem::resolvePath(context, parquetFilePath);
-        auto tempReader = std::make_unique<ParquetReader>(resolvedPath, columnSkips, context);
+        auto tempReader = std::make_unique<ParquetReader>(parquetFilePath, columnSkips, context);
         return tempReader->getNumRowGroups();
     } catch (const std::exception& e) {
         return 1; // Fallback
     }
 }
 
-void ParquetNodeTable::initParquetScanForRowGroup(Transaction* transaction,
-    ParquetNodeTableScanState& scanState) const {
+void IceDiskNodeTable::initParquetScanForRowGroup(Transaction* transaction,
+    IceDiskNodeTableScanState& iceDiskScanState) const {
     auto context = transaction->getClientContext();
     if (!context) {
         return;
@@ -117,60 +116,62 @@ void ParquetNodeTable::initParquetScanForRowGroup(Transaction* transaction,
     }
 
     // Defensive check: ensure parquet reader exists
-    if (!scanState.parquetReader) {
+    if (!iceDiskScanState.parquetReader) {
         return;
     }
 
     // Defensive check: ensure parquet scan state exists
-    if (!scanState.parquetScanState) {
+    if (!iceDiskScanState.parquetScanState) {
         return;
     }
 
     std::vector<uint64_t> groupsToRead;
 
     // Use shared state to get the next available row group for this scan state
-    if (scanState.nodeGroupIdx == INVALID_NODE_GROUP_IDX) {
+    if (iceDiskScanState.nodeGroupIdx == INVALID_NODE_GROUP_IDX) {
         common::node_group_idx_t assignedRowGroup;
-        if (dynamic_cast<ParquetNodeTableScanSharedState*>(tableScanSharedState.get())
+        if (dynamic_cast<IceDiskNodeTableScanSharedState*>(tableScanSharedState.get())
                 ->getNextBatch(assignedRowGroup)) {
-            scanState.nodeGroupIdx = assignedRowGroup;
+            iceDiskScanState.nodeGroupIdx = assignedRowGroup;
             groupsToRead.push_back(assignedRowGroup);
         } else {
             // No more row groups available - mark scan as completed
-            scanState.scanCompleted = true;
+            iceDiskScanState.scanCompleted = true;
             // Still need to initialize the scan state with empty groups so reader is in valid state
-            scanState.parquetReader->initializeScan(*scanState.parquetScanState, groupsToRead, vfs);
+            iceDiskScanState.parquetReader->initializeScan(*iceDiskScanState.parquetScanState,
+                groupsToRead, vfs);
             return;
         }
     } else {
         // Row group already assigned (e.g., by external morsel system or re-initialization)
-        groupsToRead.push_back(scanState.nodeGroupIdx);
+        groupsToRead.push_back(iceDiskScanState.nodeGroupIdx);
     }
 
     // Re-initialize scan for the specific row groups
     // Note: initializeScan can be called multiple times; the first call populates column metadata
-    scanState.parquetReader->initializeScan(*scanState.parquetScanState, groupsToRead, vfs);
+    iceDiskScanState.parquetReader->initializeScan(*iceDiskScanState.parquetScanState, groupsToRead,
+        vfs);
 }
 
-bool ParquetNodeTable::scanInternal(Transaction* transaction, TableScanState& scanState) {
-    auto& parquetScanState = static_cast<ParquetNodeTableScanState&>(scanState);
+bool IceDiskNodeTable::scanInternal(Transaction* transaction, TableScanState& scanState) {
+    auto& iceDiskScanState = static_cast<IceDiskNodeTableScanState&>(scanState);
 
     // Check if this particular scan state has already completed
-    if (parquetScanState.scanCompleted) {
+    if (iceDiskScanState.scanCompleted) {
         return false;
     }
 
     scanState.resetOutVectors();
 
     // Read all data once into scan state
-    if (!parquetScanState.dataRead) {
+    if (!iceDiskScanState.dataRead) {
         // Only the first thread reads the parquet data
-        if (!parquetScanState.initialized) {
+        if (!iceDiskScanState.initialized) {
             return false;
         }
 
         // Create a data chunk for reading parquet data
-        auto numColumns = parquetScanState.parquetReader->getNumColumns();
+        auto numColumns = iceDiskScanState.parquetReader->getNumColumns();
 
         // Defensive check: ensure parquet file has at least one column
         if (numColumns == 0) {
@@ -184,7 +185,7 @@ bool ParquetNodeTable::scanInternal(Transaction* transaction, TableScanState& sc
         // Always create the data chunk to match the exact number of parquet columns
         // to prevent crashes in the parquet reader when accessing result vectors
         for (uint32_t i = 0; i < numColumns; ++i) {
-            const auto& parquetColumnType = parquetScanState.parquetReader->getColumnType(i);
+            const auto& parquetColumnType = iceDiskScanState.parquetReader->getColumnType(i);
             auto columnType = parquetColumnType.copy();
             auto vector = std::make_shared<ValueVector>(std::move(columnType),
                 MemoryManager::Get(*transaction->getClientContext()), scanState.outState);
@@ -192,13 +193,13 @@ bool ParquetNodeTable::scanInternal(Transaction* transaction, TableScanState& sc
         }
 
         // Read from parquet
-        parquetScanState.parquetReader->scan(*parquetScanState.parquetScanState, parquetDataChunk);
+        iceDiskScanState.parquetReader->scan(*iceDiskScanState.parquetScanState, parquetDataChunk);
 
         auto selSize = parquetDataChunk.state->getSelVector().getSelSize();
         if (selSize > 0) {
-            parquetScanState.allData.resize(selSize);
+            iceDiskScanState.allData.resize(selSize);
             for (size_t row = 0; row < selSize; ++row) {
-                parquetScanState.allData[row].resize(
+                iceDiskScanState.allData[row].resize(
                     scanState.outputVectors
                         .size()); // Use output vector count, not parquet column count
 
@@ -217,7 +218,7 @@ bool ParquetNodeTable::scanInternal(Transaction* transaction, TableScanState& sc
 
                     // Get parquet column name and find its corresponding column ID
                     std::string parquetColumnName =
-                        parquetScanState.parquetReader->getColumnName(parquetCol);
+                        iceDiskScanState.parquetReader->getColumnName(parquetCol);
                     auto nodeTableEntry = this->nodeTableCatalogEntry;
 
                     // Check if the column exists first before calling getColumnID
@@ -240,44 +241,44 @@ bool ParquetNodeTable::scanInternal(Transaction* transaction, TableScanState& sc
 
                     // Only copy data if we found a matching output position
                     if (outputCol != INVALID_COLUMN_ID &&
-                        outputCol < parquetScanState.allData[row].size()) {
+                        outputCol < iceDiskScanState.allData[row].size()) {
                         // Defensive check: ensure the row index is valid for the source vector
                         if (row >= srcVector.state->getSelVector().getSelSize()) {
                             continue;
                         }
 
                         if (srcVector.isNull(row)) {
-                            parquetScanState.allData[row][outputCol] =
+                            iceDiskScanState.allData[row][outputCol] =
                                 std::make_unique<Value>(Value::createNullValue());
                         } else {
-                            parquetScanState.allData[row][outputCol] =
+                            iceDiskScanState.allData[row][outputCol] =
                                 std::make_unique<Value>(*srcVector.getAsValue(row));
                         }
                     }
                 }
             }
-            parquetScanState.totalRows = selSize;
+            iceDiskScanState.totalRows = selSize;
         }
-        parquetScanState.dataRead = true;
+        iceDiskScanState.dataRead = true;
     }
 
     // Now distribute one row to this scan state
-    if (parquetScanState.nextRowToDistribute >= parquetScanState.totalRows) {
-        parquetScanState.scanCompleted = true;
+    if (iceDiskScanState.nextRowToDistribute >= iceDiskScanState.totalRows) {
+        iceDiskScanState.scanCompleted = true;
         return false; // No more rows to distribute
     }
 
-    size_t rowIndex = parquetScanState.nextRowToDistribute++;
+    size_t rowIndex = iceDiskScanState.nextRowToDistribute++;
 
     // Copy one row to output vectors
     // Defensive checks: ensure valid row index and handle empty data gracefully
-    if (rowIndex >= parquetScanState.allData.size()) {
-        parquetScanState.scanCompleted = true;
+    if (rowIndex >= iceDiskScanState.allData.size()) {
+        iceDiskScanState.scanCompleted = true;
         return false;
     }
 
     auto numColumns =
-        std::min(scanState.outputVectors.size(), parquetScanState.allData[rowIndex].size());
+        std::min(scanState.outputVectors.size(), iceDiskScanState.allData[rowIndex].size());
     for (size_t col = 0; col < numColumns; ++col) {
         // Defensive check: ensure output vector exists
         if (col >= scanState.outputVectors.size() || !scanState.outputVectors[col]) {
@@ -287,13 +288,13 @@ bool ParquetNodeTable::scanInternal(Transaction* transaction, TableScanState& sc
         auto& dstVector = *scanState.outputVectors[col];
 
         // Defensive check: ensure value exists for this column
-        if (col >= parquetScanState.allData[rowIndex].size() ||
-            !parquetScanState.allData[rowIndex][col]) {
+        if (col >= iceDiskScanState.allData[rowIndex].size() ||
+            !iceDiskScanState.allData[rowIndex][col]) {
             dstVector.setNull(0, true);
             continue;
         }
 
-        auto& value = *parquetScanState.allData[rowIndex][col];
+        auto& value = *iceDiskScanState.allData[rowIndex][col];
 
         if (value.isNull()) {
             dstVector.setNull(0, true);
@@ -312,7 +313,7 @@ bool ParquetNodeTable::scanInternal(Transaction* transaction, TableScanState& sc
     return true;
 }
 
-row_idx_t ParquetNodeTable::getTotalRowCount(const Transaction* transaction) const {
+row_idx_t IceDiskNodeTable::getTotalRowCount(const Transaction* transaction) const {
     const auto cached = cachedRowCount.load(std::memory_order_relaxed);
     if (cached != INVALID_ROW_IDX) {
         return cached;
@@ -326,8 +327,7 @@ row_idx_t ParquetNodeTable::getTotalRowCount(const Transaction* transaction) con
     std::vector<bool> columnSkips;
 
     try {
-        auto resolvedPath = VirtualFileSystem::resolvePath(context, parquetFilePath);
-        auto tempReader = std::make_unique<ParquetReader>(resolvedPath, columnSkips, context);
+        auto tempReader = std::make_unique<ParquetReader>(parquetFilePath, columnSkips, context);
         if (!tempReader) {
             return 0;
         }
@@ -341,12 +341,12 @@ row_idx_t ParquetNodeTable::getTotalRowCount(const Transaction* transaction) con
     }
 }
 
-bool ParquetNodeTable::isVisible(const transaction::Transaction* transaction,
+bool IceDiskNodeTable::isVisible(const transaction::Transaction* transaction,
     common::offset_t offset) const {
     return offset < getTotalRowCount(transaction);
 }
 
-bool ParquetNodeTable::isVisibleNoLock(const transaction::Transaction* transaction,
+bool IceDiskNodeTable::isVisibleNoLock(const transaction::Transaction* transaction,
     common::offset_t offset) const {
     return offset < getTotalRowCount(transaction);
 }
diff --git a/src/storage/table/parquet_rel_table.cpp b/src/storage/table/ice_disk_rel_table.cpp
similarity index 71%
rename from src/storage/table/parquet_rel_table.cpp
rename to src/storage/table/ice_disk_rel_table.cpp
index 7237716d54..a87c68eac7 100644
--- a/src/storage/table/parquet_rel_table.cpp
+++ b/src/storage/table/ice_disk_rel_table.cpp
@@ -1,4 +1,6 @@
-#include "storage/table/parquet_rel_table.h"
+#include "storage/table/ice_disk_rel_table.h"
+
+#include <filesystem>
 
 #include "catalog/catalog_entry/rel_group_catalog_entry.h"
 #include "common/data_chunk/sel_vector.h"
@@ -7,6 +9,7 @@
 #include "main/client_context.h"
 #include "processor/operator/persistent/reader/parquet/parquet_reader.h"
 #include "storage/storage_manager.h"
+#include "storage/table/ice_disk_utils.h"
 #include "transaction/transaction.h"
 
 using namespace lbug::catalog;
@@ -17,7 +20,7 @@ using namespace lbug::transaction;
 namespace lbug {
 namespace storage {
 
-void ParquetRelTableScanState::setToTable(const Transaction* transaction, Table* table_,
+void IceDiskRelTableScanState::setToTable(const Transaction* transaction, Table* table_,
     std::vector<column_id_t> columnIDs_, std::vector<ColumnPredicateSet> columnPredicateSets_,
     RelDataDirection direction_) {
     // Call base class implementation but skip local table setup
@@ -36,48 +39,48 @@ void ParquetRelTableScanState::setToTable(const Transaction* transaction, Table*
     csrOffsetColumn = table->cast<RelTable>().getCSROffsetColumn(direction);
     csrLengthColumn = table->cast<RelTable>().getCSRLengthColumn(direction);
     nodeGroupIdx = INVALID_NODE_GROUP_IDX;
-    // ParquetRelTable does not support local storage, so we skip the local table initialization
+    // IceDiskRelTable does not support local storage, so we skip the local table initialization
 }
 
-ParquetRelTable::ParquetRelTable(RelGroupCatalogEntry* relGroupEntry, table_id_t fromTableID,
-    table_id_t toTableID, const StorageManager* storageManager, MemoryManager* memoryManager)
+IceDiskRelTable::IceDiskRelTable(RelGroupCatalogEntry* relGroupEntry, table_id_t fromTableID,
+    table_id_t toTableID, const StorageManager* storageManager, MemoryManager* memoryManager,
+    main::ClientContext* context)
     : ColumnarRelTableBase{relGroupEntry, fromTableID, toTableID, storageManager, memoryManager} {
-    std::string storage = relGroupEntry->getStorage();
-    if (storage.empty()) {
-        throw RuntimeException("Parquet file path is empty for parquet-backed rel table");
-    }
+    auto paths = IceDiskUtils::constructCSRPaths(relGroupEntry->getStorage(),
+        relGroupEntry->getName(), ".parquet");
+
+    auto resolvedIndicesPath = VirtualFileSystem::resolvePath(context, paths.indices);
+    IceDiskUtils::checkVersionCompatibility(context, resolvedIndicesPath);
 
-    // Use base class helper to construct CSR file paths
-    auto paths = constructCSRPaths(storage, ".parquet");
-    indicesFilePath = paths.indices;
-    indptrFilePath = paths.indptr;
+    auto resolvedIndptrPath = VirtualFileSystem::resolvePath(context, paths.indptr);
+    IceDiskUtils::checkVersionCompatibility(context, resolvedIndptrPath);
+    indicesFilePath = resolvedIndicesPath;
+    indptrFilePath = resolvedIndptrPath;
 }
 
-void ParquetRelTable::initScanState(Transaction* transaction, TableScanState& scanState,
+void IceDiskRelTable::initScanState(Transaction* transaction, TableScanState& scanState,
     bool resetCachedBoundNodeSelVec) const {
-    // For parquet tables, we create our own scan state
+    // For  tables, we create our own scan state
     auto& relScanState = scanState.cast<RelTableScanState>();
     relScanState.source = TableScanSource::COMMITTED;
     relScanState.nodeGroup = nullptr;
     relScanState.nodeGroupIdx = INVALID_NODE_GROUP_IDX;
 
     // Initialize ParquetReaders for this scan state (per-thread)
-    auto& parquetRelScanState = static_cast<ParquetRelTableScanState&>(relScanState);
+    auto& iceDiskScanState = static_cast<IceDiskRelTableScanState&>(relScanState);
 
     // Initialize readers if not already done for this scan state
-    if (!parquetRelScanState.indicesReader) {
+    if (!iceDiskScanState.indicesReader) {
         std::vector<bool> columnSkips; // Read all columns
         auto context = transaction->getClientContext();
-        auto resolvedPath = VirtualFileSystem::resolvePath(context, indicesFilePath);
-        parquetRelScanState.indicesReader =
-            std::make_unique<ParquetReader>(resolvedPath, columnSkips, context);
+        iceDiskScanState.indicesReader =
+            std::make_unique<ParquetReader>(indicesFilePath, columnSkips, context);
     }
-    if (!indptrFilePath.empty() && !parquetRelScanState.indptrReader) {
+    if (!indptrFilePath.empty() && !iceDiskScanState.indptrReader) {
         std::vector<bool> columnSkips; // Read all columns
         auto context = transaction->getClientContext();
-        auto resolvedPath = VirtualFileSystem::resolvePath(context, indptrFilePath);
-        parquetRelScanState.indptrReader =
-            std::make_unique<ParquetReader>(resolvedPath, columnSkips, context);
+        iceDiskScanState.indptrReader =
+            std::make_unique<ParquetReader>(indptrFilePath, columnSkips, context);
     }
 
     // Load shared indptr data - thread-safe to read
@@ -103,37 +106,34 @@ void ParquetRelTable::initScanState(Transaction* transaction, TableScanState& sc
 
     // 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)
-    parquetRelScanState.currentRowGroup = 0;
-    parquetRelScanState.endRowGroup = parquetRelScanState.indicesReader ?
-                                          parquetRelScanState.indicesReader->getNumRowGroups() :
-                                          0;
+    iceDiskScanState.currentRowGroup = 0;
+    iceDiskScanState.endRowGroup =
+        iceDiskScanState.indicesReader ? iceDiskScanState.indicesReader->getNumRowGroups() : 0;
 }
 
-void ParquetRelTable::initializeParquetReaders(Transaction* transaction) const {
+void IceDiskRelTable::initializeParquetReaders(Transaction* transaction) const {
     if (!indicesReader) {
         std::lock_guard lock(parquetReaderMutex);
         if (!indicesReader) {
             std::vector<bool> columnSkips; // Read all columns
             auto context = transaction->getClientContext();
-            auto resolvedPath = VirtualFileSystem::resolvePath(context, indicesFilePath);
-            indicesReader = std::make_unique<ParquetReader>(resolvedPath, columnSkips, context);
+            indicesReader = std::make_unique<ParquetReader>(indicesFilePath, columnSkips, context);
         }
     }
 }
 
-void ParquetRelTable::initializeIndptrReader(Transaction* transaction) const {
+void IceDiskRelTable::initializeIndptrReader(Transaction* transaction) const {
     if (!indptrFilePath.empty() && !indptrReader) {
         std::lock_guard lock(parquetReaderMutex);
         if (!indptrReader) {
             std::vector<bool> columnSkips; // Read all columns
             auto context = transaction->getClientContext();
-            auto resolvedPath = VirtualFileSystem::resolvePath(context, indptrFilePath);
-            indptrReader = std::make_unique<ParquetReader>(resolvedPath, columnSkips, context);
+            indptrReader = std::make_unique<ParquetReader>(indptrFilePath, columnSkips, context);
         }
     }
 }
 
-void ParquetRelTable::loadIndptrData(Transaction* transaction) const {
+void IceDiskRelTable::loadIndptrData(Transaction* transaction) const {
     if (indptrData.empty() && !indptrFilePath.empty()) {
         std::lock_guard lock(indptrDataMutex);
         if (indptrData.empty()) {
@@ -186,11 +186,11 @@ void ParquetRelTable::loadIndptrData(Transaction* transaction) const {
     }
 }
 
-bool ParquetRelTable::scanInternal(Transaction* transaction, TableScanState& scanState) {
+bool IceDiskRelTable::scanInternal(Transaction* transaction, TableScanState& scanState) {
     auto& relScanState = scanState.cast<RelTableScanState>();
 
-    // Get the ParquetRelTableScanState
-    auto& parquetRelScanState = static_cast<ParquetRelTableScanState&>(relScanState);
+    // Get the IceDiskRelTableScanState
+    auto& iceDiskScanState = static_cast<IceDiskRelTableScanState&>(relScanState);
 
     // Load shared indptr data - thread-safe to read
     if (!indptrFilePath.empty()) {
@@ -199,49 +199,49 @@ bool ParquetRelTable::scanInternal(Transaction* transaction, TableScanState& sca
 
     // 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, parquetRelScanState);
+    return scanInternalByRowGroups(transaction, iceDiskScanState);
 }
 
-bool ParquetRelTable::scanInternalByRowGroups(Transaction* transaction,
-    ParquetRelTableScanState& parquetRelScanState) {
+bool IceDiskRelTable::scanInternalByRowGroups(Transaction* transaction,
+    IceDiskRelTableScanState& iceDiskScanState) {
     // True morsel-driven parallelism: process assigned row groups and collect relationships for
     // bound nodes
 
     // Check if we have any row groups left to process
-    if (parquetRelScanState.currentRowGroup >= parquetRelScanState.endRowGroup) {
+    if (iceDiskScanState.currentRowGroup >= iceDiskScanState.endRowGroup) {
         // No more row groups to process
-        parquetRelScanState.outState->getSelVectorUnsafe().setToFiltered(0);
+        iceDiskScanState.outState->getSelVectorUnsafe().setToFiltered(0);
         return false;
     }
 
     // Process the current row group
-    std::vector<uint64_t> rowGroupsToProcess = {parquetRelScanState.currentRowGroup};
+    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 < parquetRelScanState.cachedBoundNodeSelVector.getSelSize(); ++i) {
-        common::sel_t boundNodeIdx = parquetRelScanState.cachedBoundNodeSelVector[i];
-        const auto boundNodeID = parquetRelScanState.nodeIDVector->getValue<nodeID_t>(boundNodeIdx);
+    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, parquetRelScanState, rowGroupsToProcess,
+    bool hasData = scanRowGroupForBoundNodes(transaction, iceDiskScanState, rowGroupsToProcess,
         boundNodeOffsets);
 
     // Move to next row group for next call
-    parquetRelScanState.currentRowGroup++;
+    iceDiskScanState.currentRowGroup++;
 
     return hasData;
 }
 
-common::offset_t ParquetRelTable::findSourceNodeForRow(common::offset_t globalRowIdx) const {
+common::offset_t IceDiskRelTable::findSourceNodeForRow(common::offset_t globalRowIdx) const {
     // Use base class helper for binary search
     return findSourceNodeForRowInternal(globalRowIdx, indptrData);
 }
 
-bool ParquetRelTable::scanRowGroupForBoundNodes(Transaction* transaction,
-    ParquetRelTableScanState& parquetRelScanState, const std::vector<uint64_t>& rowGroupsToProcess,
+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
@@ -250,24 +250,24 @@ bool ParquetRelTable::scanRowGroupForBoundNodes(Transaction* transaction,
     // Initialize scan state for the assigned row groups
     auto context = transaction->getClientContext();
     auto vfs = VirtualFileSystem::GetUnsafe(*context);
-    parquetRelScanState.indicesReader->initializeScan(*parquetRelScanState.parquetScanState,
+    iceDiskScanState.indicesReader->initializeScan(*iceDiskScanState.parquetScanState,
         rowGroupsToProcess, vfs);
 
     // Create DataChunk matching the indices parquet file schema
-    auto numIndicesColumns = parquetRelScanState.indicesReader->getNumColumns();
+    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 = parquetRelScanState.indicesReader->getColumnType(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);
     }
 
     // Scan the row groups and collect relationships for bound nodes.
-    const auto isFwd = parquetRelScanState.direction != RelDataDirection::BWD;
+    const auto isFwd = iceDiskScanState.direction != RelDataDirection::BWD;
     uint64_t totalRowsCollected = 0;
     const uint64_t maxRowsPerCall = DEFAULT_VECTOR_CAPACITY;
     uint64_t currentGlobalRowIdx = 0;
@@ -277,14 +277,14 @@ bool ParquetRelTable::scanRowGroupForBoundNodes(Transaction* transaction,
 
     // Calculate the starting global row index for the first row group
     if (!rowGroupsToProcess.empty()) {
-        auto metadata = parquetRelScanState.indicesReader->getMetadata();
+        auto metadata = iceDiskScanState.indicesReader->getMetadata();
         for (uint64_t rgIdx = 0; rgIdx < rowGroupsToProcess[0]; ++rgIdx) {
             currentGlobalRowIdx += metadata->row_groups[rgIdx].num_rows;
         }
     }
 
     while (totalRowsCollected < maxRowsPerCall &&
-           parquetRelScanState.indicesReader->scanInternal(*parquetRelScanState.parquetScanState,
+           iceDiskScanState.indicesReader->scanInternal(*iceDiskScanState.parquetScanState,
                indicesChunk)) {
 
         auto selSize = indicesChunk.state->getSelVector().getSelSize();
@@ -319,31 +319,31 @@ bool ParquetRelTable::scanRowGroupForBoundNodes(Transaction* transaction,
             auto nbrNodeID = internalID_t(nbrOffset, nbrTableID);
 
             // outputVectors[0] is the neighbor node ID, if requested.
-            if (!parquetRelScanState.outputVectors.empty()) {
-                parquetRelScanState.outputVectors[0]->setValue(totalRowsCollected, nbrNodeID);
+            if (!iceDiskScanState.outputVectors.empty()) {
+                iceDiskScanState.outputVectors[0]->setValue(totalRowsCollected, nbrNodeID);
             }
 
             // Copy edge properties to output vectors.
             // Catalog col IDs: NBR_ID=0, REL_ID=1 (virtual), user props=2,3,...
             // Parquet cols:     target=0,              user props=1,2,...
             // So parquet_col = catalog_col_id - 1 for user properties.
-            for (uint64_t outCol = 1; outCol < parquetRelScanState.outputVectors.size(); ++outCol) {
-                if (outCol >= parquetRelScanState.columnIDs.size()) {
+            for (uint64_t outCol = 1; outCol < iceDiskScanState.outputVectors.size(); ++outCol) {
+                if (outCol >= iceDiskScanState.columnIDs.size()) {
                     continue;
                 }
-                const auto colID = parquetRelScanState.columnIDs[outCol];
+                const auto colID = iceDiskScanState.columnIDs[outCol];
                 if (colID == INVALID_COLUMN_ID || colID == ROW_IDX_COLUMN_ID ||
                     colID == NBR_ID_COLUMN_ID) {
                     continue;
                 }
                 if (colID == REL_ID_COLUMN_ID) {
                     // REL_ID is not stored in parquet; synthesize from the global row index.
-                    parquetRelScanState.outputVectors[outCol]->setValue<internalID_t>(
+                    iceDiskScanState.outputVectors[outCol]->setValue<internalID_t>(
                         totalRowsCollected, internalID_t{currentGlobalRowIdx, getTableID()});
                     continue;
                 }
 
-                parquetRelScanState.outputVectors[outCol]->copyFromVectorData(totalRowsCollected,
+                iceDiskScanState.outputVectors[outCol]->copyFromVectorData(totalRowsCollected,
                     &indicesChunk.getValueVector(colID - 1), i);
             }
 
@@ -353,22 +353,22 @@ bool ParquetRelTable::scanRowGroupForBoundNodes(Transaction* transaction,
 
     // Set up the output state
     if (totalRowsCollected > 0) {
-        auto& selVector = parquetRelScanState.outState->getSelVectorUnsafe();
+        auto& selVector = iceDiskScanState.outState->getSelVectorUnsafe();
         selVector.setToFiltered(totalRowsCollected);
         for (uint64_t i = 0; i < totalRowsCollected; ++i) {
             selVector[i] = i;
         }
-        parquetRelScanState.setNodeIDVectorToFlat(activeBoundSelPos);
+        iceDiskScanState.setNodeIDVectorToFlat(activeBoundSelPos);
 
         return true;
     } else {
         // No data found
-        parquetRelScanState.outState->getSelVectorUnsafe().setToFiltered(0);
+        iceDiskScanState.outState->getSelVectorUnsafe().setToFiltered(0);
         return false;
     }
 }
 
-row_idx_t ParquetRelTable::getTotalRowCount(const Transaction* transaction) const {
+row_idx_t IceDiskRelTable::getTotalRowCount(const Transaction* transaction) const {
     initializeParquetReaders(const_cast<Transaction*>(transaction));
     if (!indicesReader) {
         return 0;
diff --git a/src/storage/wal/records/alter_table_entry_record_replay.cpp b/src/storage/wal/records/alter_table_entry_record_replay.cpp
index f1c33c942e..ff90e2a294 100644
--- a/src/storage/wal/records/alter_table_entry_record_replay.cpp
+++ b/src/storage/wal/records/alter_table_entry_record_replay.cpp
@@ -61,7 +61,7 @@ void WALReplayer::replayAlterTableEntryRecord(const WALRecord& walRecord) const
                                  ->ptrCast<RelGroupCatalogEntry>();
         auto relEntryInfo =
             relGroupEntry->getRelEntryInfo(extraInfo->fromTableID, extraInfo->toTableID);
-        storageManager->addRelTable(relGroupEntry, *relEntryInfo);
+        storageManager->addRelTable(relGroupEntry, *relEntryInfo, &clientContext);
     } break;
     default:
         break;
diff --git a/src/storage/wal/records/create_catalog_entry_record_replay.cpp b/src/storage/wal/records/create_catalog_entry_record_replay.cpp
index 2a9f002d14..b46e0b6245 100644
--- a/src/storage/wal/records/create_catalog_entry_record_replay.cpp
+++ b/src/storage/wal/records/create_catalog_entry_record_replay.cpp
@@ -24,7 +24,7 @@ void WALReplayer::replayCreateCatalogEntryRecord(WALRecord& walRecord) const {
         auto& entry = record.ownedCatalogEntry->constCast<TableCatalogEntry>();
         auto newEntry = catalog->createTableEntry(transaction,
             entry.getBoundCreateTableInfo(transaction, record.isInternal));
-        storageManager->createTable(newEntry->ptrCast<TableCatalogEntry>());
+        storageManager->createTable(newEntry->ptrCast<TableCatalogEntry>(), &clientContext);
     } break;
     case CatalogEntryType::SCALAR_MACRO_ENTRY: {
         auto& macroEntry = record.ownedCatalogEntry->constCast<ScalarMacroCatalogEntry>();
diff --git a/test/include/test_runner/test_group.h b/test/include/test_runner/test_group.h
index b0ba4663cd..d0263a7a48 100644
--- a/test/include/test_runner/test_group.h
+++ b/test/include/test_runner/test_group.h
@@ -116,7 +116,7 @@ struct TestGroup {
         LBUG,
         JSON,
         CSV_TO_JSON,
-        GRAPH_STD
+        ICEBUG_DISK
     };
     DatasetType datasetType;
 
diff --git a/test/runner/e2e_test.cpp b/test/runner/e2e_test.cpp
index 5c78c5d9fc..2a26f9bf5d 100644
--- a/test/runner/e2e_test.cpp
+++ b/test/runner/e2e_test.cpp
@@ -39,8 +39,8 @@ class EndToEndTest final : public DBTest {
         }
         createDB(checkpointWaitTimeout);
         createConns(connNames);
-        if (datasetType == TestGroup::DatasetType::GRAPH_STD) {
-            // For GRAPH_STD, only run schema.cypher (which contains WITH storage = ... clauses)
+        if (datasetType == TestGroup::DatasetType::ICEBUG_DISK) {
+            // For ICEBUG_DISK, only run schema.cypher (which contains WITH storage = ... clauses)
             // No copy.cypher needed as data is in external parquet files
             lbug::main::Connection* connection =
                 conn ? conn.get() : (connMap.begin()->second).get();
diff --git a/test/storage/CMakeLists.txt b/test/storage/CMakeLists.txt
index e63f929d90..83efdeabba 100644
--- a/test/storage/CMakeLists.txt
+++ b/test/storage/CMakeLists.txt
@@ -7,5 +7,6 @@ add_lbug_test(rel_tests rel_scan_test.cpp rel_delete_test.cpp)
 add_lbug_test(node_update_test node_update_test.cpp)
 add_lbug_test(detach_delete_test detach_delete_test.cpp)
 add_lbug_test(storage_utils_test storage_utils_test.cpp)
+add_lbug_test(ice_disk_utils_test ice_disk_utils_test.cpp)
 
 target_include_directories(compression_test PRIVATE ${PROJECT_SOURCE_DIR}/third_party/alp/include)
diff --git a/test/storage/ice_disk_utils_test.cpp b/test/storage/ice_disk_utils_test.cpp
new file mode 100644
index 0000000000..83693bd6d3
--- /dev/null
+++ b/test/storage/ice_disk_utils_test.cpp
@@ -0,0 +1,155 @@
+#include <filesystem>
+#include <string>
+
+#include "common/exception/copy.h"
+#include "common/exception/io.h"
+#include "common/exception/runtime.h"
+#include "common/file_system/virtual_file_system.h"
+#include "graph_test/private_graph_test.h"
+#include "gtest/gtest.h"
+#include "main/client_context.h"
+#include "main/connection.h"
+#include "main/database.h"
+#include "storage/table/ice_disk_utils.h"
+#include "test_helper/test_helper.h"
+
+using namespace lbug::common;
+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");
+
+// ─────────────────────────────────────────────────────────────
+// joinPath
+// ─────────────────────────────────────────────────────────────
+TEST(IceDiskUtils_JoinPath, EmptyBase) {
+    EXPECT_EQ("file.parquet", IceDiskUtils::joinPath("", "file.parquet"));
+}
+
+TEST(IceDiskUtils_JoinPath, BaseWithoutTrailingSlash) {
+    EXPECT_EQ("/base/file.parquet", IceDiskUtils::joinPath("/base", "file.parquet"));
+}
+
+TEST(IceDiskUtils_JoinPath, BaseWithTrailingSlash) {
+    EXPECT_EQ("/base/file.parquet", IceDiskUtils::joinPath("/base/", "file.parquet"));
+}
+
+TEST(IceDiskUtils_JoinPath, BaseWithBackslash) {
+    EXPECT_EQ("base\\file.parquet", IceDiskUtils::joinPath("base\\", "file.parquet"));
+}
+
+TEST(IceDiskUtils_JoinPath, S3URI) {
+    EXPECT_EQ("s3://bucket/prefix/file.parquet",
+        IceDiskUtils::joinPath("s3://bucket/prefix", "file.parquet"));
+}
+
+TEST(IceDiskUtils_JoinPath, HttpsURI) {
+    EXPECT_EQ("https://host/path/file.parquet",
+        IceDiskUtils::joinPath("https://host/path", "file.parquet"));
+}
+
+// ─────────────────────────────────────────────────────────────
+// constructNodeTablePath
+// ─────────────────────────────────────────────────────────────
+TEST(IceDiskUtils_ConstructNodeTablePath, EmptyDir) {
+    EXPECT_EQ("nodes_city.parquet", IceDiskUtils::constructNodeTablePath("", "city", ".parquet"));
+}
+
+TEST(IceDiskUtils_ConstructNodeTablePath, WithDir) {
+    EXPECT_EQ("/some/dir/nodes_user.parquet",
+        IceDiskUtils::constructNodeTablePath("/some/dir", "user", ".parquet"));
+}
+
+TEST(IceDiskUtils_ConstructNodeTablePath, S3URI) {
+    EXPECT_EQ("s3://bucket/data/nodes_user.parquet",
+        IceDiskUtils::constructNodeTablePath("s3://bucket/data", "user", ".parquet"));
+}
+
+// ─────────────────────────────────────────────────────────────
+// constructCSRPaths
+// ─────────────────────────────────────────────────────────────
+TEST(IceDiskUtils_ConstructCSRPaths, EmptyDir) {
+    auto paths = IceDiskUtils::constructCSRPaths("", "follows", ".parquet");
+    EXPECT_EQ("indices_follows.parquet", paths.indices);
+    EXPECT_EQ("indptr_follows.parquet", paths.indptr);
+}
+
+TEST(IceDiskUtils_ConstructCSRPaths, WithDir) {
+    auto paths = IceDiskUtils::constructCSRPaths("/some/dir", "knows", ".parquet");
+    EXPECT_EQ("/some/dir/indices_knows.parquet", paths.indices);
+    EXPECT_EQ("/some/dir/indptr_knows.parquet", paths.indptr);
+}
+
+TEST(IceDiskUtils_ConstructCSRPaths, S3URI) {
+    auto paths = IceDiskUtils::constructCSRPaths("s3://bucket/data", "follows", ".parquet");
+    EXPECT_EQ("s3://bucket/data/indices_follows.parquet", paths.indices);
+    EXPECT_EQ("s3://bucket/data/indptr_follows.parquet", paths.indptr);
+}
+
+// ─────────────────────────────────────────────────────────────
+// checkVersionCompatibility
+// ─────────────────────────────────────────────────────────────
+class IceDiskCheckVersionTest : public EmptyDBTest {
+protected:
+    void SetUp() override {
+        EmptyDBTest::SetUp();
+        createDBAndConn();
+        context = conn->getClientContext();
+    }
+
+    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"),
+        RuntimeException);
+}
+
+TEST_F(IceDiskCheckVersionTest, FileDoesNotExist) {
+    EXPECT_THROW(IceDiskUtils::checkVersionCompatibility(context,
+                     FIXTURES_DIR + "/nodes_nonexistent.parquet"),
+        IOException);
+}
+
+TEST_F(IceDiskCheckVersionTest, NotAParquetFile) {
+    EXPECT_THROW(IceDiskUtils::checkVersionCompatibility(context,
+                     FIXTURES_DIR + "/nodes_notparquet.parquet"),
+        CopyException);
+}
+
+TEST_F(IceDiskCheckVersionTest, MissingVersionKey) {
+    try {
+        IceDiskUtils::checkVersionCompatibility(context, FIXTURES_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);
+    }
+}
+
+TEST_F(IceDiskCheckVersionTest, WrongVersionValue) {
+    try {
+        IceDiskUtils::checkVersionCompatibility(context,
+            FIXTURES_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") !=
+                    std::string::npos);
+    }
+}
+
+TEST_F(IceDiskCheckVersionTest, UppercaseVersionSucceeds) {
+    // "V1" should match "v1" case-insensitively
+    EXPECT_NO_THROW(IceDiskUtils::checkVersionCompatibility(context,
+        FIXTURES_DIR + "/nodes_upperversion.parquet"));
+}
+
+TEST_F(IceDiskCheckVersionTest, ValidV1Succeeds) {
+    EXPECT_NO_THROW(IceDiskUtils::checkVersionCompatibility(context,
+        DEMO_DB_ICEBUG_DISK + "/nodes_user.parquet"));
+}
diff --git a/test/test_files/demo_db/demo_db_graph_std.test b/test/test_files/demo_db/demo_db_icebug_disk.test
similarity index 72%
rename from test/test_files/demo_db/demo_db_graph_std.test
rename to test/test_files/demo_db/demo_db_icebug_disk.test
index d22cca98ab..18462de648 100644
--- a/test/test_files/demo_db/demo_db_graph_std.test
+++ b/test/test_files/demo_db/demo_db_icebug_disk.test
@@ -1,8 +1,48 @@
--DATASET GRAPH-STD demo-db/icebug-disk
+-DATASET ICEBUG-DISK demo-db/icebug-disk
 
 --
 
--CASE DemoDBGraphStdTest
+-CASE DemoDBIcebugDiskTest
+
+-LOG RenameUserNodeTableFails
+-STATEMENT ALTER TABLE user RENAME TO user2
+---- error
+Binder exception: Cannot alter table user: icebug-disk tables are immutable.
+
+-LOG RenameFollowsRelTableFails
+-STATEMENT ALTER TABLE follows RENAME TO follows2
+---- error
+Binder exception: Cannot alter table follows: icebug-disk tables are immutable.
+
+-LOG InsertIntoUserNodeTableFails
+-STATEMENT CREATE (u:user {id: 350, name: 'Alice', age: 35});
+---- error(regex)
+^Runtime exception: Cannot insert.*
+
+-LOG InsertIntoFollowsRelTableFails
+-STATEMENT MATCH (u1:User), (u2:User) WHERE u1.name = 'Adam' AND u2.name = 'Noura' CREATE (u1)-[:Follows {since: 2011}]->(u2);
+---- error(regex)
+^Runtime exception: Cannot insert.*
+
+-LOG UpdateUserNodeFails
+-STATEMENT MATCH (u:user {name: 'Adam'}) SET u.age = 31
+---- error(regex)
+^Runtime exception: Cannot update.*
+
+-LOG UpdateFollowsRelFails
+-STATEMENT MATCH (u1:User)-[f:Follows]->(u2:User) WHERE u1.name = 'Adam' AND u2.name = 'Karissa' SET f.since = 2011
+---- error(regex)
+^Runtime exception: Cannot update.*
+
+-LOG DeleteUserNodeFails
+-STATEMENT MATCH (u:user {name: 'Adam'}) DELETE u
+---- error(regex)
+^Runtime exception: Cannot delete.*
+
+-LOG DeleteFollowsRelFails
+-STATEMENT MATCH (u1:User)-[f:Follows]->(u2:User) WHERE u1.name = 'Adam' AND u2.name = 'Karissa' DELETE f
+---- error(regex)
+^Runtime exception: Cannot delete.*
 
 -LOG MatchUserLivesInCity
 -STATEMENT MATCH (u:user)-[l:livesin]->(c:city) RETURN u.name, u.age, c.name;
diff --git a/test/test_files/ice_disk/ice_disk_invalid_storage.test b/test/test_files/ice_disk/ice_disk_invalid_storage.test
new file mode 100644
index 0000000000..255a290d19
--- /dev/null
+++ b/test/test_files/ice_disk/ice_disk_invalid_storage.test
@@ -0,0 +1,32 @@
+-DATASET CSV empty
+--
+
+-CASE IceDiskNonExistentDirNodeFails
+
+-LOG IceDiskNonExistentDirMissingParquet
+-STATEMENT CREATE NODE TABLE t(id INT64 PRIMARY KEY) WITH (storage = '/nonexistent/dir/that/does/not/exist', format = 'icebug-disk')
+---- error(regex)
+.*[Nn]o such file.*|.*[Cc]annot open.*|.*nodes_t\.parquet.*
+
+-CASE IceDiskRemoteURINodeFails
+
+-LOG IceDiskRemoteURIMissingParquet
+-STATEMENT CREATE NODE TABLE t(id INT64 PRIMARY KEY) WITH (storage = 'https://example.com/path', format = 'icebug-disk')
+---- error(regex)
+.*[Nn]o such file.*|.*[Cc]annot open.*|.*nodes_t\.parquet.*
+
+-CASE IceDiskS3URINodeFails
+
+-LOG IceDiskS3URIMissingParquet
+-STATEMENT CREATE NODE TABLE t(id INT64 PRIMARY KEY) WITH (storage = 's3://my-bucket/data', format = 'icebug-disk')
+---- error(regex)
+.*[Nn]o such file.*|.*[Cc]annot open.*|.*nodes_t\.parquet.*
+
+-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')
+---- ok
+-STATEMENT CREATE REL TABLE rel(FROM upperversion TO upperversion) WITH (format = 'icebug-disk')
+---- error(regex)
+.*[Nn]o such file.*|.*[Cc]annot open.*|.*indices_rel\.parquet.*
diff --git a/test/test_files/ice_disk/ice_disk_mix_tables.test b/test/test_files/ice_disk/ice_disk_mix_tables.test
new file mode 100644
index 0000000000..dd88abbc63
--- /dev/null
+++ b/test/test_files/ice_disk/ice_disk_mix_tables.test
@@ -0,0 +1,33 @@
+-DATASET CSV empty
+--
+
+-CASE RelIceDiskNodeRegular
+
+-LOG CreateRegularNodeTable
+-STATEMENT CREATE NODE TABLE person(id INT64 PRIMARY KEY, name STRING)
+---- 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')
+---- 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')
+---- ok
+
+-LOG CreateRegularNodeTable
+-STATEMENT CREATE NODE TABLE person(id INT64 PRIMARY KEY, name STRING)
+---- ok
+
+-LOG RegularRelWithIceDiskNodeFails
+-STATEMENT CREATE REL TABLE rel(FROM person TO upperversion)
+---- error
+Binder exception: Cannot mix icebug-disk tables with non-icebug-disk tables in CREATE REL TABLE.
+
+-LOG RegularRelWithBothIceDiskNodeFails
+-STATEMENT CREATE REL TABLE rel(FROM upperversion TO upperversion)
+---- error
+Binder exception: Cannot mix icebug-disk tables with non-icebug-disk tables in CREATE REL TABLE.
diff --git a/test/test_helper/test_helper.cpp b/test/test_helper/test_helper.cpp
index 6155b1faed..8d0e92c751 100644
--- a/test/test_helper/test_helper.cpp
+++ b/test/test_helper/test_helper.cpp
@@ -45,6 +45,8 @@ void TestHelper::executeScript(const std::string& cypherScript, Connection& conn
     }
     std::string line;
     while (getline(file, line)) {
+        // replace single quote with double
+        std::replace(line.begin(), line.end(), '\'', '"');
         // If this is a COPY statement, we need to append the LBUG_ROOT_DIRECTORY to the csv
         // file path. There maybe multiple csv files in the line, so we need to find all of them.
         std::vector<std::string> csvFilePaths;
@@ -86,37 +88,52 @@ void TestHelper::executeScript(const std::string& cypherScript, Connection& conn
             fullPath = normalizePathForCypher(std::move(fullPath));
             line.replace(line.find(csvFilePath), csvFilePath.length(), fullPath);
         }
-        // Also handle storage = 'path' for parquet tables
-        std::vector<std::string> storagePaths;
-        size_t storageIndex = 0;
-        while (true) {
-            size_t start = line.find("storage = '", storageIndex);
-            if (start == std::string::npos) {
-                break;
-            }
-            start += 11; // length of "storage = '"
-            size_t end = line.find("'", start);
-            if (end == std::string::npos) {
-                break;
-            }
-            std::string storagePath = line.substr(start, end - start);
-            storagePaths.push_back(storagePath);
-            storageIndex = end + 1;
-        }
-        for (auto& storagePath : storagePaths) {
-            auto fullPath = storagePath;
-            if (std::filesystem::path(storagePath).is_relative()) {
-                if (std::filesystem::path(storagePath).parent_path().empty()) {
-                    fullPath = (cypherDir / storagePath).string();
-                } else {
-                    fullPath = appendLbugRootPath(storagePath);
-                }
-            }
-            fullPath = normalizePathForCypher(std::move(fullPath));
 
-            size_t pos = line.find(storagePath);
-            if (pos != std::string::npos) {
-                line.replace(pos, storagePath.length(), fullPath);
+        // handle icebug-disk storage format: resolve storage = "<path>" for icebug-disk tables
+        {
+            size_t fmtStart = line.find("format = \"", 0);
+            if (fmtStart != std::string::npos) {
+                fmtStart += 10; // length of "format = \""
+                size_t fmtEnd = line.find("\"", fmtStart);
+                if (fmtEnd != std::string::npos) {
+                    std::string storageFormat = line.substr(fmtStart, fmtEnd - fmtStart);
+                    if (storageFormat.find("icebug-disk") != std::string::npos) {
+                        std::vector<std::string> storagePaths;
+                        size_t storageIndex = 0;
+                        while (true) {
+                            size_t sStart = line.find("storage = \"", storageIndex);
+                            if (sStart == std::string::npos) {
+                                break;
+                            }
+                            sStart += 11; // length of "storage = \""
+                            size_t sEnd = line.find("\"", sStart);
+                            if (sEnd == std::string::npos) {
+                                break;
+                            }
+                            storagePaths.push_back(line.substr(sStart, sEnd - sStart));
+                            storageIndex = sEnd + 1;
+                        }
+                        for (auto& storagePath : storagePaths) {
+                            // Remote URIs (e.g. s3://, https://) are passed through unchanged.
+                            if (storagePath.find("://") != std::string::npos) {
+                                continue;
+                            }
+                            auto fullPath = storagePath;
+                            if (std::filesystem::path(storagePath).is_relative()) {
+                                if (std::filesystem::path(storagePath).parent_path().empty()) {
+                                    fullPath = (cypherDir / storagePath).string();
+                                } else {
+                                    fullPath = appendLbugRootPath(storagePath);
+                                }
+                            }
+                            fullPath = normalizePathForCypher(std::move(fullPath));
+                            size_t pos = line.find(storagePath);
+                            if (pos != std::string::npos) {
+                                line.replace(pos, storagePath.length(), fullPath);
+                            }
+                        }
+                    }
+                }
             }
         }
 #ifdef __STATIC_LINK_EXTENSION_TEST__
diff --git a/test/test_runner/test_parser.cpp b/test/test_runner/test_parser.cpp
index dea0ef047d..3d4f4cb082 100644
--- a/test/test_runner/test_parser.cpp
+++ b/test/test_runner/test_parser.cpp
@@ -88,8 +88,8 @@ void TestParser::extractDataset() {
             testGroup->datasetType = TestGroup::DatasetType::JSON;
             testGroup->dataset = currentToken.params[2];
         }
-    } else if (datasetType == "GRAPH-STD") {
-        testGroup->datasetType = TestGroup::DatasetType::GRAPH_STD;
+    } else if (datasetType == "ICEBUG-DISK") {
+        testGroup->datasetType = TestGroup::DatasetType::ICEBUG_DISK;
         testGroup->dataset = currentToken.params[2];
     } else {
         throw TestException(