feat: Add Cube ADBC Driver for CubeSQL

.github/workflows/ci.yml

@@ -40,6 +40,18 @@ jobs:
           otp-version: "26.2"
           elixir-version: "1.15.8"
 
+      - name: Install FlatBuffers 2.0.8 from source
+        run: |
+          sudo apt-get update && sudo apt install -y build-essential cmake
+          cd /tmp
+          wget https://github.com/google/flatbuffers/archive/v2.0.8.tar.gz
+          tar -xzf v2.0.8.tar.gz
+          cd flatbuffers-2.0.8
+          cmake -G "Unix Makefiles" -DCMAKE_BUILD_TYPE=Release -DCMAKE_INSTALL_PREFIX=/usr/local
+          make -j$(nproc)
+          sudo make install
+          sudo ldconfig
+
       - name: Compile and Test
         run: |
           mix deps.get
@@ -70,7 +82,7 @@ jobs:
 
   macos:
     if: contains(github.event.pull_request.labels.*.name, 'skip ci') != true
-    runs-on: macos-13
+    runs-on: macos-14
     env:
       MIX_ENV: test
       OTP_VERSION: "26.2"
@@ -80,6 +92,9 @@ jobs:
       - name: Checkout
         uses: actions/checkout@v4
 
+      - name: Install Flatbuffers using Homebrew
+        run: brew install gcc flatbuffers
+
       - name: Install OTP and Elixir
         run: |
           curl -fsSO https://elixir-lang.org/install.sh

.gitignore

@@ -36,3 +36,14 @@ adbc-*.tar
 
 # checksum
 checksum.exs
+
+# Make temporary
+
+/cmake_adbc/
+/cmake_adbc_nif/
+**/CMakeFiles/*
+cmake_install.cmake
+CTestTestfile.cmake
+
+# Build artifacts
+/priv/

3rd_party/apache-arrow-adbc/c/CMakeLists.txt

@@ -101,6 +101,12 @@ if(ADBC_DRIVER_BIGQUERY)
   add_subdirectory(driver/bigquery)
 endif()
 
+if(ADBC_DRIVER_CUBE)
+  install(FILES "${REPOSITORY_ROOT}/c/include/arrow-adbc/driver/cube.h"
+          DESTINATION include/arrow-adbc/driver)
+  add_subdirectory(driver/cube)
+endif()
+
 if(ADBC_INTEGRATION_DUCKDB)
   add_subdirectory(integration/duckdb)
 endif()
@@ -155,6 +161,10 @@ LIBRARY=$<TARGET_FILE:adbc_driver_${TARGET}_shared>" ${Python3_EXECUTABLE} -m pi
   if(ADBC_DRIVER_BIGQUERY)
     adbc_install_python_package(bigquery)
   endif()
+
+  if(ADBC_DRIVER_CUBE)
+    adbc_install_python_package(cube)
+  endif()
 endif()
 
 validate_config()

3rd_party/apache-arrow-adbc/c/driver/postgresql/copy/statement.h

@@ -0,0 +1,181 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#pragma once
+
+#include <cstring>
+#include <memory>
+#include <string>
+#include <vector>
+
+#include <arrow-adbc/adbc.h>
+#include <libpq-fe.h>
+#include <nanoarrow/nanoarrow.h>
+
+#include "copy/reader.h"
+#include "driver/common/utils.h"
+#include "postgres_type.h"
+
+#define ADBC_POSTGRESQL_OPTION_BATCH_SIZE_HINT_BYTES \
+  "adbc.postgresql.batch_size_hint_bytes"
+
+#define ADBC_POSTGRESQL_OPTION_USE_COPY "adbc.postgresql.use_copy"
+
+namespace adbcpq {
+class PostgresConnection;
+class PostgresStatement;
+
+constexpr static int64_t kDefaultBatchSizeHintBytes = 16777216;
+
+/// \brief An ArrowArrayStream that reads tuples from a PGresult.
+class TupleReader final : public std::enable_shared_from_this<TupleReader> {
+ public:
+  TupleReader(PGconn* conn)
+      : status_(ADBC_STATUS_OK),
+        error_(ADBC_ERROR_INIT),
+        conn_(conn),
+        result_(nullptr),
+        pgbuf_(nullptr),
+        copy_reader_(nullptr),
+        row_id_(-1),
+        batch_size_hint_bytes_(kDefaultBatchSizeHintBytes),
+        is_finished_(false) {
+    ArrowErrorInit(&na_error_);
+    data_.data.as_char = nullptr;
+    data_.size_bytes = 0;
+  }
+
+  int GetSchema(struct ArrowSchema* out);
+  int GetNext(struct ArrowArray* out);
+  const char* last_error() const { return error_.message; }
+  void Release();
+  void ExportTo(struct ArrowArrayStream* stream);
+
+  static const struct AdbcError* ErrorFromArrayStream(struct ArrowArrayStream* stream,
+                                                      AdbcStatusCode* status);
+
+ private:
+  friend class PostgresStatement;
+
+  int GetCopyData();
+  int AppendRowAndFetchNext();
+  int BuildOutput(struct ArrowArray* out);
+
+  static int GetSchemaTrampoline(struct ArrowArrayStream* self, struct ArrowSchema* out);
+  static int GetNextTrampoline(struct ArrowArrayStream* self, struct ArrowArray* out);
+  static const char* GetLastErrorTrampoline(struct ArrowArrayStream* self);
+  static void ReleaseTrampoline(struct ArrowArrayStream* self);
+
+  AdbcStatusCode status_;
+  struct AdbcError error_;
+  struct ArrowError na_error_;
+  PGconn* conn_;
+  PGresult* result_;
+  char* pgbuf_;
+  struct ArrowBufferView data_;
+  std::unique_ptr<PostgresCopyStreamReader> copy_reader_;
+  int64_t row_id_;
+  int64_t batch_size_hint_bytes_;
+  bool is_finished_;
+};
+
+class PostgresStatement {
+ public:
+  PostgresStatement()
+      : connection_(nullptr),
+        query_(),
+        prepared_(false),
+        use_copy_(-1),
+        reader_(nullptr),
+        batch_size_hint_bytes_(kDefaultBatchSizeHintBytes) {
+    std::memset(&bind_, 0, sizeof(bind_));
+  }
+
+  // ---------------------------------------------------------------------
+  // ADBC API implementation
+
+  AdbcStatusCode Bind(struct ArrowArray* values, struct ArrowSchema* schema,
+                      struct AdbcError* error);
+  AdbcStatusCode Bind(struct ArrowArrayStream* stream, struct AdbcError* error);
+  AdbcStatusCode Cancel(struct AdbcError* error);
+  AdbcStatusCode ExecuteQuery(struct ArrowArrayStream* stream, int64_t* rows_affected,
+                              struct AdbcError* error);
+  AdbcStatusCode ExecuteSchema(struct ArrowSchema* schema, struct AdbcError* error);
+  AdbcStatusCode GetOption(const char* key, char* value, size_t* length,
+                           struct AdbcError* error);
+  AdbcStatusCode GetOptionBytes(const char* key, uint8_t* value, size_t* length,
+                                struct AdbcError* error);
+  AdbcStatusCode GetOptionDouble(const char* key, double* value, struct AdbcError* error);
+  AdbcStatusCode GetOptionInt(const char* key, int64_t* value, struct AdbcError* error);
+  AdbcStatusCode GetParameterSchema(struct ArrowSchema* schema, struct AdbcError* error);
+  AdbcStatusCode New(struct AdbcConnection* connection, struct AdbcError* error);
+  AdbcStatusCode Prepare(struct AdbcError* error);
+  AdbcStatusCode Release(struct AdbcError* error);
+  AdbcStatusCode SetOption(const char* key, const char* value, struct AdbcError* error);
+  AdbcStatusCode SetOptionBytes(const char* key, const uint8_t* value, size_t length,
+                                struct AdbcError* error);
+  AdbcStatusCode SetOptionDouble(const char* key, double value, struct AdbcError* error);
+  AdbcStatusCode SetOptionInt(const char* key, int64_t value, struct AdbcError* error);
+  AdbcStatusCode SetSqlQuery(const char* query, struct AdbcError* error);
+
+  // ---------------------------------------------------------------------
+  // Helper methods
+
+  void ClearResult();
+  AdbcStatusCode CreateBulkTable(const std::string& current_schema,
+                                 const struct ArrowSchema& source_schema,
+                                 std::string* escaped_table,
+                                 std::string* escaped_field_list,
+                                 struct AdbcError* error);
+  AdbcStatusCode ExecuteIngest(struct ArrowArrayStream* stream, int64_t* rows_affected,
+                               struct AdbcError* error);
+  AdbcStatusCode ExecuteBind(struct ArrowArrayStream* stream, int64_t* rows_affected,
+                             struct AdbcError* error);
+
+ private:
+  std::shared_ptr<PostgresTypeResolver> type_resolver_;
+  std::shared_ptr<PostgresConnection> connection_;
+
+  // Query state
+  std::string query_;
+  bool prepared_;
+  struct ArrowArrayStream bind_;
+
+  // Bulk ingest state
+  enum class IngestMode {
+    kCreate,
+    kAppend,
+    kReplace,
+    kCreateAppend,
+  };
+
+  // Options
+  int use_copy_;
+
+  struct {
+    std::string db_schema;
+    std::string target;
+    IngestMode mode = IngestMode::kCreate;
+    bool temporary = false;
+  } ingest_;
+
+  std::shared_ptr<TupleReader> reader_;
+  int64_t batch_size_hint_bytes_;
+
+  int UseCopy();
+};
+}  // namespace adbcpq

CMakeLists.txt

@@ -13,6 +13,19 @@ if(NOT DEFINED PRIV_DIR)
 endif()
 message(STATUS "Using PRIV_DIR: ${PRIV_DIR}")
 
+if(UNIX AND NOT APPLE)
+    # Find the FlatBuffers package
+    include_directories(/usr/lib/${CMAKE_LIBRARY_ARCHITECTURE}/cmake/flatbuffers)
+    find_package(FlatBuffers CONFIG REQUIRED)
+    # Print the directory where the config file was found
+    if(FlatBuffers_FOUND)
+          message(STATUS "FlatBuffers found in directory: ${FlatBuffers_DIR}")
+          message(STATUS "FlatBuffers config file path: ${FlatBuffers_CONFIG}")
+    else()
+          message(FATAL_ERROR "Could not find FlatBuffers package")
+    endif()
+endif()
+
 if(DEFINED ERTS_INCLUDE_DIR AND NOT "${ERTS_INCLUDE_DIR}" STREQUAL "")
     set(ERTS_INCLUDE_DIR "${ERTS_INCLUDE_DIR}")
 else()
@@ -35,7 +48,11 @@ if(DEFINED ENV{TARGET_GCC_FLAGS})
     set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} $ENV{TARGET_GCC_FLAGS}")
 endif()
 
-message(STATUS "CMAKE_TOOLCHAIN_FILE: ${CMAKE_TOOLCHAIN_FILE}")
+if(DEFINED CMAKE_TOOLCHAIN_FILE AND NOT "${CMAKE_TOOLCHAIN_FILE}" STREQUAL "")
+    message(STATUS "CMAKE_TOOLCHAIN_FILE: ${CMAKE_TOOLCHAIN_FILE}")
+else()
+    message(STATUS "CMAKE_TOOLCHAIN_FILE: not set")
+endif()
 
 find_package(AdbcDriverManager REQUIRED PATHS "${PRIV_DIR}/" NO_DEFAULT_PATH)
 include_directories("${PRIV_DIR}/include")

Makefile

@@ -34,11 +34,12 @@ ifdef CMAKE_TOOLCHAIN_FILE
 endif
 
 CMAKE_BUILD_TYPE ?= Release
-DEFAULT_JOBS ?= 1
-CMAKE_ADBC_BUILD_DIR = $(MIX_APP_PATH)/cmake_adbc
-CMAKE_ADBC_OPTIONS ?= ""
-CMAKE_ADBC_NIF_BUILD_DIR = $(MIX_APP_PATH)/cmake_adbc_nif
-CMAKE_ADBC_NIF_OPTIONS ?= ""
+DEFAULT_JOBS ?= 22
+BUILD_DIR = $(MIX_APP_PATH)/_build
+CMAKE_ADBC_BUILD_DIR = $(BUILD_DIR)/cmake/adbc
+CMAKE_ADBC_OPTIONS ?=
+CMAKE_ADBC_NIF_BUILD_DIR = $(BUILD_DIR)/cmake/nif
+CMAKE_ADBC_NIF_OPTIONS ?=
 MAKE_BUILD_FLAGS ?= -j$(DEFAULT_JOBS)
 
 .DEFAULT_GLOBAL := build
@@ -51,8 +52,7 @@ build: $(NIF_SO_REL)
 clean:
 	@rm -rf "$(NIF_SO_REL)"
 	@rm -rf "$(NIF_SO)"
-	@rm -rf "$(CMAKE_ADBC_NIF_BUILD_DIR)"
-	@rm -rf "$(CMAKE_ADBC_BUILD_DIR)"
+	@rm -rf "$(BUILD_DIR)"
 	@rm -rf "$(PRIV_DIR)"
 
 priv_dir:
@@ -70,6 +70,7 @@ adbc: priv_dir
 		cmake --no-warn-unused-cli \
 			-DADBC_BUILD_SHARED="ON" \
 			-DADBC_DRIVER_MANAGER="ON" \
+			-DADBC_DRIVER_CUBE="OFF" \
 			-DADBC_DRIVER_POSTGRESQL="OFF" \
 			-DADBC_DRIVER_SQLITE="OFF" \
 			-DADBC_DRIVER_FLIGHTSQL="OFF" \
@@ -89,12 +90,12 @@ $(NIF_SO_REL): priv_dir adbc $(C_SRC_REL)/adbc_nif_resource.hpp $(C_SRC_REL)/adb
 	@ mkdir -p "$(CMAKE_ADBC_NIF_BUILD_DIR)" && \
 	cd "$(CMAKE_ADBC_NIF_BUILD_DIR)" && \
 	cmake --no-warn-unused-cli \
-		-D CMAKE_BUILD_TYPE="$(CMAKE_BUILD_TYPE)" \
-		-D C_SRC="$(C_SRC)" \
-		-D ADBC_SRC="$(ADBC_SRC)" \
-		-D MIX_APP_PATH="$(MIX_APP_PATH)" \
-		-D PRIV_DIR="$(PRIV_DIR)" \
-		-D ERTS_INCLUDE_DIR="$(ERTS_INCLUDE_DIR)" \
+		-DCMAKE_BUILD_TYPE="$(CMAKE_BUILD_TYPE)" \
+		-DC_SRC="$(C_SRC)" \
+		-DADBC_SRC="$(ADBC_SRC)" \
+		-DMIX_APP_PATH="$(MIX_APP_PATH)" \
+		-DPRIV_DIR="$(PRIV_DIR)" \
+		$(if $(ERTS_INCLUDE_DIR),-DERTS_INCLUDE_DIR="$(ERTS_INCLUDE_DIR)") \
 		$(CMAKE_CONFIGURE_FLAGS) $(CMAKE_ADBC_NIF_OPTIONS) "$(shell pwd)" && \
 	make "$(MAKE_BUILD_FLAGS)" && \
 	cp "$(CMAKE_ADBC_NIF_BUILD_DIR)/adbc_nif.so" "$(NIF_SO)"

c_src/adbc_arrow_schema.hpp

@@ -110,7 +110,7 @@ static int get_map_schema(ErlNifEnv *env, struct ArrowSchema * schema, uint64_t
         return erlang::nif::error(env, "invalid ArrowSchema (map), its entries n_children != 2");
     }
 
-    ERL_NIF_TERM key_schema, value_schema;
+    ERL_NIF_TERM key_schema = kAtomNil, value_schema = kAtomNil;
     int kv = 0;
     for (int64_t child_i = 0; child_i < 2; child_i++) {
         struct ArrowSchema * entry_schema = entries_schema->children[child_i];
@@ -141,7 +141,7 @@ static int get_map_schema(ErlNifEnv *env, struct ArrowSchema * schema, uint64_t
     }
 
     ERL_NIF_TERM map_kv_keys[] = { kAtomKey, kAtomValue };
-    ERL_NIF_TERM map_kv_values[] = { key_schema, value_schema };
+    ERL_NIF_TERM map_kv_values[] = { key_schema, value_schema };  // Now guaranteed to be initialized
     // only fail if there are duplicated keys
     // so we don't need to check the return value
     enif_make_map_from_arrays(env, map_kv_keys, map_kv_values, 2, &map_kv_schema);

c_src/adbc_nif_resource.hpp

@@ -61,7 +61,7 @@ template <typename T> struct NifRes {
       error = erlang::nif::error(env, "cannot allocate Nif resource\n");
       return res;
     }
-    memset(&res->val, 0, sizeof(val_type));
+    res->val = val_type();
     res->private_data = nullptr;
     return res;
   }