duckdb: enable csr_rel_ traversal and fix join push-down naming

duckdb/src/catalog/duckdb_catalog.cpp

@@ -1,8 +1,11 @@
 #include "catalog/duckdb_catalog.h"
 
+#include <optional>
 #include <regex>
+#include <utility>
 
 #include "binder/bound_attach_info.h"
+#include "binder/expression/variable_expression.h"
 #include "catalog/catalog_entry/node_table_catalog_entry.h"
 #include "catalog/catalog_entry/rel_group_catalog_entry.h"
 #include "catalog/duckdb_table_catalog_entry.h"
@@ -21,9 +24,10 @@ namespace duckdb_extension {
 
 DuckDBCatalog::DuckDBCatalog(std::string dbPath, std::string catalogName,
     std::string defaultSchemaName, main::ClientContext* context, const DuckDBConnector& connector,
-    const binder::AttachOption& attachOption)
+    const binder::AttachOption& attachOption, std::string attachedDbName)
     : CatalogExtension{}, dbPath{std::move(dbPath)}, catalogName{std::move(catalogName)},
       defaultSchemaName{std::move(defaultSchemaName)},
+      dbName{std::move(attachedDbName)},
       tableNamesVector{common::LogicalType::STRING(), storage::MemoryManager::Get(*context)},
       connector{connector}, context_{context} {
     skipUnsupportedTable = DuckDBStorageExtension::SKIP_UNSUPPORTED_TABLE_DEFAULT_VAL;
@@ -57,14 +61,37 @@ void DuckDBCatalog::init() {
     DuckDBResultConverter::getDuckDBVectorConversionFunc(common::PhysicalTypeID::STRING,
         conversionFunc);
     conversionFunc(resultChunk->data[0], tableNamesVector, resultChunk->size());
+    // Two-pass initialization: node tables must be registered before rel tables
+    // so that rel tables can resolve their src/dst node table IDs. The table
+    // enumeration order is alphabetical, which can put rel_* tables before the
+    // node tables they reference.
+    // First pass: register node tables (everything that is not a rel table).
+    for (auto i = 0u; i < resultChunk->size(); i++) {
+        auto tableName = tableNamesVector.getValue<common::string_t>(i).getAsString();
+        auto lowerName = tableName;
+        common::StringUtils::toLower(lowerName);
+        if (lowerName.rfind("rel_", 0) == 0 || lowerName.rfind("csr_rel_", 0) == 0) {
+            continue;
+        }
+        createForeignTable(tableName);
+    }
+    // Second pass: register rel tables.
     for (auto i = 0u; i < resultChunk->size(); i++) {
         auto tableName = tableNamesVector.getValue<common::string_t>(i).getAsString();
         auto lowerName = tableName;
         common::StringUtils::toLower(lowerName);
         if (lowerName.rfind("rel_", 0) == 0) {
-            createForeignRelTable(tableName);
-        } else {
-            createForeignTable(tableName);
+            // Foreign-key-based rel table: scan-driven, optimizer generates a join.
+            // No CSR columns; backed by a ForeignRelTable. Foreign keys reference
+            // primary keys, which are not guaranteed to equal node offsets.
+            createForeignRelTable(tableName, false /* internalIDContract */);
+        } else if (lowerName.rfind("csr_rel_", 0) == 0) {
+            // CSR-based rel table: materialized into a local on-disk CSR rel table.
+            // TODO: COPY data from DuckDB into a local RelTable.
+            // The csr_rel_ prefix promises that the foreign keys are usable as
+            // node offsets directly (dense, gapless, aligned with the node
+            // tables' internal ID scheme), enabling MATCH traversal.
+            createForeignRelTable(tableName, true /* internalIDContract */);
         }
     }
 }
@@ -115,7 +142,9 @@ void DuckDBCatalog::createForeignTable(const std::string& tableName) {
 
     // Create DuckDB scan function for SQL pushdown
     auto scanFunction = getScanFunction(duckdbTableInfo);
-    auto foreignDatabaseName = std::format("{}.{}", catalogName, tableName);
+    // Must be the attached-database name: the join-push-down optimizer uses it
+    // as a lookup key into DatabaseManager::getAttachedDatabase().
+    auto foreignDatabaseName = dbName;
     auto mainTableEntry = std::make_unique<catalog::NodeTableCatalogEntry>(info->tableName,
         primaryKeyName, foreignDatabaseName, catalog::ShadowTag{});
     for (auto& definition : extraInfo->propertyDefinitions) {
@@ -128,32 +157,35 @@ void DuckDBCatalog::createForeignTable(const std::string& tableName) {
     lbug::storage::StorageManager::Get(*context_)->createTable(mainEntry);
 }
 
-void DuckDBCatalog::createForeignRelTable(const std::string& tableName) {
-    // Query foreign key info to find src/dst node tables
-    auto fkQuery = std::format("SELECT kcu.column_name, ccu.table_name "
-                               "FROM information_schema.table_constraints tc "
-                               "JOIN information_schema.key_column_usage kcu "
-                               "  ON tc.constraint_name = kcu.constraint_name "
-                               "  AND tc.table_schema = kcu.table_schema "
-                               "JOIN information_schema.constraint_column_usage ccu "
-                               "  ON ccu.constraint_name = tc.constraint_name "
-                               "  AND ccu.table_schema = tc.table_schema "
-                               "WHERE tc.constraint_type = 'FOREIGN KEY' "
-                               "  AND tc.table_name = '{}'",
+void DuckDBCatalog::createForeignRelTable(const std::string& tableName, bool internalIDContract) {
+    // Query foreign key info to find src/dst node tables.
+    //
+    // information_schema.constraint_column_usage is unusable for this in
+    // DuckDB: for FK constraints it reports the constraint's own table (the
+    // referencing side), not the referenced one. duckdb_constraints() exposes
+    // the referenced table directly; unnest() flattens the column list so each
+    // row yields (fk_column, referenced_table).
+    auto fkQuery = std::format("SELECT unnest(constraint_column_names) AS column_name, "
+                               "referenced_table FROM duckdb_constraints() "
+                               "WHERE constraint_type = 'FOREIGN KEY' "
+                               "AND referenced_table IS NOT NULL AND table_name = '{}'",
         tableName);
     auto fkResult = connector.executeQuery(fkQuery);
 
     std::string srcTableName, dstTableName;
+    std::string srcColName, dstColName;
     for (auto i = 0u; i < fkResult->RowCount(); i++) {
         auto colName = fkResult->GetValue(0, i).GetValue<std::string>();
         auto refTable = fkResult->GetValue(1, i).GetValue<std::string>();
         auto lowerCol = colName;
         common::StringUtils::toLower(lowerCol);
         if (lowerCol == "src_id" || lowerCol.find("src") == 0) {
             srcTableName = refTable;
+            srcColName = colName;
         } else if (lowerCol == "dst_id" || lowerCol.find("dst") == 0 ||
                    lowerCol.find("dest") == 0) {
             dstTableName = refTable;
+            dstColName = colName;
         }
     }
 
@@ -166,8 +198,15 @@ void DuckDBCatalog::createForeignRelTable(const std::string& tableName) {
     std::vector<binder::PropertyDefinition> propertyDefinitions;
     bindPropertyDefinitions(tableName, propertyDefinitions);
 
-    // Determine the node table IDs from the main catalog
+    // Determine the node table IDs from the main catalog. containsTable() must
+    // be checked first: getTableCatalogEntry() throws when the table is
+    // missing, and a rel table may reference tables that were not registered.
     auto* catalog = context_->getDatabase()->getCatalog();
+    if (!catalog->containsTable(&transaction::DUMMY_TRANSACTION, srcTableName) ||
+        !catalog->containsTable(&transaction::DUMMY_TRANSACTION, dstTableName)) {
+        createForeignTable(tableName);
+        return;
+    }
     auto* srcEntry = catalog->getTableCatalogEntry(&transaction::DUMMY_TRANSACTION, srcTableName);
     auto* dstEntry = catalog->getTableCatalogEntry(&transaction::DUMMY_TRANSACTION, dstTableName);
     if (srcEntry == nullptr || dstEntry == nullptr) {
@@ -178,14 +217,21 @@ void DuckDBCatalog::createForeignRelTable(const std::string& tableName) {
     common::table_id_t srcTableID = srcEntry->getTableID();
     common::table_id_t dstTableID = dstEntry->getTableID();
 
-    // Build query and scan info
+    // Build columns for the scan bind data (must happen before columnTypes is
+    // moved into the scan info below).
     std::vector<common::LogicalType> columnTypes;
     std::vector<std::string> columnNames;
     for (auto& def : propertyDefinitions) {
         columnNames.push_back(def.getName());
         columnTypes.push_back(def.getType().copy());
     }
+    binder::expression_vector columns;
+    for (auto i = 0u; i < columnTypes.size(); i++) {
+        columns.push_back(std::make_shared<binder::VariableExpression>(columnTypes[i].copy(),
+            columnNames[i], columnNames[i]));
+    }
 
+    // Build query and scan info
     auto queryStr =
         std::format("SELECT * FROM \"{}\".{}.{}", catalogName, defaultSchemaName, tableName);
     auto duckdbTableInfo = std::make_shared<DuckDBTableScanInfo>(queryStr, std::move(columnTypes),
@@ -200,13 +246,32 @@ void DuckDBCatalog::createForeignRelTable(const std::string& tableName) {
     }
     tables->createEntry(&transaction::DUMMY_TRANSACTION, std::move(tableEntry));
 
-    // Create bind data for the scan function
-    binder::expression_vector emptyColumns;
-    auto bindData =
-        std::make_shared<DuckDBScanBindData>(queryStr, columnNames, connector, emptyColumns);
-
-    // Create RelGroupCatalogEntry
-    auto foreignDatabaseName = std::format("{}.{}", catalogName, tableName);
+    std::optional<std::pair<common::column_id_t, common::column_id_t>> srcDstColumnPositions;
+    if (internalIDContract) {
+        auto findPos = [&columnNames](
+                           const std::string& name) -> std::optional<common::column_id_t> {
+            for (auto i = 0u; i < columnNames.size(); i++) {
+                auto lower = columnNames[i];
+                common::StringUtils::toLower(lower);
+                auto lowerName = name;
+                common::StringUtils::toLower(lowerName);
+                if (lower == lowerName) {
+                    return i;
+                }
+            }
+            return std::nullopt;
+        };
+        auto srcPos = findPos(srcColName);
+        auto dstPos = findPos(dstColName);
+        if (srcPos && dstPos) {
+            srcDstColumnPositions = std::make_pair(*srcPos, *dstPos);
+        }
+    }
+    auto bindData = std::make_shared<DuckDBScanBindData>(queryStr, columnNames, connector,
+        std::move(columns), srcDstColumnPositions);
+    // Create RelGroupCatalogEntry. foreignDatabaseName must be the
+    // attached-database name (see the node shadow comment above).
+    auto foreignDatabaseName = dbName;
 
     std::vector<catalog::RelTableCatalogInfo> relTableInfos;
     auto info = bindCreateTableInfo(tableName);
@@ -217,8 +282,10 @@ void DuckDBCatalog::createForeignRelTable(const std::string& tableName) {
     auto relGroupEntry =
         std::make_unique<catalog::RelGroupCatalogEntry>(tableName, common::RelMultiplicity::MANY,
             common::RelMultiplicity::MANY, common::ExtendDirection::BOTH, std::move(relTableInfos),
-            "", // storage
-            common::StorageFormat::NONE, scanFunc, bindData, std::move(foreignDatabaseName));
+            // "<attachedDb>.<table>" matches the pg_client convention; the
+            // join-push-down optimizer extracts the table name after the dot.
+            dbName + "." + tableName, common::StorageFormat::NONE, scanFunc, bindData,
+            std::move(foreignDatabaseName));
 
     for (auto& def : propertyDefinitions) {
         relGroupEntry->addProperty(def);

duckdb/src/include/catalog/duckdb_catalog.h

@@ -33,7 +33,7 @@ class DuckDBCatalog : public extension::CatalogExtension {
 public:
     DuckDBCatalog(std::string dbPath, std::string catalogName, std::string defaultSchemaName,
         main::ClientContext* context, const DuckDBConnector& connector,
-        const binder::AttachOption& attachOption);
+        const binder::AttachOption& attachOption, std::string attachedDbName);
 
     void init() override;
 
@@ -50,12 +50,17 @@ class DuckDBCatalog : public extension::CatalogExtension {
 
 private:
     void createForeignTable(const std::string& tableName);
-    void createForeignRelTable(const std::string& tableName);
+    void createForeignRelTable(const std::string& tableName, bool internalIDContract);
 
 protected:
     std::string dbPath;
     std::string catalogName;
+    // The name this database was attached under (e.g. 'g' in ATTACH ... as g).
+    // The join-push-down optimizer uses it as a lookup key into
+    // DatabaseManager::getAttachedDatabase(), so shadow entries and rel group
+    // entries must store this name -- not the schema-qualified catalog name.
     std::string defaultSchemaName;
+    std::string dbName;
     common::ValueVector tableNamesVector;
     bool skipUnsupportedTable;
     const DuckDBConnector& connector;

duckdb/src/include/function/duckdb_scan.h

@@ -1,5 +1,8 @@
 #pragma once
 
+#include <optional>
+#include <utility>
+
 #include "binder/expression/expression_util.h"
 #include "common/types/types.h"
 #include "connector/duckdb_result_converter.h"
@@ -56,16 +59,33 @@ struct DuckDBScanBindData : function::TableFuncBindData {
     std::vector<std::string> columnNamesInDuckDB;
     const DuckDBConnector& connector;
     DuckDBResultConverter converter;
+    // Positions of the src/dst columns in the scan output when the scan
+    // output carries node offsets directly (csr_rel_* contract); nullopt
+    // otherwise. See TableFuncBindData::getSrcDstColumnPositions.
+    std::optional<std::pair<common::column_id_t, common::column_id_t>> srcDstColumnPositions;
 
     DuckDBScanBindData(std::string query, std::vector<std::string> columnNamesInDuckDB,
         const DuckDBConnector& connector, binder::expression_vector columns)
         : function::TableFuncBindData{std::move(columns), 0 /* numRows */}, query{std::move(query)},
           columnNamesInDuckDB{std::move(columnNamesInDuckDB)}, connector{connector},
           converter{binder::ExpressionUtil::getDataTypes(this->columns)} {}
+    DuckDBScanBindData(std::string query, std::vector<std::string> columnNamesInDuckDB,
+        const DuckDBConnector& connector, binder::expression_vector columns,
+        std::optional<std::pair<common::column_id_t, common::column_id_t>> srcDstColumnPositions)
+        : function::TableFuncBindData{std::move(columns), 0 /* numRows */}, query{std::move(query)},
+          columnNamesInDuckDB{std::move(columnNamesInDuckDB)}, connector{connector},
+          converter{binder::ExpressionUtil::getDataTypes(this->columns)},
+          srcDstColumnPositions{srcDstColumnPositions} {}
     DuckDBScanBindData(const DuckDBScanBindData& other)
         : TableFuncBindData{other}, query{other.query},
           columnNamesInDuckDB{other.columnNamesInDuckDB}, connector{other.connector},
-          converter{binder::ExpressionUtil::getDataTypes(this->columns)} {}
+          converter{binder::ExpressionUtil::getDataTypes(this->columns)},
+          srcDstColumnPositions{other.srcDstColumnPositions} {}
+
+    std::optional<std::pair<common::column_id_t, common::column_id_t>>
+    getSrcDstColumnPositions() const override {
+        return srcDstColumnPositions;
+    }
 
     std::string getColumnsToSelect() const;
     std::vector<uint32_t> getColumnIndicesToSelect() const;

duckdb/src/storage/duckdb_storage.cpp

@@ -30,7 +30,7 @@ std::unique_ptr<main::AttachedDatabase> attachDuckDB(std::string dbName, std::st
     connector->connect(dbPath, catalogName, schemaName, clientContext);
 
     auto duckdbCatalog = std::make_unique<DuckDBCatalog>(std::move(dbPath), std::move(catalogName),
-        schemaName, clientContext, *connector, attachOption);
+        schemaName, clientContext, *connector, attachOption, dbName);
     duckdbCatalog->init();
     return std::make_unique<AttachedDuckDBDatabase>(dbName, DuckDBStorageExtension::DB_TYPE,
         std::move(duckdbCatalog), std::move(connector));

duckdb/test/test_files/duckdb_rel.test

@@ -0,0 +1,64 @@
+-DATASET CSV empty
+
+--
+
+-CASE DuckdbAutoDetectRelTables
+-SKIP_FSM_LEAK_CHECK
+-LOAD_DYNAMIC_EXTENSION duckdb
+-STATEMENT ATTACH '${LBUG_ROOT_DIRECTORY}/extension/duckdb/test/test_files/users_sessions.db' as g (dbtype duckdb);
+---- 1
+Attached database successfully.
+-STATEMENT LOAD FROM g.users RETURN name ORDER BY name;
+---- 3
+Alice
+Bob
+Carol
+-STATEMENT LOAD FROM g.sessions RETURN device ORDER BY device;
+---- 4
+desktop
+laptop
+mobile
+tablet
+-STATEMENT LOAD FROM g.rel_user_owns_session RETURN src_user, dst_session, since ORDER BY src_user, dst_session;
+---- 5
+0|1|2024-01-15
+0|2|2024-02-20
+1|0|2024-03-05
+2|1|2024-05-25
+2|3|2024-04-10
+-STATEMENT MATCH (u:g.users) RETURN count(*);
+---- 1
+3
+-STATEMENT MATCH (u:g.users) WHERE u.name = 'Alice' RETURN count(*);
+---- 1
+1
+-STATEMENT DETACH g;
+---- ok
+
+-CASE DuckdbCsrRelPrefixRegistersAsRelTable
+-SKIP_FSM_LEAK_CHECK
+-LOAD_DYNAMIC_EXTENSION duckdb
+-STATEMENT ATTACH '${LBUG_ROOT_DIRECTORY}/extension/duckdb/test/test_files/users_sessions.db' as g (dbtype duckdb);
+---- 1
+Attached database successfully.
+-STATEMENT LOAD FROM g.csr_rel_user_blocks_user RETURN src_user, dst_blocked ORDER BY src_user, dst_blocked;
+---- 2
+0|1
+2|1
+-STATEMENT MATCH (u:g.users) RETURN count(*);
+---- 1
+3
+-STATEMENT DETACH g;
+---- ok
+
+-CASE DuckdbCsrRelTraversal
+-SKIP_FSM_LEAK_CHECK
+-LOAD_DYNAMIC_EXTENSION duckdb
+-STATEMENT ATTACH '${LBUG_ROOT_DIRECTORY}/extension/duckdb/test/test_files/users_sessions.db' as g (dbtype duckdb);
+---- 1
+Attached database successfully.
+-STATEMENT MATCH (u:g.users)-[b:csr_rel_user_blocks_user]->(v:g.users) RETURN count(*);
+---- 1
+2
+-STATEMENT DETACH g;
+---- ok

postgres/src/storage/postgres_storage.cpp

@@ -33,7 +33,7 @@ std::unique_ptr<main::AttachedDatabase> attachPostgres(std::string dbName, std::
     auto connector = std::make_unique<PostgresConnector>();
     connector->connect(dbPath, catalogName, schemaName, clientContext);
     auto catalog = std::make_unique<duckdb_extension::DuckDBCatalog>(dbPath, catalogName,
-        schemaName, clientContext, *connector, attachOption);
+        schemaName, clientContext, *connector, attachOption, dbName);
     catalog->init();
     return std::make_unique<AttachedPostgresDatabase>(dbName, PostgresStorageExtension::DB_TYPE,
         std::move(catalog), std::move(connector), catalogName);

sqlite/src/storage/sqlite_storage.cpp

@@ -27,7 +27,8 @@ std::unique_ptr<main::AttachedDatabase> attachSqlite(std::string dbName, std::st
     connector->connect(dbPath, catalogName, SqliteStorageExtension::DEFAULT_SCHEMA_NAME,
         clientContext);
     auto catalog = std::make_unique<duckdb_extension::DuckDBCatalog>(dbPath, catalogName,
-        SqliteStorageExtension::DEFAULT_SCHEMA_NAME, clientContext, *connector, attachOption);
+        SqliteStorageExtension::DEFAULT_SCHEMA_NAME, clientContext, *connector, attachOption,
+        dbName);
     catalog->init();
     return std::make_unique<duckdb_extension::AttachedDuckDBDatabase>(dbName,
         SqliteStorageExtension::DB_TYPE, std::move(catalog), std::move(connector));

unity_catalog/src/storage/unity_catalog_storage.cpp

@@ -19,7 +19,8 @@ std::unique_ptr<main::AttachedDatabase> attachUnityCatalog(std::string dbName, s
     connector->connect(dbPath, dbName, UnityCatalogStorageExtension::DEFAULT_SCHEMA_NAME,
         clientContext);
     auto catalog = std::make_unique<duckdb_extension::DuckDBCatalog>(dbPath, dbPath,
-        UnityCatalogStorageExtension::DEFAULT_SCHEMA_NAME, clientContext, *connector, attachOption);
+        UnityCatalogStorageExtension::DEFAULT_SCHEMA_NAME, clientContext, *connector, attachOption,
+        dbName);
     catalog->init();
     return std::make_unique<duckdb_extension::AttachedDuckDBDatabase>(dbName,
         UnityCatalogStorageExtension::DB_TYPE, std::move(catalog), std::move(connector));