Mercurial > hg > MonetDB

changeset 86156:1e594fe37abb smart-merge-jan22

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
Simplify commit API signature: remove redundant latest.
author Aris Koning <aris.koning@monetdbsolutions.com>
date Tue, 26 Jul 2022 11:00:18 +0200
parents 4412f52fd48a
children b59b4dbf66b3
files sql/include/sql_catalog.h sql/storage/bat/bat_storage.c sql/storage/objectset.c sql/storage/store.c
diffstat 4 files changed, 40 insertions(+), 54 deletions(-) [+]
line wrap: on
line diff
--- 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, ulng *active, ulng latest);/* commit/rollback changes */
+typedef int (*tc_commit_fptr) (struct sql_trans *tr, struct sql_change *c, ulng commit_ts, ulng oldest, ulng *active);/* 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);
--- 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, 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 commit_update_col( sql_trans *tr, sql_change *c, ulng commit_ts, ulng oldest, ulng *active);
+static int commit_update_idx( sql_trans *tr, sql_change *c, ulng commit_ts, ulng oldest, ulng *active);
+static int commit_update_del( sql_trans *tr, sql_change *c, ulng commit_ts, ulng oldest, ulng *active);
 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, 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 commit_create_col(sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active);
+static int commit_create_idx(sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active);
+static int commit_create_del(sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active);
 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, ulng *active, ulng latest)
+merge_segments(storage *s, sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active)
 {
 	segment *cur = s->segs->h, *seg = NULL;
 	for (; cur; cur = cur->next) {
@@ -402,11 +402,9 @@ merge_segments(storage *s, sql_trans *tr
 			/* 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) {
+		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;
 				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 */
@@ -428,17 +426,11 @@ merge_segments(storage *s, sql_trans *tr
 					else
 						mark4destroy(cur, change, commit_ts);
 					cur = seg;
-				}
-				/* skip merge */
-				else {
-					seg = cur;
+					continue;
 				}
 			}
-			/* skip merge */
-			else {
-				seg = cur;
-			}
 		}
+		seg = cur;
 	}
 }
 
@@ -3038,12 +3030,11 @@ log_create_col(sql_trans *tr, sql_change
 }
 
 static int
-commit_create_col_( sql_trans *tr, sql_column *c, ulng commit_ts, ulng oldest, ulng *active, ulng latest)
+commit_create_col_( sql_trans *tr, sql_column *c, ulng commit_ts, ulng oldest, ulng *active)
 {
 	int ok = LOG_OK;
 	(void)oldest;
 	(void)active;
-	(void)latest;
 
 	if(!isTempTable(c->t)) {
 		sql_delta *delta = ATOMIC_PTR_GET(&c->data);
@@ -3061,12 +3052,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, ulng *active, ulng latest)
+commit_create_col( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active)
 {
 	sql_column *c = (sql_column*)change->obj;
 	if (!tr->parent)
 		c->base.new = 0;
-	return commit_create_col_( tr, c, commit_ts, oldest, active, latest);
+	return commit_create_col_( tr, c, commit_ts, oldest, active);
 }
 
 /* will be called for new idx's and when new index columns are created */
@@ -3146,12 +3137,11 @@ log_create_idx(sql_trans *tr, sql_change
 }
 
 static int
-commit_create_idx_( sql_trans *tr, sql_idx *i, ulng commit_ts, ulng oldest, ulng *active, ulng latest)
+commit_create_idx_( sql_trans *tr, sql_idx *i, ulng commit_ts, ulng oldest, ulng *active)
 {
 	int ok = LOG_OK;
 	(void)oldest;
 	(void)active;
-	(void)latest;
 
 	if(!isTempTable(i->t)) {
 		sql_delta *delta = ATOMIC_PTR_GET(&i->data);
@@ -3168,12 +3158,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, ulng *active, ulng latest)
+commit_create_idx( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active)
 {
 	sql_idx *i = (sql_idx*)change->obj;
 	if (!tr->parent)
 		i->base.new = 0;
-	return commit_create_idx_(tr, i, commit_ts, oldest, active, latest);
+	return commit_create_idx_(tr, i, commit_ts, oldest, active);
 }
 
 static int
@@ -3385,7 +3375,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, ulng *active, ulng latest)
+commit_create_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active)
 {
 	int ok = LOG_OK;
 	sql_table *t = (sql_table*)change->obj;
@@ -3398,21 +3388,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, active, latest/* create is we are alone */ /*oldest*/);
+		merge_segments(dbat, tr, change, commit_ts, commit_ts, active);
 		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, active, latest);
+				ok = commit_create_col_(tr, c, commit_ts, oldest, active);
 			}
 			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, active, latest);
+						ok = commit_create_idx_(tr, i, commit_ts, oldest, active);
 				}
 			}
 			if (!tr->parent)
@@ -3550,14 +3540,13 @@ log_destroy_del(sql_trans *tr, sql_chang
 }
 
 static int
-commit_destroy_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active, ulng latest)
+commit_destroy_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active)
 {
 	(void)tr;
 	(void)change;
 	(void)commit_ts;
 	(void)oldest;
 	(void)active;
-	(void)latest;
 	return 0;
 }
 
@@ -4003,14 +3992,13 @@ log_update_col( sql_trans *tr, sql_chang
 }
 
 static int
-commit_update_col_( sql_trans *tr, sql_column *c, ulng commit_ts, ulng oldest, ulng *active, ulng latest)
+commit_update_col_( sql_trans *tr, sql_column *c, ulng commit_ts, ulng oldest, ulng *active)
 {
 	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) {
@@ -4064,14 +4052,14 @@ tc_gc_rollbacked_storage( sql_store Stor
 
 
 static int
-commit_update_col( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active, ulng latest)
+commit_update_col( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active)
 {
 	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, active, latest);
+		return commit_update_col_(tr, c, commit_ts, oldest, active);
 	if (commit_ts)
 		delta->cs.ts = commit_ts;
 	if (!commit_ts) { /* rollback */
@@ -4116,7 +4104,7 @@ log_update_idx( sql_trans *tr, sql_chang
 }
 
 static int
-commit_update_idx_( sql_trans *tr, sql_idx *i, ulng commit_ts, ulng oldest, ulng *active, ulng latest)
+commit_update_idx_( sql_trans *tr, sql_idx *i, ulng commit_ts, ulng oldest, ulng *active)
 {
 	int ok = LOG_OK;
 	sql_delta *delta = ATOMIC_PTR_GET(&i->data);
@@ -4124,7 +4112,6 @@ commit_update_idx_( sql_trans *tr, sql_i
 
 	(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) {
@@ -4147,14 +4134,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, ulng *active, ulng latest)
+commit_update_idx( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active)
 {
 	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, active, latest);
+		return commit_update_idx_( tr, i, commit_ts, oldest, active);
 	if (commit_ts)
 		delta->cs.ts = commit_ts;
 	if (!commit_ts) { /* rollback */
@@ -4231,7 +4218,7 @@ commit_storage(sql_trans *tr, storage *d
 }
 
 static int
-commit_update_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active, ulng latest)
+commit_update_del( sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active)
 {
 	int ok = LOG_OK;
 	sql_table *t = (sql_table*)change->obj;
@@ -4284,11 +4271,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, active, latest);
+			merge_segments(dbat, tr, change, commit_ts, oldest, active);
 		if (ok == LOG_OK && dbat == d && oldest == commit_ts)
 			ok = merge_storage(dbat);
 	} else if (ok == LOG_OK && tr->parent) {/* cleanup older save points */
-		merge_segments(dbat, tr, change, commit_ts, oldest, active, latest);
+		merge_segments(dbat, tr, change, commit_ts, oldest, active);
 		ATOMIC_PTR_SET(&t->data, savepoint_commit_storage(dbat, commit_ts));
 	}
 	unlock_table(tr->store, t->base.id);
--- a/sql/storage/objectset.c
+++ b/sql/storage/objectset.c
@@ -607,7 +607,7 @@ tc_gc_objectversion(sql_store store, sql
 }
 
 static int
-tc_commit_objectversion(sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active_, ulng latest)
+tc_commit_objectversion(sql_trans *tr, sql_change *change, ulng commit_ts, ulng oldest, ulng *active_)
 {
 	objectversion *ov = (objectversion*)change->data;
 	if (commit_ts) {
@@ -616,7 +616,6 @@ tc_commit_objectversion(sql_trans *tr, s
 		change->committed = commit_ts < TRANSACTION_ID_BASE ? true: false;
 		(void)oldest;
 		(void)active_;
-		(void)latest;
 		if (!tr->parent)
 			change->obj->new = 0;
 	}
--- a/sql/storage/store.c
+++ b/sql/storage/store.c
@@ -3640,9 +3640,9 @@ sql_trans_rollback(sql_trans *tr, bool c
 			sql_change *c = n->data;
 
 			if (c->commit) {
-				ulng latest = store_get_timestamp(store); /* get latest timestamp (to merge segments) */
-				ulng *active = store_get_active(store); /* get active transactions (to merge segments) */
-				c->commit(tr, c, 0 /* ie rollback */, oldest, active, latest);
+				ulng *active = malloc(sizeof(ulng) * (store->active->cnt + 1));
+				active = store_get_active(store); /* get active transactions (to merge segments) */
+				c->commit(tr, c, 0 /* ie rollback */, oldest, active);
 				free(active);
 			}
 			c->ts = commit_ts;
@@ -3994,9 +3994,9 @@ sql_trans_commit(sql_trans *tr)
 			sql_change *c = n->data;
 
 			if (c->commit && ok == LOG_OK) {
-				ulng latest = store_get_timestamp(store); /* get latest timestamp (to merge segments) */
-				ulng *active = store_get_active(store); /* get active transactions (to merge segments) */
-				ok = c->commit(tr, c, commit_ts, oldest, active, latest);
+				ulng *active = malloc(sizeof(ulng) * (store->active->cnt + 1));
+				active = store_get_active(store); /* get active transactions (to merge segments) */
+				ok = c->commit(tr, c, commit_ts, oldest, active);
 				free(active);
 			}
 			else