From 19bd1f8cc2247cf7da678909a83abfe9861d5f01 Mon Sep 17 00:00:00 2001 From: Dorin Hogea Date: Wed, 10 Jun 2026 14:17:39 -0400 Subject: [PATCH 1/2] Enforce unique indexes across all shards of a TRUNCATE time partition Time partitions spread data across multiple shard tables to implement data retention. Until now, unique indexes were only enforced within the shard being written, making it possible to insert the same key into different shards without conflict. A new 'partition_unique' tunable enables cross-shard unique enforcement for TRUNCATE-rollout partitions. When on, any write that would violate a unique constraint in any sibling shard is rejected with the same error that a within-shard violation produces. The enforcement covers inserts and updates that change a key column. The feature carries a measurable write cost proportional to the number of sibling shards and unique indexes, driven by the extra index lookups per write and the larger BDB lock footprint per transaction. A new 'partition_unique_debug' tunable traces the full lifecycle: OSQL_PARTITION_SHARDS send on the replicant, receive/store and free on the master, and each cross-shard index probe. Known limitations: - Enabling the tunable does not validate pre-existing data. - UPSERT is not supported on time partitions when enabled (enforced at write time with a clear error). - ON UPDATE CASCADE is handled master-side in constraints.c without going through the OSQL stream, so cross-shard enforcement does not apply to cascaded key changes until that path is updated. Bug fixes applied during review: - Endian-safe nshards in OSQL_PARTITION_SHARDS wire format (htonl/ntohl) - Bounds-check nshards and validate shard name NUL terminators in receiver - Fail writes on malloc failure in timepart_get_shard_names via errstat - Remove redundant forward declaration of check_cross_shard_unique Refactoring: - Consolidate 7 copy-pasted reqerrstr blocks into reqerrstr_dup_key / reqerrstr_uncommittable_dup helpers in indices.c Tests: add partition_unique_check correctness test and partition_unique_perf performance test. Signed-off-by: Dorin Hogea --- db/comdb2.c | 2 + db/comdb2.h | 6 + db/constraints.c | 25 +- db/db_tunables.h | 9 + db/indices.c | 123 ++++- db/indices.h | 5 + db/osqlblockproc.c | 2 + db/osqlcomm.c | 132 +++++- db/osqlcomm.h | 9 + db/osqlrpltypes.h | 3 +- db/osqlsqlthr.c | 55 +++ db/sql.h | 1 + db/sqlglue.c | 8 + db/toblock.c | 7 +- db/views.c | 32 ++ db/views.h | 9 + docs/pages/programming/timepart.md | 4 +- tests/partition_unique_check.test/Makefile | 9 + tests/partition_unique_check.test/runit | 421 ++++++++++++++++++ tests/partition_unique_perf.test/Makefile | 9 + tests/partition_unique_perf.test/README | 28 ++ tests/partition_unique_perf.test/runit | 369 +++++++++++++++ tests/tunables.test/t00_all_tunables.expected | 1 + 23 files changed, 1231 insertions(+), 38 deletions(-) create mode 100644 tests/partition_unique_check.test/Makefile create mode 100755 tests/partition_unique_check.test/runit create mode 100644 tests/partition_unique_perf.test/Makefile create mode 100644 tests/partition_unique_perf.test/README create mode 100755 tests/partition_unique_perf.test/runit diff --git a/db/comdb2.c b/db/comdb2.c index 9239cc6b78..5fead094ca 100644 --- a/db/comdb2.c +++ b/db/comdb2.c @@ -680,6 +680,8 @@ int gbl_requeue_on_tran_dispatch = 1; int gbl_crc32c = 1; int gbl_repscore = 0; int gbl_surprise = 1; // TODO: change name to something else +int gbl_partition_unique = 0; +int gbl_partition_unique_debug = 0; int gbl_check_wrong_db = 1; int gbl_broken_max_rec_sz = 0; int gbl_private_blkseq = 1; diff --git a/db/comdb2.h b/db/comdb2.h index 0b3ef2141f..c44d6f743d 100644 --- a/db/comdb2.h +++ b/db/comdb2.h @@ -1432,6 +1432,10 @@ struct ireq { uint8_t **idxInsert; uint8_t **idxDelete; + /* shard dbtables for cross-shard unique check (TRUNCATE partitions) */ + struct dbtable **partition_shards; + int npartition_shards; + /* osql prefault step index */ int *osql_step_ix; @@ -3549,6 +3553,8 @@ extern int gbl_repscore; extern int gbl_max_verify_retries; extern int gbl_surprise; +extern int gbl_partition_unique; +extern int gbl_partition_unique_debug; extern int gbl_check_wrong_db; diff --git a/db/constraints.c b/db/constraints.c index 1a628bedc4..03630b5f9a 100644 --- a/db/constraints.c +++ b/db/constraints.c @@ -243,6 +243,11 @@ static inline void free_cached_delayed_indexes(struct ireq *iq) free(iq->idxDelete); iq->idxInsert = iq->idxDelete = NULL; } + if (gbl_partition_unique_debug && iq->partition_shards) + logmsg(LOGMSG_USER, "%s: [master] freeing partition_shards nshards=%d\n", __func__, iq->npartition_shards); + free(iq->partition_shards); + iq->partition_shards = NULL; + iq->npartition_shards = 0; } enum ct_etype { CTE_ADD = 1, CTE_DEL, CTE_UPD }; @@ -1408,9 +1413,19 @@ int delayed_key_adds(struct ireq *iq, void *trans, int *blkpos, int *ixout, /* light the prefault kill bit for this subop - newkeys */ prefault_kill_bits(iq, doidx, PFRQ_NEWKEY); + /* cross-shard unique check for TRUNCATE partitions */ + rc = check_cross_shard_unique(iq, trans, doidx, key, ixkeylen, errout); + if (rc == RC_INTERNAL_RETRY) { + *blkpos = curop->blkpos; + close_constraint_table_cursor(cur); + free_cached_delayed_indexes(iq); + ERR(rc, "deadlock", 0); + } + /* add the key */ - rc = ix_addk(iq, trans, key, doidx, genid, addrrn, od_dta_tail, - od_tail_len, ix_isnullk(iq->usedb, key, doidx)); + if (!rc) + rc = ix_addk(iq, trans, key, doidx, genid, addrrn, od_dta_tail, od_tail_len, + ix_isnullk(iq->usedb, key, doidx)); if (iq->vfy_idx_track) { dup_txn_insert = track_record_index(iq, doidx, key, ixkeylen); @@ -1431,11 +1446,7 @@ int delayed_key_adds(struct ireq *iq, void *trans, int *blkpos, int *ixout, *errout = OP_FAILED_VERIFY; rc = ERR_VERIFY; } else { - reqerrstr(iq, COMDB2_CSTRT_RC_DUP, - "add key constraint duplicate key '%s' on table " - "'%s' index %d", - get_keynm_from_db_idx(iq->usedb, doidx), - iq->usedb->tablename, doidx); + reqerrstr_dup_key(iq, iq->usedb, doidx); *errout = OP_FAILED_UNIQ; } diff --git a/db/db_tunables.h b/db/db_tunables.h index 7b01c0f83e..97f582fcfa 100644 --- a/db/db_tunables.h +++ b/db/db_tunables.h @@ -2573,6 +2573,15 @@ REGISTER_TUNABLE("transaction_grace_period", "Time to wait for connections with pending transactions to go away on exit. (Default: 60)", TUNABLE_INTEGER, &gbl_transaction_grace_period, 0, NULL, NULL, NULL, NULL); +REGISTER_TUNABLE("partition_unique", + "Enforce unique index constraints across all shards of a TRUNCATE time partition (Default: off). " + "Does not validate pre-existing data; cross-shard duplicates present before enabling this tunable " + "are the user's responsibility to detect and resolve.", + TUNABLE_BOOLEAN, &gbl_partition_unique, NOARG, NULL, NULL, NULL, NULL); +REGISTER_TUNABLE("partition_unique_debug", + "Enable debug tracing for partition_unique cross-shard unique enforcement (Default: off). " + "Logs packet build/send on replicant, receive/store and free on master, and each cross-shard check.", + TUNABLE_BOOLEAN, &gbl_partition_unique_debug, NOARG, NULL, NULL, NULL, NULL); REGISTER_TUNABLE("partition_sc_reorder", "If the schema change is serialized for a partition, run current shard last", TUNABLE_BOOLEAN, &gbl_partition_sc_reorder, 0, NULL, NULL, NULL, NULL); REGISTER_TUNABLE("partition_retroactively", diff --git a/db/indices.c b/db/indices.c index 6332f3dca9..124c662507 100644 --- a/db/indices.c +++ b/db/indices.c @@ -349,7 +349,6 @@ static inline void append_genid_to_key(dtikey_t *ditk, int ixkeylen) goto done; \ } while (0); - int add_record_indices(struct ireq *iq, void *trans, blob_buffer_t *blobs, size_t maxblobs, int *opfailcode, int *ixfailnum, int *rrn, unsigned long long *genid, @@ -440,6 +439,14 @@ int add_record_indices(struct ireq *iq, void *trans, blob_buffer_t *blobs, gbl_osqlpf_step[*(iq->osql_step_ix)].step += 2; if (reorder) { + /* cross-shard unique check before deferring the key */ + rc = check_cross_shard_unique(iq, trans, ixnum, key, ixkeylen, opfailcode); + if (rc == RC_INTERNAL_RETRY) { + ERR(rc, "deadlock", 0); + } else if (rc) { + *ixfailnum = ixnum; + ERR(rc, "partition unique constraint violation", 0); + } // if not datacopy, no need to save od_dta_tail void *data = NULL; int datalen = 0; @@ -488,6 +495,15 @@ int add_record_indices(struct ireq *iq, void *trans, blob_buffer_t *blobs, vgenid = 0; // no need to verify again } + /* cross-shard unique check for TRUNCATE partitions */ + rc = check_cross_shard_unique(iq, trans, ixnum, key, ixkeylen, opfailcode); + if (rc == RC_INTERNAL_RETRY) { + ERR(rc, "deadlock", 0); + } else if (rc) { + *ixfailnum = ixnum; + ERR(rc, "partition unique constraint violation", 0); + } + /* add the key */ rc = ix_addk(iq, trans, key, ixnum, *genid, *rrn, od_dta_tail, od_tail_len, isnullk); @@ -533,6 +549,71 @@ int add_record_indices(struct ireq *iq, void *trans, blob_buffer_t *blobs, return rc; } +/* + * Cross-shard unique check for TRUNCATE time partitions. + * Returns IX_DUP if the key exists in any sibling shard, RC_INTERNAL_RETRY on + * deadlock, 0 if no violation found. Sets *opfailcode on violation. + */ +int check_cross_shard_unique(struct ireq *iq, void *trans, int ixnum, char *key, int keylen, int *opfailcode) +{ + if (!gbl_partition_unique || !iq->partition_shards) + return 0; + if (iq->usedb->ix_dupes[ixnum] != 0) + return 0; + if (ix_isnullk(iq->usedb, key, ixnum)) + return 0; + + int rc = 0; + struct dbtable *saveddb = iq->usedb; + if (gbl_partition_unique_debug) + logmsg(LOGMSG_USER, "%s: [master] cross-shard check table='%s' ix=%d nshards=%d\n", __func__, + saveddb->tablename, ixnum, iq->npartition_shards); + for (int s = 0; s < iq->npartition_shards && !rc; s++) { + if (iq->partition_shards[s] == saveddb) + continue; + int fndrrn = 0; + unsigned long long fndgenid = 0; + iq->usedb = iq->partition_shards[s]; + int src = ix_find_by_key_tran(iq, key, keylen, ixnum, NULL, &fndrrn, &fndgenid, NULL, NULL, 0, trans); + if (gbl_partition_unique_debug) + logmsg(LOGMSG_USER, "%s: [master] checked shard='%s' ix=%d rc=%d\n", __func__, iq->usedb->tablename, + ixnum, src); + if (src == IX_FND) { + *opfailcode = OP_FAILED_UNIQ; + rc = IX_DUP; + } else if (src == RC_INTERNAL_RETRY) { + rc = RC_INTERNAL_RETRY; + } + } + iq->usedb = saveddb; + return rc; +} + +void reqerrstr_dup_key(struct ireq *iq, struct dbtable *db, int ixnum) +{ + if (gbl_partition_unique && db->timepartition_name) + reqerrstr(iq, COMDB2_CSTRT_RC_DUP, + "add key constraint duplicate key '%s' on partition '%s' shard '%s' index %d", + get_keynm_from_db_idx(db, ixnum), db->timepartition_name, db->tablename, ixnum); + else + reqerrstr(iq, COMDB2_CSTRT_RC_DUP, "add key constraint duplicate key '%s' on table '%s' index %d", + get_keynm_from_db_idx(db, ixnum), db->tablename, ixnum); +} + +void reqerrstr_uncommittable_dup(struct ireq *iq, struct dbtable *db, int ixnum) +{ + if (gbl_partition_unique && db->timepartition_name) + reqerrstr(iq, COMDB2_CSTRT_RC_DUP, + "Transaction is uncommittable: Duplicate insert on key '%s' " + "on partition '%s' shard '%s' index %d", + get_keynm_from_db_idx(db, ixnum), db->timepartition_name, db->tablename, ixnum); + else + reqerrstr(iq, COMDB2_CSTRT_RC_DUP, + "Transaction is uncommittable: Duplicate insert on key '%s' " + "in table '%s' index %d", + get_keynm_from_db_idx(db, ixnum), db->tablename, ixnum); +} + /* * Add an individual key. The operation * is defered until the end of the block op (we call insert_add_op). @@ -560,8 +641,19 @@ static int add_key(struct ireq *iq, void *trans, int ixnum, return ERR_INTERNAL; } - rc = ix_addk(iq, trans, newkey, ixnum, genid, rrn, od_dta_tail, - od_tail_len, ix_isnullk(iq->usedb, newkey, ixnum)); + /* cross-shard unique check for TRUNCATE partitions. + * NOTE: this check is currently a no-op for the cascade path. + * ON UPDATE CASCADE is handled entirely on the master side in + * constraints.c without going through the OSQL stream, so + * iq->partition_shards is never populated here and + * check_cross_shard_unique returns 0 immediately. + * This should become effective once cascading on time partitions + * is implemented and the master cascade processor populates + * iq->partition_shards before invoking upd_record_indices. */ + rc = check_cross_shard_unique(iq, trans, ixnum, newkey, getkeysize(iq->usedb, ixnum), opfailcode); + if (!rc) + rc = ix_addk(iq, trans, newkey, ixnum, genid, rrn, od_dta_tail, od_tail_len, + ix_isnullk(iq->usedb, newkey, ixnum)); if (iq->debug) { reqprintf(iq, "ix_addk IX %d RRN %d KEY ", ixnum, rrn); reqdumphex(iq, newkey, getkeysize(iq->usedb, ixnum)); @@ -569,7 +661,7 @@ static int add_key(struct ireq *iq, void *trans, int ixnum, } if (rc == IX_DUP) *opfailcode = OP_FAILED_UNIQ; - else if (rc != 0) + else if (rc != 0 && rc != RC_INTERNAL_RETRY) *opfailcode = OP_FAILED_INTERNAL; } @@ -1371,6 +1463,19 @@ int process_defered_table(struct ireq *iq, void *trans, int *blkpos, int *ixout, if (ditk->type == DIT_ADD) { int addrrn = 2; + /* cross-shard unique check for TRUNCATE partitions */ + int opfailcode_tmp = 0; + rc = check_cross_shard_unique(iq, trans, ditk->ixnum, ditk->ixkey, getkeysize(iq->usedb, ditk->ixnum), + &opfailcode_tmp); + if (rc == IX_DUP) { + reqerrstr_dup_key(iq, ditk->usedb, ditk->ixnum); + *errout = OP_FAILED_UNIQ; + *ixout = ditk->ixnum; + goto done; + } else if (rc == RC_INTERNAL_RETRY) { + *errout = OP_FAILED_INTERNAL; + goto done; + } /* add the key */ rc = ix_addk(iq, trans, ditk->ixkey, ditk->ixnum, ditk->genid, addrrn, od_dta_tail, od_tail_len, @@ -1385,15 +1490,7 @@ int process_defered_table(struct ireq *iq, void *trans, int *blkpos, int *ixout, } if (rc == IX_DUP) { - reqerrstr(iq, COMDB2_CSTRT_RC_DUP, - "add key constraint " - "duplicate key '%s' on " - "table '%s' index %d", - get_keynm_from_db_idx(ditk->usedb, ditk->ixnum), - ditk->usedb->tablename, ditk->ixnum); - - //*blkpos = curop->blkpos; - + reqerrstr_dup_key(iq, ditk->usedb, ditk->ixnum); *errout = OP_FAILED_UNIQ; *ixout = ditk->ixnum; goto done; diff --git a/db/indices.h b/db/indices.h index 08398b6a68..355473cae0 100644 --- a/db/indices.h +++ b/db/indices.h @@ -59,4 +59,9 @@ int del_new_record_indices(struct ireq *iq, void *trans, unsigned long long del_keys, blob_buffer_t *del_idx_blobs, int verify_retry); +int check_cross_shard_unique(struct ireq *iq, void *trans, int ixnum, char *key, int keylen, int *opfailcode); + +void reqerrstr_dup_key(struct ireq *iq, struct dbtable *db, int ixnum); +void reqerrstr_uncommittable_dup(struct ireq *iq, struct dbtable *db, int ixnum); + #endif diff --git a/db/osqlblockproc.c b/db/osqlblockproc.c index 44724684fa..325fbacbbf 100644 --- a/db/osqlblockproc.c +++ b/db/osqlblockproc.c @@ -506,6 +506,7 @@ static int setup_reorder_key(blocksql_tran_t *tran, int type, osql_sess_t *sess, break; } /* This doesn't touch btrees and should be processed first */ + case OSQL_PARTITION_SHARDS: case OSQL_SERIAL: case OSQL_SELECTV: key->tbl_idx = 0; @@ -518,6 +519,7 @@ static int setup_reorder_key(blocksql_tran_t *tran, int type, osql_sess_t *sess, } switch (type) { + case OSQL_PARTITION_SHARDS: case OSQL_QBLOB: case OSQL_DELIDX: case OSQL_INSIDX: diff --git a/db/osqlcomm.c b/db/osqlcomm.c index d7163f90c7..6caf9d0008 100644 --- a/db/osqlcomm.c +++ b/db/osqlcomm.c @@ -58,6 +58,7 @@ #include "sc_logic.h" #include "eventlog.h" #include +#include "indices.h" #define MAX_CLUSTER REPMAX @@ -4123,6 +4124,60 @@ int osql_send_usedb(osql_target_t *target, unsigned long long rqid, uuid_t uuid, return rc; } +/** + * Send OSQL_PARTITION_SHARDS op + * Informs the master of all shard table names belonging to a TRUNCATE + * partition so it can lock them before any writes arrive. + * Payload: 4-byte big-endian shard count followed by nshards null-terminated names. + */ +int osql_send_partition_shards(osql_target_t *target, unsigned long long rqid, uuid_t uuid, char **shards, int nshards, + int type) +{ + if (check_master(target)) + return OSQL_SEND_ERROR_WRONGMASTER; + + /* compute total payload size: 4 bytes for count + names */ + int payloadsz = sizeof(int); + for (int i = 0; i < nshards; i++) + payloadsz += strlen(shards[i]) + 1; + + char *payload = malloc(payloadsz); + if (!payload) + return -1; + + char *p = payload; + int nshards_wire = htonl(nshards); + memcpy(p, &nshards_wire, sizeof(int)); + p += sizeof(int); + for (int i = 0; i < nshards; i++) { + int len = strlen(shards[i]) + 1; + memcpy(p, shards[i], len); + p += len; + } + + uint8_t buf[OSQLCOMM_UUID_RPL_TYPE_LEN]; + uint8_t *p_buf = buf; + uint8_t *p_buf_end = buf + sizeof(buf); + if (rqid == OSQL_RQID_USE_UUID) { + osql_uuid_rpl_t hdr = {0}; + hdr.type = OSQL_PARTITION_SHARDS; + comdb2uuidcpy(hdr.uuid, uuid); + type = osql_net_type_to_net_uuid_type(NET_OSQL_SOCK_RPL); + osqlcomm_uuid_rpl_type_put(&hdr, p_buf, p_buf_end); + } else { + osql_rpl_t hdr_rqid = {0}; + hdr_rqid.type = OSQL_PARTITION_SHARDS; + hdr_rqid.sid = rqid; + osqlcomm_rpl_type_put(&hdr_rqid, p_buf, p_buf_end); + } + + int rc = target->send(target, type, buf, sizeof(buf), 0, payload, payloadsz); + free(payload); + if (rc) + logmsg(LOGMSG_ERROR, "%s target->send returns rc=%d\n", __func__, rc); + return rc; +} + /** * Send UPDCOLS op * It handles remote/local connectivity @@ -7326,11 +7381,7 @@ int osql_process_packet(struct ireq *iq, uuid_t uuid, void *trans, char **pmsg, if (err->errcode == OP_FAILED_UNIQ) { if (iq->vfy_idx_track == 1 && iq->dup_key_insert == 1) { rc = ERR_UNCOMMITTABLE_TXN; - reqerrstr(iq, COMDB2_CSTRT_RC_DUP, "Transaction is uncommittable: " - "Duplicate insert on key '%s' " - "in table '%s' index %d", - get_keynm_from_db_idx(iq->usedb, err->ixnum), - iq->usedb->tablename, err->ixnum); + reqerrstr_uncommittable_dup(iq, iq->usedb, err->ixnum); err->errcode = ERR_UNCOMMITTABLE_TXN; goto done_delete; } @@ -7361,11 +7412,7 @@ int osql_process_packet(struct ireq *iq, uuid_t uuid, void *trans, char **pmsg, if (rc != ERR_VERIFY) { /* this can happen if we're skipping delayed key adds */ - reqerrstr(iq, COMDB2_CSTRT_RC_DUP, "add key constraint " - "duplicate key '%s' on " - "table '%s' index %d", - get_keynm_from_db_idx(iq->usedb, err->ixnum), - iq->usedb->tablename, err->ixnum); + reqerrstr_dup_key(iq, iq->usedb, err->ixnum); } } else if (rc != RC_INTERNAL_RETRY) { errstat_cat_strf(&iq->errstat, " unable to add record rc = %d", @@ -7709,6 +7756,71 @@ int osql_process_packet(struct ireq *iq, uuid_t uuid, void *trans, char **pmsg, ); memcpy(pIdx, pData, dt.nData); } break; + case OSQL_PARTITION_SHARDS: { + /* Payload: 4-byte shard count followed by null-terminated shard names. + * Lock each shard table and store its dbtable pointer for use by + * add_record_indices() during cross-shard unique checks. */ + const uint8_t *local_buf_end = (const uint8_t *)msg + msglen; + if (local_buf_end - p_buf < (ptrdiff_t)sizeof(int)) { + logmsg(LOGMSG_ERROR, "%s: OSQL_PARTITION_SHARDS truncated\n", __func__); + return ERR_INTERNAL; + } + int nshards; + memcpy(&nshards, p_buf, sizeof(int)); + nshards = ntohl(nshards); + if (nshards < 0 || nshards > 1024) { + logmsg(LOGMSG_ERROR, "%s: OSQL_PARTITION_SHARDS bad nshards=%d\n", __func__, nshards); + return ERR_INTERNAL; + } + const char *name = (const char *)p_buf + sizeof(int); + const char *payload_end = (const char *)local_buf_end; + /* free previous partition shard list if any */ + free(iq->partition_shards); + iq->partition_shards = NULL; + iq->npartition_shards = 0; + + iq->partition_shards = malloc(nshards * sizeof(struct dbtable *)); + if (!iq->partition_shards) + return ERR_INTERNAL; + + for (int i = 0; i < nshards; i++) { + /* ensure there is at least one byte and a NUL terminator in bounds */ + const char *nul = memchr(name, '\0', payload_end - name); + if (!nul) { + logmsg(LOGMSG_ERROR, "%s: OSQL_PARTITION_SHARDS payload truncated at shard %d\n", __func__, i); + free(iq->partition_shards); + iq->partition_shards = NULL; + iq->npartition_shards = 0; + return ERR_INTERNAL; + } + rc = bdb_lock_tablename_read(thedb->bdb_env, name, trans); + if (rc == BDBERR_DEADLOCK) { + if (iq->debug) + reqprintf(iq, "LOCK PARTITION SHARD READ DEADLOCK"); + free(iq->partition_shards); + iq->partition_shards = NULL; + iq->npartition_shards = 0; + return RC_INTERNAL_RETRY; + } else if (rc) { + if (iq->debug) + reqprintf(iq, "LOCK PARTITION SHARD READ ERROR: %d", rc); + free(iq->partition_shards); + iq->partition_shards = NULL; + iq->npartition_shards = 0; + return ERR_INTERNAL; + } + iq->partition_shards[iq->npartition_shards] = get_dbtable_by_name(name); + if (iq->partition_shards[iq->npartition_shards]) + iq->npartition_shards++; + name = nul + 1; + } + if (gbl_partition_unique_debug) { + logmsg(LOGMSG_USER, "%s: [master] received OSQL_PARTITION_SHARDS nshards=%d stored=%d\n", __func__, nshards, + iq->npartition_shards); + for (int i = 0; i < iq->npartition_shards; i++) + logmsg(LOGMSG_USER, "%s: [master] shard[%d]='%s'\n", __func__, i, iq->partition_shards[i]->tablename); + } + } break; case OSQL_QBLOB: { osql_qblob_t dt = {0}; const uint8_t *p_buf_end = p_buf + sizeof(osql_qblob_t), diff --git a/db/osqlcomm.h b/db/osqlcomm.h index b5ea4c8e2a..e0e2ca06aa 100644 --- a/db/osqlcomm.h +++ b/db/osqlcomm.h @@ -60,6 +60,15 @@ int osql_comm_send_socksqlreq(osql_target_t *target, const char *sql, int sqlen, int osql_send_usedb(osql_target_t *target, unsigned long long rqid, uuid_t uuid, char *tablename, int type, unsigned long long version); +/** + * Send OSQL_PARTITION_SHARDS op + * Sends the list of shard table names for a TRUNCATE partition so the master + * can lock them before any writes arrive. + * + */ +int osql_send_partition_shards(osql_target_t *target, unsigned long long rqid, uuid_t uuid, char **shards, int nshards, + int type); + /** * Send INDEX op * It handles remote/local connectivity diff --git a/db/osqlrpltypes.h b/db/osqlrpltypes.h index 04148179ab..413a059405 100644 --- a/db/osqlrpltypes.h +++ b/db/osqlrpltypes.h @@ -54,7 +54,8 @@ XMACRO_OSQL_RPL_TYPES( OSQL_DONE_WITH_EFFECTS, 28, "OSQL_DONE_WITH_EFFECTS" ) XMACRO_OSQL_RPL_TYPES( OSQL_PREPARE, 29, "OSQL_PREPARE" ) /* participant should prepare */ \ XMACRO_OSQL_RPL_TYPES( OSQL_DIST_TXNID, 30, "OSQL_DIST_TXNID" ) /* send dist-txnid to coordinator */ \ XMACRO_OSQL_RPL_TYPES( OSQL_PARTICIPANT, 31, "OSQL_PARTICIPANT" ) /* a participant (to coordinator) */ \ -XMACRO_OSQL_RPL_TYPES( MAX_OSQL_TYPES, 32, "OSQL_MAX") +XMACRO_OSQL_RPL_TYPES( OSQL_PARTITION_SHARDS, 32, "OSQL_PARTITION_SHARDS" ) /* shard table list for a TRUNCATE partition */ \ +XMACRO_OSQL_RPL_TYPES( MAX_OSQL_TYPES, 33, "OSQL_MAX") // clang-format on diff --git a/db/osqlsqlthr.c b/db/osqlsqlthr.c index 0eda3a8475..abb2905ab3 100644 --- a/db/osqlsqlthr.c +++ b/db/osqlsqlthr.c @@ -943,6 +943,10 @@ static int osql_sock_restart(struct sqlclntstate *clnt, int maxretries, int keep osql->tablename = NULL; osql->tablenamelen = 0; } + if (osql->partition_name) { + free(osql->partition_name); + osql->partition_name = NULL; + } if (!keep_session) { if (gbl_master_swing_osql_verbose) @@ -1414,6 +1418,10 @@ int osql_sock_abort(struct sqlclntstate *clnt, int type) clnt->osql.tablename = NULL; clnt->osql.tablenamelen = 0; } + if (clnt->osql.partition_name) { + free(clnt->osql.partition_name); + clnt->osql.partition_name = NULL; + } if (gbl_debug_disttxn_trace) { uuidstr_t us; @@ -1473,6 +1481,49 @@ static int osql_send_usedb_logic_int(char *tablename, struct sqlclntstate *clnt, osql->replicant_numops++; DEBUG_PRINT_NUMOPS(); + if (rc != SQLITE_OK) + return rc; + + /* If this table is a shard of a TRUNCATE partition with unique indexes, + * send the full shard list so the master can lock all sibling shards. + * Deduplicate by partition name — only send once per partition per txn. */ + if (gbl_partition_unique) { + struct dbtable *db = get_dbtable_by_name(tablename); + const char *partname = db ? db->timepartition_name : NULL; + if (partname && (!osql->partition_name || strcmp(osql->partition_name, partname) != 0)) { + int nshards = 0; + struct errstat shard_err = {0}; + char **shards = timepart_get_shard_names(partname, &nshards, &shard_err); + if (!shards && shard_err.errval != 0) { + logmsg(LOGMSG_ERROR, "%s: %s\n", __func__, shard_err.errstr); + return SQLITE_NOMEM; + } + if (shards) { + if (gbl_partition_unique_debug) { + logmsg(LOGMSG_USER, "%s: [replicant] sending OSQL_PARTITION_SHARDS partition='%s' nshards=%d\n", + __func__, partname, nshards); + for (int i = 0; i < nshards; i++) + logmsg(LOGMSG_USER, "%s: [replicant] shard[%d]='%s'\n", __func__, i, shards[i]); + } + do { + rc = osql_send_partition_shards(&osql->target, osql->rqid, osql->uuid, shards, nshards, nettype); + RESTART_SOCKSQL; + } while (restarted); + + for (int i = 0; i < nshards; i++) + free(shards[i]); + free(shards); + + if (rc == SQLITE_OK) { + free(osql->partition_name); + osql->partition_name = strdup(partname); + } + osql->replicant_numops++; + DEBUG_PRINT_NUMOPS(); + } + } + } + return rc; } @@ -1608,6 +1659,10 @@ static int osql_send_commit_logic(struct sqlclntstate *clnt, int retries, int ne osql->tablename = NULL; osql->tablenamelen = 0; } + if (osql->partition_name) { + free(osql->partition_name); + osql->partition_name = NULL; + } osql->tran_ops = 0; /* reset transaction size counter*/ if (!clnt->dbtran.trans_has_sp) { diff --git a/db/sql.h b/db/sql.h index c4e04c306a..a034839b76 100644 --- a/db/sql.h +++ b/db/sql.h @@ -144,6 +144,7 @@ typedef struct osqlstate { uuid_t uuid; /* session id, take 2 */ char *tablename; /* malloc-ed cache of send tablename for usedb */ int tablenamelen; /* tablename length */ + char *partition_name; /* malloc-ed cache of last sent partition name for OSQL_PARTITION_SHARDS */ int sentops; /* number of operations per statement */ int tran_ops; /* actual number of operations for a transaction */ int replicant_numops; /* total num of ops sent by replicant to master which diff --git a/db/sqlglue.c b/db/sqlglue.c index b8005c9965..bcabe7a937 100644 --- a/db/sqlglue.c +++ b/db/sqlglue.c @@ -9193,6 +9193,14 @@ int sqlite3BtreeInsert( } } + if ((rec_flags & OSQL_IGNORE_FAILURE) && gbl_partition_unique && pCur->db->timepartition_name) { + sqlite3VdbeError(pCur->vdbe, + "UPSERT is not supported on time partition '%s' " + "when partition_unique is enabled", + pCur->db->timepartition_name); + return SQLITE_ERROR; + } + if (is_update) { /* Updating an existing record. */ rc = osql_updrec(pCur, thd, pCur->ondisk_buf, getdatsize(pCur->db), pCur->vdbe->updCols, pblobs, MAXBLOBS, rec_flags); diff --git a/db/toblock.c b/db/toblock.c index 6d36a09c94..dda7a254ee 100644 --- a/db/toblock.c +++ b/db/toblock.c @@ -80,6 +80,7 @@ #include "schemachange.h" #include "views.h" #include +#include "indices.h" #if 0 #define TEST_OSQL @@ -4971,11 +4972,7 @@ static int toblock_main_int(struct javasp_trans_state *javasp_trans_handle, stru if (iq->vfy_idx_track == 1 && iq->dup_key_insert == 1) { rc = ERR_UNCOMMITTABLE_TXN; errout = ERR_UNCOMMITTABLE_TXN; - reqerrstr(iq, COMDB2_CSTRT_RC_DUP, "Transaction is uncommittable: " - "Duplicate insert on key '%s' " - "in table '%s' index %d", - get_keynm_from_db_idx(iq->usedb, ixout), - iq->usedb->tablename, ixout); + reqerrstr_uncommittable_dup(iq, iq->usedb, ixout); } else { check_serializability = 1; } diff --git a/db/views.c b/db/views.c index 6ae3961cf2..5095bee7ce 100644 --- a/db/views.c +++ b/db/views.c @@ -3613,6 +3613,38 @@ struct dbtable *timepart_retro_route(struct timepart_retro *retros, unsigned lon return ret; } +/** + * Return a heap-allocated array of strdup'd shard table names for a TRUNCATE + * partition. *nshards is set to the number of entries. Returns NULL if the + * partition does not exist or is not a TRUNCATE partition. On allocation + * failure, returns NULL and sets err. The caller must free each string and + * then the array itself. + * + */ +char **timepart_get_shard_names(const char *partname, int *nshards, struct errstat *err) +{ + *nshards = 0; + + views_lock(); + timepart_view_t *view = _get_view(thedb->timepart_views, partname); + if (!view || view->rolltype != TIMEPART_ROLLOUT_TRUNCATE) { + views_unlock(); + return NULL; + } + + char **shards = malloc(view->nshards * sizeof(char *)); + if (!shards) { + errstat_set_rcstrf(err, VIEW_ERR_MALLOC, "out of memory getting shard names for partition '%s'", partname); + views_unlock(); + return NULL; + } + for (int i = 0; i < view->nshards; i++) + shards[(*nshards)++] = strdup(view->shards[i].tblname); + views_unlock(); + + return shards; +} + #include "views_systable.c" #include "views_serial.c" diff --git a/db/views.h b/db/views.h index ccf6b4d507..f3fcbb4512 100644 --- a/db/views.h +++ b/db/views.h @@ -562,4 +562,13 @@ int timepart_analyze_partition(char *name, void *td, struct sqlclntstate *clnt, */ struct dbtable *timepart_retro_route(struct timepart_retro *retros, unsigned long long genid, const char *f, int l); +/** + * Copy the table names of all shards belonging to a TRUNCATE partition into + * the caller-supplied array. Returns the number of shards written (0 if the + * partition does not exist or is not a TRUNCATE partition). The caller must + * supply an array of at least MAXTABLELEN+1 bytes per slot. + * + */ +char **timepart_get_shard_names(const char *partname, int *nshards, struct errstat *err); + #endif diff --git a/docs/pages/programming/timepart.md b/docs/pages/programming/timepart.md index b6614136c6..7a82a6ee69 100644 --- a/docs/pages/programming/timepart.md +++ b/docs/pages/programming/timepart.md @@ -48,8 +48,8 @@ If the client would choose periodicity `daily`, and retention 31, at all time th ## Current limitations * The name space for tables and partitions is the same. Creating a partition name cannot reuse an existing table name. This is inconvenient and it will be addressed by future efforts. -* Enforcing unique constraints is not possible across shards of a time partition. An option to allow this at the expense of commit performance will be goal of a future project. -* UPSERT is not supported +* Cross-shard unique constraint enforcement is available for TRUNCATE partitions via the `partition_unique` tunable. When enabled, every unique index is enforced across all shards at the cost of additional index lookups per write. Note that enabling this tunable does not validate pre-existing data: any cross-shard duplicates present before the tunable is turned on will remain and are the user's responsibility to detect and resolve. +* UPSERT is not supported on time partitions. ## Rollout implementation details diff --git a/tests/partition_unique_check.test/Makefile b/tests/partition_unique_check.test/Makefile new file mode 100644 index 0000000000..07180ce670 --- /dev/null +++ b/tests/partition_unique_check.test/Makefile @@ -0,0 +1,9 @@ +ifeq ($(TESTSROOTDIR),) + include ../testcase.mk +else + include $(TESTSROOTDIR)/testcase.mk +endif + +ifeq ($(TEST_TIMEOUT),) + export TEST_TIMEOUT=5m +endif diff --git a/tests/partition_unique_check.test/runit b/tests/partition_unique_check.test/runit new file mode 100755 index 0000000000..a00d0dbc9a --- /dev/null +++ b/tests/partition_unique_check.test/runit @@ -0,0 +1,421 @@ +#!/usr/bin/env bash +bash -n "$0" | exit 1 +source ${TESTSROOTDIR}/tools/runit_common.sh + +# Correctness test for partition_unique cross-shard unique enforcement. +# +# Tests: +# 1. Standalone INSERT violation (inline path in add_record_indices) +# 2. Standalone UPDATE violation (deferred path in delayed_key_adds) +# 3. INSERT violations inside BEGIN/COMMIT (violation at 1st, 2nd, and 3rd write) +# 4. UPDATE violations inside BEGIN/COMMIT (violation at 1st and 2nd update) +# 5. Non-violating inserts and updates succeed +# 6. Cascade ON UPDATE path (RECFLAGS_UPD_CASCADE -> do_inline=1 in add_key) + +dbname=$1 +cmd="cdb2sql ${CDB2_OPTIONS} $dbname default" +cmdt="cdb2sql -tabs ${CDB2_OPTIONS} $dbname default" + +pass=0 +fail=0 + +# expect_ok : run sql, fail test if it errors +expect_ok() { + local desc=$1 sql=$2 + if ! $cmd "$sql" > /dev/null 2>&1; then + echo "FAIL [$desc]: expected success, got error" + fail=$(( fail + 1 )) + else + echo "PASS [$desc]" + pass=$(( pass + 1 )) + fi +} + +# expect_err : run sql, fail test if it succeeds +expect_err() { + local desc=$1 sql=$2 + if $cmd "$sql" > /dev/null 2>&1; then + echo "FAIL [$desc]: expected error (dup key), got success" + fail=$(( fail + 1 )) + else + echo "PASS [$desc]" + pass=$(( pass + 1 )) + fi +} + +# expect_txn_err : run statements in BEGIN/COMMIT, expect COMMIT to fail +expect_txn_err() { + local desc=$1; shift + local stmts=("$@") + local sql="begin; " + for s in "${stmts[@]}"; do sql+="$s; "; done + sql+="commit" + if $cmd "$sql" > /dev/null 2>&1; then + echo "FAIL [$desc]: expected txn to fail, got success" + fail=$(( fail + 1 )) + else + echo "PASS [$desc]" + pass=$(( pass + 1 )) + fi +} + +# put_tunable_all : set a tunable on every cluster node. +# On a cluster, tunables are node-local; the master must also have the value. +put_tunable_all() { + local tunable=$1 value=$2 + if [[ -n "$CLUSTER" ]]; then + for node in $CLUSTER; do + cdb2sql ${CDB2_OPTIONS} $dbname @$node "put tunable $tunable $value" > /dev/null 2>&1 + done + else + $cmd "put tunable $tunable $value" > /dev/null 2>&1 + fi +} + +# run_txn_file : run a SQL file, expect success +run_txn_file() { + local desc=$1 file=$2 + local out + out=$(cdb2sql ${CDB2_OPTIONS} $dbname default -f "$file" 2>&1) + if (( $? != 0 )); then + echo "FAIL [$desc]: expected success, got: ${out}" + fail=$(( fail + 1 )) + else + echo "PASS [$desc]" + pass=$(( pass + 1 )) + fi +} + +# ============================================================ +# Setup: 2-shard TRUNCATE partition +# +# Start time is computed as now-25h. With period 'daily' and retention 2: +# older shard: [now-25h, now-1h) +# current shard: [now-1h, now+23h) +# next rollout: now+23h (far enough away that no test run will hit it) +# +# Schema: a int unique, b int unique, c int +# older shard (inserted directly): a=1..20, b=1..20 +# current shard (via partition view): a=101..120, b=101..120 +# ============================================================ + +# Clean up from any previous run so the test is independently re-runnable. +put_tunable_all partition_unique 0 +$cmd "drop table t" > /dev/null 2>&1 +$cmd "drop table p" > /dev/null 2>&1 + +# Compute start 25 hours ago: guarantees 2 shards and ~23h until next rollout. +start_ts=$(date -d "25 hours ago" "+%Y%m%dT%H%M%S") + +$cmd "create table t (a int, b int, c int, unique(a), unique(b)) + partitioned by time period 'daily' retention 2 start '${start_ts}'" \ + > /dev/null || failexit "create partition failed" + +# Identify the older shard (the one not receiving new inserts). +$cmd "insert into t values (9999, 9999, 0)" > /dev/null || failexit "probe insert failed" +older_shard=$($cmdt "select shardname from comdb2_timepartshards where name='t' + order by shardname" | while read shard; do + cnt=$($cmdt "select count(*) from \"${shard}\" where a=9999") + [[ "$cnt" == "0" ]] && echo "$shard" && break +done) +$cmd "delete from t where a=9999" > /dev/null +[[ -z "$older_shard" ]] && failexit "could not identify older shard" + +# Seed the older shard directly: a=1..20, b=1..20 +$cmd "insert into \"${older_shard}\" + select value, value, 0 from generate_series(1, 20)" \ + > /dev/null || failexit "seed older shard failed" + +# Seed the current shard via partition view: a=101..120, b=101..120 +$cmd "insert into t select value+100, value+100, 0 from generate_series(1, 20)" \ + > /dev/null || failexit "seed current shard failed" + +# Enable cross-shard unique enforcement on every node (tunable is node-local). +put_tunable_all partition_unique 1 + +echo "" +echo "=== 1. Standalone INSERT violations ===" +expect_err "insert a conflicts older shard" "insert into t values (5, 200, 0)" +expect_err "insert b conflicts older shard" "insert into t values (200, 5, 0)" +expect_err "insert a conflicts current shard" "insert into t values (105, 200, 0)" +expect_err "insert b conflicts current shard" "insert into t values (200, 105, 0)" + +echo "" +echo "=== 2. Standalone UPDATE violations (deferred key path) ===" +# Update a key in the current shard to a value that exists in the older shard. +expect_err "update a to older shard value" "update t set a=5 where a=101" +expect_err "update b to older shard value" "update t set b=5 where b=101" +# Update a key in the older shard to a value that exists in the current shard. +$cmd "insert into \"${older_shard}\" values (50, 50, 0)" > /dev/null +expect_err "update a to current shard value" "update t set a=105 where a=50" +expect_err "update b to current shard value" "update t set b=105 where b=50" +$cmd "delete from t where a=50" > /dev/null + +echo "" +echo "=== 3. INSERT violations inside BEGIN/COMMIT ===" +# Violation is the only (first) write in the transaction. +expect_txn_err "txn 1-write: violates a" \ + "insert into t values (5, 200, 0)" +expect_txn_err "txn 1-write: violates b" \ + "insert into t values (200, 5, 0)" +# Violation is the second write (first write is conflict-free). +expect_txn_err "txn 2-writes: second violates a" \ + "insert into t values (300, 300, 0)" \ + "insert into t values (5, 400, 0)" +expect_txn_err "txn 2-writes: second violates b" \ + "insert into t values (300, 300, 0)" \ + "insert into t values (400, 5, 0)" +# Violation is the third write (first two are conflict-free). +expect_txn_err "txn 3-writes: third violates a" \ + "insert into t values (300, 300, 0)" \ + "insert into t values (400, 400, 0)" \ + "insert into t values (5, 500, 0)" +expect_txn_err "txn 3-writes: third violates b" \ + "insert into t values (300, 300, 0)" \ + "insert into t values (400, 400, 0)" \ + "insert into t values (500, 5, 0)" + +echo "" +echo "=== 4. UPDATE violations inside BEGIN/COMMIT ===" +# First update violates. +expect_txn_err "txn 1-update: violates a" \ + "update t set a=5 where a=101" +expect_txn_err "txn 1-update: violates b" \ + "update t set b=5 where b=101" +# Second update violates (first targets a different row and is conflict-free). +expect_txn_err "txn 2-updates: second violates a" \ + "update t set a=999 where a=101" \ + "update t set a=5 where a=102" +expect_txn_err "txn 2-updates: second violates b" \ + "update t set b=999 where b=101" \ + "update t set b=5 where b=102" + +echo "" +echo "=== 5. Non-violations (should succeed) ===" +expect_ok "insert unique values" "insert into t values (500, 600, 0)" +expect_ok "update a to unique value" "update t set a=700 where a=500" +expect_ok "update b to unique value" "update t set b=800 where b=600" +$cmd "delete from t where a=700" > /dev/null + +# For success-case transactions use SQL files to avoid cdb2sql multi-statement +# handle state issues (result rows from prior statements must be drained before +# sending the next request). +txnfile=$(mktemp /tmp/puc_XXXXXX.sql) + +# Insert two rows with values that don't conflict in any shard. +printf 'begin\ninsert into t values (900, 1000, 0)\ninsert into t values (1100, 1200, 0)\ncommit\n' \ + > "$txnfile" +run_txn_file "txn insert: 2 unique rows" "$txnfile" + +# Update two *different* rows to new unique values. +# (900, 1000): update a -> 1300 +# (1100, 1200): update b -> 1400 +# Targeting different rows avoids OCC vgenid staleness on the second update. +printf 'begin\nupdate t set a=1300 where a=900\nupdate t set b=1400 where b=1200\ncommit\n' \ + > "$txnfile" +run_txn_file "txn update: 2 different rows to unique values" "$txnfile" + +rm -f "$txnfile" + +# Cleanup rows added in section 5. +$cmd "delete from t where a >= 500" > /dev/null + +echo "" +echo "=== 6. Cascade ON UPDATE path (RECFLAGS_UPD_CASCADE -> do_inline=1) ===" +# Requires a parent table with a PK and the partition having a FK with +# ON UPDATE CASCADE. The cascaded column must also be UNIQUE so that +# check_cross_shard_unique fires via the add_key do_inline=1 branch. +# +# Schema: +# parent p(pk int unique) +# partition t: b is FK -> p(pk) ON UPDATE CASCADE, b is also UNIQUE +# +# Setup: p has pk=5, pk=10, pk=200. +# older_shard has b=10 (FK to p.pk=10) and b=5 (extra unique value). +# current shard has b=200 (FK to p.pk=200). +# +# Violation: UPDATE p SET pk=200->5 cascades b=200->5 in the current shard, +# but b=5 already exists in the older shard. +# Non-violation: UPDATE p SET pk=200->999 cascades b=200->999 (unique). + +$cmd "create table p (pk int, unique(pk))" > /dev/null 2>&1 +if (( $? != 0 )); then + echo "SKIP [cascade tests]: could not create parent table" +else + # Re-create partition t with FK on b. + $cmd "drop table t" > /dev/null 2>&1 + rc=0 + $cmd "create table t (a int, b int, c int, unique(a), unique(b), + foreign key(b) references p(pk) on update cascade) + partitioned by time period 'daily' retention 2 start '${start_ts}'" \ + > /dev/null 2>&1 || rc=$? + + if (( rc != 0 )); then + echo "SKIP [cascade tests]: FK constraints on time partitions not supported" + else + $cmd "insert into p values (5)" > /dev/null + $cmd "insert into p values (10)" > /dev/null + $cmd "insert into p values (200)" > /dev/null + + # Identify the older shard for the newly re-created partition. + $cmd "insert into t values (9998, 200, 0)" > /dev/null + older_shard2=$($cmdt "select shardname from comdb2_timepartshards where name='t' + order by shardname" | while read shard; do + cnt=$($cmdt "select count(*) from \"${shard}\" where a=9998") + [[ "$cnt" == "0" ]] && echo "$shard" && break + done) + $cmd "delete from t where a=9998" > /dev/null + + if [[ -z "$older_shard2" ]]; then + echo "SKIP [cascade tests]: could not identify older shard for FK partition" + else + # older shard: b=10 (FK to p.pk=10), b=5 (extra unique value) + $cmd "insert into \"${older_shard2}\" values (1, 10, 0)" > /dev/null + $cmd "insert into \"${older_shard2}\" values (2, 5, 0)" > /dev/null + # current shard: b=200 (FK to p.pk=200) + $cmd "insert into t values (100, 200, 0)" > /dev/null + + # Cascade violation: UPDATE p pk=200->5 cascades b=200->5 in current + # shard, but b=5 exists in older shard -> cross-shard unique violation. + expect_err "cascade update b to cross-shard dup" \ + "update p set pk=5 where pk=200" + + # Cascade non-violation: UPDATE p pk=200->999, b=200->999 is unique. + expect_ok "cascade update b to unique value" \ + "update p set pk=999 where pk=200" + fi + fi +fi + +echo "" +echo "=== 7. UPSERT rejected on partition when partition_unique is enabled ===" +expect_err "upsert DO NOTHING on partition" \ + "insert into t values (5, 200, 0) on conflict do nothing" +expect_err "upsert DO NOTHING with target on partition" \ + "insert into t values (5, 200, 0) on conflict(a) do nothing" + +echo "" +echo "=== 8. Debug trace (partition_unique_debug tunable) ===" + +# Fresh 2-shard partition for this section (no FK constraints). +$cmd "drop table t" > /dev/null 2>&1 +$cmd "create table t (a int, b int, c int, unique(a), unique(b)) + partitioned by time period 'daily' retention 2 start '${start_ts}'" \ + > /dev/null || failexit "section 8: create partition failed" + +$cmd "insert into t values (9999, 9999, 0)" > /dev/null +older_shard8=$($cmdt "select shardname from comdb2_timepartshards where name='t' + order by shardname" | while read shard; do + cnt=$($cmdt "select count(*) from \"${shard}\" where a=9999") + [[ "$cnt" == "0" ]] && echo "$shard" && break +done) +$cmd "delete from t where a=9999" > /dev/null +[[ -z "$older_shard8" ]] && failexit "section 8: could not identify older shard" + +$cmd "insert into \"${older_shard8}\" select value, value, 0 from generate_series(1, 20)" \ + > /dev/null || failexit "section 8: seed older shard failed" +$cmd "insert into t select value+100, value+100, 0 from generate_series(1, 20)" \ + > /dev/null || failexit "section 8: seed current shard failed" + +put_tunable_all partition_unique_debug 1 + +logfile="${TESTDIR}/logs/${DBNAME}.db" +if [[ -n "$CLUSTER" ]]; then + master=$($cmdt "select host from comdb2_cluster where is_master='Y'" 2>/dev/null | head -1) + [[ -n "$master" ]] && logfile="${TESTDIR}/logs/${DBNAME}.${master}.db" +fi + +# check_trace: verify a pattern appears at least once in the server log. +check_trace() { + local desc=$1 pattern=$2 + if grep -q "$pattern" "$logfile"; then + echo "PASS [trace: $desc]" + pass=$(( pass + 1 )) + else + echo "FAIL [trace: $desc]: pattern '$pattern' not found in $logfile" + fail=$(( fail + 1 )) + fi +} + +# check_path_fired : run sql, verify cross-shard check count increases. +check_path_fired() { + local desc=$1 sql=$2 + local before after + before=$(grep -c "cross-shard check" "$logfile" 2>/dev/null || echo 0) + $cmd "$sql" > /dev/null 2>&1 + sleep 0.3 + after=$(grep -c "cross-shard check" "$logfile" 2>/dev/null || echo 0) + if (( after > before )); then + echo "PASS [trace: $desc]" + pass=$(( pass + 1 )) + else + echo "FAIL [trace: $desc]: cross-shard check not fired (before=$before after=$after)" + fail=$(( fail + 1 )) + fi +} + +# --- 8a. Verify all 4 trace categories fire (send, receive, check, free) --- +# One violating INSERT exercises all 4 trace points in sequence. +$cmd "insert into t values (5, 200, 0)" > /dev/null 2>&1 +sleep 0.3 + +check_trace "replicant sending OSQL_PARTITION_SHARDS" "\[replicant\] sending OSQL_PARTITION_SHARDS" +check_trace "replicant lists shard" "\[replicant\].*shard\[0\]=" +check_trace "master received OSQL_PARTITION_SHARDS" "\[master\] received OSQL_PARTITION_SHARDS" +check_trace "master lists stored shard" "\[master\].*shard\[0\]=" +check_trace "master cross-shard check" "\[master\] cross-shard check" +check_trace "master checked shard" "\[master\].*checked shard=" +check_trace "master freeing partition_shards" "\[master\] freeing partition_shards" + +# --- 8b. non-reorder add_record_indices: standalone INSERT (reorder off by default) --- +check_path_fired "non-reorder INSERT (add_record_indices)" \ + "insert into t values (200, 300, 0)" + +# --- 8c. delayed_key_adds: standalone UPDATE --- +check_path_fired "delayed_key_adds (UPDATE)" \ + "update t set a=201 where a=200" + +# --- 8d. reorder add_record_indices + process_defered_table: +# multi-row txn with reorder_idx_writes=1 (tran_rows > 1 triggers reorder). +# Use a SQL file to avoid cdb2sql multi-statement drain issues. --- +put_tunable_all reorder_idx_writes 1 +reorder_txn=$(mktemp /tmp/puc_reorder_XXXXXX.sql) +printf 'begin\ninsert into t values (400, 500, 0)\ninsert into t values (401, 501, 0)\ncommit\n' > "$reorder_txn" + +before=$(grep -c "cross-shard check" "$logfile" 2>/dev/null || echo 0) +cdb2sql ${CDB2_OPTIONS} $dbname default -f "$reorder_txn" > /dev/null 2>&1 +sleep 0.3 +after=$(grep -c "cross-shard check" "$logfile" 2>/dev/null || echo 0) +rm -f "$reorder_txn" +if (( after > before )); then + echo "PASS [trace: reorder add_record_indices + process_defered_table (multi-row txn)]" + pass=$(( pass + 1 )) +else + echo "FAIL [trace: reorder add_record_indices + process_defered_table (multi-row txn)]: cross-shard check not fired (before=$before after=$after)" + fail=$(( fail + 1 )) +fi +put_tunable_all reorder_idx_writes 0 + +# --- 8e. add_key do_inline (ON UPDATE CASCADE) --- +# The ON UPDATE CASCADE is handled entirely on the master side in constraints.c, +# not via the OSQL stream. iq->partition_shards is never populated for the +# cascade path, so check_cross_shard_unique returns early there. +# This call site is a known limitation: cascade cross-shard unique enforcement +# requires the master constraint processor to populate iq->partition_shards, +# which is not currently implemented. +echo "SKIP [trace: add_key do_inline (cascade)]: cascade is master-side; iq->partition_shards not set in that path" + +put_tunable_all partition_unique_debug 0 + +echo "" +echo "=== Results ===" +echo " passed: $pass" +echo " failed: $fail" +echo "" + +[[ $fail -eq 0 ]] || failexit "$fail test(s) failed" + +put_tunable_all partition_unique 0 + +echo "Testcase passed." +exit 0 diff --git a/tests/partition_unique_perf.test/Makefile b/tests/partition_unique_perf.test/Makefile new file mode 100644 index 0000000000..d690074bc3 --- /dev/null +++ b/tests/partition_unique_perf.test/Makefile @@ -0,0 +1,9 @@ +ifeq ($(TESTSROOTDIR),) + include ../testcase.mk +else + include $(TESTSROOTDIR)/testcase.mk +endif + +ifeq ($(TEST_TIMEOUT),) + export TEST_TIMEOUT=10m +endif diff --git a/tests/partition_unique_perf.test/README b/tests/partition_unique_perf.test/README new file mode 100644 index 0000000000..84575230be --- /dev/null +++ b/tests/partition_unique_perf.test/README @@ -0,0 +1,28 @@ +Performance test for the partition_unique tunable. + +The partition_unique tunable enforces unique index constraints across all +shards of a TRUNCATE time partition. When enabled, each write that touches +a unique index performs additional ix_find_by_key_tran lookups against every +sibling shard before calling ix_addk on the target shard. This test measures +the overhead of those extra lookups. + +Setup: + - Creates a TRUNCATE time partition with two shards (start date two days + in the past so both shards exist immediately without waiting for rollouts) + - Seeds the older shard with 1000 rows so the cross-shard BTree lookups + traverse real pages rather than empty trees + +Benchmark (NROWS=5000 per run): + 1. INSERT NROWS rows with unique id and name columns + 2. UPDATE val on all inserted rows (non-key column, no uniqueness check) + 3. DELETE all inserted rows + +The benchmark is run twice: once with partition_unique disabled (baseline) +and once with it enabled. Timing for each operation and the total overhead +are printed to stdout. + +The test only fails on operational errors (a write returning a non-zero +exit code). It always exits 0 on success regardless of the timing results, +making it safe to run in CI without introducing flaky failures. + +To increase the signal-to-noise ratio, raise NROWS in runit. diff --git a/tests/partition_unique_perf.test/runit b/tests/partition_unique_perf.test/runit new file mode 100755 index 0000000000..2e12afb7ac --- /dev/null +++ b/tests/partition_unique_perf.test/runit @@ -0,0 +1,369 @@ +#!/usr/bin/env bash +bash -n "$0" | exit 1 +source ${TESTSROOTDIR}/tools/runit_common.sh +. ${TESTSROOTDIR}/tools/hrtime.sh + +# Performance test for partition_unique tunable. +# +# Measures two distinct impacts of cross-shard unique enforcement: +# +# Part 1 — Extra processing (BTree lookups): +# Single-threaded inserts with a warmup pass to prime the buffer pool. +# Measures the per-row cost of the extra ix_find_by_key_tran calls in +# check_cross_shard_unique() across multiple iterations. +# +# Part 2 — Lock footprint (contention): +# Concurrent workers each inserting a slice of rows simultaneously. +# Measures total throughput (rows/sec) to capture the impact of the +# BDB lock manager handling more lock objects per transaction. +# +# The test only fails on operational errors, not on performance results. + +dbname=$1 +cmd="cdb2sql ${CDB2_OPTIONS} $dbname default" +cmdt="cdb2sql -tabs ${CDB2_OPTIONS} $dbname default" + +# Part 1 parameters +NROWS=30000 # rows per single-threaded iteration +ITERATIONS=3 # iterations to average over + +# Part 2 parameters +NWORKERS=8 # concurrent insert workers +ROWS_PER_WORKER=5000 + +# Part 3 parameters +NWORKERS_UPD=8 # concurrent update workers +ROWS_FOR_UPD=20000 # total rows pre-inserted for update benchmark + +# ============================================================ +# Helpers +# ============================================================ + +function run_or_die +{ + local sql=$1 + local msg=$2 + $cmd "$sql" > /dev/null || failexit "$msg" +} + +function set_tunable +{ + local val=$1 + if [[ -n "$CLUSTER" ]]; then + for node in $CLUSTER; do + cdb2sql ${CDB2_OPTIONS} --host $node $dbname default \ + "put tunable partition_unique $val" > /dev/null \ + || failexit "put tunable partition_unique $val on $node" + done + else + $cmd "put tunable partition_unique $val" > /dev/null \ + || failexit "put tunable partition_unique $val" + fi +} + +# insert_rows : inserts NROWS rows starting at base_id +function insert_rows +{ + local base=$1 + $cmd "insert into t select value+${base}, 'name'||(value+${base}), value+${base} \ + from generate_series(1, ${NROWS})" > /dev/null \ + || failexit "insert failed at base ${base}" +} + +# delete_rows +function delete_rows +{ + local base=$1 + local lo=$(( base + 1 )) + local hi=$(( base + NROWS )) + $cmd "delete from t where id between ${lo} and ${hi}" > /dev/null \ + || failexit "delete failed at base ${base}" +} + +# ============================================================ +# Setup: TRUNCATE partition with 4 shards +# Start 4 days + 1 hour in the past so all 4 shards exist immediately. +# ============================================================ + +NPARTITION_SHARDS=8 + +starttime=$(get_timestamp '-(8*24*3600+3600)') +run_or_die \ + "create table t (id int, name cstring(64), val int, + unique (id), unique (name)) + partitioned by time period 'daily' retention ${NPARTITION_SHARDS} start '${starttime}'" \ + "create partitioned table failed" + +nshards=$($cmdt "select count(*) from comdb2_timepartshards where name='t'") +if [[ "$nshards" != "$NPARTITION_SHARDS" ]]; then + failexit "expected ${NPARTITION_SHARDS} shards, got '$nshards'" +fi + +# Identify the current shard (the one that receives partition view inserts) +# by inserting a sentinel row and checking which shard contains it. +$cmd "insert into t values (-1, 'sentinel', -1)" > /dev/null \ + || failexit "sentinel insert failed" +current_shard="" +while IFS= read -r shard; do + cnt=$($cmdt "select count(*) from \"${shard}\" where id = -1") + if [[ "$cnt" == "1" ]]; then + current_shard="$shard" + break + fi +done < <($cmdt "select shardname from comdb2_timepartshards where name='t'") +$cmd "delete from t where id = -1" > /dev/null +[[ -z "$current_shard" ]] && failexit "could not identify current shard" + +# Seed the 3 older shards with non-conflicting IDs so cross-shard lookups +# traverse real BTree pages. Use IDs well above the benchmark range. +seed_base=$(( NROWS * (ITERATIONS + NWORKERS + 10) )) +seed_slot=0 +while IFS= read -r shard; do + [[ "$shard" == "$current_shard" ]] && continue + slot_base=$(( seed_base + seed_slot * 2000 )) + $cmd "insert into \"${shard}\" \ + select value+${slot_base}, 'seed'||(value+${slot_base}), value+${slot_base} \ + from generate_series(1, 2000)" > /dev/null \ + || failexit "seed insert into shard ${shard} failed" + seed_slot=$(( seed_slot + 1 )) +done < <($cmdt "select shardname from comdb2_timepartshards where name='t'") + +SIBLING_SHARDS=$(( NPARTITION_SHARDS - 1 )) +EXTRA_LOOKUPS_PER_ROW=$(( 2 * SIBLING_SHARDS )) # 2 unique indexes x sibling shards + +echo "Partition: ${NPARTITION_SHARDS} shards, ${SIBLING_SHARDS} older shards each seeded with 2000 rows" +echo "Schema: 2 unique indexes (id, name) — each insert triggers ${EXTRA_LOOKUPS_PER_ROW} cross-shard BTree lookups" +echo "" + +# ============================================================ +# PART 1: Single-threaded — isolate extra BTree lookup cost +# ============================================================ + +echo "======================================================" +echo "PART 1: Single-threaded insert benchmark" +echo " ${NROWS} rows x ${ITERATIONS} iterations, buffer pool warmed up" +echo "======================================================" + +# Warmup: one pass without measuring to prime the buffer pool +set_tunable 0 +insert_rows 0 +delete_rows 0 + +# Measure without partition_unique +set_tunable 0 +total_off=0 +for i in $(seq 1 $ITERATIONS); do + t0=$(timems) + insert_rows $(( (i - 1) * NROWS )) + t1=$(timems) + ms=$(( t1 - t0 )) + echo " [off] iter $i: ${ms} ms" + total_off=$(( total_off + ms )) + delete_rows $(( (i - 1) * NROWS )) +done +avg_off=$(( total_off / ITERATIONS )) +echo " [off] average: ${avg_off} ms" +echo "" + +# Measure with partition_unique +set_tunable 1 +total_on=0 +for i in $(seq 1 $ITERATIONS); do + t0=$(timems) + insert_rows $(( (i - 1) * NROWS )) + t1=$(timems) + ms=$(( t1 - t0 )) + echo " [on] iter $i: ${ms} ms" + total_on=$(( total_on + ms )) + delete_rows $(( (i - 1) * NROWS )) +done +avg_on=$(( total_on / ITERATIONS )) +echo " [on] average: ${avg_on} ms" +echo "" + +p1_overhead=$(( avg_on - avg_off )) +p1_pct=$(( avg_off > 0 ? p1_overhead * 100 / avg_off : 0 )) +echo " Part 1 overhead: ${p1_overhead} ms avg per run (${p1_pct}% of baseline)" +echo " Extra BTree lookups per run: $(( NROWS * EXTRA_LOOKUPS_PER_ROW )) (${NROWS} rows x 2 unique indexes x ${SIBLING_SHARDS} sibling shards)" +echo "" + +# ============================================================ +# PART 2: Concurrent — measure lock footprint under contention +# ============================================================ + +echo "======================================================" +echo "PART 2: Concurrent insert benchmark" +echo " ${NWORKERS} workers x ${ROWS_PER_WORKER} rows each" +echo "======================================================" + +# Each worker inserts its own non-overlapping ID range to avoid real conflicts. +# Worker w uses IDs: w*ROWS_PER_WORKER+1 .. (w+1)*ROWS_PER_WORKER + +function run_concurrent +{ + local label=$1 + local pids=() + local t0 t1 failed=0 + local total_rows=$(( NWORKERS * ROWS_PER_WORKER )) + + t0=$(timems) + for w in $(seq 0 $(( NWORKERS - 1 ))); do + local base=$(( w * ROWS_PER_WORKER )) + $cmd "insert into t \ + select value+${base}, 'n'||(value+${base}), value+${base} \ + from generate_series(1, ${ROWS_PER_WORKER})" > /dev/null \ + || failexit "[${label}] worker ${w} insert failed" & + pids+=($!) + done + + for pid in "${pids[@]}"; do + wait "$pid" || failed=1 + done + t1=$(timems) + + [[ $failed -ne 0 ]] && failexit "[${label}] one or more concurrent workers failed" + + # store results in globals for use by caller + concurrent_ms=$(( t1 - t0 )) + concurrent_rps=$(( total_rows * 1000 / (concurrent_ms > 0 ? concurrent_ms : 1) )) + echo " elapsed: ${concurrent_ms} ms | total rows: ${total_rows} | throughput: ${concurrent_rps} rows/sec" + + $cmd "delete from t where id between 1 and ${total_rows}" \ + > /dev/null || failexit "[${label}] cleanup failed" +} + +# Without partition_unique +set_tunable 0 +echo " [off]" +run_concurrent "off" +off_rps=$concurrent_rps +off_ms=$concurrent_ms + +echo "" + +# With partition_unique +set_tunable 1 +echo " [on]" +run_concurrent "on" +on_rps=$concurrent_rps +on_ms=$concurrent_ms + +echo "" +p2_overhead=$(( on_ms - off_ms )) +p2_pct=$(( off_ms > 0 ? p2_overhead * 100 / off_ms : 0 )) +echo " Part 2 overhead: ${p2_overhead} ms (${p2_pct}% of baseline)" +echo " Throughput change: ${off_rps} -> ${on_rps} rows/sec" +echo "" + +# ============================================================ +# PART 3: Concurrent updates — non-key column vs unique key column +# ============================================================ + +echo "======================================================" +echo "PART 3: Concurrent update benchmark" +echo " ${NWORKERS_UPD} workers, ${ROWS_FOR_UPD} rows pre-inserted" +echo " 3a: update non-key column (val) — baseline, no uniqueness check" +echo " 3b: update unique key column (name) — triggers cross-shard check" +echo "======================================================" + +# Pre-insert rows into the current shard for the update benchmark. +# Use an ID range that doesn't overlap with Part 2's range. +UPD_BASE=$(( NWORKERS * ROWS_PER_WORKER + 1 )) +UPD_HI=$(( UPD_BASE + ROWS_FOR_UPD - 1 )) +SLICE=$(( ROWS_FOR_UPD / NWORKERS_UPD )) + +$cmd "insert into t \ + select value+${UPD_BASE}-1, 'n'||(value+${UPD_BASE}-1), value+${UPD_BASE}-1 \ + from generate_series(1, ${ROWS_FOR_UPD})" > /dev/null \ + || failexit "part3 setup insert failed" + +# run_concurrent_update