diff --git a/partition-routing-hook.md b/partition-routing-hook.md new file mode 100644 index 000000000..f54cc2bae --- /dev/null +++ b/partition-routing-hook.md @@ -0,0 +1,126 @@ +# Design: opaque partition-routing hook for remote partition subgraphs + +## Goal + +Partitioned node tables store their rows in partition subgraphs (`_p` node +tables). Today every subgraph is local. In a distributed deployment a partition may live +on another host. We want ladybugdb to stay **embedded and distribution-agnostic**: the +core never learns about hosts, sockets, or serialization. Instead it exposes one opaque +interface that a *distributed wrapper* installs at startup; every place the engine would +touch a partition subgraph consults the interface first, and falls back to local storage +when the interface is absent (the default, and the behavior of every existing test). + +## Inventory: every embedded "subgraph call" that needs a seam + +| # | Seam | Location | What happens today | +|---|------|----------|--------------------| +| 1 | Partition lifecycle (create/drop/rename subgraphs, subgraph registration) | `src/catalog/catalog.cpp:179-243, 615-669` | Parent DDL creates/drops/renames `_p` node-table subgraphs | +| 2 | Local storage creation for partitions | `src/storage/storage_manager.cpp:271` | A `NodeTable` is created per partition subgraph | +| 3 | Read binding: expand parent → partitions | `expandPartitionedNodeTables`, `src/binder/bind/bind_graph_pattern.cpp:779-860` | Pattern on a partitioned parent is rewritten into a multi-table scan over all child entries | +| 4 | Rel endpoint binding: parent → n×m FROM/TO pairs | `resolveRelEndpoints`, `src/binder/bind/bind_ddl.cpp:236-255` | Rel tables attach to each partition subgraph | +| 5 | Point-write routing (INSERT/SET/MERGE) | `NodeInsertExecutor::resolveTargetTable` / `resolveTableForNodeID`, `src/processor/operator/persistent/insert_executor.cpp:95-116` | `computePartitionIndexes` picks the child table for the evaluated key | +| 6 | Bulk-write routing (INSERT ... FROM / COPY FROM) | `NodeBatchInsert` targets + `computePartitionIndexes`, `src/processor/operator/persistent/node_batch_insert.cpp:~500-680`; binder carries `NodePartitionWriteInfo` (`bind_copy_from.cpp:182-193`) | Rows are hash/range-routed into per-partition targets | + +Everything else (planner, processor, WAL, GDS `graph::Graph`) is already subgraph-blind +or consumes the same bound entries, so these six seams are the complete surface. + +## The interface + +One plain struct of plain function pointers plus an opaque context handle — no virtual +inheritance. `nullptr` members mean "handle locally", so the wrapper only overrides what +it owns. See `src/include/common/partition_routing_hook.h` for the authoritative +definition; summary: + +| Hook | Seam | Contract | +|------|------|----------| +| `locate(ctx, ref, &handle)` | placement | Called before any touch of partition `ref`. true + handle = wrapper owns it (remote); false = local. Must be consistent; answers are cached. | +| `onPartitionCreate(ctx, ref, handle)` | lifecycle | Fires for **every** partition creation — this is how a wrapper learns about new partitions and decides placement. | +| `onPartitionDrop(ctx, ref, handle)` | lifecycle | Fires for claimed partitions when their subgraph entry is dropped. Renames are not reported (`PartitionRef` is ID-based and IDs survive renames). | +| `bindScan(ctx, ref, handle, &spec)` | reads (bind time) | Wrapper fills a `PartitionScanSpec`: its table function + a bind-data factory keyed by the node's unique expression name. The engine attaches the spec to an internal clone of the partition's catalog entry (preserving schema, table ID, lineage). Bind columns must be named `.` / `._ID` (same convention as extension foreign tables). | +| `insertRow(ctx, ref, handle, tx, keyVec, colVecs)` | point writes | Single already-evaluated row; wrapper ships it and returns the remotely-assigned nodeID. | +| `insertChunk(ctx, ref, handle, tx, keyVec, colVecs, startRow, numRows)` | bulk writes | Run of rows; row j lives at selection position `selVector[startRow + j]` (same convention as `InMemChunkedNodeGroup::append`). | +| `lookupRow(ctx, ref, handle, tx, nodeID, outVecs)` | MERGE lookups | Fetch an existing remote row into output vectors. | + +Registration is process-global (`setPartitionRoutingHooks`), must happen before the first +Database is opened, and the hooks object must outlive the registration. `nullptr` hooks +(default) preserve embedded behavior bit-for-bit. + +### Registration and lifetime + +- `main::Database` gains `setPartitionRouting(const PartitionRoutingHooks*)`, callable + only between construction and the first query / recovery start. The pointer is then + immutable and copied into each `ClientContext` (read path) and handed to `Catalog` / + `StorageManager` (DDL path) — no locks on the hot path. +- Default is `nullptr` everywhere: an un-hooked build behaves bit-for-bit as today. +- The wrapper registers its own scan/insert table functions with the normal function + registry before opening the database, so `bindScan` only needs to name them. + +## How each seam changes + +1. **Catalog lifecycle** — `createNodeTableSubgraph` / drop / rename paths call + `onPartition*` after (or instead of, when `locate` claims the partition) the local + catalog mutation. The catalog metadata (child table IDs, partition method, key + column) is *always* recorded locally: it is the distributed system's source of truth + for placement, and it survives restarts so `locate` can be re-consulted. +2. **Storage manager** — when creating tables for partition children + (`storage_manager.cpp:271`), skip local storage for partitions claimed by `locate`. + No local files, no WAL records, no checkpoint work for them. (Checkpoint/replay must + consult `locate` before assuming a child table has local state — the one place the + recovery path needs the hook.) +3. **Read binding** — in `expandPartitionedNodeTables`, a claimed partition contributes + the wrapper's scan function entry instead of the local child entry; the existing + multi-table union scan handles the mix of local and remote partitions unchanged. +4. **Rel endpoints** — `resolveRelEndpoints` does the same substitution. Note: for a + rel table between two remote-partitioned parents this expands to n×m pairs; if that + becomes a problem, the optional escape hatch is a second hook that lets the wrapper + bind the whole rel table at once, but start without it. +5. **Point writes** — `resolveTargetTable()` first computes the partition index (core + logic, unchanged), then consults `locate`; claimed partitions go through + `insertRow` and the returned nodeID flows into the existing output-vector path. +6. **Bulk writes** — `NodeBatchInsert` keeps `computePartitionIndexes` as-is, then + partitions each key chunk's selection into local targets (existing code path) and + remote targets (one `insertChunk` call per claimed partition per chunk). + +## Invariants the core keeps (why this stays "no distribution knowledge") + +- The engine owns the **partition function** (hash/range over the key column). Placement + is therefore computable anywhere without RPC; the wrapper only owns *where* a + partition lives, never *which* partition a row belongs to. +- The engine owns the **catalog** (parent/child IDs, schemas). Remote partitions are + first-class catalog entries with local metadata. +- All transport concerns — hosts, connections, serialization, retries, pushing down + predicates — live behind `PartitionHandle` in the wrapper. Ladybug passes the handle + back verbatim and never inspects it. +- Every hook is optional and every callback site falls through to the current local + code path when unclaimed. + +## Alternatives considered + +- **`RemoteNodeTable : storage::NodeTable`** — most transparent (no binder/executor + changes), but drags WAL, checkpoint, versioning, and scan-state internals into the + public seam, coupling the wrapper to ladybug's storage ABI release-to-release. +- **Subclass `graph::Graph`** — covers only the GDS neighbor-scan interface, not + INSERT/COPY/DDL, which are the majority of subgraph touch points. +- **Distributed planning in the core** — the thing we explicitly do not want. + +## Implementation notes / deltas from the first sketch + +- `bindScan` hands over a `PartitionScanSpec` (function + bind-data factory) instead of a + catalog entry: the engine clones the partition's own catalog entry and stamps the + wrapper's scan onto it, so schema/table-ID/lineage stay consistent and write paths can + resolve the parent from catalog truth even when pattern entries are substituted. +- `NodeTableCatalogEntry::CreateBindDataFunc` now receives `nodeUniqueName` so + foreign-backed entries can name their output columns the way the planner expects + (mirrors what the duckdb/postgres extensions do inside their own `getBoundScanInfo`). +- `NodePartitionWriteInfo` carries the parent table ID so executors can build + `PartitionRef`s without re-deriving lineage. +- Mixed local/remote scans of one parent are rejected at bind time (the multi-entry + `ScanNodeTable` union cannot host scan-function-backed entries); fully-claimed parents + collapse to a single substitute entry and use the existing table-function scan path. +- Checkpoint, metadata-snapshot serialization, rollback, and storage creation all skip + claimed partitions (no local table/WAL/checkpoint state exists for them). +- Not wired in this first landing (documented limitations): UPDATE/DELETE on remote rows, + rel tables referencing remote-partitioned parents, GDS algorithms over remote + partitions, and direct writes to individual remote partition subgraphs by name. + +## Suggested landing order diff --git a/src/binder/bind/bind_graph_pattern.cpp b/src/binder/bind/bind_graph_pattern.cpp index ef8addab0..3d6f2d637 100644 --- a/src/binder/bind/bind_graph_pattern.cpp +++ b/src/binder/bind/bind_graph_pattern.cpp @@ -1,3 +1,6 @@ +#include +#include + #include "binder/binder.h" #include "binder/expression/expression_util.h" #include "binder/expression/path_expression.h" @@ -9,6 +12,7 @@ #include "common/constants.h" #include "common/enums/rel_direction.h" #include "common/exception/binder.h" +#include "common/partition_routing_hook.h" #include "common/types/types.h" #include "common/utils.h" #include "function/cast/functions/cast_from_string_functions.h" @@ -775,12 +779,67 @@ static std::vector sortEntries(const table_catalog_entry_set return entries; } +namespace { + +// Scan-substitute entries (local child clones carrying a wrapper scan function) must outlive +// the bound statement that references them. Cache them per PartitionRef so repeated binds of +// the same remote partition reuse one stable entry. Note: a wrapper that swaps its scan +// function for an already-bound partition within one process lifetime will keep serving the +// function captured at first bind. +std::mutex& scanEntryMutex() { + static std::mutex mtx; + return mtx; +} + +struct ScanEntryKey { + const void* database; + const void* scanFunction; + bool operator==(const ScanEntryKey& other) const { + return database == other.database && scanFunction == other.scanFunction; + } +}; + +struct ScanEntryKeyHasher { + uint64_t operator()(const ScanEntryKey& key) const { + return std::hash{}(key.database) * 31 + + std::hash{}(key.scanFunction); + } +}; + +std::unordered_map, ScanEntryKeyHasher>& +scanEntryCache() { + static std::unordered_map, ScanEntryKeyHasher> + cache; + return cache; +} + +// Entries are cached per (database, wrapper scan function) so repeated binds reuse one stable +// substitute without leaking state across databases. +TableCatalogEntry* retainPartitionedScanEntry(std::unique_ptr entry, + main::ClientContext* clientContext, const void* scanFunctionIdentity) { + ScanEntryKey key{clientContext->getDatabase(), scanFunctionIdentity}; + std::lock_guard lck{scanEntryMutex()}; + auto [it, inserted] = scanEntryCache().emplace(key, std::move(entry)); + return it->second.get(); +} + +} // namespace + // A partitioned parent node table owns no physical storage; its records live across its // partition subgraphs. When a node label resolves to a partitioned parent we expand it into the // child partition tables so the (existing) multi-table node scan unions over every partition. +// +// If a routing wrapper (see common/partition_routing_hook.h) claims a partition via `locate`, +// the local child owns no storage and cannot be scanned; instead the wrapper supplies a scan +// function through `bindScan`, which the engine attaches to an internal clone of the child's +// catalog entry (keeping schema, table ID, and partition lineage). Scanning a parent that +// mixes claimed and unclaimed partitions cannot be planned, so it is rejected at bind time. static table_catalog_entry_set_t expandPartitionedNodeTables(catalog::Catalog* catalog, - const transaction::Transaction* transaction, const table_catalog_entry_set_t& entrySet) { + const transaction::Transaction* transaction, const table_catalog_entry_set_t& entrySet, + main::ClientContext* clientContext) { table_catalog_entry_set_t expanded; + bool anyClaimed = false; + std::string claimedParentName; for (auto entry : entrySet) { if (entry->getType() != CatalogEntryType::NODE_TABLE_ENTRY) { expanded.insert(entry); @@ -791,11 +850,63 @@ static table_catalog_entry_set_t expandPartitionedNodeTables(catalog::Catalog* c expanded.insert(entry); continue; } + const auto* hooks = common::getPartitionRoutingHooks(); for (auto childID : nodeEntry->getChildTableIDs()) { auto* child = catalog->getTableCatalogEntry(transaction, childID); - expanded.insert(child); + const auto ref = common::PartitionRef{nodeEntry->getTableID(), + child->ptrCast()->getPartitionIndex()}; + common::PartitionHandle handle = nullptr; + if (hooks == nullptr || hooks->locate == nullptr || + !hooks->locate(hooks->context, ref, &handle)) { + expanded.insert(child); + continue; + } + anyClaimed = true; + claimedParentName = nodeEntry->getName(); + if (hooks->bindScan == nullptr) { + throw BinderException( + std::format("Partition index {} of table {} is routed remotely, but the " + "partition routing hooks do not provide bindScan.", + ref.partitionIndex, nodeEntry->getName())); + } + common::PartitionScanSpec spec; + if (!hooks->bindScan(hooks->context, ref, handle, &spec)) { + throw BinderException(std::format("Partition routing hooks did not provide a " + "scan for partition index {} of table {}.", + ref.partitionIndex, nodeEntry->getName())); + } + if (spec.scanFunction == nullptr || spec.createBindData == nullptr) { + throw BinderException(std::format("Partition routing hooks provided an invalid " + "scan for partition index {} of table {}.", + ref.partitionIndex, nodeEntry->getName())); + } + // Attach the wrapper's scan to a clone of the child entry so the substitute keeps + // the parent's schema, table ID, and partition lineage. + auto patched = child->copy(); + auto* patchedNode = patched->ptrCast(); + patchedNode->setScanFunction(*spec.scanFunction); + patchedNode->setCreateBindDataFunc( + [createBindData = std::move(spec.createBindData)](main::ClientContext*, + const std::string& nodeUniqueName) { return createBindData(nodeUniqueName); }); + if (patchedNode->getBoundScanInfo(clientContext, "") == nullptr) { + throw BinderException(std::format( + "The scan function provided by the partition routing hooks for partition " + "index {} of table {} did not produce a valid scan.", + ref.partitionIndex, nodeEntry->getName())); + } + // Partitions routed to the same wrapper scan share one substitute entry, so a + // fully-claimed parent collapses to a single entry in the set below. + expanded.insert( + retainPartitionedScanEntry(std::move(patched), clientContext, spec.scanFunction)); } } + if (anyClaimed && expanded.size() > 1) { + throw BinderException(std::format( + "Table {}: scanning a mix of locally stored and remotely routed partitions is not " + "supported. A routing wrapper must claim either all or none of a scanned parent's " + "partitions and expose them as one consolidated scan entry.", + claimedParentName)); + } return expanded; } @@ -851,7 +962,7 @@ Binder::bindNodeTableEntries(const std::vector& tableNames) const { } } // Expand partitioned parents into their partition subgraphs for scanning. - entrySet = expandPartitionedNodeTables(catalog, transaction, entrySet); + entrySet = expandPartitionedNodeTables(catalog, transaction, entrySet, clientContext); return {sortEntries(entrySet), std::move(dbNames)}; } diff --git a/src/binder/bind/copy/bind_copy_from.cpp b/src/binder/bind/copy/bind_copy_from.cpp index cbbe699c8..a2715cabc 100644 --- a/src/binder/bind/copy/bind_copy_from.cpp +++ b/src/binder/bind/copy/bind_copy_from.cpp @@ -186,8 +186,8 @@ std::unique_ptr Binder::bindCopyNodeFrom(const Statement& statem if (nodeTableEntry.isPartitioned()) { partitionWriteInfo = NodePartitionWriteInfo{ static_cast(*nodeTableEntry.getPartitionMethod()), - nodeTableEntry.getPartitionColumnID(), nodeTableEntry.getNumPartitions(), - nodeTableEntry.getChildTableIDs()}; + nodeTableEntry.getTableID(), nodeTableEntry.getPartitionColumnID(), + nodeTableEntry.getNumPartitions(), nodeTableEntry.getChildTableIDs()}; } // Check extension secondary index loaded auto catalog = Catalog::Get(*clientContext); diff --git a/src/catalog/catalog.cpp b/src/catalog/catalog.cpp index 0a50a2bd9..0c4cdd87e 100644 --- a/src/catalog/catalog.cpp +++ b/src/catalog/catalog.cpp @@ -12,6 +12,7 @@ #include "catalog/catalog_entry/type_catalog_entry.h" #include "common/exception/catalog.h" #include "common/exception/runtime.h" +#include "common/partition_routing_hook.h" #include "common/serializer/deserializer.h" #include "common/serializer/serializer.h" #include "extension/extension_manager.h" @@ -29,6 +30,42 @@ using namespace lbug::transaction; namespace lbug { namespace catalog { +namespace { + +// Notify the routing wrapper (if any) that a partition subgraph entry now exists +// or is about to be dropped, so it can provision/decommission the remote copy. +// Creation notifications fire for every partition regardless of placement - this is +// how a wrapper learns about newly provisioned partitions and decides where they +// live; later locate() calls are answered from the wrapper's own records. +void notifyPartitionCreated(common::table_id_t parentTableID, uint64_t partitionIndex) { + const auto* hooks = common::getPartitionRoutingHooks(); + if (hooks == nullptr || hooks->onPartitionCreate == nullptr) { + return; + } + common::PartitionHandle handle = nullptr; + if (hooks->locate != nullptr) { + hooks->locate(hooks->context, common::PartitionRef{parentTableID, partitionIndex}, &handle); + } + hooks->onPartitionCreate(hooks->context, common::PartitionRef{parentTableID, partitionIndex}, + handle); +} + +void notifyPartitionDropped(common::table_id_t parentTableID, uint64_t partitionIndex) { + const auto* hooks = common::getPartitionRoutingHooks(); + if (hooks == nullptr || hooks->onPartitionDrop == nullptr) { + return; + } + common::PartitionHandle handle = nullptr; + if (hooks->locate != nullptr && + hooks->locate(hooks->context, common::PartitionRef{parentTableID, partitionIndex}, + &handle)) { + hooks->onPartitionDrop(hooks->context, common::PartitionRef{parentTableID, partitionIndex}, + handle); + } +} + +} // namespace + Catalog::Catalog() : version{0} { initCatalogSets(); registerBuiltInFunctions(); @@ -185,6 +222,8 @@ void Catalog::dropTableEntry(Transaction* transaction, const TableCatalogEntry* auto* child = getTableCatalogEntry(transaction, childID); dropAllIndexes(transaction, childID); dropSerialSequence(transaction, child); + notifyPartitionDropped(nodeEntry->getTableID(), + child->ptrCast()->getPartitionIndex()); if (tables->containsEntry(transaction, child->getName())) { tables->dropEntry(transaction, child->getName(), child->getOID()); } else { @@ -668,6 +707,10 @@ CatalogEntry* Catalog::createNodeTableEntry(Transaction* transaction, // Each partition subgraph is a node table and therefore its own subgraph. createNodeTableSubgraph(transaction, childName); parent->addChildTableID(childOID); + // Let the routing wrapper provision remote storage for this partition. + // Renames are not reported: PartitionRef is ID-based and IDs survive + // renames. + notifyPartitionCreated(parent->getTableID(), i); } } return parentEntry; diff --git a/src/catalog/catalog_entry/node_table_catalog_entry.cpp b/src/catalog/catalog_entry/node_table_catalog_entry.cpp index 09833f983..499b5783d 100644 --- a/src/catalog/catalog_entry/node_table_catalog_entry.cpp +++ b/src/catalog/catalog_entry/node_table_catalog_entry.cpp @@ -170,15 +170,11 @@ std::string NodeTableCatalogEntry::toCypher(const ToCypherInfo& /*info*/) const return base + ";"; } -std::optional NodeTableCatalogEntry::getScanFunction() const { - return scanFunction; -} - std::unique_ptr NodeTableCatalogEntry::getBoundScanInfo( - main::ClientContext* context, [[maybe_unused]] const std::string& nodeUniqueName) { + main::ClientContext* context, const std::string& nodeUniqueName) { if (scanFunction.has_value()) { // Foreign table - call the extension's bind data function - auto bindData = createBindDataFunc(context); + auto bindData = createBindDataFunc(context, nodeUniqueName); return std::make_unique(*scanFunction, std::move(bindData)); } // Check referenced entry (shadow tables: NodeTableCatalogEntry that wraps a foreign entry) diff --git a/src/common/CMakeLists.txt b/src/common/CMakeLists.txt index e9b541656..9d8fcba02 100644 --- a/src/common/CMakeLists.txt +++ b/src/common/CMakeLists.txt @@ -23,6 +23,7 @@ add_library(lbug_common md5.cpp metric.cpp null_mask.cpp + partition_routing_hook.cpp profiler.cpp random_engine.cpp roaring_mask.cpp diff --git a/src/common/partition_routing_hook.cpp b/src/common/partition_routing_hook.cpp new file mode 100644 index 000000000..91fc7d950 --- /dev/null +++ b/src/common/partition_routing_hook.cpp @@ -0,0 +1,29 @@ +#include "common/partition_routing_hook.h" + +#include "common/exception/exception.h" + +namespace lbug { +namespace common { + +namespace { + +const PartitionRoutingHooks* hooks_ = nullptr; + +} // namespace + +void setPartitionRoutingHooks(const PartitionRoutingHooks* hooks) { + if (hooks != nullptr && hooks_ != nullptr) { + // A wrapper may reset to nullptr (e.g. between tests) but two different + // hook sets must never be live at once: routing answers are cached by + // the engine and assumed stable. + throw Exception("Partition routing hooks are already installed."); + } + hooks_ = hooks; +} + +const PartitionRoutingHooks* getPartitionRoutingHooks() { + return hooks_; +} + +} // namespace common +} // namespace lbug 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 61107dede..2e2733484 100644 --- a/src/include/catalog/catalog_entry/node_table_catalog_entry.h +++ b/src/include/catalog/catalog_entry/node_table_catalog_entry.h @@ -25,10 +25,13 @@ struct SortedByProperty { static SortedByProperty deserialize(common::Deserializer& deserializer); }; -// Callback to create bind data for foreign tables -// This allows extensions to provide bind data creation without core needing to know extension types -using CreateBindDataFunc = - std::function(main::ClientContext* context)>; +// Callback to create bind data for foreign tables. +// `nodeUniqueName` is the unique expression name of the node pattern being bound (empty for +// standalone function binds). Implementations must name their output columns +// "." and "._ID" so planner schema lookups on node +// property expressions resolve against the scan output. +using CreateBindDataFunc = std::function( + main::ClientContext* context, const std::string& nodeUniqueName)>; // Tag for shadow table constructor struct ShadowTag {}; @@ -90,8 +93,13 @@ class LBUG_API NodeTableCatalogEntry final : public TableCatalogEntry { void setSortedByProperties(std::vector properties) { sortedByProperties = std::move(properties); } - std::optional getScanFunction() const override; - const CreateBindDataFunc& getCreateBindDataFunc() const { return createBindDataFunc; } + std::optional getScanFunction() const override { return scanFunction; } + void setScanFunction(function::TableFunction scanFunction_) { + scanFunction = std::move(scanFunction_); + } + void setCreateBindDataFunc(CreateBindDataFunc createBindDataFunc_) { + createBindDataFunc = std::move(createBindDataFunc_); + } const std::string& getForeignDatabaseName() const { return foreignDatabaseName; } void setReferencedEntry(TableCatalogEntry* entry) { referencedEntry = entry; } diff --git a/src/include/common/partition_routing.h b/src/include/common/partition_routing.h index 5df5c7399..bfc516a17 100644 --- a/src/include/common/partition_routing.h +++ b/src/include/common/partition_routing.h @@ -19,15 +19,18 @@ enum class PartitionMethod : uint8_t { HASH = 0, RANGE = 1 }; // partition order (index == partition index). struct NodePartitionWriteInfo { PartitionMethod method; + // Owning partitioned parent. Used to address partitions via PartitionRef. + common::table_id_t parentTableID; common::column_id_t partitionKeyColumnID; uint64_t numPartitions; std::vector partitionTableIDs; NodePartitionWriteInfo() = default; - NodePartitionWriteInfo(PartitionMethod method, common::column_id_t partitionKeyColumnID, - uint64_t numPartitions, std::vector partitionTableIDs) - : method{method}, partitionKeyColumnID{partitionKeyColumnID}, numPartitions{numPartitions}, - partitionTableIDs{std::move(partitionTableIDs)} {} + NodePartitionWriteInfo(PartitionMethod method, common::table_id_t parentTableID, + common::column_id_t partitionKeyColumnID, uint64_t numPartitions, + std::vector partitionTableIDs) + : method{method}, parentTableID{parentTableID}, partitionKeyColumnID{partitionKeyColumnID}, + numPartitions{numPartitions}, partitionTableIDs{std::move(partitionTableIDs)} {} }; } // namespace common diff --git a/src/include/common/partition_routing_hook.h b/src/include/common/partition_routing_hook.h new file mode 100644 index 000000000..350a32339 --- /dev/null +++ b/src/include/common/partition_routing_hook.h @@ -0,0 +1,135 @@ +#pragma once + +#include + +#include "common/api.h" +#include "common/types/types.h" +#include "common/vector/value_vector.h" +#include + +namespace lbug { +namespace function { +struct TableFunction; +struct TableFuncBindData; +} // namespace function +namespace transaction { +class Transaction; +} // namespace transaction + +namespace common { + +// --------------------------------------------------------------------------- +// Partition routing hooks +// +// Partitioned node tables keep their rows in partition subgraphs (`_p` +// node tables). In a distributed deployment a partition may live on a remote +// host. These hooks let a distributed wrapper intercept every subgraph access +// and route it remotely, while ladybugdb itself stays embedded and knows +// nothing about hosts, transport, or serialization. +// +// Contract: +// * The engine owns the partition function (hash/range over the key column) +// and the full catalog metadata (parent/child table IDs, schemas) for every +// partition, including remote ones. A wrapper therefore never decides +// *which* partition a row belongs to - only *where* it lives. +// * `PartitionHandle` is opaque: the engine receives it from `locate()` and +// hands it back verbatim afterwards. The wrapper interprets it (e.g. as a +// connection or host descriptor). +// * Every member is optional; NULL means "handle locally", which is the +// default and preserves the embedded behavior bit-for-bit. +// * Hooks must be installed before the first Database is opened (recovery, +// checkpointing and query planning all consult them). They are process +// global and must not change afterwards. +// --------------------------------------------------------------------------- + +// Identifies one partition subgraph: (partitioned parent table ID, partition index). +struct PartitionRef { + table_id_t parentTableID = INVALID_TABLE_ID; + uint64_t partitionIndex = 0; +}; + +using PartitionHandle = void*; + +// Everything the engine needs to read one remotely-routed partition. +// `scanFunction` must point to storage owned by the wrapper that stays valid for the +// process lifetime (a static object is fine). +// `createBindData(nodeUniqueName)` must return bind data whose columns expose one +// INTERNAL-ID column named "._ID" plus one column per parent +// property named ".", in parent schema order - the same +// convention extension-provided foreign tables follow - so planner schema lookups on +// node property expressions resolve against the scan output. +struct PartitionScanSpec { + const function::TableFunction* scanFunction = nullptr; + std::function(const std::string& nodeUniqueName)> + createBindData = nullptr; +}; + +struct PartitionRoutingHooks { + // Wrapper-owned state, passed back to every callback. + void* context = nullptr; + + // --- Placement ------------------------------------------------------------ + // Called before the engine touches partition `ref` for any purpose (storage + // creation, scan binding, write routing, lifecycle notification). + // Return true + set *handleOut -> the wrapper owns this partition (remote). + // Return false -> local storage, exactly as today. + // Must be side-effect free and consistent: the same ref must always yield + // the same answer, and the same handle for claimed refs. + bool (*locate)(void* context, PartitionRef ref, PartitionHandle* handleOut) = nullptr; + + // --- Lifecycle ------------------------------------------------------------ + // Notification that a partition subgraph entry was created / dropped in the + // catalog. Create fires for EVERY partition (this is how a wrapper learns + // about newly provisioned partitions and decides where they live); drop + // fires for partitions the wrapper claims via locate(). Fire-and-forget; + // the engine proceeds regardless of the outcome. Renames are not reported + // because PartitionRef is ID-based and IDs survive renames. + void (*onPartitionCreate)(void* context, PartitionRef ref, PartitionHandle handle) = nullptr; + void (*onPartitionDrop)(void* context, PartitionRef ref, PartitionHandle handle) = nullptr; + + // --- Reads ---------------------------------------------------------------- + // Called at bind time for each claimed partition of a scanned parent. The + // wrapper fills *specOut with its own scan function plus a bind-data factory. + // The engine attaches them to an internal clone of the partition's catalog + // entry (keeping schema, table ID, and partition lineage). Mixed local/remote + // scans of one parent are rejected at bind time. + bool (*bindScan)(void* context, PartitionRef ref, PartitionHandle handle, + PartitionScanSpec* specOut) = nullptr; + + // --- Point writes (INSERT / MERGE-create) ---------------------------------- + // Called instead of a local NodeTable::insert for a claimed partition. The + // key and column vectors are already evaluated and hold exactly one row. + // The wrapper ships the row and returns the remotely-assigned node ID + // (offset + child table ID), which flows into the output vector as usual. + nodeID_t (*insertRow)(void* context, PartitionRef ref, PartitionHandle handle, + transaction::Transaction* tx, const ValueVector* keyVector, + std::span columnVectors) = nullptr; + + // --- Bulk writes (COPY FROM / INSERT ... SELECT) --------------------------- + // Same contract as insertRow but for a run of numRows consecutive logical rows starting at + // startRow. Row j of the run lives at selection position + // `vec->state->getSelVector()[startRow + j]` of every column vector - the same convention + // InMemChunkedNodeGroup::append uses. PK/uniqueness validation is the wrapper's + // responsibility for claimed partitions. + void (*insertChunk)(void* context, PartitionRef ref, PartitionHandle handle, + transaction::Transaction* tx, const ValueVector* keyVector, + std::span columnVectors, uint64_t startRow, uint64_t numRows) = nullptr; + + // --- Lookups (MERGE-match output materialization) --------------------------- + // Fetch an existing remote row by node ID into the output vectors. Return + // false if the row does not exist. If unset, lookups against claimed + // partitions fail with "not supported". + bool (*lookupRow)(void* context, PartitionRef ref, PartitionHandle handle, + transaction::Transaction* tx, nodeID_t nodeID, + std::span outputVectors) = nullptr; +}; + +// Process-global registration. Call before opening any Database; installing +// hooks twice without resetting to nullptr first is an error. The registry keeps +// the raw pointer, so the hooks object must outlive the registration (typically +// static or owned by the wrapper for the process lifetime). +LBUG_API void setPartitionRoutingHooks(const PartitionRoutingHooks* hooks); +LBUG_API const PartitionRoutingHooks* getPartitionRoutingHooks(); + +} // namespace common +} // namespace lbug diff --git a/src/include/processor/operator/persistent/insert_executor.h b/src/include/processor/operator/persistent/insert_executor.h index bfc8c2c8b..f90a0ef86 100644 --- a/src/include/processor/operator/persistent/insert_executor.h +++ b/src/include/processor/operator/persistent/insert_executor.h @@ -1,6 +1,7 @@ #pragma once #include "common/enums/conflict_action.h" +#include "common/partition_routing_hook.h" #include "expression_evaluator/expression_evaluator.h" #include "processor/execution_context.h" #include "storage/table/node_table.h" @@ -48,7 +49,15 @@ struct NodeTableInsertInfo { // derive the PK vector position) and `partitionTables` holds every partition subgraph in // partition order. `partitionKeyColumnID` indexes the partition-key column among // columnDataVectors. Empty `partitionTables` means a plain single-table write. + // + // Partition subgraphs whose storage is routed remotely (see + // common/partition_routing_hook.h) carry a null table pointer; their entries in + // `partitionChildIDs` / `partitionRefs` / `partitionHandles` remain valid and describe + // how to route rows through the hooks instead. std::vector partitionTables; + std::vector partitionChildIDs; + std::vector partitionRefs; + std::vector partitionHandles; common::column_id_t partitionKeyColumnID = common::INVALID_COLUMN_ID; NodeTableInsertInfo(storage::NodeTable* table, @@ -61,7 +70,9 @@ struct NodeTableInsertInfo { private: NodeTableInsertInfo(const NodeTableInsertInfo& other) : table{other.table}, columnDataEvaluators{copyVector(other.columnDataEvaluators)}, - pkVector{nullptr}, partitionTables{other.partitionTables}, + pkVector{nullptr}, columnDataVectors{other.columnDataVectors}, columnIDs{other.columnIDs}, + partitionTables{other.partitionTables}, partitionChildIDs{other.partitionChildIDs}, + partitionRefs{other.partitionRefs}, partitionHandles{other.partitionHandles}, partitionKeyColumnID{other.partitionKeyColumnID} {} }; @@ -87,8 +98,14 @@ class NodeInsertExecutor { bool checkConflict(const transaction::Transaction* transaction, storage::NodeTable* table) const; + // Computes the partition subgraph index for the current (already evaluated) partition-key + // value. + uint64_t currentPartitionIndex() const; // Resolves the partition subgraph for the current (already evaluated) partition-key value. + // Returns nullptr when that partition is routed remotely. storage::NodeTable* resolveTargetTable() const; + // Ships the current row to the routing wrapper for remote partition `index`. + common::nodeID_t insertRemotely(uint64_t index, transaction::Transaction* transaction) const; storage::NodeTable* resolveTableForNodeID(common::nodeID_t nodeID) const; private: diff --git a/src/include/processor/operator/persistent/node_batch_insert.h b/src/include/processor/operator/persistent/node_batch_insert.h index f182801a8..ab4011342 100644 --- a/src/include/processor/operator/persistent/node_batch_insert.h +++ b/src/include/processor/operator/persistent/node_batch_insert.h @@ -2,6 +2,7 @@ #include "common/enums/column_evaluate_type.h" #include "common/partition_routing.h" +#include "common/partition_routing_hook.h" #include "common/types/types.h" #include "expression_evaluator/expression_evaluator.h" #include "processor/operator/persistent/batch_insert.h" @@ -96,7 +97,13 @@ struct NodeBatchInsertSharedState final : BatchInsertSharedState { std::vector mainDataColumns; // One write target per partition subgraph (or exactly one for a non-partitioned table). + // Targets whose storage is routed remotely (see common/partition_routing_hook.h) carry a + // null table pointer; their aligned entries below describe how to route rows through the + // hooks instead. std::vector targets; + // Aligned with `targets`; meaningful only when the write goes into a partitioned parent. + std::vector partitionRefs; + std::vector partitionHandles; // Index of the partition-key column among the evaluated column vectors; INVALID when the // write is not into a partitioned parent. common::column_id_t partitionKeyColumnIdx = common::INVALID_COLUMN_ID; diff --git a/src/planner/join_order/cardinality_estimator.cpp b/src/planner/join_order/cardinality_estimator.cpp index a2ddb96ab..de1d8f4cb 100644 --- a/src/planner/join_order/cardinality_estimator.cpp +++ b/src/planner/join_order/cardinality_estimator.cpp @@ -2,6 +2,7 @@ #include "binder/expression/property_expression.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/table_catalog_entry.h" #include "common/enums/extend_direction_util.h" @@ -60,8 +61,11 @@ void CardinalityEstimator::init(const NodeExpression& node) { auto key = node.getInternalID()->getUniqueName(); cardinality_t numNodes = 0u; for (auto entry : node.getEntries()) { - // Skip foreign tables - they don't have storage in the local database - if (entry->getType() == catalog::CatalogEntryType::FOREIGN_TABLE_ENTRY) { + // Skip foreign tables and other scan-function-backed entries (e.g. remotely routed + // partition substitutes) - they don't have storage in the local database. + if (entry->getType() == catalog::CatalogEntryType::FOREIGN_TABLE_ENTRY || + (entry->getType() == catalog::CatalogEntryType::NODE_TABLE_ENTRY && + entry->ptrCast()->getScanFunction().has_value())) { continue; } auto tableID = entry->getTableID(); diff --git a/src/planner/plan/plan_join_order.cpp b/src/planner/plan/plan_join_order.cpp index 8b5176285..1c2029922 100644 --- a/src/planner/plan/plan_join_order.cpp +++ b/src/planner/plan/plan_join_order.cpp @@ -3,6 +3,7 @@ #include "binder/bound_scan_source.h" #include "binder/expression_visitor.h" #include "catalog/catalog_entry/catalog_entry_type.h" +#include "catalog/catalog_entry/node_table_catalog_entry.h" #include "common/enums/join_type.h" #include "common/enums/rel_direction.h" #include "common/enums/table_type.h" @@ -309,6 +310,17 @@ void Planner::planNodeScan(uint32_t nodePos) { node.get()); } } else { + // Defensive: entries that provide their own scan function (e.g. remotely routed + // partition substitutes installed by the binder) can only be planned as the single + // entry of a pattern. The binder rejects mixes earlier; this guards against paths + // that assemble entry sets without going through that expansion. + for (auto* entry : node->getEntries()) { + if (entry->getType() == catalog::CatalogEntryType::NODE_TABLE_ENTRY && + entry->ptrCast()->getScanFunction().has_value()) { + throw RuntimeException( + "Cannot scan a mix of local tables and scan-function-backed entries."); + } + } appendScanNodeTable(node->getInternalID(), node->getTableIDs(), properties, plan, node.get()); } diff --git a/src/processor/map/map_insert.cpp b/src/processor/map/map_insert.cpp index 47ec4d070..87c99afdd 100644 --- a/src/processor/map/map_insert.cpp +++ b/src/processor/map/map_insert.cpp @@ -1,6 +1,7 @@ #include "binder/expression/rel_expression.h" #include "catalog/catalog.h" #include "catalog/catalog_entry/node_table_catalog_entry.h" +#include "common/partition_routing_hook.h" #include "main/client_context.h" #include "planner/operator/persistent/logical_insert.h" #include "processor/expression_mapper.h" @@ -39,30 +40,61 @@ NodeInsertExecutor PlanMapper::getNodeInsertExecutor(const LogicalInsertInfo* bo auto columnsPos = populateReturnColumnsPos(*boundInfo, outSchema); auto info = NodeInsertInfo(nodeIDPos, columnsPos, boundInfo->conflictAction); auto storageManager = StorageManager::Get(*clientContext); - auto table = storageManager->getTable(node.getEntry(0)->getTableID())->ptrCast(); evaluator_vector_t evaluators; auto exprMapper = ExpressionMapper(&inSchema); for (auto& expr : boundInfo->columnDataExprs) { evaluators.push_back(exprMapper.getEvaluator(expr)); } - auto tableInfo = NodeTableInsertInfo(table, std::move(evaluators)); - // A partitioned parent is resolved into its partition subgraphs during binding. Route the - // row into the partition matching its partition-key value at insert time. - if (node.getNumEntries() > 1) { - const auto* firstEntry = node.getEntry(0)->ptrCast(); + // A partitioned parent is resolved from catalog truth (not the pattern's entries, which + // may contain routing-wrapper scan substitutes). Route each row into the partition matching + // its partition-key value at insert time; partitions claimed by routing hooks carry a null + // table pointer and are shipped remotely. + const auto* firstEntry = node.getEntry(0)->ptrCast(); + // A pattern on the parent itself expands either to all partition children or, when every + // partition is routed remotely, to a single wrapper-provided substitute (which carries its + // own scan function). A pattern naming one partition subgraph directly is a plain write. + const bool parentPattern = + firstEntry->isPartitionChild() && + (node.getNumEntries() > 1 || firstEntry->getScanFunction().has_value()); + if (parentPattern) { const auto parentID = firstEntry->getParentTableID(); DASSERT(parentID != INVALID_TABLE_ID); auto transaction = transaction::Transaction::Get(*clientContext); const auto* parent = Catalog::Get(*clientContext) ->getTableCatalogEntry(transaction, parentID) ->ptrCast(); + DASSERT(parent->isPartitioned()); + const auto childTableIDs = parent->getChildTableIDs(); + DASSERT(!childTableIDs.empty()); + const auto* hooks = common::getPartitionRoutingHooks(); + common::PartitionHandle handle = nullptr; + const bool firstClaimed = + hooks != nullptr && hooks->locate != nullptr && + hooks->locate(hooks->context, common::PartitionRef{parentID, 0}, &handle); + // The "first partition" table is only used to derive the PK vector position; it can be + // null when every partition is routed remotely. + auto tableInfo = NodeTableInsertInfo( + firstClaimed ? nullptr : + storageManager->getTable(childTableIDs[0])->ptrCast(), + std::move(evaluators)); tableInfo.partitionKeyColumnID = parent->getPartitionColumnID(); - tableInfo.partitionTables.reserve(node.getNumEntries()); - for (auto i = 0u; i < node.getNumEntries(); ++i) { + tableInfo.partitionTables.reserve(childTableIDs.size()); + for (auto i = 0u; i < childTableIDs.size(); ++i) { + const auto ref = common::PartitionRef{parentID, i}; + common::PartitionHandle partHandle = nullptr; + const bool claimed = hooks != nullptr && hooks->locate != nullptr && + hooks->locate(hooks->context, ref, &partHandle); tableInfo.partitionTables.push_back( - storageManager->getTable(node.getEntry(i)->getTableID())->ptrCast()); + claimed ? nullptr : + storageManager->getTable(childTableIDs[i])->ptrCast()); + tableInfo.partitionChildIDs.push_back(childTableIDs[i]); + tableInfo.partitionRefs.push_back(ref); + tableInfo.partitionHandles.push_back(claimed ? partHandle : nullptr); } + return NodeInsertExecutor(std::move(info), std::move(tableInfo)); } + auto table = storageManager->getTable(firstEntry->getTableID())->ptrCast(); + auto tableInfo = NodeTableInsertInfo(table, std::move(evaluators)); return NodeInsertExecutor(std::move(info), std::move(tableInfo)); } diff --git a/src/processor/operator/persistent/insert_executor.cpp b/src/processor/operator/persistent/insert_executor.cpp index ed1fdeb69..26c7ba23e 100644 --- a/src/processor/operator/persistent/insert_executor.cpp +++ b/src/processor/operator/persistent/insert_executor.cpp @@ -1,5 +1,6 @@ #include "processor/operator/persistent/insert_executor.h" +#include "common/exception/runtime.h" #include "processor/partition_routing.h" #include "transaction/transaction.h" @@ -42,7 +43,8 @@ void NodeTableInsertInfo::init(const ResultSet& resultSet, main::ClientContext* columnDataVectors.push_back(evaluator->resultVector.get()); columnIDs.push_back(columnIDs.size()); } - pkVector = columnDataVectors[table->getPKColumnID()]; + // Null table means every partition is routed remotely; PK validation is the wrapper's job. + pkVector = table == nullptr ? nullptr : columnDataVectors[table->getPKColumnID()]; } void NodeInsertExecutor::init(ResultSet* resultSet, const ExecutionContext* context) { @@ -92,25 +94,47 @@ void NodeInsertExecutor::setNodeIDVectorToNonNull() const { info.nodeIDVector->setNull(info.nodeIDVector->state->getSelVector()[0], false); } +uint64_t NodeInsertExecutor::currentPartitionIndex() const { + auto* keyVector = tableInfo.columnDataVectors[tableInfo.partitionKeyColumnID]; + DASSERT(keyVector->state->getSelVector().getSelSize() == 1); + std::vector partitionIndexes; + computePartitionIndexes(*keyVector, tableInfo.partitionTables.size(), partitionIndexes); + return partitionIndexes[0]; +} + storage::NodeTable* NodeInsertExecutor::resolveTargetTable() const { if (tableInfo.partitionTables.empty()) { return tableInfo.table; } + // Null table = remotely routed partition; callers branch on that. + return tableInfo.partitionTables[currentPartitionIndex()]; +} + +nodeID_t NodeInsertExecutor::insertRemotely(uint64_t index, + transaction::Transaction* transaction) const { + const auto* hooks = common::getPartitionRoutingHooks(); + if (hooks == nullptr || hooks->insertRow == nullptr) { + throw RuntimeException( + "Partition is routed remotely but no routing hooks with insertRow are installed."); + } auto* keyVector = tableInfo.columnDataVectors[tableInfo.partitionKeyColumnID]; - DASSERT(keyVector->state->getSelVector().getSelSize() == 1); - std::vector partitionIndexes; - computePartitionIndexes(*keyVector, tableInfo.partitionTables.size(), partitionIndexes); - return tableInfo.partitionTables[partitionIndexes[0]]; + return hooks->insertRow(hooks->context, tableInfo.partitionRefs[index], + tableInfo.partitionHandles[index], transaction, keyVector, tableInfo.columnDataVectors); } storage::NodeTable* NodeInsertExecutor::resolveTableForNodeID(common::nodeID_t nodeID) const { if (tableInfo.partitionTables.empty()) { return tableInfo.table; } - for (auto* table : tableInfo.partitionTables) { - if (table->getTableID() == nodeID.tableID) { - return table; + for (auto i = 0u; i < tableInfo.partitionTables.size(); ++i) { + if (tableInfo.partitionChildIDs[i] != nodeID.tableID) { + continue; + } + if (tableInfo.partitionTables[i] != nullptr) { + return tableInfo.partitionTables[i]; } + // Remotely routed partition: the caller must go through the hooks. + return nullptr; } return tableInfo.table; } @@ -120,16 +144,24 @@ nodeID_t NodeInsertExecutor::insert(main::ClientContext* context) { evaluator->evaluate(); } auto transaction = Transaction::Get(*context); - auto* targetTable = resolveTargetTable(); - if (checkConflict(transaction, targetTable)) { - return info.getNodeID(); + nodeID_t resultNodeID; + if (!tableInfo.partitionTables.empty() && resolveTargetTable() == nullptr) { + // Remotely routed partition: the wrapper owns conflict handling and returns the + // remotely-assigned node ID. + resultNodeID = insertRemotely(currentPartitionIndex(), transaction); + } else { + auto* targetTable = resolveTargetTable(); + if (checkConflict(transaction, targetTable)) { + return info.getNodeID(); + } + storage::NodeTableInsertState insertState{*info.nodeIDVector, *tableInfo.pkVector, + tableInfo.columnDataVectors}; + targetTable->initInsertState(context, insertState); + targetTable->insert(transaction, insertState); + resultNodeID = info.getNodeID(); } - storage::NodeTableInsertState insertState{*info.nodeIDVector, *tableInfo.pkVector, - tableInfo.columnDataVectors}; - targetTable->initInsertState(context, insertState); - targetTable->insert(transaction, insertState); writeColumnVectors(info.columnVectors, tableInfo.columnDataVectors); - return info.getNodeID(); + return resultNodeID; } void NodeInsertExecutor::skipInsert() const { @@ -156,6 +188,22 @@ void NodeInsertExecutor::skipInsert(nodeID_t nodeID, main::ClientContext* contex } auto transaction = Transaction::Get(*context); auto* table = resolveTableForNodeID(nodeID); + if (table == nullptr) { + // Remotely routed partition: fetch the row through the routing hooks. + const auto* hooks = common::getPartitionRoutingHooks(); + if (hooks == nullptr || hooks->lookupRow == nullptr) { + throw RuntimeException("Partition is routed remotely but no routing hooks with " + "lookupRow are installed."); + } + for (auto i = 0u; i < tableInfo.partitionTables.size(); ++i) { + if (tableInfo.partitionChildIDs[i] == nodeID.tableID) { + hooks->lookupRow(hooks->context, tableInfo.partitionRefs[i], + tableInfo.partitionHandles[i], transaction, nodeID, outputVectors); + return; + } + } + return; + } storage::NodeTableScanState scanState{info.nodeIDVector, std::move(outputVectors), info.nodeIDVector->state}; scanState.setToTable(transaction, table, std::move(columnIDs), {}); diff --git a/src/processor/operator/persistent/node_batch_insert.cpp b/src/processor/operator/persistent/node_batch_insert.cpp index a982fbef4..fde6f4483 100644 --- a/src/processor/operator/persistent/node_batch_insert.cpp +++ b/src/processor/operator/persistent/node_batch_insert.cpp @@ -500,10 +500,21 @@ void NodeBatchInsert::initGlobalStateInternal(ExecutionContext* context) { if (nodeInfo->partitionInfo.has_value()) { const auto& partitionInfo = *nodeInfo->partitionInfo; nodeSharedState->targets.reserve(partitionInfo.numPartitions); - for (auto tableID : partitionInfo.partitionTableIDs) { + const auto* hooks = common::getPartitionRoutingHooks(); + for (auto i = 0u; i < partitionInfo.partitionTableIDs.size(); ++i) { + const auto tableID = partitionInfo.partitionTableIDs[i]; NodeBatchInsertTarget target; - target.table = storageManager->getTable(tableID)->ptrCast(); + const auto ref = common::PartitionRef{partitionInfo.parentTableID, i}; + common::PartitionHandle handle = nullptr; + const bool claimed = hooks != nullptr && hooks->locate != nullptr && + hooks->locate(hooks->context, ref, &handle); + // Remotely routed partitions own no local table; rows are shipped through the + // routing hooks in copyToNodeGroup. + target.table = + claimed ? nullptr : storageManager->getTable(tableID)->ptrCast(); nodeSharedState->targets.push_back(std::move(target)); + nodeSharedState->partitionRefs.push_back(ref); + nodeSharedState->partitionHandles.push_back(claimed ? handle : nullptr); } // The partition-key column id is the catalog property id. Property columns are evaluated // in schema order, so its index among the column evaluators equals its position in @@ -520,6 +531,9 @@ void NodeBatchInsert::initGlobalStateInternal(ExecutionContext* context) { } for (auto& target : nodeSharedState->targets) { + if (target.table == nullptr) { + continue; // remotely routed: PK handling is the wrapper's job + } nodeSharedState->initTargetPKIndex(context, target); } } @@ -535,13 +549,19 @@ void NodeBatchInsert::initLocalStateInternal(ResultSet* resultSet, ExecutionCont nodeLocalState->stats.emplace( std::span{nodeInfo->columnTypes.begin(), nodeInfo->outputDataColumns.size()}); nodeLocalState->targets.reserve(nodeSharedState->targets.size()); - for (auto& sharedTarget : nodeSharedState->targets) { + for (auto i = 0u; i < nodeSharedState->targets.size(); ++i) { + auto& sharedTarget = nodeSharedState->targets[i]; NodeBatchInsertLocalTarget localTarget; if (sharedTarget.globalIndexBuilder) { localTarget.localIndexBuilder = sharedTarget.globalIndexBuilder->clone(); } - localTarget.errorHandler = - createErrorHandler(context, sharedTarget.table, &nodeLocalState->duplicatePKSkipResult); + // Remotely routed targets never receive local appends, so they carry no error handler. + if (sharedTarget.table == nullptr) { + localTarget.errorHandler = std::nullopt; + } else { + localTarget.errorHandler = createErrorHandler(context, sharedTarget.table, + &nodeLocalState->duplicatePKSkipResult); + } nodeLocalState->targets.push_back(std::move(localTarget)); } nodeLocalState->optimisticAllocator = @@ -683,6 +703,21 @@ void NodeBatchInsert::copyToNodeGroup(transaction::Transaction* transaction, } auto& target = nodeLocalState->targets[partitionIdx]; + if (nodeSharedState->targets[partitionIdx].table == nullptr) { + // Remotely routed partition: ship the run through the routing hooks. Rows are + // addressed by their logical index in the evaluated row space (the same convention + // InMemChunkedNodeGroup::append uses). + const auto* hooks = common::getPartitionRoutingHooks(); + if (hooks == nullptr || hooks->insertChunk == nullptr) { + throw RuntimeException("Partition is routed remotely but no routing hooks with " + "insertChunk are installed."); + } + hooks->insertChunk(hooks->context, nodeSharedState->partitionRefs[partitionIdx], + nodeSharedState->partitionHandles[partitionIdx], transaction, &keyVector, + nodeLocalState->columnVectors, i, runEnd - i); + i = runEnd; + continue; + } auto numAppendedTuples = 0ul; while (numAppendedTuples < runEnd - i) { if (!target.chunkedGroup) { @@ -841,6 +876,10 @@ void NodeBatchInsert::finalize(ExecutionContext* context) { auto& pageAllocator = *transaction->getLocalStorage()->addOptimisticAllocator(); for (auto targetIdx = 0u; targetIdx < nodeSharedState->targets.size(); ++targetIdx) { auto& sharedTarget = nodeSharedState->targets[targetIdx]; + if (sharedTarget.table == nullptr) { + // Remotely routed partition: the wrapper owns index finalization. + continue; + } auto errorHandler = createErrorHandler(context, sharedTarget.table, nodeSharedState->duplicatePKSkipResult.get()); if (sharedTarget.sharedNodeGroup) { diff --git a/src/storage/storage_manager.cpp b/src/storage/storage_manager.cpp index af0565014..1be020c17 100644 --- a/src/storage/storage_manager.cpp +++ b/src/storage/storage_manager.cpp @@ -7,6 +7,7 @@ #include "common/constants.h" #include "common/enums/storage_format.h" #include "common/file_system/virtual_file_system.h" +#include "common/partition_routing_hook.h" #include "common/random_engine.h" #include "common/serializer/in_mem_file_writer.h" #include "main/attached_database.h" @@ -141,9 +142,40 @@ void StorageManager::recover(main::ClientContext& clientContext, bool throwOnWal walReplayer->replay(throwOnWalReplayFailure, enableChecksums); } +namespace { + +// True if the entry is a partition subgraph whose storage is routed elsewhere by +// an installed PartitionRoutingHooks. Such partitions keep full catalog metadata +// but own no local table/WAL/checkpoint state. +bool isRemotelyRoutedPartition(const catalog::TableCatalogEntry* entry) { + if (entry->getType() != CatalogEntryType::NODE_TABLE_ENTRY) { + return false; + } + auto* nodeEntry = entry->constPtrCast(); + if (!nodeEntry->isPartitionChild()) { + return false; + } + const auto* hooks = common::getPartitionRoutingHooks(); + if (hooks == nullptr || hooks->locate == nullptr) { + return false; + } + common::PartitionHandle handle = nullptr; + return hooks->locate(hooks->context, + common::PartitionRef{nodeEntry->getParentTableID(), nodeEntry->getPartitionIndex()}, + &handle); +} + +} // namespace + void StorageManager::createNodeTable(NodeTableCatalogEntry* entry, main::ClientContext* context) { tableNameCache[entry->getTableID()] = entry->getName(); + // A partition subgraph claimed by the routing wrapper owns no local storage: + // no table object, no WAL records, no checkpoint work. + if (isRemotelyRoutedPartition(entry)) { + return; + } + if (entry->getStorageFormat() != StorageFormat::NONE) { if (entry->getStorageFormat() == StorageFormat::ICEBUG_DISK) { // Create icebug-disk-backed node table @@ -357,6 +389,10 @@ bool StorageManager::checkpoint(main::ClientContext* context, const Catalog& cat std::shared_lock lck{mtx}; for (const auto entry : nodeTableEntries) { + // Partition subgraphs claimed by the routing wrapper hold no local state. + if (isRemotelyRoutedPartition(entry)) { + continue; + } if (!tables.contains(entry->getTableID())) { throw RuntimeException(std::format( "Checkpoint failed: table {} not found in storage manager.", entry->getName())); @@ -392,6 +428,10 @@ bool StorageManager::checkpoint(main::ClientContext* context, const Catalog& cat std::shared_lock lck{mtx}; for (const auto entry : nodeTableEntries) { + // Partition subgraphs claimed by the routing wrapper hold no local state. + if (isRemotelyRoutedPartition(entry)) { + continue; + } if (!tables.contains(entry->getTableID())) { throw RuntimeException(std::format( "Checkpoint failed: table {} not found in storage manager.", entry->getName())); @@ -445,6 +485,10 @@ void StorageManager::rollbackCheckpoint(const Catalog& catalog) { tableEntry->ptrCast()->isPartitioned()) { continue; } + // Remotely routed partition subgraphs hold no local state. + if (isRemotelyRoutedPartition(tableEntry)) { + continue; + } DASSERT(tables.contains(tableEntry->getTableID())); tables.at(tableEntry->getTableID())->rollbackCheckpoint(); } @@ -474,6 +518,10 @@ void StorageManager::serialize(const Catalog& catalog, Serializer& ser) { ser.writeDebuggingInfo("num_node_tables"); ser.write(nodeTableEntries.size()); for (const auto tableEntry : nodeTableEntries) { + // Remotely routed partition subgraphs hold no local state to serialize. + if (isRemotelyRoutedPartition(tableEntry)) { + continue; + } DASSERT(tables.contains(tableEntry->getTableID())); ser.writeDebuggingInfo("table_id"); ser.write(tableEntry->getTableID()); @@ -510,6 +558,10 @@ void StorageManager::serialize(const Catalog& catalog, const Transaction& snapsh ser.writeDebuggingInfo("num_node_tables"); ser.write(nodeTableEntries.size()); for (const auto tableEntry : nodeTableEntries) { + // Remotely routed partition subgraphs hold no local state to serialize. + if (isRemotelyRoutedPartition(tableEntry)) { + continue; + } DASSERT(tables.contains(tableEntry->getTableID())); ser.writeDebuggingInfo("table_id"); ser.write(tableEntry->getTableID()); diff --git a/test/api/CMakeLists.txt b/test/api/CMakeLists.txt index f636439df..2d445e66f 100644 --- a/test/api/CMakeLists.txt +++ b/test/api/CMakeLists.txt @@ -1,6 +1,7 @@ add_lbug_api_test(api_test api_test.cpp file_search_path_test.cpp + partition_routing_test.cpp system_config_test.cpp arrow_test.cpp arrow_node_table_test.cpp diff --git a/test/api/partition_routing_test.cpp b/test/api/partition_routing_test.cpp new file mode 100644 index 000000000..187cb7230 --- /dev/null +++ b/test/api/partition_routing_test.cpp @@ -0,0 +1,387 @@ +#include +#include +#include +#include + +#include "api_test/api_test.h" +#include "binder/binder.h" +#include "binder/bound_table_scan_info.h" +#include "binder/ddl/property_definition.h" +#include "binder/expression/variable_expression.h" +#include "catalog/catalog.h" +#include "catalog/catalog_entry/node_table_catalog_entry.h" +#include "common/constants.h" +#include "common/partition_routing_hook.h" +#include "function/table/bind_data.h" +#include "function/table/simple_table_function.h" +#include "storage/storage_manager.h" +#include "storage/table/node_table.h" +#include "transaction/transaction.h" +#include + +using namespace lbug::catalog; +using namespace lbug::common; +using namespace lbug::function; +using namespace lbug::main; +using namespace lbug::testing; +using namespace lbug::binder; +using namespace lbug::storage; +using namespace lbug::transaction; + +namespace { + +// --------------------------------------------------------------------------- +// Mock distributed wrapper. +// +// Models the minimal behavior of a real wrapper: partitions are claimed for +// remote placement when their parent is provisioned remotely; writes are +// shipped to an in-memory sink; reads are served through a wrapper-registered +// table function exposed via a foreign-backed catalog entry. +// --------------------------------------------------------------------------- + +struct RemoteRow { + int64_t id; + int64_t amount; +}; + +struct MockWrapper { + // When true, provisioning notifications claim the parent for remote placement. + bool provisionRemotely = false; + std::unordered_set claimedParents; + // Force-claims a single partition of a parent (to produce local/remote mixes). + std::optional> forcedClaim; + std::vector createdRefs; + std::vector droppedRefs; + std::vector pointRows; // captured via insertRow + std::vector chunkRows; // captured via insertChunk + PartitionScanSpec scanSpec; + TableFunction ownedScanFunction; + + void reset() { + provisionRemotely = false; + claimedParents.clear(); + forcedClaim.reset(); + createdRefs.clear(); + droppedRefs.clear(); + pointRows.clear(); + chunkRows.clear(); + scanSpec = PartitionScanSpec{}; + ownedScanFunction = TableFunction{}; + } + + std::vector allRows() const { + std::vector combined = pointRows; + combined.insert(combined.end(), chunkRows.begin(), chunkRows.end()); + return combined; + } +}; + +MockWrapper mock; + +std::string refString(PartitionRef ref) { + return std::format("{}:{}", ref.parentTableID, ref.partitionIndex); +} + +bool locateHook(void* /*context*/, PartitionRef ref, PartitionHandle* handleOut) { + // Any stable non-null marker stands in for a connection/host descriptor. + *handleOut = reinterpret_cast(static_cast(0xC0FFEE)); + if (mock.claimedParents.contains(ref.parentTableID)) { + return true; + } + if (mock.forcedClaim.has_value() && mock.forcedClaim->first == ref.parentTableID && + mock.forcedClaim->second == ref.partitionIndex) { + return true; + } + return false; +} + +void onPartitionCreateHook(void* /*context*/, PartitionRef ref, PartitionHandle /*handle*/) { + mock.createdRefs.push_back(refString(ref)); + if (mock.provisionRemotely) { + mock.claimedParents.insert(ref.parentTableID); + } +} + +void onPartitionDropHook(void* /*context*/, PartitionRef ref, PartitionHandle /*handle*/) { + mock.droppedRefs.push_back(refString(ref)); +} + +// Point-write sink. Column vectors hold exactly one evaluated row in schema order. +nodeID_t insertRowHook(void* /*context*/, PartitionRef ref, PartitionHandle /*handle*/, + Transaction* /*tx*/, const ValueVector* keyVector, + std::span columnVectors) { + const auto idPos = columnVectors[0]->state->getSelVector()[0]; + const auto keySel = keyVector->state->getSelVector()[0]; + mock.pointRows.push_back( + {columnVectors[0]->getValue(idPos), keyVector->getValue(keySel)}); + return {static_cast(mock.pointRows.size() - 1), ref.parentTableID}; +} + +// Bulk-write sink. Row j of the run lives at selection position [startRow + j]. +void insertChunkHook(void* /*context*/, PartitionRef /*ref*/, PartitionHandle /*handle*/, + Transaction* /*tx*/, const ValueVector* keyVector, std::span columnVectors, + uint64_t startRow, uint64_t numRows) { + const auto& keySel = keyVector->state->getSelVector(); + const auto& idSel = columnVectors[0]->state->getSelVector(); + for (uint64_t j = 0; j < numRows; ++j) { + mock.chunkRows.push_back({columnVectors[0]->getValue(idSel[startRow + j]), + keyVector->getValue(keySel[startRow + j])}); + } +} + +// Read path: serve the captured rows through a wrapper-owned table function. +// Column layout: [rowid, id, amount]. +offset_t remoteScanInternalFunc(const TableFuncMorsel& morsel, const TableFuncInput& /*input*/, + DataChunk& output) { + if (!morsel.hasMoreToOutput()) { + return 0; + } + const auto rows = mock.allRows(); + for (auto i = 0u; i < morsel.getMorselSize(); ++i) { + output.getValueVectorMutable(0).setValue(i, (int64_t)(morsel.startOffset + i)); + output.getValueVectorMutable(1).setValue(i, rows[morsel.startOffset + i].id); + output.getValueVectorMutable(2).setValue(i, rows[morsel.startOffset + i].amount); + } + return morsel.getMorselSize(); +} + +expression_vector remoteScanColumns(const std::string& nodeUniqueName) { + expression_vector columns; + columns.push_back(std::make_shared(LogicalType::INT64(), + nodeUniqueName + "." + std::string(InternalKeyword::ID), "rowid")); + columns.push_back( + std::make_shared(LogicalType::INT64(), nodeUniqueName + ".id", "id")); + columns.push_back(std::make_shared(LogicalType::INT64(), + nodeUniqueName + ".amount", "amount")); + return columns; +} + +std::unique_ptr remoteScanBindFunc(const ClientContext* /*context*/, + const TableFuncBindInput* input) { + std::vector names{"rowid", "id", "amount"}; + std::vector types; + types.emplace_back(LogicalType::INT64()); + types.emplace_back(LogicalType::INT64()); + types.emplace_back(LogicalType::INT64()); + names = TableFunction::extractYieldVariables(names, input->yieldVariables); + auto columns = input->binder->createVariables(names, types); + return std::make_unique(std::move(columns), mock.allRows().size()); +} + +TableFunction makeRemoteScanFunction() { + TableFunction func("test_partition_remote_scan", std::vector{}); + func.tableFunc = SimpleTableFunc::getTableFunc(remoteScanInternalFunc); + func.bindFunc = remoteScanBindFunc; + func.initSharedStateFunc = SimpleTableFunc::initSharedState; + func.initLocalStateFunc = TableFunction::initEmptyLocalState; + return func; +} + +// Register the wrapper's scan function in this database and keep a copy for bindScan to +// hand out. The engine attaches it to clones of the partition's own catalog entries. +void setupRemoteScan(Connection* con) { + auto* context = con->getClientContext(); + auto catalog = Catalog::Get(*context); + // Function/entry registration happens outside any active query, so use the dummy + // transaction like core bootstrap code does. + auto* transaction = &DUMMY_CHECKPOINT_TRANSACTION; + + if (!catalog->containsFunction(transaction, "test_partition_remote_scan")) { + function_set fs; + fs.push_back(std::make_unique(makeRemoteScanFunction())); + catalog->addFunction(transaction, CatalogEntryType::TABLE_FUNCTION_ENTRY, + "test_partition_remote_scan", std::move(fs), true /* isInternal */); + } + + if (mock.scanSpec.scanFunction != nullptr) { + return; + } + mock.ownedScanFunction = makeRemoteScanFunction(); + mock.scanSpec.scanFunction = &mock.ownedScanFunction; + mock.scanSpec.createBindData = [](const std::string& nodeUniqueName) { + return std::make_unique(remoteScanColumns(nodeUniqueName), + mock.allRows().size()); + }; +} + +bool bindScanHook(void* /*context*/, PartitionRef /*ref*/, PartitionHandle /*handle*/, + PartitionScanSpec* specOut) { + *specOut = mock.scanSpec; + return true; +} + +struct HooksGuard { + HooksGuard() { setupHooks(); } + ~HooksGuard() { + setPartitionRoutingHooks(nullptr); + mock.reset(); + } + // Must outlive the registration: the registry keeps the raw pointer. + static PartitionRoutingHooks& hookStruct() { + static PartitionRoutingHooks hooks; + return hooks; + } + static void setupHooks() { + auto& hooks = hookStruct(); + hooks.context = &mock; + hooks.locate = locateHook; + hooks.onPartitionCreate = onPartitionCreateHook; + hooks.onPartitionDrop = onPartitionDropHook; + hooks.bindScan = bindScanHook; + hooks.insertRow = insertRowHook; + hooks.insertChunk = insertChunkHook; + setPartitionRoutingHooks(&hooks); + } +}; + +table_id_t getTableID(Connection* con, const std::string& name) { + auto* context = con->getClientContext(); + return Catalog::Get(*context) + ->getTableCatalogEntry(&DUMMY_CHECKPOINT_TRANSACTION, name) + ->getTableID(); +} + +bool hasLocalStorage(Connection* con, table_id_t tableID) { + return StorageManager::Get(*con->getClientContext())->containsTable(tableID); +} + +std::string sortLines(std::string s) { + std::vector lines; + std::istringstream iss{s}; + for (std::string line; std::getline(iss, line);) { + lines.push_back(line); + } + std::sort(lines.begin(), lines.end()); + std::ostringstream oss; + for (const auto& line : lines) { + oss << line << '\n'; + } + return oss.str(); +} + +} // namespace + +class PartitionRoutingTest : public ApiTest { + void SetUp() override { + ApiTest::SetUp(); + setupRemoteScan(conn.get()); + } +}; + +TEST_F(PartitionRoutingTest, LifecycleAndStorageSkip) { + HooksGuard guard; + + // Unclaimed partitioned table: fully local, unchanged behavior. + ASSERT_TRUE(conn->query("CREATE NODE TABLE LocalP (id INT64 PRIMARY KEY, amount INT64) " + "PARTITION BY HASH(amount) PARTITIONS 2;") + ->isSuccess()); + ASSERT_EQ(mock.createdRefs.size(), 2u); + + // Claimed partitioned table: catalog metadata stays local, storage is skipped and the + // wrapper sees provisioning notifications. + mock.createdRefs.clear(); + mock.provisionRemotely = true; + ASSERT_TRUE(conn->query("CREATE NODE TABLE RemoteP (id INT64 PRIMARY KEY, amount INT64) " + "PARTITION BY HASH(amount) PARTITIONS 3;") + ->isSuccess()); + mock.provisionRemotely = false; + ASSERT_EQ(mock.createdRefs.size(), 3u); + + const auto remoteParentID = getTableID(conn.get(), "RemoteP"); + for (auto i = 0u; i < 3; ++i) { + EXPECT_EQ(mock.createdRefs[i], refString(PartitionRef{remoteParentID, i})); + } + + // Catalog keeps full metadata for every partition... + for (auto i = 0u; i < 3; ++i) { + ASSERT_NO_THROW(getTableID(conn.get(), std::format("RemoteP_p{}", i))); + } + // ...but no local storage exists for claimed partitions. + for (auto i = 0u; i < 3; ++i) { + const auto childID = getTableID(conn.get(), std::format("RemoteP_p{}", i)); + EXPECT_FALSE(hasLocalStorage(conn.get(), childID)) + << "partition " << i << " should have no local storage"; + } + // The unclaimed sibling owns local storage as usual. + EXPECT_TRUE(hasLocalStorage(conn.get(), getTableID(conn.get(), "LocalP_p0"))); + + // Dropping the parent notifies the wrapper for every partition subgraph. + ASSERT_TRUE(conn->query("DROP TABLE RemoteP;")->isSuccess()); + ASSERT_EQ(mock.droppedRefs.size(), 3u); + for (auto i = 0u; i < 3; ++i) { + EXPECT_EQ(mock.droppedRefs[i], refString(PartitionRef{remoteParentID, i})); + } +} + +TEST_F(PartitionRoutingTest, FullyRemotePointInsertAndScan) { + HooksGuard guard; + + mock.provisionRemotely = true; + ASSERT_TRUE(conn->query("CREATE NODE TABLE RP (id INT64 PRIMARY KEY, amount INT64) " + "PARTITION BY HASH(amount) PARTITIONS 3;") + ->isSuccess()); + mock.provisionRemotely = false; + + // Point inserts route to the wrapper, which assigns node IDs. + for (auto id = 1; id <= 3; ++id) { + ASSERT_TRUE(conn->query(std::format("CREATE (:RP {{id: {}, amount: {}}});", id, id * 10)) + ->isSuccess()) + << std::format("insert {}", id); + } + ASSERT_EQ(mock.pointRows.size(), 3u); + + // Reads are served by the wrapper's consolidated scan entry. + auto result = conn->query("MATCH (n:RP) RETURN n.id, n.amount ORDER BY n.id;"); + ASSERT_TRUE(result->isSuccess()) << result->toString(); + ASSERT_EQ(result->getNumTuples(), 3u); + EXPECT_EQ(sortLines(result->toString()), "1|10\n2|20\n3|30\nn.id|n.amount\n"); +} + +TEST_F(PartitionRoutingTest, FullyRemoteBulkInsert) { + HooksGuard guard; + + mock.provisionRemotely = true; + ASSERT_TRUE(conn->query("CREATE NODE TABLE RP (id INT64 PRIMARY KEY, amount INT64) " + "PARTITION BY HASH(amount) PARTITIONS 3;") + ->isSuccess()); + mock.provisionRemotely = false; + + const auto csvPath = + TestHelper::appendLbugRootPath("test/test_files/partition_routing/routing_bulk.csv"); + auto copyResult = conn->query(std::format("COPY RP FROM '{}';", csvPath)); + ASSERT_TRUE(copyResult->isSuccess()) << copyResult->toString(); + ASSERT_EQ(mock.chunkRows.size(), 4u); + + auto result = conn->query("MATCH (n:RP) RETURN count(*) AS c;"); + ASSERT_TRUE(result->isSuccess()) << result->toString(); + ASSERT_EQ(result->getNumTuples(), 1u); + EXPECT_EQ(result->getNext()->getValue(0)->getValue(), 4); +} + +TEST_F(PartitionRoutingTest, MixedLocalRemoteScanRejected) { + HooksGuard guard; + + // Created while nothing is claimed: all partitions are local. + ASSERT_TRUE(conn->query("CREATE NODE TABLE MP (id INT64 PRIMARY KEY, amount INT64) " + "PARTITION BY HASH(amount) PARTITIONS 2;") + ->isSuccess()); + ASSERT_TRUE(conn->query("CREATE (:MP {id: 1, amount: 10});")->isSuccess()); + + // Now claim exactly one partition behind the engine's back (models a wrapper that only + // owns part of a table). Scanning such a parent cannot be planned and must be rejected + // at bind time rather than silently returning wrong results. + mock.forcedClaim = {getTableID(conn.get(), "MP"), 1}; + auto result = conn->query("MATCH (n:MP) RETURN n.id;"); + ASSERT_FALSE(result->isSuccess()); + EXPECT_NE(result->toString().find("mix of locally stored and remotely routed"), + std::string::npos) + << result->toString(); + + // Writes still route correctly under partial claims: each row lands either locally or at + // the wrapper according to the partition function. + mock.forcedClaim.reset(); + ASSERT_TRUE(conn->query("CREATE (:MP {id: 2, amount: 20});")->isSuccess()); + auto result2 = conn->query("MATCH (n:MP) RETURN n.id ORDER BY n.id;"); + ASSERT_TRUE(result2->isSuccess()) << result2->toString(); + ASSERT_EQ(result2->getNumTuples(), 2u); +} diff --git a/test/test_files/partition_routing/routing_bulk.csv b/test/test_files/partition_routing/routing_bulk.csv new file mode 100644 index 000000000..6f57a8323 --- /dev/null +++ b/test/test_files/partition_routing/routing_bulk.csv @@ -0,0 +1,4 @@ +1,10 +2,20 +3,30 +4,40