Changeset: fdfc6bf915d8 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=fdfc6bf915d8
Modified Files:
gdk/gdk_logger.c
gdk/gdk_logger.h
gdk/gdk_logger_internals.h
sql/backends/monet5/rel_bin.c
sql/backends/monet5/sql.c
sql/backends/monet5/sql.mal
sql/backends/monet5/sql_result.c
sql/backends/monet5/sql_result.h
sql/common/sql_types.c
sql/include/sql_relation.h
sql/server/rel_updates.c
sql/server/sql_parser.y
sql/server/sql_scan.c
sql/storage/bat/bat_logger.c
sql/storage/bat/bat_storage.c
sql/storage/sql_storage.h
sql/storage/store.c
Branch: unlock
Log Message:
lots of cleanup of very old log versions (we will not support with this change)
removed LOCKED mode for loading.
Status is still far from functional
diffs (truncated from 3182 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
@@ -217,18 +217,19 @@ static int
log_read_format(logger *l, logformat *data)
{
assert(!l->inmemory);
- return mnstr_read(l->log, &data->flag, 1, 1) == 1 &&
- mnstr_readLng(l->log, &data->nr) == 1 &&
- mnstr_readInt(l->log, &data->tid) == 1;
+ return mnstr_read(l->input_log, &data->flag, 1, 1) == 1 &&
+ mnstr_readLng(l->input_log, &data->nr) == 1 &&
+ mnstr_readInt(l->input_log, &data->tid) == 1;
}
static gdk_return
log_write_format(logger *l, logformat *data)
{
assert(!l->inmemory);
- if (mnstr_write(l->log, &data->flag, 1, 1) == 1 &&
- mnstr_writeLng(l->log, data->nr) &&
- mnstr_writeInt(l->log, data->tid))
+
+ if (mnstr_write(l->output_log, &data->flag, 1, 1) == 1 &&
+ mnstr_writeLng(l->output_log, data->nr) &&
+ mnstr_writeInt(l->output_log, data->tid))
return GDK_SUCCEED;
fprintf(stderr, "!ERROR: log_write_format: write failed\n");
return GDK_FAIL;
@@ -242,9 +243,8 @@ log_read_string(logger *l)
char *buf;
assert(!l->inmemory);
- if (mnstr_readInt(l->log, &len) != 1) {
+ if (mnstr_readInt(l->input_log, &len) != 1) {
fprintf(stderr, "!ERROR: log_read_string: read failed\n");
-//MK This leads to non-repeatable log structure?
return NULL;
}
if (len == 0)
@@ -256,7 +256,7 @@ log_read_string(logger *l)
return (char *) -1;
}
- if ((nr = mnstr_read(l->log, buf, 1, len)) != (ssize_t) len) {
+ if ((nr = mnstr_read(l->input_log, buf, 1, len)) != (ssize_t) len) {
buf[len - 1] = 0;
fprintf(stderr, "!ERROR: log_read_string: couldn't read name
(%s) %zd\n", buf, nr);
GDKfree(buf);
@@ -274,8 +274,8 @@ log_write_string(logger *l, const char *
assert(!l->inmemory);
assert(len > 1);
assert(len <= INT_MAX);
- if (!mnstr_writeInt(l->log, (int) len) ||
- mnstr_write(l->log, n, 1, len) != (ssize_t) len) {
+ if (!mnstr_writeInt(l->output_log, (int) len) ||
+ mnstr_write(l->output_log, n, 1, len) != (ssize_t) len) {
fprintf(stderr, "!ERROR: log_write_string: write failed\n");
return GDK_FAIL;
}
@@ -287,6 +287,10 @@ log_read_clear(logger *lg, trans *tr, ch
{
if (lg->debug & 1)
fprintf(stderr, "#logger found log_read_clear %s\n", NAME(name,
tpe, id));
+
+ if (tr->tid <= lg->saved_tid) /* skipping */
+ return LOG_OK;
+
if (tr_grow(tr) != GDK_SUCCEED)
return LOG_ERR;
tr->changes[tr->nr].type = LOG_CLEAR;
@@ -339,7 +343,7 @@ la_bat_clear(logger *lg, logaction *la)
}
static log_return
-log_read_seq(logger *lg, logformat *l)
+log_read_seq(logger *lg, trans *tr, logformat *l)
{
int seq = (int) l->nr;
lng val;
@@ -347,11 +351,14 @@ log_read_seq(logger *lg, logformat *l)
assert(!lg->inmemory);
assert(l->nr <= (lng) INT_MAX);
- if (mnstr_readLng(lg->log, &val) != 1) {
+ if (mnstr_readLng(lg->input_log, &val) != 1) {
fprintf(stderr, "!ERROR: log_read_seq: read failed\n");
return LOG_EOF;
}
+ if (tr->tid <= lg->saved_tid) /* skipping */
+ return LOG_OK;
+
if ((p = log_find(lg->seqs_id, lg->dseqs, seq)) != BUN_NONE &&
p >= lg->seqs_id->batInserted) {
if (BUNinplace(lg->seqs_val, p, &val, false) != GDK_SUCCEED)
@@ -375,8 +382,8 @@ log_write_id(logger *l, char tpe, oid id
lng lid = id;
assert(!l->inmemory);
assert(lid >= 0);
- if (mnstr_writeChr(l->log, tpe) &&
- mnstr_writeLng(l->log, lid))
+ if (mnstr_writeChr(l->output_log, tpe) &&
+ mnstr_writeLng(l->output_log, lid))
return GDK_SUCCEED;
fprintf(stderr, "!ERROR: log_write_id: write failed\n");
return GDK_FAIL;
@@ -388,8 +395,8 @@ log_read_id(logger *lg, char *tpe, oid *
lng lid;
assert(!lg->inmemory);
- if (mnstr_readChr(lg->log, tpe) != 1 ||
- mnstr_readLng(lg->log, &lid) != 1) {
+ if (mnstr_readChr(lg->input_log, tpe) != 1 ||
+ mnstr_readLng(lg->input_log, &lid) != 1) {
fprintf(stderr, "!ERROR: log_read_id: read failed\n");
return LOG_EOF;
}
@@ -517,7 +524,7 @@ log_read_updates(logger *lg, trans *tr,
return LOG_ERR;
if (ht >= 0 && tt >= 0) {
BAT *uid = NULL;
- BAT *r;
+ BAT *r = NULL;
void *(*rt) (ptr, stream *, size_t) = BATatoms[tt].atomRead;
void *tv = NULL;
@@ -535,29 +542,31 @@ log_read_updates(logger *lg, trans *tr,
#endif
assert(l->nr <= (lng) BUN_MAX);
- if (l->flag == LOG_UPDATE) {
- uid = COLnew(0, ht, (BUN) l->nr, PERSISTENT);
- if (uid == NULL) {
+ if (tr->tid > lg->saved_tid) {
+ if (l->flag == LOG_UPDATE) {
+ uid = COLnew(0, ht, (BUN) l->nr, PERSISTENT);
+ if (uid == NULL) {
+ logbat_destroy(b);
+ return LOG_ERR;
+ }
+ } else {
+ assert(ht == TYPE_void);
+ }
+
+ r = COLnew(0, tt, (BUN) l->nr, PERSISTENT);
+ if (r == NULL) {
+ BBPreclaim(uid);
logbat_destroy(b);
return LOG_ERR;
}
- } else {
- assert(ht == TYPE_void);
+ if (tseq)
+ BATtseqbase(r, 0);
}
- r = COLnew(0, tt, (BUN) l->nr, PERSISTENT);
- if (r == NULL) {
- BBPreclaim(uid);
- logbat_destroy(b);
- return LOG_ERR;
- }
-
- if (tseq)
- BATtseqbase(r, 0);
if (ht == TYPE_void && l->flag == LOG_INSERT) {
lng nr = l->nr;
for (; res == LOG_OK && nr > 0; nr--) {
- void *t = rt(tv, lg->log, 1);
+ void *t = rt(tv, lg->input_log, 1);
if (t == NULL) {
/* see if failure was due to
@@ -585,8 +594,8 @@ log_read_updates(logger *lg, trans *tr,
if (!pax) {
lng nr = l->nr;
for (; res == LOG_OK && nr > 0; nr--) {
- void *h = rh(hv, lg->log, 1);
- void *t = rt(tv, lg->log, 1);
+ void *h = rh(hv, lg->input_log, 1);
+ void *t = rt(tv, lg->input_log, 1);
if (h == NULL)
res = LOG_EOF;
@@ -595,8 +604,8 @@ log_read_updates(logger *lg, trans *tr,
res = LOG_EOF;
else
res = LOG_ERR;
- } else if (BUNappend(uid, h, true) !=
GDK_SUCCEED ||
- BUNappend(r, t, true) !=
GDK_SUCCEED)
+ } else if ((uid && BUNappend(uid, h,
true) != GDK_SUCCEED) ||
+ (r && BUNappend(r, t, true) !=
GDK_SUCCEED))
res = LOG_ERR;
if (t != tv)
GDKfree(t);
@@ -605,39 +614,39 @@ log_read_updates(logger *lg, trans *tr,
char compressed = 0;
lng nr = l->nr;
- if (mnstr_read(lg->log, &compressed, 1, 1) != 1)
+ if (mnstr_read(lg->input_log, &compressed, 1,
1) != 1)
return LOG_ERR;
if (compressed) {
- void *h = rh(hv, lg->log, 1);
+ void *h = rh(hv, lg->input_log, 1);
assert(uid->ttype == TYPE_void);
if (h == NULL)
res = LOG_EOF;
- else {
+ else if (uid) {
BATtseqbase(uid, *(oid*)h);
BATsetcount(uid, (BUN) l->nr);
}
} else {
for (; res == LOG_OK && nr > 0; nr--) {
- void *h = rh(hv, lg->log, 1);
+ void *h = rh(hv, lg->input_log,
1);
if (h == NULL)
res = LOG_EOF;
- else if (BUNappend(uid, h,
true) != GDK_SUCCEED)
+ else if (uid && BUNappend(uid,
h, true) != GDK_SUCCEED)
res = LOG_ERR;
}
}
nr = l->nr;
for (; res == LOG_OK && nr > 0; nr--) {
- void *t = rt(tv, lg->log, 1);
+ void *t = rt(tv, lg->input_log, 1);
if (t == NULL) {
if (strstr(GDKerrbuf, "malloc")
== NULL)
res = LOG_EOF;
else
res = LOG_ERR;
- } else if (BUNappend(r, t, true) !=
GDK_SUCCEED)
+ } else if (r && BUNappend(r, t, true)
!= GDK_SUCCEED)
res = LOG_ERR;
if (t != tv)
GDKfree(t);
@@ -648,6 +657,8 @@ log_read_updates(logger *lg, trans *tr,
if (tv != lg->buf)
GDKfree(tv);
+ if (tr->tid <= lg->saved_tid) /* skipping */
+ return res;
if (res == LOG_OK) {
if (tr_grow(tr) == GDK_SUCCEED) {
tr->changes[tr->nr].type = l->flag;
@@ -741,7 +752,9 @@ la_bat_updates(logger *lg, logaction *la
static log_return
log_read_destroy(logger *lg, trans *tr, char *name, char tpe, oid id)
{
- (void) lg;
+ if (tr->tid <= lg->saved_tid) /* skipping */
+ return LOG_OK;
+
assert(!lg->inmemory);
if (tr_grow(tr) == GDK_SUCCEED) {
tr->changes[tr->nr].type = LOG_DESTROY;
@@ -790,6 +803,9 @@ log_read_create(logger *lg, trans *tr, c
int ht, tt;
char *ha, *ta;
+ if (tr->tid <= lg->saved_tid) /* skipping */
+ return LOG_OK;
+
assert(!lg->inmemory);
if (lg->debug & 1)
fprintf(stderr, "#log_read_create %s\n", name);
@@ -855,7 +871,8 @@ la_bat_create(logger *lg, logaction *la)
static log_return
log_read_use(logger *lg, trans *tr, logformat *l, char *name, char tpe, oid id)
{
- (void) lg;
+ if (tr->tid <= lg->saved_tid) /* skipping */
+ return LOG_OK;
assert(!lg->inmemory);
if (tr_grow(tr) != GDK_SUCCEED)
@@ -1055,6 +1072,7 @@ tr_commit(logger *lg, trans *tr)
}
la_destroy(&tr->changes[i]);
}
+ lg->saved_tid = tr->tid;
return tr_destroy(tr);
}
@@ -1063,7 +1081,7 @@ tr_commit(logger *lg, trans *tr)
#endif
static gdk_return
-logger_open(logger *lg)
+logger_open_output(logger *lg)
{
int len;
char id[BUFSIZ];
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
