Skip to content

Enforce unique indexes across all shards of a TRUNCATE time partition #6001

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dorinhogea wants to merge 1 commit into
base: main
Choose a base branch
from
Open

Enforce unique indexes across all shards of a TRUNCATE time partition #6001

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions db/comdb2.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down
6 changes: 6 additions & 0 deletions db/comdb2.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;

Expand Down Expand Up @@ -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;

Expand Down
25 changes: 18 additions & 7 deletions db/constraints.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 };
Expand Down Expand Up @@ -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);
Expand All @@ -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;
}

Expand Down
9 changes: 9 additions & 0 deletions db/db_tunables.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
123 changes: 110 additions & 13 deletions db/indices.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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,
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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);
Expand Down Expand Up @@ -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).
Expand Down Expand Up @@ -560,16 +641,27 @@ 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));
reqmoref(iq, " RC %d", rc);
}
if (rc == IX_DUP)
*opfailcode = OP_FAILED_UNIQ;
else if (rc != 0)
else if (rc != 0 && rc != RC_INTERNAL_RETRY)
*opfailcode = OP_FAILED_INTERNAL;
}

Expand Down Expand Up @@ -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,
Expand All @@ -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;
Expand Down
5 changes: 5 additions & 0 deletions db/indices.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
2 changes: 2 additions & 0 deletions db/osqlblockproc.c
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand All @@ -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:
Expand Down
Loading
Loading