Added support for DuckDB on the Edge

Everything works except triggers that are not supported by DuckDB

Author: marcobambini

src/cloudsync.c

@@ -113,7 +113,8 @@ struct cloudsync_context {
     int         debug;
     bool        merge_equal_values;
     void        *aux_data;
-    
+    int         step_depth;
+
     // stmts and context values
     dbvm_t      *schema_version_stmt;
     dbvm_t      *data_version_stmt;
@@ -430,6 +431,23 @@ int64_t cloudsync_pk_context_dbversion (cloudsync_pk_decode_bind_context *ctx) {
     return ctx->db_version;
 }
 
+int64_t cloudsync_pk_context_colversion (cloudsync_pk_decode_bind_context *ctx) {
+    return ctx->col_version;
+}
+
+int64_t cloudsync_pk_context_seq (cloudsync_pk_decode_bind_context *ctx) {
+    return ctx->seq;
+}
+
+void *cloudsync_pk_context_siteid (cloudsync_pk_decode_bind_context *ctx, int64_t *siteid_len) {
+    *siteid_len = ctx->site_id_len;
+    return (void *)ctx->site_id;
+}
+
+dbvm_t *cloudsync_pk_context_vm (cloudsync_pk_decode_bind_context *ctx) {
+    return ctx->vm;
+}
+
 // MARK: - CloudSync Context -
 
 int cloudsync_insync (cloudsync_context *data) {
@@ -564,6 +582,14 @@ void cloudsync_set_auxdata (cloudsync_context *data, void *xdata) {
     data->aux_data = xdata;
 }
 
+int cloudsync_step_depth (cloudsync_context *data) {
+    return data->step_depth;
+}
+
+void cloudsync_set_step_depth (cloudsync_context *data, int depth) {
+    data->step_depth = depth;
+}
+
 void cloudsync_set_schema (cloudsync_context *data, const char *schema) {
     if (data->current_schema && schema && strcmp(data->current_schema, schema) == 0) return;
     if (data->current_schema) cloudsync_memory_free(data->current_schema);
@@ -1011,6 +1037,11 @@ bool table_add_to_context (cloudsync_context *data, table_algo algo, const char
         table->npks = 1; // rowid
         #endif
     }
+
+    // NOTE: pk_name array is populated lazily (e.g. in DuckDB's
+    // BuildChangesSelectSQL) rather than here, because table_add_to_context
+    // can be called from database_exec_callback (settings load) where
+    // issuing another query on the same connection would recurse.
 
     int ncols = database_count_nonpk(data, table_name, table->schema);
     if (ncols < 0) {cloudsync_set_dberror(data); goto abort_add_table;}
@@ -1093,6 +1124,23 @@ const char *table_colname (cloudsync_table_context *table, int index) {
     return table->col_name[index];
 }
 
+const char *table_name (cloudsync_table_context *table) {
+    return table->name;
+}
+
+const char *table_metaref (cloudsync_table_context *table) {
+    return table->meta_ref;
+}
+
+int cloudsync_table_count (cloudsync_context *data) {
+    return data->tables_count;
+}
+
+cloudsync_table_context *cloudsync_table_at (cloudsync_context *data, int index) {
+    if (index < 0 || index >= data->tables_count) return NULL;
+    return data->tables[index];
+}
+
 bool table_pk_exists (cloudsync_table_context *table, const char *value, size_t len) {
     // check if a row with the same primary key already exists
     // if so, this means the row might have been previously deleted (sentinel)
@@ -2224,6 +2272,42 @@ int cloudsync_payload_encode_step (cloudsync_payload_context *payload, cloudsync
     return DBRES_OK;
 }
 
+int cloudsync_payload_encode_combine (cloudsync_payload_context *target, cloudsync_payload_context *source) {
+    if (!source || source->nrows == 0) return DBRES_OK;
+    if (!target) return DBRES_ERROR;
+
+    // If target is empty, just take over source's data
+    if (target->nrows == 0) {
+        target->buffer = source->buffer;
+        target->bsize = source->bsize;
+        target->balloc = source->balloc;
+        target->bused = source->bused;
+        target->nrows = source->nrows;
+        target->ncols = source->ncols;
+        // Clear source so it won't free the buffer
+        source->buffer = NULL;
+        source->bsize = 0;
+        source->balloc = 0;
+        source->bused = 0;
+        source->nrows = 0;
+        return DBRES_OK;
+    }
+
+    // Append source buffer to target
+    size_t needed = target->bused + source->bused;
+    if (needed > target->balloc) {
+        size_t new_alloc = needed * 2;
+        char *new_buf = cloudsync_memory_realloc(target->buffer, (uint64_t)new_alloc);
+        if (!new_buf) return DBRES_NOMEM;
+        target->buffer = new_buf;
+        target->balloc = new_alloc;
+    }
+    memcpy(target->buffer + target->bused, source->buffer, source->bused);
+    target->bused += source->bused;
+    target->nrows += source->nrows;
+    return DBRES_OK;
+}
+
 int cloudsync_payload_encode_final (cloudsync_payload_context *payload, cloudsync_context *data) {
     DEBUG_FUNCTION("cloudsync_payload_encode_final");
 
@@ -2559,7 +2643,7 @@ int cloudsync_payload_get (cloudsync_context *data, char **blob, int *blob_size,
     // retrieve BLOB
     char sql[1024];
     snprintf(sql, sizeof(sql), "WITH max_db_version AS (SELECT MAX(db_version) AS max_db_version FROM cloudsync_changes WHERE site_id=cloudsync_siteid()) "
-                               "SELECT * FROM (SELECT cloudsync_payload_encode(tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq) AS payload, max_db_version AS max_db_version, MAX(IIF(db_version = max_db_version, seq, 0)) FROM cloudsync_changes, max_db_version WHERE site_id=cloudsync_siteid() AND (db_version>%d OR (db_version=%d AND seq>%d))) WHERE payload IS NOT NULL", *db_version, *db_version, *seq);
+                               "SELECT * FROM (SELECT cloudsync_payload_encode(tbl, pk, col_name, col_value, col_version, db_version, site_id, cl, seq) AS payload, MAX(max_db_version) AS max_db_version, MAX(CASE WHEN db_version = max_db_version THEN seq ELSE 0 END) FROM cloudsync_changes, max_db_version WHERE site_id=cloudsync_siteid() AND (db_version>%d OR (db_version=%d AND seq>%d))) WHERE payload IS NOT NULL", *db_version, *db_version, *seq);
 
     int64_t len = 0;
     int rc = database_select_blob_2int(data, sql, blob, &len, new_db_version, new_seq);
@@ -2726,6 +2810,8 @@ int cloudsync_cleanup_internal (cloudsync_context *data, cloudsync_table_context
     return DBRES_OK;
 }
 
+static void cloudsync_finalize_context_stmts (cloudsync_context *data);
+
 int cloudsync_cleanup (cloudsync_context *data, const char *table_name) {
     cloudsync_table_context *table = table_lookup(data, table_name);
     if (!table) return DBRES_OK;
@@ -2742,6 +2828,7 @@ int cloudsync_cleanup (cloudsync_context *data, const char *table_name) {
         // cleanup database on last table
         cloudsync_reset_siteid(data);
         dbutils_settings_cleanup(data);
+        cloudsync_finalize_context_stmts(data);
     } else {
         if (database_internal_table_exists(data, CLOUDSYNC_TABLE_SETTINGS_NAME) == true) {
             cloudsync_update_schema_hash(data);
@@ -2755,38 +2842,38 @@ int cloudsync_cleanup_all (cloudsync_context *data) {
     return database_cleanup(data);
 }
 
+// Finalize and NULL out all context-level prepared statements and cached schema.
+// Shared by cloudsync_cleanup (last table) and cloudsync_terminate.
+static void cloudsync_finalize_context_stmts (cloudsync_context *data) {
+    if (data->schema_version_stmt) { databasevm_finalize(data->schema_version_stmt); data->schema_version_stmt = NULL; }
+    if (data->data_version_stmt) { databasevm_finalize(data->data_version_stmt); data->data_version_stmt = NULL; }
+    if (data->db_version_stmt) { databasevm_finalize(data->db_version_stmt); data->db_version_stmt = NULL; }
+    if (data->getset_siteid_stmt) { databasevm_finalize(data->getset_siteid_stmt); data->getset_siteid_stmt = NULL; }
+    if (data->current_schema) { cloudsync_memory_free(data->current_schema); data->current_schema = NULL; }
+}
+
 int cloudsync_terminate (cloudsync_context *data) {
     // can't use for/loop here because data->tables_count is changed by table_remove
     while (data->tables_count > 0) {
         cloudsync_table_context *t = data->tables[data->tables_count - 1];
         table_remove(data, t);
         table_free(t);
     }
-    
-    if (data->schema_version_stmt) databasevm_finalize(data->schema_version_stmt);
-    if (data->data_version_stmt) databasevm_finalize(data->data_version_stmt);
-    if (data->db_version_stmt) databasevm_finalize(data->db_version_stmt);
-    if (data->getset_siteid_stmt) databasevm_finalize(data->getset_siteid_stmt);
-    if (data->current_schema) cloudsync_memory_free(data->current_schema);
-    
-    data->schema_version_stmt = NULL;
-    data->data_version_stmt = NULL;
-    data->db_version_stmt = NULL;
-    data->getset_siteid_stmt = NULL;
-    data->current_schema = NULL;
-    
+
+    cloudsync_finalize_context_stmts(data);
+
     // reset the site_id so the cloudsync_context_init will be executed again
     // if any other cloudsync function is called after terminate
     data->site_id[0] = 0;
-    
+
     return 1;
 }
 
 int cloudsync_init_table (cloudsync_context *data, const char *table_name, const char *algo_name, bool skip_int_pk_check) {
     // sanity check table and its primary key(s)
     int rc = cloudsync_table_sanity_check(data, table_name, skip_int_pk_check);
     if (rc != DBRES_OK) return rc;
-    
+
     // init cloudsync_settings
     if (cloudsync_context_init(data) == NULL) {
         return cloudsync_set_error(data, "Unable to initialize cloudsync context", DBRES_MISUSE);
@@ -2812,7 +2899,7 @@ int cloudsync_init_table (cloudsync_context *data, const char *table_name, const
 
     // check if table name was already augmented
     table_algo algo_current = dbutils_table_settings_get_algo(data, table_name);
-    
+
     // sanity check algorithm
     if ((algo_new == algo_current) && (algo_current != table_algo_none)) {
         // if table algorithms and the same and not none, do nothing
@@ -2845,16 +2932,16 @@ int cloudsync_init_table (cloudsync_context *data, const char *table_name, const
     // check triggers
     rc = database_create_triggers(data, table_name, algo_new, init_filter);
     if (rc != DBRES_OK) return cloudsync_set_error(data, "An error occurred while creating triggers", DBRES_MISUSE);
-    
+
     // check meta-table
     rc = database_create_metatable(data, table_name);
     if (rc != DBRES_OK) return cloudsync_set_error(data, "An error occurred while creating metatable", DBRES_MISUSE);
-    
+
     // add prepared statements
     if (cloudsync_add_dbvms(data) != DBRES_OK) {
         return cloudsync_set_error(data, "An error occurred while trying to compile prepared SQL statements", DBRES_MISUSE);
     }
-    
+
     // add table to in-memory data context
     if (table_add_to_context(data, algo_new, table_name) == false) {
         char buffer[1024];
@@ -2865,6 +2952,6 @@ int cloudsync_init_table (cloudsync_context *data, const char *table_name, const
     if (cloudsync_refill_metatable(data, table_name) != DBRES_OK) {
         return cloudsync_set_error(data, "An error occurred while trying to fill the augmented table", DBRES_MISUSE);
     }
-        
+
     return DBRES_OK;
 }

src/cloudsync.h

@@ -61,6 +61,7 @@ int64_t cloudsync_dbversion_next (cloudsync_context *data, int64_t merging_versi
 int64_t cloudsync_dbversion (cloudsync_context *data);
 void cloudsync_update_schema_hash (cloudsync_context *data);
 int cloudsync_dbversion_check_uptodate (cloudsync_context *data);
+int cloudsync_dbversion_rerun (cloudsync_context *data);
 bool cloudsync_config_exists (cloudsync_context *data);
 dbvm_t *cloudsync_colvalue_stmt (cloudsync_context *data, const char *tbl_name, bool *persistent);
 
@@ -72,6 +73,8 @@ int cloudsync_commit_alter (cloudsync_context *data, const char *table_name);
 void *cloudsync_db (cloudsync_context *data);
 void *cloudsync_auxdata (cloudsync_context *data);
 void cloudsync_set_auxdata (cloudsync_context *data, void *xdata);
+int cloudsync_step_depth (cloudsync_context *data);
+void cloudsync_set_step_depth (cloudsync_context *data, int depth);
 int cloudsync_set_error (cloudsync_context *data, const char *err_user, int err_code);
 int cloudsync_set_dberror (cloudsync_context *data);
 const char *cloudsync_errmsg (cloudsync_context *data);
@@ -86,6 +89,7 @@ const char *cloudsync_table_schema (cloudsync_context *data, const char *table_n
 // Payload
 int    cloudsync_payload_apply (cloudsync_context *data, const char *payload, int blen, int *nrows);
 int    cloudsync_payload_encode_step (cloudsync_payload_context *payload, cloudsync_context *data, int argc, dbvalue_t **argv);
+int    cloudsync_payload_encode_combine (cloudsync_payload_context *target, cloudsync_payload_context *source);
 int    cloudsync_payload_encode_final (cloudsync_payload_context *payload, cloudsync_context *data);
 char  *cloudsync_payload_blob (cloudsync_payload_context *payload, int64_t *blob_size, int64_t *nrows);
 size_t cloudsync_payload_context_size (size_t *header_size);
@@ -98,6 +102,10 @@ void *table_column_lookup (cloudsync_table_context *table, const char *col_name,
 bool table_enabled (cloudsync_table_context *table);
 void table_set_enabled (cloudsync_table_context *table, bool value);
 bool table_add_to_context (cloudsync_context *data, table_algo algo, const char *table_name);
+const char *table_name (cloudsync_table_context *table);
+const char *table_metaref (cloudsync_table_context *table);
+int cloudsync_table_count (cloudsync_context *data);
+cloudsync_table_context *cloudsync_table_at (cloudsync_context *data, int index);
 bool table_pk_exists (cloudsync_table_context *table, const char *value, size_t len);
 int table_count_cols (cloudsync_table_context *table);
 int table_count_pks (cloudsync_table_context *table);
@@ -130,6 +138,10 @@ void *cloudsync_pk_context_pk (cloudsync_pk_decode_bind_context *ctx, int64_t *p
 char *cloudsync_pk_context_colname (cloudsync_pk_decode_bind_context *ctx, int64_t *colname_len);
 int64_t cloudsync_pk_context_cl (cloudsync_pk_decode_bind_context *ctx);
 int64_t cloudsync_pk_context_dbversion (cloudsync_pk_decode_bind_context *ctx);
+int64_t cloudsync_pk_context_colversion (cloudsync_pk_decode_bind_context *ctx);
+int64_t cloudsync_pk_context_seq (cloudsync_pk_decode_bind_context *ctx);
+void *cloudsync_pk_context_siteid (cloudsync_pk_decode_bind_context *ctx, int64_t *siteid_len);
+dbvm_t *cloudsync_pk_context_vm (cloudsync_pk_decode_bind_context *ctx);
 
 #ifdef __cplusplus
 }

src/duckdb/CMakeLists.txt

@@ -0,0 +1,49 @@
+cmake_minimum_required(VERSION 3.5)
+
+set(TARGET_NAME cloudsync)
+
+set(EXTENSION_NAME ${TARGET_NAME}_extension)
+set(LOADABLE_EXTENSION_NAME ${TARGET_NAME}_loadable_extension)
+
+project(${TARGET_NAME})
+include_directories(src/include)
+# Resolve the real source directory (in case we're accessed via symlink)
+get_filename_component(CLOUDSYNC_REAL_DIR "${CMAKE_CURRENT_LIST_DIR}" REALPATH)
+get_filename_component(CLOUDSYNC_PARENT_DIR "${CLOUDSYNC_REAL_DIR}/.." REALPATH)
+
+include_directories(${CLOUDSYNC_PARENT_DIR})
+
+# CloudSync core sources (shared with SQLite/PostgreSQL)
+set(CLOUDSYNC_CORE_SOURCES
+    ${CLOUDSYNC_PARENT_DIR}/cloudsync.c
+    ${CLOUDSYNC_PARENT_DIR}/dbutils.c
+    ${CLOUDSYNC_PARENT_DIR}/lz4.c
+    ${CLOUDSYNC_PARENT_DIR}/pk.c
+    ${CLOUDSYNC_PARENT_DIR}/utils.c
+)
+
+# DuckDB-specific sources
+set(EXTENSION_SOURCES
+    cloudsync_duckdb.cpp
+    database_duckdb.cpp
+    sql_duckdb.c
+    ${CLOUDSYNC_CORE_SOURCES}
+)
+
+build_static_extension(${TARGET_NAME} ${EXTENSION_SOURCES})
+build_loadable_extension(${TARGET_NAME} " " ${EXTENSION_SOURCES})
+
+# Link Security framework on macOS (for SecRandomCopyBytes in utils.c)
+if(APPLE)
+    find_library(SECURITY_FRAMEWORK Security)
+    if(SECURITY_FRAMEWORK)
+        target_link_libraries(${LOADABLE_EXTENSION_NAME} ${SECURITY_FRAMEWORK})
+        target_link_libraries(${EXTENSION_NAME} ${SECURITY_FRAMEWORK})
+    endif()
+endif()
+
+install(
+  TARGETS ${EXTENSION_NAME}
+  EXPORT "${DUCKDB_EXPORT_SET}"
+  LIBRARY DESTINATION "${INSTALL_LIB_DIR}"
+  ARCHIVE DESTINATION "${INSTALL_LIB_DIR}")