Re: [Monetdb-developers] MonetDB: default - Cleanup of logger code to split head and tail...

18 Jul 2012

On Wed, Jul 18, 2012 at 12:55:31PM +0200, Niels Nes wrote:
...
Changeset: 982fdf69e13b for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=982fdf69e13b
Modified Files:
   gdk/gdk_logger.c
   gdk/gdk_logger.h
   sql/server/rel_optimizer.c
   sql/storage/bat/bat_logger.c
   sql/storage/restrict/restrict_logger.c
Branch: default
Log Message:
Cleanup of logger code to split head and tail bat into 2 void headed bats
This checkin breaks the current on disk format, ie if your using the
defaul branch a reload is needed.
Before the (september) release we need to add updating functions
for atleast SQL.
Niels
...
diffs (truncated from 723 to 300 lines):

diff --git a/gdk/gdk_logger.c b/gdk/gdk_logger.c
--- a/gdk/gdk_logger.c
+++ b/gdk/gdk_logger.c
@@ -73,7 +73,10 @@
 #include "gdk_logger.h"
 #include <string.h>



+static BUN BUNfndT( BAT *b, ptr v)
+{

return BUNfnd(BATmirror(b), v);

+}
 /*

@-
The log record encoding is geared at reduced storage space, but

@@ -218,12 +221,11 @@ la_bat_clear(logger *lg, logaction *la)
   BAT *b;
/* do we need to skip these old updates */

if (BATcount(lg->snapshots)) {
BUN p = BUNfnd(lg->snapshots, &bid);




if (BATcount(lg->snapshots_bid)) {

BUN p = BUNfndT(lg->snapshots_bid, &bid);


if (p != BUN_NONE) {



	BATiter i = bat_iterator(lg->snapshots);


	int tid = *(int *) BUNtloc(i, p);




	int tid = *(int *) Tloc(lg->snapshots_tid, p);

if (lg->tid <= tid)
	return;



@@ -244,16 +246,18 @@ static int
 log_read_seq(logger *lg, logformat *l)
 {
   int seq = l->nr;

lng id;


lng val;
BUN p;


if (!mnstr_readLng(lg->log, &id))


if (!mnstr_readLng(lg->log, &val))
return LOG_ERR;


if (BUNfnd(lg->seqs, &seq) != BUN_NONE) {
BUNdelHead(lg->seqs, &seq, FALSE);




if ((p = BUNfndT(lg->seqs_id, &seq)) != BUN_NONE) {
BUNdelete(lg->seqs_id, p, FALSE);


BUNdelete(lg->seqs_val, p, FALSE);

}


BUNins(lg->seqs, &seq, &id, FALSE);



BUNappend(lg->seqs_id, &seq, FALSE);
BUNappend(lg->seqs_val, &val, FALSE);
return LOG_OK;

}
@@ -404,12 +408,11 @@ la_bat_updates(logger *lg, logaction *la
   	return;		/* ignore bats no longer in the catalog */
/* do we need to skip these old updates */

if (BATcount(lg->snapshots)) {
BUN p = BUNfnd(lg->snapshots, &bid);




if (BATcount(lg->snapshots_bid)) {

BUN p = BUNfndT(lg->snapshots_bid, &bid);


if (p != BUN_NONE) {



	BATiter i = bat_iterator(lg->snapshots);


	int tid = *(int *) BUNtloc(i, p);




	int tid = *(int *) Tloc(lg->snapshots_tid, p);

if (lg->tid <= tid)
	return;



@@ -477,10 +480,14 @@ la_bat_destroy(logger *lg, logaction *la
   log_bid bid = logger_find_bat(lg, la->name);
if (bid) {

BUN p;


logger_del_bat(lg, bid);


if (BUNfnd(lg->snapshots, &bid) != BUN_NONE) {


	BUNdelHead(lg->snapshots, &bid, FALSE);


	BUNins(lg->snapshots, &bid, &lg->tid, FALSE);




if ((p = BUNfndT(lg->snapshots_bid, &bid)) != BUN_NONE) {


	BUNdelete(lg->snapshots_bid, p, FALSE);


	BUNdelete(lg->snapshots_tid, p, FALSE);


	BUNappend(lg->snapshots_bid, &bid, FALSE);


	BUNappend(lg->snapshots_tid, &lg->tid, FALSE);

}
}

}
@@ -564,15 +571,19 @@ la_bat_use(logger *lg, logaction *la)
 {
   log_bid bid = la->nr;
   BAT *b = BATdescriptor(bid);

BUN p;
if (!b) {
GDKerror("logger: could not use bat (%d) for %s\n", (int) bid, la->name);
return;
}
logger_add_bat(lg, b, la->name);



if (BUNfnd(lg->snapshots, &b->batCacheid) != BUN_NONE)
BUNdelHead(lg->snapshots, &b->batCacheid, FALSE);


BUNins(lg->snapshots, &b->batCacheid, &lg->tid, FALSE);


if ((p = BUNfndT(lg->snapshots_bid, &b->batCacheid)) != BUN_NONE) {
BUNdelete(lg->snapshots_bid, p, FALSE);


BUNdelete(lg->snapshots_tid, p, FALSE);


}
BUNappend(lg->snapshots_bid, &b->batCacheid, FALSE);
BUNappend(lg->snapshots_tid, &lg->tid, FALSE);
logbat_destroy(b);

}
@@ -875,17 +886,23 @@ static int
 logger_commit(logger *lg)
 {
   int id = LOG_SID;

BUN p;
if (lg->debug & 1)
fprintf(stderr, "logger_commit\n");



BUNdelHead(lg->seqs, &id, FALSE);
BUNins(lg->seqs, &id, &lg->id, FALSE);


p = BUNfndT(lg->seqs_id, &id);

BUNdelete(lg->seqs_id, p, FALSE);

BUNdelete(lg->seqs_val, p, FALSE);

BUNappend(lg->seqs_id, &id, FALSE);

BUNappend(lg->seqs_val, &lg->id, FALSE);
/* cleanup old snapshots */



if (BATcount(lg->snapshots)) {
BATclear(lg->snapshots, FALSE);


BATcommit(lg->snapshots);




if (BATcount(lg->snapshots_bid)) {
BATclear(lg->snapshots_bid, FALSE);


BATclear(lg->snapshots_tid, FALSE);


BATcommit(lg->snapshots_bid);


BATcommit(lg->snapshots_tid);

}
return bm_commit(lg);

}
@@ -919,34 +936,34 @@ check_version(logger *lg, FILE *fp)
 }
static int
-bm_subcommit(BAT *list, BAT *catalog, BAT *extra, int debug)
+bm_subcommit(BAT *list_bid, BAT *list_nme, BAT *catalog_bid, BAT *catalog_nme, BAT *extra, int debug)
 {
   BUN p, q;

BUN nn = 2 + (list->batFirst > list->batDeleted ? list->batFirst - list->batDeleted : 0) + BATcount(list) + (extra ? BATcount(extra) : 0);


BUN nn = 3 + (list_bid->batFirst > list_bid->batDeleted ? list_bid->batFirst - list_bid->batDeleted : 0) + BATcount(list_bid) + (extra ? BATcount(extra) : 0);
bat *n = (bat*)GDKmalloc(sizeof(bat) * nn);
int i = 0;


BATiter iter = bat_iterator(list);


BATiter iter = (list_nme)?bat_iterator(list_nme):bat_iterator(list_bid);
int res;
n[i++] = 0;		/* n[0] is not used */
/* first loop over deleted then over current and new */



for (p = list->batDeleted; p < list->batFirst; p++) {
bat col = *(log_bid *) BUNhead(iter, p);




for (p = list_bid->batDeleted; p < list_bid->batFirst; p++) {

bat col = *(log_bid *) Tloc(list_bid, p);


if (debug & 1)
	fprintf(stderr, "commit deleted %s (%d) %s\n",
		BBPname(col), col,



		(list == catalog) ? BUNtail(iter, p) : "snapshot");




		(list_bid == catalog_bid) ? BUNtail(iter, p) : "snapshot");

n[i++] = ABS(col);
}


BATloop(list, p, q) {
bat col = *(log_bid *) BUNhead(iter, p);




BATloop(list_bid, p, q) {

bat col = *(log_bid *) Tloc(list_bid, p);


if (debug & 1)
	fprintf(stderr, "commit new %s (%d) %s\n",
		BBPname(col), col,



		(list == catalog) ? BUNtail(iter, p) : "snapshot");




		(list_bid == catalog_bid) ? BUNtail(iter, p) : "snapshot");

n[i++] = ABS(col);
}
if (extra) {

@@ -957,14 +974,16 @@ bm_subcommit(BAT *list, BAT *catalog, BA
   		if (debug & 1)
   			fprintf(stderr, "commit extra %s %s\n",
   				name,

			(list == catalog) ? BUNtail(iter, p) : "snapshot");




			(list_bid == catalog_bid) ? BUNtail(iter, p) : "snapshot");
n[i++] = ABS(BBPindex(name));

}
}
/* now commit catalog, so it's also up to date on disk */


n[i++] = ABS(catalog->batCacheid);


n[i++] = ABS(catalog_bid->batCacheid);
n[i++] = ABS(catalog_nme->batCacheid);
assert((BUN) i <= nn);


BATcommit(catalog);


BATcommit(catalog_bid);
BATcommit(catalog_nme);
res = TMsubcommit_list(n, i);
GDKfree(n);
return res;

@@ -992,8 +1011,8 @@ logger_new(int debug, char *fn, char *lo
   FILE *fp;
   char filename[BUFSIZ];
   char bak[BUFSIZ];

log_bid seqs = 0;
bat catalog;


log_bid seqs_id = 0;

bat catalog_bid, catalog_nme;
if (lg == NULL)
return NULL;


@@ -1022,9 +1041,12 @@ logger_new(int debug, char *fn, char *lo
   lg->postfuncp = postfuncp;
   lg->log = NULL;
   lg->end = 0;

lg->catalog = NULL;
lg->snapshots = NULL;
lg->seqs = NULL;


lg->catalog_bid = NULL;

lg->catalog_nme = NULL;

lg->snapshots_bid = NULL;

lg->snapshots_tid = NULL;

lg->seqs_id = NULL;

lg->seqs_val = NULL;
snprintf(filename, BUFSIZ, "%s%s", lg->dir, LOGFILE);
snprintf(bak, BUFSIZ, "%s.bak", filename);


@@ -1040,10 +1062,10 @@ logger_new(int debug, char *fn, char *lo
   }
   fp = fopen(filename, "r");

snprintf(bak, BUFSIZ, "%s_catalog", fn);
catalog = BBPindex(bak);


snprintf(bak, BUFSIZ, "%s_catalog_bid", fn);
catalog_bid = BBPindex(bak);


if (catalog == 0) {


if (catalog_bid == 0) {
log_bid bid = 0;
/* catalog does not exist, so the log file also shouldn't exist */


@@ -1055,16 +1077,24 @@ logger_new(int debug, char *fn, char *lo
   		goto error;
   	}

lg->catalog = logbat_new(TYPE_int, TYPE_str, BATSIZE);




lg->catalog_bid = logbat_new(TYPE_void, TYPE_int, BATSIZE);


lg->catalog_nme = logbat_new(TYPE_void, TYPE_str, BATSIZE);

if (debug)
	fprintf(stderr, "create %s catalog\n", fn);


bid = lg->catalog->batCacheid;




/* Make persistent */


bid = lg->catalog_bid->batCacheid;


BBPincref(bid, TRUE);


BATmode(lg->catalog_bid, PERSISTENT);


snprintf(bak, BUFSIZ, "%s_catalog_bid", fn);


BBPrename(lg->catalog_bid->batCacheid, bak);


/* Make persistent */

bid = lg->catalog_nme->batCacheid;

BBPincref(bid, TRUE);



BATmode(lg->catalog, PERSISTENT);


BBPrename(lg->catalog->batCacheid, bak);




BATmode(lg->catalog_nme, PERSISTENT);


snprintf(bak, BUFSIZ, "%s_catalog_nme", fn);


BBPrename(lg->catalog_nme->batCacheid, bak);


if (!GDKcreatedir(filename)) {
	logger_fatal("logger_new: cannot create directory for log file %s\n",


@@ -1082,7 +1112,7 @@ logger_new(int debug, char *fn, char *lo
   	fclose(fp);
   	fp = NULL;

if (bm_subcommit(lg->catalog, lg->catalog, NULL, lg->debug) != 0) {




if (bm_subcommit(lg->catalog_bid, lg->catalog_nme, lg->catalog_bid, lg->catalog_nme, NULL, lg->debug) != 0) {
/* cannot commit catalog, so remove log */
unlink(filename);
goto error;



@@ -1092,8 +1122,11 @@ logger_new(int debug, char *fn, char *lo
   	   require a logical reference we also add a logical
   	   reference for the persistent bats */
   	BUN p, q;

BAT *b = BATdescriptor(catalog);


BATiter bi = bat_iterator(b);




BAT *b = BATdescriptor(catalog_bid), *n;



snprintf(bak, BUFSIZ, "%s_catalog_nme", fn);


catalog_nme = BBPindex(bak);


n = BATdescriptor(catalog_nme);


/* the catalog exists, and so should the log file */
if (fp == NULL) {


@@ -1107,42 +1140,63 @@ logger_new(int debug, char *fn, char *lo
_______________________________________________
Checkin-list mailing list
Checkin-list@monetdb.org
http://mail.monetdb.org/mailman/listinfo/checkin-list
-- 
Niels Nes, Centrum Wiskunde & Informatica (CWI)
Science Park 123, 1098 XG Amsterdam, The Netherlands
room L3.14,  phone ++31 20 592-4098 	sip:4098@sip.cwi.nl
url: http://www.cwi.nl/~niels   e-mail: Niels.Nes@cwi.nl

    