Changeset: 46beead61552 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/46beead61552
Modified Files:
sql/include/sql_catalog.h
sql/storage/bat/bat_storage.c
sql/storage/objectset.c
sql/storage/sql_storage.h
sql/storage/store.c
Branch: smart-merge-jan22
Log Message:
Implement smarter segment merging
diffs (truncated from 389 to 300 lines):
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
@@ -233,7 +233,7 @@ struct os_iter {
/* transaction changes */
typedef int (*tc_valid_fptr) (struct sql_trans *tr, struct sql_change *c/*,
ulng commit_ts, ulng oldest*/);
typedef int (*tc_log_fptr) (struct sql_trans *tr, struct sql_change *c);
/* write changes to the
log */
-typedef int (*tc_commit_fptr) (struct sql_trans *tr, struct sql_change *c,
ulng commit_ts, ulng oldest);/* commit/rollback changes */
+typedef int (*tc_commit_fptr) (struct sql_trans *tr, struct sql_change *c,
ulng commit_ts, ulng oldest, ulng *active, ulng latest);/* commit/rollback
changes */
typedef int (*tc_cleanup_fptr) (sql_store store, struct sql_change *c, ulng
oldest); /* garbage collection, ie cleanup structures when possible */
typedef void (*destroy_fptr)(sql_store store, sql_base *b);
typedef int (*validate_fptr)(struct sql_trans *tr, sql_base *b, int delete);
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
@@ -19,15 +19,15 @@
static int log_update_col( sql_trans *tr, sql_change *c);
static int log_update_idx( sql_trans *tr, sql_change *c);
static int log_update_del( sql_trans *tr, sql_change *c);
-static int commit_update_col( sql_trans *tr, sql_change *c, ulng commit_ts,
ulng oldest);
-static int commit_update_idx( sql_trans *tr, sql_change *c, ulng commit_ts,
ulng oldest);
-static int commit_update_del( sql_trans *tr, sql_change *c, ulng commit_ts,
ulng oldest);
+static int commit_update_col( sql_trans *tr, sql_change *c, ulng commit_ts,
ulng oldest, ulng *active, ulng latest);
+static int commit_update_idx( sql_trans *tr, sql_change *c, ulng commit_ts,
ulng oldest, ulng *active, ulng latest);
+static int commit_update_del( sql_trans *tr, sql_change *c, ulng commit_ts,
ulng oldest, ulng *active, ulng latest);
static int log_create_col(sql_trans *tr, sql_change *change);
static int log_create_idx(sql_trans *tr, sql_change *change);
static int log_create_del(sql_trans *tr, sql_change *change);
-static int commit_create_col(sql_trans *tr, sql_change *change, ulng
commit_ts, ulng oldest);
-static int commit_create_idx(sql_trans *tr, sql_change *change, ulng
commit_ts, ulng oldest);
-static int commit_create_del(sql_trans *tr, sql_change *change, ulng
commit_ts, ulng oldest);
+static int commit_create_col(sql_trans *tr, sql_change *change, ulng
commit_ts, ulng oldest, ulng *active, ulng latest);
+static int commit_create_idx(sql_trans *tr, sql_change *change, ulng
commit_ts, ulng oldest, ulng *active, ulng latest);
+static int commit_create_del(sql_trans *tr, sql_change *change, ulng
commit_ts, ulng oldest, ulng *active, ulng latest);
static int tc_gc_col( sql_store Store, sql_change *c, ulng oldest);
static int tc_gc_idx( sql_store Store, sql_change *c, ulng oldest);
static int tc_gc_del( sql_store Store, sql_change *c, ulng oldest);
@@ -389,7 +389,7 @@ segments2cs(sql_trans *tr, segments *seg
/* TODO return LOG_OK/ERR */
static void
-merge_segments(storage *s, sql_trans *tr, sql_change *change, ulng commit_ts,
ulng oldest)
+merge_segments(storage *s, sql_trans *tr, sql_change *change, ulng commit_ts,
ulng oldest, ulng *active, ulng latest)
{
segment *cur = s->segs->h, *seg = NULL;
for (; cur; cur = cur->next) {
@@ -398,22 +398,45 @@ 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 */
+ if (!seg) {
+ /* first segment */
+ seg = cur;
+ }
+ else {
+ /* possible merge since both deleted flags are equal
+ and the timestamp is lesser than the latest one
+ when the active transactions were computed */
+ if (seg->deleted == cur->deleted && seg->ts <= latest
&& cur->ts < TRANSACTION_ID_BASE) {
+ int merge = 1;
+ for (int i = 0; active[i] != 0; i++) {
+ if (active[i] != tr->ts /* the
transaction can ignore its own active timestamp since it's committing */
+ && ((active[i] >= seg->ts &&
active[i] <= cur->ts)
+ || (active[i] <=
seg->ts && active[i] >= cur->ts))) {
+ /* cannot safely merge since
there is an active transaction between the segments */
+ merge = 0;
+ 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;
+ }
+ /* skip merge */
+ else {
+ seg = cur;
+ }
+ }
+ /* skip merge */
+ else {
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 */
}
}
}
@@ -3015,10 +3038,12 @@ log_create_col(sql_trans *tr, sql_change
}
static int
-commit_create_col_( sql_trans *tr, sql_column *c, ulng commit_ts, ulng oldest)
+commit_create_col_( sql_trans *tr, sql_column *c, ulng commit_ts, ulng oldest,
ulng *active, ulng latest)
{
int ok = LOG_OK;
(void)oldest;
+ (void)active;
+ (void)latest;
if(!isTempTable(c->t)) {
sql_delta *delta = ATOMIC_PTR_GET(&c->data);
@@ -3036,12 +3061,12 @@ commit_create_col_( sql_trans *tr, sql_c
}
static int
-commit_create_col( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest)
+commit_create_col( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest, ulng *active, ulng latest)
{
sql_column *c = (sql_column*)change->obj;
if (!tr->parent)
c->base.new = 0;
- return commit_create_col_( tr, c, commit_ts, oldest);
+ return commit_create_col_( tr, c, commit_ts, oldest, active, latest);
}
/* will be called for new idx's and when new index columns are created */
@@ -3121,10 +3146,12 @@ log_create_idx(sql_trans *tr, sql_change
}
static int
-commit_create_idx_( sql_trans *tr, sql_idx *i, ulng commit_ts, ulng oldest)
+commit_create_idx_( sql_trans *tr, sql_idx *i, ulng commit_ts, ulng oldest,
ulng *active, ulng latest)
{
int ok = LOG_OK;
(void)oldest;
+ (void)active;
+ (void)latest;
if(!isTempTable(i->t)) {
sql_delta *delta = ATOMIC_PTR_GET(&i->data);
@@ -3141,12 +3168,12 @@ commit_create_idx_( sql_trans *tr, sql_i
}
static int
-commit_create_idx( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest)
+commit_create_idx( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest, ulng *active, ulng latest)
{
sql_idx *i = (sql_idx*)change->obj;
if (!tr->parent)
i->base.new = 0;
- return commit_create_idx_(tr, i, commit_ts, oldest);
+ return commit_create_idx_(tr, i, commit_ts, oldest, active, latest);
}
static int
@@ -3358,7 +3385,7 @@ log_create_del(sql_trans *tr, sql_change
}
static int
-commit_create_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest)
+commit_create_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest, ulng *active, ulng latest)
{
int ok = LOG_OK;
sql_table *t = (sql_table*)change->obj;
@@ -3371,21 +3398,21 @@ commit_create_del( sql_trans *tr, sql_ch
assert(ok == LOG_OK);
if (ok != LOG_OK)
return ok;
- merge_segments(dbat, tr, change, commit_ts, commit_ts/* create
is we are alone */ /*oldest*/);
+ merge_segments(dbat, tr, change, commit_ts, commit_ts, active,
latest/* create is we are alone */ /*oldest*/);
assert(dbat->cs.ts == tr->tid);
dbat->cs.ts = commit_ts;
if (ok == LOG_OK) {
for(node *n = ol_first_node(t->columns); n && ok ==
LOG_OK; n = n->next) {
sql_column *c = n->data;
- ok = commit_create_col_(tr, c, commit_ts,
oldest);
+ ok = commit_create_col_(tr, c, commit_ts,
oldest, active, latest);
}
if (t->idxs) {
for(node *n = ol_first_node(t->idxs); n && ok
== LOG_OK; n = n->next) {
sql_idx *i = n->data;
if (ATOMIC_PTR_GET(&i->data))
- ok = commit_create_idx_(tr, i,
commit_ts, oldest);
+ ok = commit_create_idx_(tr, i,
commit_ts, oldest, active, latest);
}
}
if (!tr->parent)
@@ -3523,12 +3550,14 @@ log_destroy_del(sql_trans *tr, sql_chang
}
static int
-commit_destroy_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest)
+commit_destroy_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest, ulng *active, ulng latest)
{
(void)tr;
(void)change;
(void)commit_ts;
(void)oldest;
+ (void)active;
+ (void)latest;
return 0;
}
@@ -3974,12 +4003,14 @@ log_update_col( sql_trans *tr, sql_chang
}
static int
-commit_update_col_( sql_trans *tr, sql_column *c, ulng commit_ts, ulng oldest)
+commit_update_col_( sql_trans *tr, sql_column *c, ulng commit_ts, ulng oldest,
ulng *active, ulng latest)
{
int ok = LOG_OK;
sql_delta *delta = ATOMIC_PTR_GET(&c->data);
(void)oldest;
+ (void)active;
+ (void)latest;
if (isTempTable(c->t)) {
if (commit_ts) { /* commit */
if (c->t->commit_action == CA_COMMIT ||
c->t->commit_action == CA_PRESERVE) {
@@ -4033,14 +4064,14 @@ tc_gc_rollbacked_storage( sql_store Stor
static int
-commit_update_col( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest)
+commit_update_col( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest, ulng *active, ulng latest)
{
int ok = LOG_OK;
sql_column *c = (sql_column*)change->obj;
sql_delta *delta = ATOMIC_PTR_GET(&c->data);
if (isTempTable(c->t))
- return commit_update_col_(tr, c, commit_ts, oldest);
+ return commit_update_col_(tr, c, commit_ts, oldest, active,
latest);
if (commit_ts)
delta->cs.ts = commit_ts;
if (!commit_ts) { /* rollback */
@@ -4085,13 +4116,15 @@ log_update_idx( sql_trans *tr, sql_chang
}
static int
-commit_update_idx_( sql_trans *tr, sql_idx *i, ulng commit_ts, ulng oldest)
+commit_update_idx_( sql_trans *tr, sql_idx *i, ulng commit_ts, ulng oldest,
ulng *active, ulng latest)
{
int ok = LOG_OK;
sql_delta *delta = ATOMIC_PTR_GET(&i->data);
int type = (oid_index(i->type))?TYPE_oid:TYPE_lng;
(void)oldest;
+ (void)active;
+ (void)latest;
if (isTempTable(i->t)) {
if (commit_ts) { /* commit */
if (i->t->commit_action == CA_COMMIT ||
i->t->commit_action == CA_PRESERVE) {
@@ -4114,14 +4147,14 @@ commit_update_idx_( sql_trans *tr, sql_i
}
static int
-commit_update_idx( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest)
+commit_update_idx( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest, ulng *active, ulng latest)
{
int ok = LOG_OK;
sql_idx *i = (sql_idx*)change->obj;
sql_delta *delta = ATOMIC_PTR_GET(&i->data);
if (isTempTable(i->t))
- return commit_update_idx_( tr, i, commit_ts, oldest);
+ return commit_update_idx_( tr, i, commit_ts, oldest, active,
latest);
if (commit_ts)
delta->cs.ts = commit_ts;
if (!commit_ts) { /* rollback */
@@ -4198,7 +4231,7 @@ commit_storage(sql_trans *tr, storage *d
}
static int
-commit_update_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest)
+commit_update_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng
oldest, ulng *active, ulng latest)
{
int ok = LOG_OK;
sql_table *t = (sql_table*)change->obj;
@@ -4251,11 +4284,11 @@ commit_update_del( sql_trans *tr, sql_ch
ok = segments2cs(tr, dbat->segs, &dbat->cs);
assert(ok == LOG_OK);
if (ok == LOG_OK)
- merge_segments(dbat, tr, change, commit_ts, oldest);
+ merge_segments(dbat, tr, change, commit_ts, oldest,
active, latest);
if (ok == LOG_OK && dbat == d && oldest == commit_ts)
ok = merge_storage(dbat);
} else if (ok == LOG_OK && tr->parent) {/* cleanup older save points */
_______________________________________________
checkin-list mailing list -- checkin-list@monetdb.org
To unsubscribe send an email to checkin-list-le...@monetdb.org
