fix: use the correct schema for previously initialized tables on new connections to the database

Author: andinux

src/cloudsync.c

@@ -167,7 +167,7 @@ struct cloudsync_table_context {
     #endif
 
     char        **pk_name;                      // array of primary key names
-    
+
     // precompiled statements
     dbvm_t      *meta_pkexists_stmt;            // check if a primary key already exist in the augmented table
     dbvm_t      *meta_sentinel_update_stmt;     // update a local sentinel row
@@ -573,6 +573,13 @@ const char *cloudsync_schema (cloudsync_context *data) {
     return data->current_schema;
 }
 
+const char *cloudsync_table_schema (cloudsync_context *data, const char *table_name) {
+    cloudsync_table_context *table = table_lookup(data, table_name);
+    if (!table) return NULL;
+
+    return table->schema;
+}
+
 // MARK: - Table Utils -
 
 void table_pknames_free (char **names, int nrows) {
@@ -589,7 +596,7 @@ char *table_build_mergedelete_sql (cloudsync_table_context *table) {
     }
     #endif
 
-    return sql_build_delete_by_pk(table->context, table->name);
+    return sql_build_delete_by_pk(table->context, table->name, table->schema);
 }
 
 char *table_build_mergeinsert_sql (cloudsync_table_context *table, const char *colname) {
@@ -610,9 +617,9 @@ char *table_build_mergeinsert_sql (cloudsync_table_context *table, const char *c
 
     if (colname == NULL) {
         // is sentinel insert
-        sql = sql_build_insert_pk_ignore(table->context, table->name);
+        sql = sql_build_insert_pk_ignore(table->context, table->name, table->schema);
     } else {
-        sql = sql_build_upsert_pk_and_col(table->context, table->name, colname);
+        sql = sql_build_upsert_pk_and_col(table->context, table->name, colname, table->schema);
     }
     return sql;
 }
@@ -627,7 +634,7 @@ char *table_build_value_sql (cloudsync_table_context *table, const char *colname
     #endif
 
     // SELECT age FROM customers WHERE first_name=? AND last_name=?;
-    return sql_build_select_cols_by_pk(table->context, table->name, colname);
+    return sql_build_select_cols_by_pk(table->context, table->name, colname, table->schema);
 }
 
 cloudsync_table_context *table_create (cloudsync_context *data, const char *name, table_algo algo) {
@@ -639,7 +646,18 @@ cloudsync_table_context *table_create (cloudsync_context *data, const char *name
     table->context = data;
     table->algo = algo;
     table->name = cloudsync_string_dup_lowercase(name);
-    table->schema = (data->current_schema) ? cloudsync_string_dup(data->current_schema) : NULL;
+
+    // Detect schema from metadata table location. If metadata table doesn't
+    // exist yet (during initialization), fall back to cloudsync_schema() which
+    // returns the explicitly set schema or current_schema().
+    table->schema = database_table_schema(name);
+    if (!table->schema) {
+        const char *fallback_schema = cloudsync_schema(data);
+        if (fallback_schema) {
+            table->schema = cloudsync_string_dup(fallback_schema);
+        }
+    }
+
     if (!table->name) {
         cloudsync_memory_free(table);
         return NULL;
@@ -827,7 +845,7 @@ int table_add_stmts (cloudsync_table_context *table, int ncols) {
 
     // precompile the get column value statement
     if (ncols > 0) {
-        sql = sql_build_select_nonpk_by_pk(data, table->name);
+        sql = sql_build_select_nonpk_by_pk(data, table->name, table->schema);
         if (!sql) {rc = DBRES_NOMEM; goto cleanup;}
         DEBUG_SQL("real_col_values_stmt: %s", sql);
 
@@ -954,8 +972,14 @@ bool table_ensure_capacity (cloudsync_context *data) {
 
 bool table_add_to_context (cloudsync_context *data, table_algo algo, const char *table_name) {
     DEBUG_DBFUNCTION("cloudsync_context_add_table %s", table_name);
-    
-    // check if table is already in the global context and in that case just return
+
+    // Check if table already initialized in this connection's context.
+    // Note: This prevents same-connection duplicate initialization.
+    // SQLite clients cannot distinguish schemas, so having 'public.users'
+    // and 'auth.users' would cause sync ambiguity. Users should avoid
+    // initializing tables with the same name in different schemas.
+    // If two concurrent connections initialize tables with the same name
+    // in different schemas, the behavior is undefined.
     cloudsync_table_context *table = table_lookup(data, table_name);
     if (table) return true;
 
@@ -967,7 +991,7 @@ bool table_add_to_context (cloudsync_context *data, table_algo algo, const char
     if (!table) return false;
 
     // fill remaining metadata in the table
-    int count = database_count_pk(data, table_name, false);
+    int count = database_count_pk(data, table_name, false, table->schema);
     if (count < 0) {cloudsync_set_dberror(data); goto abort_add_table;}
     table->npks = count;
     if (table->npks == 0) {
@@ -979,7 +1003,7 @@ bool table_add_to_context (cloudsync_context *data, table_algo algo, const char
         #endif
     }
 
-    int ncols = database_count_nonpk(data, table_name);
+    int ncols = database_count_nonpk(data, table_name, table->schema);
     if (ncols < 0) {cloudsync_set_dberror(data); goto abort_add_table;}
     int rc = table_add_stmts(table, ncols);
     if (rc != DBRES_OK) goto abort_add_table;
@@ -997,8 +1021,11 @@ bool table_add_to_context (cloudsync_context *data, table_algo algo, const char
 
         table->col_value_stmt = (dbvm_t **)cloudsync_memory_alloc((uint64_t)(sizeof(void *) * ncols));
         if (!table->col_value_stmt) goto abort_add_table;
-        
-        char *sql = cloudsync_memory_mprintf(SQL_PRAGMA_TABLEINFO_LIST_NONPK_NAME_CID, table_name, table_name);
+
+        // Pass empty string when schema is NULL; SQL will fall back to current_schema()
+        const char *schema = table->schema ? table->schema : "";
+        char *sql = cloudsync_memory_mprintf(SQL_PRAGMA_TABLEINFO_LIST_NONPK_NAME_CID,
+                                              table_name, schema, table_name, schema);
         if (!sql) goto abort_add_table;
         rc = database_exec_callback(data, sql, table_add_to_context_cb, (void *)table);
         cloudsync_memory_free(sql);
@@ -1678,7 +1705,8 @@ int cloudsync_finalize_alter (cloudsync_context *data, cloudsync_table_context *
     } else {
         // compact meta-table
         // delete entries for removed columns
-        char *sql = cloudsync_memory_mprintf(SQL_CLOUDSYNC_DELETE_COLS_NOT_IN_SCHEMA_OR_PKCOL, table->meta_ref, table->name, CLOUDSYNC_TOMBSTONE_VALUE);
+        const char *schema = table->schema ? table->schema : "";
+        char *sql = sql_build_delete_cols_not_in_schema_query(schema, table->name, table->meta_ref, CLOUDSYNC_TOMBSTONE_VALUE);
         rc = database_exec(data, sql);
         cloudsync_memory_free(sql);
         if (rc != DBRES_OK) {
@@ -1688,7 +1716,7 @@ int cloudsync_finalize_alter (cloudsync_context *data, cloudsync_table_context *
 
         char buffer[1024];
         char *singlequote_escaped_table_name = sql_escape_name(table->name, buffer, sizeof(buffer));
-        sql = cloudsync_memory_mprintf(SQL_PRAGMA_TABLEINFO_PK_QUALIFIED_COLLIST_FMT, singlequote_escaped_table_name, singlequote_escaped_table_name);
+        sql = sql_build_pk_qualified_collist_query(schema, singlequote_escaped_table_name);
         if (!sql) {rc = DBRES_NOMEM; goto finalize;}
 
         char *pkclause = NULL;
@@ -1775,15 +1803,16 @@ int cloudsync_refill_metatable (cloudsync_context *data, const char *table_name)
     dbvm_t *vm = NULL;
     int64_t db_version = cloudsync_dbversion_next(data, CLOUDSYNC_VALUE_NOTSET);
     char *pkdecode = NULL;
-    
-    char *sql = cloudsync_memory_mprintf(SQL_PRAGMA_TABLEINFO_PK_COLLIST, table_name);
+
+    const char *schema = table->schema ? table->schema : "";
+    char *sql = sql_build_pk_collist_query(schema, table_name);
     char *pkclause_identifiers = NULL;
     int rc = database_select_text(data, sql, &pkclause_identifiers);
     cloudsync_memory_free(sql);
     if (rc != DBRES_OK) goto finalize;
     char *pkvalues_identifiers = (pkclause_identifiers) ? pkclause_identifiers : "rowid";
-    
-    sql = cloudsync_memory_mprintf(SQL_PRAGMA_TABLEINFO_PK_DECODE_SELECTLIST, table_name);
+
+    sql = sql_build_pk_decode_selectlist_query(schema, table_name);
     rc = database_select_text(data, sql, &pkdecode);
     cloudsync_memory_free(sql);
     if (rc != DBRES_OK) goto finalize;
@@ -2504,14 +2533,18 @@ int cloudsync_table_sanity_check (cloudsync_context *data, const char *name, boo
         return cloudsync_set_error(data, buffer, DBRES_ERROR);
     }
 
+    // check if already initialized
+    cloudsync_table_context *table = table_lookup(data, name);
+    if (table) return DBRES_OK;
+    
     // check if table exists
-    if (database_table_exists(data, name) == false) {
+    if (database_table_exists(data, name, cloudsync_schema(data)) == false) {
         snprintf(buffer, sizeof(buffer), "Table %s does not exist", name);
         return cloudsync_set_error(data, buffer, DBRES_ERROR);
     }
 
     // no more than 128 columns can be used as a composite primary key (SQLite hard limit)
-    int npri_keys = database_count_pk(data, name, false);
+    int npri_keys = database_count_pk(data, name, false, cloudsync_schema(data));
     if (npri_keys < 0) return cloudsync_set_dberror(data);
     if (npri_keys > 128) return cloudsync_set_error(data, "No more than 128 columns can be used to form a composite primary key", DBRES_ERROR);
 
@@ -2528,7 +2561,7 @@ int cloudsync_table_sanity_check (cloudsync_context *data, const char *name, boo
             // the affinity of a column is determined by the declared type of the column,
             // according to the following rules in the order shown:
             // 1. If the declared type contains the string "INT" then it is assigned INTEGER affinity.
-            int npri_keys_int = database_count_int_pk(data, name);
+            int npri_keys_int = database_count_int_pk(data, name, cloudsync_schema(data));
             if (npri_keys_int < 0) return cloudsync_set_dberror(data);
             if (npri_keys == npri_keys_int) {
                 snprintf(buffer, sizeof(buffer), "Table %s uses a single-column INTEGER primary key. For CRDT replication, primary keys must be globally unique. Consider using a TEXT primary key with UUIDs or ULID to avoid conflicts across nodes. If you understand the risk and still want to use this INTEGER primary key, set the third argument of the cloudsync_init function to 1 to skip this check.", name);
@@ -2540,7 +2573,7 @@ int cloudsync_table_sanity_check (cloudsync_context *data, const char *name, boo
 
     // if user declared explicit primary key(s) then make sure they are all declared as NOT NULL
     if (npri_keys > 0) {
-        int npri_keys_notnull = database_count_pk(data, name, true);
+        int npri_keys_notnull = database_count_pk(data, name, true, cloudsync_schema(data));
         if (npri_keys_notnull < 0) return cloudsync_set_dberror(data);
         if (npri_keys != npri_keys_notnull) {
             snprintf(buffer, sizeof(buffer), "All primary keys must be explicitly declared as NOT NULL (table %s)", name);
@@ -2550,7 +2583,7 @@ int cloudsync_table_sanity_check (cloudsync_context *data, const char *name, boo
 
     // check for columns declared as NOT NULL without a DEFAULT value.
     // Otherwise, col_merge_stmt would fail if changes to other columns are inserted first.
-    int n_notnull_nodefault = database_count_notnull_without_default(data, name);
+    int n_notnull_nodefault = database_count_notnull_without_default(data, name, cloudsync_schema(data));
     if (n_notnull_nodefault < 0) return cloudsync_set_dberror(data);
     if (n_notnull_nodefault > 0) {
         snprintf(buffer, sizeof(buffer), "All non-primary key columns declared as NOT NULL must have a DEFAULT value. (table %s)", name);

src/cloudsync.h

@@ -17,7 +17,7 @@
 extern "C" {
 #endif
 
-#define CLOUDSYNC_VERSION                       "0.9.82"
+#define CLOUDSYNC_VERSION                       "0.9.83"
 #define CLOUDSYNC_MAX_TABLENAME_LEN             512
 
 #define CLOUDSYNC_VALUE_NOTSET                  -1
@@ -81,6 +81,7 @@ int cloudsync_commit_hook (void *ctx);
 void cloudsync_rollback_hook (void *ctx);
 void cloudsync_set_schema (cloudsync_context *data, const char *schema);
 const char *cloudsync_schema (cloudsync_context *data);
+const char *cloudsync_table_schema (cloudsync_context *data, const char *table_name);
 
 // Payload
 int    cloudsync_payload_apply (cloudsync_context *data, const char *payload, int blen, int *nrows);

src/database.h

@@ -66,7 +66,7 @@ int  database_select_text (cloudsync_context *data, const char *sql, char **valu
 int  database_select_blob (cloudsync_context *data, const char *sql, char **value, int64_t *value_len);
 int  database_select_blob_2int (cloudsync_context *data, const char *sql, char **value, int64_t *value_len, int64_t *value2, int64_t *value3);
 int  database_write (cloudsync_context *data, const char *sql, const char **values, DBTYPE types[], int lens[], int count);
-bool database_table_exists (cloudsync_context *data, const char *table_name);
+bool database_table_exists (cloudsync_context *data, const char *table_name, const char *schema);
 bool database_internal_table_exists (cloudsync_context *data, const char *name);
 bool database_trigger_exists (cloudsync_context *data, const char *table_name);
 int  database_create_metatable (cloudsync_context *data, const char *table_name);
@@ -75,10 +75,10 @@ int  database_delete_triggers (cloudsync_context *data, const char *table_name);
 int  database_pk_names (cloudsync_context *data, const char *table_name, char ***names, int *count);
 int  database_cleanup (cloudsync_context *data);
 
-int database_count_pk (cloudsync_context *data, const char *table_name, bool not_null);
-int database_count_nonpk (cloudsync_context *data, const char *table_name);
-int database_count_int_pk (cloudsync_context *data, const char *table_name);
-int database_count_notnull_without_default (cloudsync_context *data, const char *table_name);
+int database_count_pk (cloudsync_context *data, const char *table_name, bool not_null, const char *schema);
+int database_count_nonpk (cloudsync_context *data, const char *table_name, const char *schema);
+int database_count_int_pk (cloudsync_context *data, const char *table_name, const char *schema);
+int database_count_notnull_without_default (cloudsync_context *data, const char *table_name, const char *schema);
 
 int64_t  database_schema_version (cloudsync_context *data);
 uint64_t database_schema_hash (cloudsync_context *data);
@@ -139,13 +139,18 @@ uint64_t dbmem_size (void *ptr);
 // SQL
 char *sql_build_drop_table (const char *table_name, char *buffer, int bsize, bool is_meta);
 char *sql_escape_name (const char *name, char *buffer, size_t bsize);
-char *sql_build_select_nonpk_by_pk (cloudsync_context *data, const char *table_name);
-char *sql_build_delete_by_pk (cloudsync_context *data, const char *table_name);
-char *sql_build_insert_pk_ignore (cloudsync_context *data, const char *table_name);
-char *sql_build_upsert_pk_and_col (cloudsync_context *data, const char *table_name, const char *colname);
-char *sql_build_select_cols_by_pk (cloudsync_context *data, const char *table_name, const char *colname);
+char *sql_build_select_nonpk_by_pk (cloudsync_context *data, const char *table_name, const char *schema);
+char *sql_build_delete_by_pk (cloudsync_context *data, const char *table_name, const char *schema);
+char *sql_build_insert_pk_ignore (cloudsync_context *data, const char *table_name, const char *schema);
+char *sql_build_upsert_pk_and_col (cloudsync_context *data, const char *table_name, const char *colname, const char *schema);
+char *sql_build_select_cols_by_pk (cloudsync_context *data, const char *table_name, const char *colname, const char *schema);
 char *sql_build_rekey_pk_and_reset_version_except_col (cloudsync_context *data, const char *table_name, const char *except_col);
+char *sql_build_delete_cols_not_in_schema_query(const char *schema, const char *table_name, const char *meta_ref, const char *pkcol);
+char *sql_build_pk_collist_query(const char *schema, const char *table_name);
+char *sql_build_pk_decode_selectlist_query(const char *schema, const char *table_name);
+char *sql_build_pk_qualified_collist_query(const char *schema, const char *table_name);
 
+char *database_table_schema(const char *table_name);
 char *database_build_meta_ref(const char *schema, const char *table_name);
 char *database_build_base_ref(const char *schema, const char *table_name);
 

src/postgresql/cloudsync--1.0.sql

@@ -271,3 +271,8 @@ RETURNS text
 AS 'MODULE_PATHNAME', 'pg_cloudsync_schema'
 LANGUAGE C VOLATILE;
 
+-- Get current schema name (if any)
+CREATE OR REPLACE FUNCTION cloudsync_table_schema(table_name text)
+RETURNS text
+AS 'MODULE_PATHNAME', 'pg_cloudsync_table_schema'
+LANGUAGE C VOLATILE;

src/postgresql/cloudsync_postgresql.c

@@ -1118,7 +1118,7 @@ Datum cloudsync_insert (PG_FUNCTION_ARGS) {
         if (!table) {
             char meta_name[1024];
             snprintf(meta_name, sizeof(meta_name), "%s_cloudsync", table_name);
-            if (!database_table_exists(data, meta_name)) {
+            if (!database_table_exists(data, meta_name, cloudsync_schema(data))) {
                 ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("Unable to retrieve table name %s in cloudsync_insert", table_name)));
             }
 
@@ -1212,7 +1212,7 @@ Datum cloudsync_delete (PG_FUNCTION_ARGS) {
         if (!table) {
             char meta_name[1024];
             snprintf(meta_name, sizeof(meta_name), "%s_cloudsync", table_name);
-            if (!database_table_exists(data, meta_name)) {
+            if (!database_table_exists(data, meta_name, cloudsync_schema(data))) {
                 ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("Unable to retrieve table name %s in cloudsync_delete", table_name)));
             }
 
@@ -1403,7 +1403,7 @@ Datum cloudsync_update_finalfn (PG_FUNCTION_ARGS) {
         if (!table) {
             char meta_name[1024];
             snprintf(meta_name, sizeof(meta_name), "%s_cloudsync", table_name);
-            if (!database_table_exists(data, meta_name)) {
+            if (!database_table_exists(data, meta_name, cloudsync_schema(data))) {
                 ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("Unable to retrieve table name %s in cloudsync_update", table_name)));
             }
 
@@ -1543,6 +1543,23 @@ Datum pg_cloudsync_schema (PG_FUNCTION_ARGS) {
     PG_RETURN_TEXT_P(cstring_to_text(schema));
 }
 
+PG_FUNCTION_INFO_V1(pg_cloudsync_table_schema);
+Datum pg_cloudsync_table_schema (PG_FUNCTION_ARGS) {
+    if (PG_ARGISNULL(0)) {
+        ereport(ERROR, (errcode(ERRCODE_INVALID_PARAMETER_VALUE), errmsg("table_name cannot be NULL")));
+    }
+
+    const char *table_name = text_to_cstring(PG_GETARG_TEXT_PP(0));
+    cloudsync_context *data = get_cloudsync_context();
+    const char *schema = cloudsync_table_schema(data, table_name);
+
+    if (!schema) {
+        PG_RETURN_NULL();
+    }
+    
+    PG_RETURN_TEXT_P(cstring_to_text(schema));
+}
+
 // MARK: - Changes -
 
 // Encode a single value using cloudsync pk encoding