Changeset: d3d1a38907d1 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/d3d1a38907d1
Modified Files:
sql/backends/monet5/sql.c
sql/storage/bat/bat_storage.c
sql/storage/sql_storage.h
Branch: default
Log Message:
Added bind_updates_idx interface to avoid race conditions on idxs update bats
diffs (truncated from 306 to 300 lines):
diff --git a/sql/backends/monet5/sql.c b/sql/backends/monet5/sql.c
--- a/sql/backends/monet5/sql.c
+++ b/sql/backends/monet5/sql.c
@@ -1565,16 +1565,19 @@ str
mvc_bind_idxbat_wrap(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
int upd = (pci->argc == 7 || pci->argc == 9);
- BAT *b = NULL, *bn;
+ BAT *b = NULL;
bat *bid = getArgReference_bat(stk, pci, 0);
- int coltype = getBatType(getArgType(mb, pci, 0));
mvc *m = NULL;
str msg;
- const char *sname = *getArgReference_str(stk, pci, 2 + upd);
- const char *tname = *getArgReference_str(stk, pci, 3 + upd);
- const char *iname = *getArgReference_str(stk, pci, 4 + upd);
- int access = *getArgReference_int(stk, pci, 5 + upd);
-
+ const char *sname = *getArgReference_str(stk, pci, 2 + upd);
+ const char *tname = *getArgReference_str(stk, pci, 3 + upd);
+ const char *iname = *getArgReference_str(stk, pci, 4 + upd);
+ const int access = *getArgReference_int(stk, pci, 5 + upd);
+
+ const bool partitioned_access = pci->argc == (8 + upd) &&
getArgType(mb, pci, 6 + upd) == TYPE_int;
+
+ /* This doesn't work with quick access for now... */
+ assert(access != QUICK);
if ((msg = getSQLContext(cntxt, mb, &m, NULL)) != NULL)
return msg;
if ((msg = checkSQLContext(cntxt)) != NULL)
@@ -1583,102 +1586,73 @@ mvc_bind_idxbat_wrap(Client cntxt, MalBl
sql_schema *s = mvc_bind_schema(m, sname);
sql_table *t = mvc_bind_table(m, s, tname);
if (t && !isTable(t))
- throw(SQL, "sql.tid", SQLSTATE(42000) "%s '%s' is not
persistent",
+ throw(SQL, "sql.bindidx", SQLSTATE(42000) "%s '%s' is not
persistent",
TABLE_TYPE_DESCRIPTION(t->type, t->properties),
t->base.name);
sql_idx *i = mvc_bind_idx(m, s, iname);
- b = mvc_bind_idxbat(m, sname, tname, iname, access);
- if (b && b->ttype && b->ttype != coltype) {
- BBPunfix(b->batCacheid);
- throw(SQL,"sql.bind",SQLSTATE(42000) "Index column type
mismatch %s.%s.%s",sname,tname,iname);
- }
- if (b) {
- if (pci->argc == (8 + upd) && getArgType(mb, pci, 6 + upd) ==
TYPE_int) {
- BUN cnt = store->storage_api.count_idx(m->session->tr,
i, 0), psz;
- /* partitioned access */
- int part_nr = *getArgReference_int(stk, pci, 6 + upd);
- int nr_parts = *getArgReference_int(stk, pci, 7 + upd);
-
- if (access == 0) {
- BUN l, h;
- psz = cnt ? (cnt / nr_parts) : 0;
- l = part_nr * psz;
- if (l > cnt)
- l = cnt;
- h = (part_nr + 1 == nr_parts) ? cnt : ((part_nr
+ 1) * psz);
- if (h > cnt)
- h = cnt;
- bn = BATslice(b, l, h);
- if(bn == NULL){
- BBPunfix(b->batCacheid);
- throw(SQL, "sql.bindidx",
GDK_EXCEPTION);
- }
- BAThseqbase(bn, l);
- } else {
- /* BAT b holds the UPD_ID bat */
- oid l, h;
- cnt =
store->storage_api.count_idx(m->session->tr, i, 0);
- psz = cnt ? (cnt / nr_parts) : 0;
- l = part_nr * psz;
- if (l > cnt)
- l = cnt;
- h = (part_nr + 1 == nr_parts) ? cnt : ((part_nr
+ 1) * psz);
- if (h > cnt)
- h = cnt;
- h--;
- bn = BATselect(b, NULL, &l, &h, true, true,
false);
- if(bn == NULL) {
- BBPunfix(b->batCacheid);
- throw(SQL, "sql.bindidx",
GDK_EXCEPTION);
- }
+
+ if (partitioned_access) {
+ /* partitioned access */
+ int part_nr = *getArgReference_int(stk, pci, 6 + upd);
+ int nr_parts = *getArgReference_int(stk, pci, 7 + upd);
+ BUN cnt = store->storage_api.count_idx(m->session->tr, i, 0),
psz;
+ oid l, h;
+ psz = cnt ? (cnt / nr_parts) : 0;
+ l = part_nr * psz;
+ if (l > cnt)
+ l = cnt;
+ h = (part_nr + 1 == nr_parts) ? cnt : ((part_nr + 1) * psz);
+ if (h > cnt)
+ h = cnt;
+
+ if (upd) {
+ sql_updates* updates =
store->storage_api.bind_updates_idx(m->session->tr, i);
+
+ if (!updates)
+ throw(SQL,"sql.bindidx",SQLSTATE(HY005) "Cannot
access the update columns");
+ BAT *ui = updates->ui;
+ BAT *uv = updates->uv;
+ GDKfree(updates);
+
+ h--;
+ BAT* bn = BATselect(ui, NULL, &l, &h, true, true,
false);
+ if(bn == NULL) {
+ BBPunfix(ui->batCacheid);
+ BBPunfix(uv->batCacheid);
+ throw(SQL, "sql.bindidx", GDK_EXCEPTION);
}
- BBPunfix(b->batCacheid);
- b = bn;
- } else if (upd) {
- BAT *uv = mvc_bind_idxbat(m, sname, tname, iname,
RD_UPD_VAL);
+
bat *uvl = getArgReference_bat(stk, pci, 1);
- if ( uv == NULL){
- BBPunfix(b->batCacheid);
- throw(SQL,"sql.bindidx",SQLSTATE(42000) "Cannot
access index column %s.%s.%s",sname,tname,iname);
- }
- *bid = b->batCacheid;
- BBPkeepref(b);
- *uvl = uv->batCacheid;
- BBPkeepref(uv);
- return MAL_SUCCEED;
- }
- if (upd) {
- bat *uvl = getArgReference_bat(stk, pci, 1);
-
- if (BATcount(b)) {
- BAT *uv = mvc_bind_idxbat(m, sname, tname,
iname, RD_UPD_VAL);
- BAT *ui = mvc_bind_idxbat(m, sname, tname,
iname, RD_UPD_ID);
- BAT *id, *vl;
- if ( ui == NULL || uv == NULL) {
+
+ if (BATcount(bn)) {
+ BAT *id;
+ BAT *vl;
+ if (ui == NULL || uv == NULL) {
bat_destroy(uv);
bat_destroy(ui);
- BBPunfix(b->batCacheid);
+ BBPunfix(bn->batCacheid);
throw(SQL,"sql.bindidx",SQLSTATE(42000)
"Cannot access index column %s.%s.%s",sname,tname,iname);
}
- id = BATproject(b, ui);
- vl = BATproject(b, uv);
+ assert(uv->batCount == ui->batCount);
+ id = BATproject(bn, ui);
+ vl = BATproject(bn, uv);
bat_destroy(ui);
bat_destroy(uv);
if (id == NULL || vl == NULL) {
+ BBPunfix(bn->batCacheid);
bat_destroy(id);
bat_destroy(vl);
- BBPunfix(b->batCacheid);
throw(SQL, "sql.bindidx",
GDK_EXCEPTION);
}
if ( BATcount(id) != BATcount(vl)){
- BBPunfix(b->batCacheid);
+ BBPunfix(bn->batCacheid);
bat_destroy(id);
bat_destroy(vl);
throw(SQL, "sql.bindidx",
SQLSTATE(0000) "Inconsistent BAT count");
}
+ BBPkeepref(id);
+ BBPkeepref(vl);
*bid = id->batCacheid;
- BBPkeepref(id);
*uvl = vl->batCacheid;
- BBPkeepref(vl);
} else {
*bid = e_bat(TYPE_oid);
*uvl =
e_bat((i->type==join_idx)?TYPE_oid:TYPE_lng);
@@ -1691,16 +1665,51 @@ mvc_bind_idxbat_wrap(Client cntxt, MalBl
throw(SQL, "sql.bindidx",
SQLSTATE(HY013) MAL_MALLOC_FAIL);
}
}
+ } else {
+ int idxtype = getBatType(getArgType(mb, pci, 0));
+ b = store->storage_api.bind_idx(m->session->tr, i,
access);
+
+ if (b && b->ttype && b->ttype != idxtype) {
+ BBPunfix(b->batCacheid);
+ throw(SQL,"sql.bindidx",SQLSTATE(42000) "Index
type mismatch %s.%s.%s",sname,tname,iname);
+ }
+
+ BAT* bn = BATslice(b, l, h);
+ if(bn == NULL) {
+ BBPunfix(b->batCacheid);
+ throw(SQL, "sql.bindidx", GDK_EXCEPTION);
+ }
+ BAThseqbase(bn, l);
BBPunfix(b->batCacheid);
- } else {
- *bid = b->batCacheid;
- BBPkeepref(b);
+ BBPkeepref(bn);
+ *bid = bn->batCacheid;
}
- return MAL_SUCCEED;
- }
- if (sname)
- throw(SQL, "sql.bindidx", SQLSTATE(HY005) "Cannot access column
descriptor %s for %s.%s", iname, sname, tname);
- throw(SQL, "sql.bindidx", SQLSTATE(HY005) "Cannot access column
descriptor %s for %s", iname, tname);
+ }
+ else if (upd) { /*unpartitioned access to update bats*/
+ sql_updates* updates =
store->storage_api.bind_updates_idx(m->session->tr, i);
+
+ if (!updates)
+ throw(SQL,"sql.bindidx",SQLSTATE(HY005) "Cannot access
the update columns");
+
+ bat *uvl = getArgReference_bat(stk, pci, 1);
+ BBPkeepref(updates->ui);
+ BBPkeepref(updates->uv);
+ *bid = updates->ui->batCacheid;
+ *uvl = updates->uv->batCacheid;
+ GDKfree(updates);
+ }
+ else { /*unpartitioned access to base index*/
+ int idxtype = getBatType(getArgType(mb, pci, 0));
+ b = store->storage_api.bind_idx(m->session->tr, i, access);
+
+ if (b && b->ttype && b->ttype != idxtype) {
+ BBPunfix(b->batCacheid);
+ throw(SQL,"sql.bindidx",SQLSTATE(42000) "Index type
mismatch %s.%s.%s",sname,tname,iname);
+ }
+ BBPkeepref(b);
+ *bid = b->batCacheid;
+ }
+ return MAL_SUCCEED;
}
str
diff --git a/sql/storage/bat/bat_storage.c b/sql/storage/bat/bat_storage.c
--- a/sql/storage/bat/bat_storage.c
+++ b/sql/storage/bat/bat_storage.c
@@ -1088,6 +1088,37 @@ bind_updates(sql_trans *tr, sql_column *
return upd;
}
+static void*
+bind_updates_idx(sql_trans *tr, sql_idx *i) {
+ sql_updates* upd = GDKmalloc(sizeof(sql_updates));
+ if (!upd)
+ return NULL;
+
+ lock_column(tr->store, i->base.id);
+ size_t cnt = count_idx(tr, i, 0);
+ sql_delta *d = idx_timestamp_delta(tr, i);
+ int type = oid_index(i->type)?TYPE_oid:TYPE_lng;
+
+ if (!d) {
+ unlock_column(tr->store, i->base.id);
+ GDKfree(upd);
+ return NULL;
+ }
+
+ upd->ui = bind_ubat(tr, d, isTempTable(i->t), RD_UPD_ID, type, cnt);
+ upd->uv = bind_ubat(tr, d, isTempTable(i->t), RD_UPD_VAL, type, cnt);
+
+ unlock_column(tr->store, i->base.id);
+
+ if (upd->ui == NULL || upd->uv == NULL) {
+ bat_destroy(upd->ui);
+ bat_destroy(upd->uv);
+ GDKfree(upd);
+ return NULL;
+ }
+ return upd;
+}
+
static void * /* BAT * */
bind_col(sql_trans *tr, sql_column *c, int access)
{
@@ -4895,6 +4926,7 @@ bat_storage_init( store_functions *sf)
{
sf->bind_col = &bind_col;
sf->bind_updates = &bind_updates;
+ sf->bind_updates_idx = &bind_updates_idx;
sf->bind_idx = &bind_idx;
sf->bind_cands = &bind_cands;
diff --git a/sql/storage/sql_storage.h b/sql/storage/sql_storage.h
--- a/sql/storage/sql_storage.h
+++ b/sql/storage/sql_storage.h
@@ -137,6 +137,7 @@ typedef struct table_functions {
*/
typedef void *(*bind_col_fptr) (sql_trans *tr, sql_column *c, int access);
typedef void *(*bind_updates_fptr) (sql_trans *tr, sql_column *c);
+typedef void *(*bind_updates_idx_fptr) (sql_trans *tr, sql_idx *c);
typedef void *(*bind_idx_fptr) (sql_trans *tr, sql_idx *i, int access);
typedef void *(*bind_cands_fptr) (sql_trans *tr, sql_table *t, int
nr_of_parts, int part_nr);
@@ -222,6 +223,7 @@ typedef struct store_functions {
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
