Changeset: 023b9d2e912d for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/023b9d2e912d
Removed Files:
dummy.txt
Branch: Jan2022
Log Message:
Merge with smart-merge-jan22.
diffs (truncated from 481 to 300 lines):
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
@@ -391,6 +391,7 @@ segments2cs(sql_trans *tr, segments *seg
static void
merge_segments(storage *s, sql_trans *tr, sql_change *change, ulng commit_ts,
ulng oldest)
{
+ sqlstore* store = tr->store;
segment *cur = s->segs->h, *seg = NULL;
for (; cur; cur = cur->next) {
if (cur->ts == tr->tid) {
@@ -398,24 +399,47 @@ merge_segments(storage *s, sql_trans *tr
cur->oldts = 0;
cur->ts = commit_ts;
}
- if (cur->ts <= oldest && cur->ts < TRANSACTION_ID_BASE) { /*
possibly merge range */
- if (!seg) { /* skip first */
- seg = cur;
- } else if (seg->end == cur->start && seg->deleted ==
cur->deleted) {
- /* merge with previous */
- seg->end = cur->end;
- seg->next = cur->next;
- if (cur == s->segs->t)
- s->segs->t = seg;
- if (commit_ts == oldest)
- _DELETE(cur);
- else
- mark4destroy(cur, change, commit_ts);
- cur = seg;
- } else {
- seg = cur; /* begin of new merge */
+ if (!seg) {
+ /* first segment */
+ seg = cur;
+ }
+ else if (seg->ts < TRANSACTION_ID_BASE) {
+ /* possible merge since both deleted flags are equal */
+ if (seg->deleted == cur->deleted && cur->ts <
TRANSACTION_ID_BASE) {
+ int merge = 1;
+ node *n = store->active->h;
+ for (int i = 0; i < store->active->cnt; i++, n
= n->next) {
+ ulng active = ((sql_trans*)n->data)->ts;
+ assert(active != seg->ts && active !=
cur->ts);
+
+ if (active == tr->ts)
+ continue; /* pretent that
committing transaction has already committed and is no longer active */
+ if (seg->ts < active && cur->ts <
active)
+ break;
+ if (seg->ts > active && cur->ts >
active)
+ continue;
+
+ assert((active > seg->ts && active <
cur->ts) || (active < seg->ts && active > cur->ts));
+ /* cannot safely merge since there is
an active transaction between the segments */
+ merge = false;
+ break;
+ }
+ /* merge segments */
+ if (merge) {
+ seg->end = cur->end;
+ seg->next = cur->next;
+ if (cur == s->segs->t)
+ s->segs->t = seg;
+ if (commit_ts == oldest)
+ _DELETE(cur);
+ else
+ mark4destroy(cur, change,
commit_ts);
+ cur = seg;
+ continue;
+ }
}
}
+ seg = cur;
}
}
@@ -3297,12 +3321,18 @@ static int
log_segments(sql_trans *tr, segments *segs, sqlid id)
{
/* log segments */
+ lock_table(tr->store, id);
for (segment *seg = segs->h; seg; seg=seg->next) {
+ unlock_table(tr->store, id);
if (seg->ts == tr->tid && seg->end-seg->start) {
- if (log_segment(tr, seg, id) != LOG_OK)
+ if (log_segment(tr, seg, id) != LOG_OK) {
+ unlock_table(tr->store, id);
return LOG_ERR;
+ }
}
+ lock_table(tr->store, id);
}
+ unlock_table(tr->store, id);
return LOG_OK;
}
@@ -3790,7 +3820,9 @@ log_table_append(sql_trans *tr, sql_tabl
if (isTempTable(t))
return LOG_OK;
size_t end = segs_end(segs, tr, t);
+ lock_table(tr->store, t->base.id);
for (segment *cur = segs->h; cur && ok; cur = cur->next) {
+ unlock_table(tr->store, t->base.id);
if (cur->ts == tr->tid && !cur->deleted && cur->start < end) {
for (node *n = ol_first_node(t->columns); n && ok; n =
n->next) {
sql_column *c = n->data;
@@ -3835,7 +3867,9 @@ log_table_append(sql_trans *tr, sql_tabl
}
}
}
+ lock_table(tr->store, t->base.id);
}
+ unlock_table(tr->store, t->base.id);
return ok == GDK_SUCCEED ? LOG_OK : LOG_ERR;
}
@@ -3846,8 +3880,6 @@ log_storage(sql_trans *tr, sql_table *t,
if (ok == LOG_OK && cleared)
ok = tr_log_cs(tr, t, &s->cs, s->segs->h, t->base.id);
if (ok == LOG_OK)
- ok = segments2cs(tr, s->segs, &s->cs);
- if (ok == LOG_OK)
ok = log_segments(tr, s->segs, id);
if (ok == LOG_OK && !cleared)
ok = log_table_append(tr, t, s->segs);
diff --git a/sql/storage/store.c b/sql/storage/store.c
--- a/sql/storage/store.c
+++ b/sql/storage/store.c
@@ -7006,7 +7006,7 @@ sql_trans_begin(sql_session *s)
tr->active = 1;
(void) ATOMIC_INC(&store->nr_active);
- list_append(store->active, s);
+ list_append(store->active, tr);
TRC_DEBUG(SQL_STORE, "Exit sql_trans_begin for transaction: " ULLFMT
"\n", tr->tid);
store_unlock(store);
@@ -7024,20 +7024,20 @@ sql_trans_end(sql_session *s, int ok)
sql_trans_rollback(s->tr, false);
}
assert(s->tr->active);
+ sqlstore *store = s->tr->store;
+ store_lock(store);
s->tr->active = 0;
s->tr->status = 0;
s->auto_commit = s->ac_on_commit;
- sqlstore *store = s->tr->store;
- store_lock(store);
- list_remove_data(store->active, NULL, s);
+ list_remove_data(store->active, NULL, s->tr);
ATOMIC_SET(&store->lastactive, GDKusec());
(void) ATOMIC_DEC(&store->nr_active);
ulng oldest = store_get_timestamp(store);
if (store->active && store->active->h) {
for(node *n = store->active->h; n; n = n->next) {
- sql_session *s = n->data;
- if (s->tr->ts < oldest)
- oldest = s->tr->ts;
+ sql_trans *tr = n->data;
+ if (tr->ts < oldest)
+ oldest = tr->ts;
}
}
store->oldest = oldest;
diff --git a/sql/test/concurrent/Tests/All b/sql/test/concurrent/Tests/All
--- a/sql/test/concurrent/Tests/All
+++ b/sql/test/concurrent/Tests/All
@@ -1,2 +1,3 @@
simple_select
crash_on_concurrent_use.SF-1411926
+smart-segment-merge
diff --git a/sql/test/concurrent/Tests/smart-segment-merge.test
b/sql/test/concurrent/Tests/smart-segment-merge.test
new file mode 100644
--- /dev/null
+++ b/sql/test/concurrent/Tests/smart-segment-merge.test
@@ -0,0 +1,306 @@
+# INIT
+
+@connection(id=1, username=monetdb, password=monetdb)
+statement ok
+CREATE TABLE Test (k int);
+
+@connection(id=1)
+query I
+SELECT segments FROM sys.deltas('sys', 'test');
+----
+1
+
+@connection(id=1)
+statement ok
+INSERT INTO Test SELECT value FROM generate_series(1, 11);
+
+@connection(id=1)
+query I
+SELECT segments FROM sys.deltas('sys', 'test');
+----
+1
+
+
+# TEST INSERTS
+# the four uncommitted transactions create four new segments, appended to the
tail of the segment list;
+# the first commit (id=1) does not merge any segments since active
transactions prevent it;
+# the second commit (id=3) does not merge any segments since there is an
uncommitted segment of id=2 between id=1 and id=3;
+# the third commit (id=4) is able to merge its segment plus the segment
inserted by the transaction id=3,
+# as they are next to each other and the active transaction is not able to
read them;
+# the fourth commit is able to merge everything, as there are no active
transactions and all segments are contiguous.
+
+@connection(id=1)
+statement ok
+begin transaction
+
+@connection(id=2, username=monetdb, password=monetdb)
+statement ok
+begin transaction
+
+@connection(id=3, username=monetdb, password=monetdb)
+statement ok
+begin transaction
+
+@connection(id=4, username=monetdb, password=monetdb)
+statement ok
+begin transaction
+
+@connection(id=1)
+statement ok
+INSERT INTO Test VALUES (11);
+
+@connection(id=1)
+query I
+SELECT segments FROM sys.deltas('sys', 'test');
+----
+2
+
+@connection(id=2)
+statement ok
+INSERT INTO Test VALUES (12);
+
+@connection(id=1)
+query I
+SELECT segments FROM sys.deltas('sys', 'test');
+----
+3
+
+@connection(id=3)
+statement ok
+INSERT INTO Test VALUES (13), (14);
+
+@connection(id=1)
+query I
+SELECT segments FROM sys.deltas('sys', 'test');
+----
+4
+
+@connection(id=4)
+statement ok
+INSERT INTO Test VALUES (15);
+
+@connection(id=1)
+query I
+SELECT segments FROM sys.deltas('sys', 'test');
+----
+5
+
+@connection(id=1)
+statement ok
+commit
+
+@connection(id=1)
+query I
+SELECT segments FROM sys.deltas('sys', 'test');
+----
+5
+
+@connection(id=3)
+statement ok
+commit
+
+@connection(id=1)
+query I
+SELECT segments FROM sys.deltas('sys', 'test');
+----
+5
+
+@connection(id=4)
+statement ok
+commit
+
+@connection(id=1)
+query I
+SELECT segments FROM sys.deltas('sys', 'test');
+----
+4
+
+@connection(id=1)
+query T
+SELECT listagg(k) FROM test;
+----
+1,2,3,4,5,6,7,8,9,10,11,13,14,15
+
+@connection(id=2)
+query T
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
