Add native ADBC extension

CMakeLists.txt

@@ -79,6 +79,7 @@ function(add_extension_if_enabled_and_skip_32bit extension)
 endfunction()
 
 add_extension_if_enabled_and_skip_32bit("duckdb")
+add_extension_if_enabled_and_skip_32bit("adbc")
 add_extension_if_enabled_and_skip_32bit("postgres")
 add_extension_if_enabled_and_skip_32bit("sqlite")
 add_extension_if_enabled_and_skip_32bit("delta")

adbc/CMakeLists.txt

@@ -0,0 +1,29 @@
+find_path(ADBC_INCLUDE_DIR
+        NAMES arrow-adbc/adbc.h
+        HINTS "$ENV{CONDA_PREFIX}/include")
+
+find_library(ADBC_DRIVER_MANAGER_LIBRARY
+        NAMES adbc_driver_manager
+        HINTS "$ENV{CONDA_PREFIX}/lib")
+
+if (NOT ADBC_INCLUDE_DIR OR NOT ADBC_DRIVER_MANAGER_LIBRARY)
+    message(FATAL_ERROR "ADBC extension requires libadbc-driver-manager. Install the C++ ADBC dependencies with pixi/conda.")
+endif ()
+
+include_directories(
+        src/include
+        ${CMAKE_BINARY_DIR}/src/include
+        ${PROJECT_SOURCE_DIR}/src/include
+        ${ADBC_INCLUDE_DIR})
+
+add_subdirectory(src/connector)
+add_subdirectory(src/function)
+add_subdirectory(src/catalog)
+add_subdirectory(src/storage)
+add_subdirectory(src/main)
+
+build_extension_lib(${BUILD_STATIC_EXTENSION} "adbc")
+
+target_link_libraries(lbug_${EXTENSION_LIB_NAME}_extension
+        PRIVATE
+        ${ADBC_DRIVER_MANAGER_LIBRARY})

adbc/src/catalog/CMakeLists.txt

@@ -0,0 +1,8 @@
+add_library(lbug_adbc_catalog
+        OBJECT
+        adbc_catalog.cpp
+        adbc_table_catalog_entry.cpp)
+
+set(ADBC_EXTENSION_OBJECT_FILES
+        ${ADBC_EXTENSION_OBJECT_FILES} $<TARGET_OBJECTS:lbug_adbc_catalog>
+        PARENT_SCOPE)

adbc/src/catalog/adbc_catalog.cpp

@@ -0,0 +1,53 @@
+#include "catalog/adbc_catalog.h"
+
+#include "catalog/adbc_table_catalog_entry.h"
+#include "catalog/catalog_entry/node_table_catalog_entry.h"
+#include "function/adbc_scan.h"
+#include "main/client_context.h"
+#include "main/database.h"
+#include "storage/storage_manager.h"
+#include "transaction/transaction.h"
+
+namespace lbug {
+namespace adbc_extension {
+
+void ADBCCatalog::init() {
+    for (auto& tableName : connector.getTableNames()) {
+        createForeignTable(tableName);
+    }
+}
+
+void ADBCCatalog::createForeignTable(const std::string& tableName) {
+    auto columns = connector.getTableSchema(schemaName, tableName);
+    std::vector<std::string> columnNames;
+    std::vector<common::LogicalType> columnTypes;
+    for (auto& [name, type] : columns) {
+        columnNames.push_back(name);
+        columnTypes.push_back(type.copy());
+    }
+    auto scanInfo = std::make_shared<ADBCTableScanInfo>(tableName, columnNames,
+        copyVector(columnTypes), connector);
+    auto attachedEntry = std::make_unique<catalog::ADBCTableCatalogEntry>(tableName,
+        getADBCScanFunction(scanInfo), scanInfo);
+    for (auto i = 0u; i < columnNames.size(); i++) {
+        attachedEntry->addProperty(binder::PropertyDefinition{
+            binder::ColumnDefinition{columnNames[i], columnTypes[i].copy()}});
+    }
+    auto attachedEntryPtr = attachedEntry.get();
+    tables->createEntry(&transaction::DUMMY_TRANSACTION, std::move(attachedEntry));
+    auto primaryKeyName = columnNames[0];
+    auto mainTableEntry = std::make_unique<catalog::NodeTableCatalogEntry>(tableName,
+        primaryKeyName, tableName, catalog::ShadowTag{});
+    for (auto i = 0u; i < columnNames.size(); i++) {
+        mainTableEntry->addProperty(binder::PropertyDefinition{
+            binder::ColumnDefinition{columnNames[i], columnTypes[i].copy()}});
+    }
+    mainTableEntry->setReferencedEntry(attachedEntryPtr);
+    context->getDatabase()->getCatalog()->addTableEntry(std::move(mainTableEntry));
+    auto mainEntry = context->getDatabase()->getCatalog()->getTableCatalogEntry(
+        &transaction::DUMMY_TRANSACTION, tableName);
+    lbug::storage::StorageManager::Get(*context)->createTable(mainEntry);
+}
+
+} // namespace adbc_extension
+} // namespace lbug

adbc/src/catalog/adbc_table_catalog_entry.cpp

@@ -0,0 +1,59 @@
+#include "catalog/adbc_table_catalog_entry.h"
+
+#include "binder/bound_scan_source.h"
+#include "binder/expression/variable_expression.h"
+#include "common/constants.h"
+
+namespace lbug {
+namespace catalog {
+
+ADBCTableCatalogEntry::ADBCTableCatalogEntry(std::string name,
+    std::optional<function::TableFunction> scanFunction,
+    std::shared_ptr<adbc_extension::ADBCTableScanInfo> scanInfo)
+    : TableCatalogEntry{CatalogEntryType::FOREIGN_TABLE_ENTRY, std::move(name)},
+      scanFunction{std::move(scanFunction)}, scanInfo{std::move(scanInfo)} {}
+
+common::TableType ADBCTableCatalogEntry::getTableType() const {
+    return common::TableType::FOREIGN;
+}
+
+std::unique_ptr<binder::BoundTableScanInfo> ADBCTableCatalogEntry::getBoundScanInfo(
+    main::ClientContext* /*context*/, const std::string& nodeUniqueName) {
+    binder::expression_vector columns;
+    std::vector<std::string> scanColumnNames;
+    std::vector<common::LogicalType> scanColumnTypes;
+    if (!nodeUniqueName.empty()) {
+        auto idUniqueName = nodeUniqueName + "." + std::string(common::InternalKeyword::ID);
+        columns.push_back(std::make_shared<binder::VariableExpression>(common::LogicalType::INT64(),
+            idUniqueName, scanInfo->columnNames[0]));
+        scanColumnNames.push_back(scanInfo->columnNames[0]);
+        scanColumnTypes.push_back(common::LogicalType::INT64());
+    }
+    for (auto i = 0u; i < scanInfo->columnNames.size(); i++) {
+        auto uniqueName = nodeUniqueName.empty() ? scanInfo->columnNames[i] :
+                                                   nodeUniqueName + "." + scanInfo->columnNames[i];
+        columns.push_back(std::make_shared<binder::VariableExpression>(
+            scanInfo->columnTypes[i].copy(), uniqueName, scanInfo->columnNames[i]));
+        scanColumnNames.push_back(scanInfo->columnNames[i]);
+        scanColumnTypes.push_back(scanInfo->columnTypes[i].copy());
+    }
+    auto boundScanInfo = std::make_shared<adbc_extension::ADBCTableScanInfo>(scanInfo->tableName,
+        std::move(scanColumnNames), std::move(scanColumnTypes), scanInfo->connector);
+    auto bindData = std::make_unique<adbc_extension::ADBCScanBindData>(std::move(boundScanInfo),
+        std::move(columns));
+    return std::make_unique<binder::BoundTableScanInfo>(scanFunction, std::move(bindData));
+}
+
+std::unique_ptr<TableCatalogEntry> ADBCTableCatalogEntry::copy() const {
+    auto other = std::make_unique<ADBCTableCatalogEntry>(name, scanFunction, scanInfo);
+    other->copyFrom(*this);
+    return other;
+}
+
+std::unique_ptr<binder::BoundExtraCreateCatalogEntryInfo>
+ADBCTableCatalogEntry::getBoundExtraCreateInfo(transaction::Transaction*) const {
+    UNREACHABLE_CODE;
+}
+
+} // namespace catalog
+} // namespace lbug

adbc/src/connector/CMakeLists.txt

@@ -0,0 +1,7 @@
+add_library(lbug_adbc_connector
+        OBJECT
+        adbc_connector.cpp)
+
+set(ADBC_EXTENSION_OBJECT_FILES
+        ${ADBC_EXTENSION_OBJECT_FILES} $<TARGET_OBJECTS:lbug_adbc_connector>
+        PARENT_SCOPE)