Changeset: 69da46562f4e for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=69da46562f4e
Modified Files:
sql/backends/monet5/sql.mx
sql/backends/monet5/sql_result.c
sql/include/sql_catalog.h
sql/storage/bat/bat_storage.c
sql/storage/restrict/restrict_storage.c
sql/storage/store.c
sql/test/leaks/Tests/check0.stable.out
Branch: Oct2012
Log Message:
fixed problem with concurrent inserts/updates/deletes
diffs (truncated from 1106 to 300 lines):
diff --git a/sql/backends/monet5/sql.mx b/sql/backends/monet5/sql.mx
--- a/sql/backends/monet5/sql.mx
+++ b/sql/backends/monet5/sql.mx
@@ -6347,7 +6347,7 @@ str SQLcluster1(Client cntxt, MalBlkPtr
if ( t == NULL)
throw(SQL,"sql.cluster","42S02!Table missing");
tr = m->session->tr;
- t->base.wtime = s->base.wtime = tr->wtime = tr->stime;
+ t->base.wtime = s->base.wtime = tr->wtime = tr->wstime;
t->base.rtime = s->base.rtime = tr->rtime = tr->stime;
/* actually build the hash on the multi-column primary key */
@@ -6385,7 +6385,8 @@ str SQLcluster1(Client cntxt, MalBlkPtr
d->bid = 0;
d->ibase = 0;
d->ibid = bid; /* use the insert bat */
- c->base.wtime = c->base.rtime = tr->stime;
+ c->base.wtime = tr->wstime;
+ c->base.rtime = tr->stime;
}
/* bat was cleared */
t->cleared = 1;
@@ -6422,7 +6423,7 @@ SQLcluster2(Client cntxt, MalBlkPtr mb,
throw(SQL,"sql.cluster","42S02!Table missing");
tr = m->session->tr;
- t->base.wtime = s->base.wtime = tr->wtime = tr->stime;
+ t->base.wtime = s->base.wtime = tr->wtime = tr->wstime;
t->base.rtime = s->base.rtime = tr->rtime = tr->stime;
for (o = t->columns.set->h; o; o = o->next) {
sql_delta *d;
@@ -6461,7 +6462,8 @@ SQLcluster2(Client cntxt, MalBlkPtr mb,
d->ibase = 0;
d->ibid = bid; /* use the insert bat */
- c->base.wtime = c->base.rtime = tr->stime;
+ c->base.wtime = tr->wstime;
+ c->base.rtime = tr->stime;
}
/* bat was cleared */
t->cleared = 1;
@@ -6520,7 +6522,7 @@ vacuum(Client cntxt, MalBlkPtr mb, MalSt
}
tr = m->session->tr;
- t->base.wtime = s->base.wtime = tr->wtime = tr->stime;
+ t->base.wtime = s->base.wtime = tr->wtime = tr->wstime;
t->base.rtime = s->base.rtime = tr->rtime = tr->stime;
/* get the deletions BAT*/
@@ -6569,7 +6571,8 @@ vacuum(Client cntxt, MalBlkPtr mb, MalSt
d->bid = 0;
d->ibase = 0;
d->ibid = bids[i]; /* use the insert bat */
- c->base.wtime = c->base.rtime = tr->stime;
+ c->base.wtime = tr->wstime;
+ c->base.rtime = tr->stime;
}
BATclear(del, TRUE);
BBPreleaseref(del->batCacheid);
@@ -6748,7 +6751,8 @@ compression(Client cntxt, MalBlkPtr mb,
d->bid = 0;
d->ibase = 0;
d->ibid = e->batCacheid; /* use the insert bat */
- c->base.wtime = c->base.rtime = tr->stime;
+ c->base.wtime = tr->wstime;
+ c->base.rtime = tr->stime;
snprintf(buf,BUFSIZ,"%s/%s/%s/0", *sch, *tbl, c->base.name);
if (compr)
msg = DICTcompress(&ret, &nme, &b->batCacheid);
@@ -6760,7 +6764,7 @@ compression(Client cntxt, MalBlkPtr mb,
}
/* bat was cleared */
t->cleared = 1;
- t->base.wtime = s->base.wtime = tr->wtime = tr->stime;
+ t->base.wtime = s->base.wtime = tr->wtime = tr->wstime;
t->base.rtime = s->base.rtime = tr->rtime = tr->stime;
return msg;
}
diff --git a/sql/backends/monet5/sql_result.c b/sql/backends/monet5/sql_result.c
--- a/sql/backends/monet5/sql_result.c
+++ b/sql/backends/monet5/sql_result.c
@@ -692,8 +692,9 @@ mvc_import_table(Client cntxt, mvc *m, b
BAT *b = store_funcs.bind_col(m->session->tr,
c, RDONLY);
sql_delta *d = c->data;
- c->base.wtime = c->t->base.wtime =
c->t->s->base.wtime = m->session->tr->wtime = m->session->tr->stime;
+ c->base.wtime = c->t->base.wtime =
c->t->s->base.wtime = m->session->tr->wtime = m->session->tr->wstime;
d->cnt = BATcount(b);
+ d->ibase = BATcount(b);
BBPunfix(b->batCacheid);
}
diff --git a/sql/include/sql_catalog.h b/sql/include/sql_catalog.h
--- a/sql/include/sql_catalog.h
+++ b/sql/include/sql_catalog.h
@@ -168,7 +168,8 @@ typedef size_t backend_stack;
typedef struct sql_trans {
char *name;
- int stime; /* transaction time stamp (aka start time) */
+ int stime; /* read transaction time stamp */
+ int wstime; /* write transaction time stamp */
int rtime;
int wtime;
int schema_number; /* schema timestamp */
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
@@ -217,7 +217,7 @@ update_col(sql_trans *tr, sql_column *c,
{
sql_delta *bat = c->data;
- c->base.wtime = c->t->base.wtime = c->t->s->base.wtime = tr->wtime =
tr->stime;
+ c->base.wtime = c->t->base.wtime = c->t->s->base.wtime = tr->wtime =
tr->wstime;
c->base.rtime = c->t->base.rtime = c->t->s->base.rtime = tr->rtime =
tr->stime;
if (tpe == TYPE_bat)
delta_update_bat(bat, i, isNew(c));
@@ -230,7 +230,7 @@ update_idx(sql_trans *tr, sql_idx * i, v
{
sql_delta *bat = i->data;
- i->base.wtime = i->t->base.wtime = i->t->s->base.wtime = tr->wtime =
tr->stime;
+ i->base.wtime = i->t->base.wtime = i->t->s->base.wtime = tr->wtime =
tr->wstime;
i->base.rtime = i->t->base.rtime = i->t->s->base.rtime = tr->rtime =
tr->stime;
if (tpe == TYPE_bat)
delta_update_bat(bat, ib, isNew(i));
@@ -241,12 +241,17 @@ update_idx(sql_trans *tr, sql_idx * i, v
void
delta_append_bat( sql_delta *bat, BAT *i )
{
- BAT *b = temp_descriptor(bat->ibid);
+ BAT *c = BBPquickdesc(bat->bid, 0), *b;
+
+ if (!BATcount(i))
+ return ;
+ b = temp_descriptor(bat->ibid);
if (bat->cached) {
bat_destroy(bat->cached);
bat->cached = NULL;
}
+ assert(!c || BATcount(c) == bat->ibase);
if (!isEbat(b)){
/* try to use mmap() */
if (BATcount(b)+BATcount(i) > (BUN) REMAP_PAGE_MAXSIZE) {
@@ -261,6 +266,7 @@ delta_append_bat( sql_delta *bat, BAT *i
}
BATappend(b, i, TRUE);
bat->cnt += BATcount(i);
+ assert(BUNlast(b) > b->batInserted);
bat_destroy(b);
}
@@ -268,11 +274,13 @@ void
delta_append_val( sql_delta *bat, void *i )
{
BAT *b = temp_descriptor(bat->ibid);
+ BAT *c = BBPquickdesc(bat->bid, 0);
if (bat->cached) {
bat_destroy(bat->cached);
bat->cached = NULL;
}
+ assert(!c || BATcount(c) == bat->ibase);
if (isEbat(b)) {
bat_destroy(b);
temp_destroy(bat->ibid);
@@ -280,6 +288,7 @@ delta_append_val( sql_delta *bat, void *
b = temp_descriptor(bat->ibid);
}
BUNappend(b, i, TRUE);
+ assert(BUNlast(b) > b->batInserted);
bat->cnt ++;
bat_destroy(b);
}
@@ -290,7 +299,7 @@ append_col(sql_trans *tr, sql_column *c,
sql_delta *bat = c->data;
/* appends only write */
- c->base.wtime = c->t->base.wtime = c->t->s->base.wtime = tr->wtime =
tr->stime;
+ c->base.wtime = c->t->base.wtime = c->t->s->base.wtime = tr->wtime =
tr->wstime;
if (tpe == TYPE_bat)
delta_append_bat(bat, i);
else
@@ -303,7 +312,7 @@ append_idx(sql_trans *tr, sql_idx * i, v
sql_delta *bat = i->data;
/* appends only write */
- i->base.wtime = i->t->base.wtime = i->t->s->base.wtime = tr->wtime =
tr->stime;
+ i->base.wtime = i->t->base.wtime = i->t->s->base.wtime = tr->wtime =
tr->wstime;
if (tpe == TYPE_bat)
delta_append_bat(bat, ib);
else
@@ -339,6 +348,7 @@ delta_delete_val( sql_dbat *bat, oid rid
bat_destroy(b);
b = temp_descriptor(bat->dbid);
}
+ assert(b->T->heap.storage != STORE_PRIV);
BUNappend(b, (ptr)&rid, TRUE);
bat_destroy(b);
@@ -374,7 +384,7 @@ delete_tab(sql_trans *tr, sql_table * t,
}
/* deletes only write */
- t->base.wtime = t->s->base.wtime = tr->wtime = tr->stime;
+ t->base.wtime = t->s->base.wtime = tr->wtime = tr->wstime;
if (tpe == TYPE_bat)
delta_delete_bat(bat, ib);
else
@@ -505,6 +515,7 @@ new_persistent_delta( sql_delta *bat, in
BAT *i = temp_descriptor(bat->ibid);
bat->ibase = BATcount(b);
+ bat->cnt = BATcount(b) + BATcount(i);
bat->ibid = temp_copy(i->batCacheid, FALSE);
bat_destroy(i);
i = temp_descriptor(bat->ibid);
@@ -516,7 +527,7 @@ new_persistent_delta( sql_delta *bat, in
int type = b->ttype;
bat->bid = bat->ibid;
- bat->ibase = BATcount(b);
+ bat->cnt = bat->ibase = BATcount(b);
bat_destroy(b);
i = bat_new(TYPE_void, type, sz);
@@ -890,6 +901,7 @@ int
dup_dbat( sql_trans *tr, sql_dbat *obat, sql_dbat *bat, int is_new, int temp)
{
bat->dbid = obat->dbid;
+ bat->cnt = obat->cnt;
bat->dname = _STRDUP(obat->dname);
if (bat->dbid) {
if (is_new) {
@@ -1127,7 +1139,7 @@ BATcleanProps( BAT *b )
}
static int
-gtr_update_delta( sql_trans *tr, sql_delta *cbat)
+gtr_update_delta( sql_trans *tr, sql_delta *cbat, int *changes)
{
int ok = LOG_OK;
BAT *ups, *ins, *cur;
@@ -1139,6 +1151,7 @@ gtr_update_delta( sql_trans *tr, sql_del
ins = temp_descriptor(cbat->ibid);
/* any inserts */
if (BUNlast(ins) > BUNfirst(ins)) {
+ (*changes)++;
if (BATcount(cur)+BATcount(ins) > (BUN) REMAP_PAGE_MAXSIZE) {
/* try to use mmap() */
BATmmap(cur, STORE_MMAP, STORE_MMAP,
STORE_MMAP, STORE_MMAP, 1);
}
@@ -1154,6 +1167,7 @@ gtr_update_delta( sql_trans *tr, sql_del
ups = temp_descriptor(cbat->ubid);
/* any updates */
if (BUNlast(ups) > BUNfirst(ups)) {
+ (*changes)++;
void_replace_bat(cur, ups, TRUE);
temp_destroy(cbat->ubid);
cbat->ubid = e_ubat(cur->ttype);
@@ -1164,53 +1178,72 @@ gtr_update_delta( sql_trans *tr, sql_del
}
static int
-gtr_update_table(sql_trans *tr, sql_table *t)
+gtr_update_table(sql_trans *tr, sql_table *t, int *tchanges)
{
int ok = LOG_OK;
node *n;
for (n = t->columns.set->h; ok == LOG_OK && n; n = n->next) {
+ int changes = 0;
sql_column *c = n->data;
if (!c->base.wtime)
continue;
- ok = gtr_update_delta(tr, c->data);
+ ok = gtr_update_delta(tr, c->data, &changes);
+ if (changes)
+ c->base.wtime = tr->wstime;
+ (*tchanges) |= changes;
}
if (ok == LOG_OK && t->idxs.set) {
for (n = t->idxs.set->h; ok == LOG_OK && n; n = n->next) {
+ int changes = 0;
sql_idx *ci = n->data;
/* some indices have no bats */
if (!ci->base.wtime)
continue;
- ok = gtr_update_delta(tr, ci->data);
+ ok = gtr_update_delta(tr, ci->data, &changes);
+ if (changes)
+ ci->base.wtime = tr->wstime;
+ (*tchanges) |= changes;
}
}
+ if (*tchanges)
_______________________________________________
checkin-list mailing list
[email protected]
http://mail.monetdb.org/mailman/listinfo/checkin-list
