Changeset: 707217a26951 for MonetDB
URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=707217a26951
Modified Files:
.hgignore
clients/R/MonetDB.R/R/dplyr.R
clients/Tests/exports.stable.out
gdk/gdk.h
gdk/gdk_logger.c
gdk/gdk_logger.h
gdk/gdk_private.h
gdk/gdk_storage.c
gdk/gdk_utils.c
gdk/gdk_utils.h
monetdb5/mal/mal_session.c
monetdb5/modules/kernel/logger.c
sql/backends/monet5/sql_scenario.c
sql/jdbc/tests/Tests/Test_Dobjects.stable.out
sql/scripts/99_system.sql
sql/server/sql_mvc.c
sql/storage/bat/bat_logger.c
sql/storage/bat/bat_logger.h
sql/storage/restrict/restrict_logger.c
sql/storage/sql_storage.h
sql/storage/store.c
sql/test/Tests/systemfunctions.stable.out
sql/test/leaks/Tests/check0.stable.out
sql/test/leaks/Tests/check0.stable.out.int128
sql/test/leaks/Tests/check1.stable.out
sql/test/leaks/Tests/check1.stable.out.int128
sql/test/leaks/Tests/check2.stable.out
sql/test/leaks/Tests/check2.stable.out.int128
sql/test/leaks/Tests/check3.stable.out
sql/test/leaks/Tests/check3.stable.out.int128
sql/test/leaks/Tests/check4.stable.out
sql/test/leaks/Tests/check4.stable.out.int128
sql/test/leaks/Tests/check5.stable.out
sql/test/leaks/Tests/check5.stable.out.int128
sql/test/leaks/Tests/drop3.stable.out
sql/test/leaks/Tests/drop3.stable.out.int128
sql/test/leaks/Tests/select1.stable.out
sql/test/leaks/Tests/select1.stable.out.int128
sql/test/leaks/Tests/select2.stable.out
sql/test/leaks/Tests/select2.stable.out.int128
sql/test/leaks/Tests/temp1.stable.out
sql/test/leaks/Tests/temp1.stable.out.int128
sql/test/leaks/Tests/temp2.stable.out
sql/test/leaks/Tests/temp2.stable.out.int128
sql/test/leaks/Tests/temp3.stable.out
sql/test/leaks/Tests/temp3.stable.out.int128
tools/merovingian/daemon/argvcmds.c
tools/merovingian/daemon/argvcmds.h
tools/merovingian/daemon/discoveryrunner.c
tools/merovingian/daemon/discoveryrunner.h
tools/merovingian/daemon/forkmserver.c
tools/merovingian/daemon/merovingian.c
tools/merovingian/utils/properties.c
tools/merovingian/utils/properties.h
tools/merovingian/utils/utils.c
tools/merovingian/utils/utils.h
Branch: default
Log Message:
Merge transaction-replication into default
diffs (truncated from 2391 to 300 lines):
diff --git a/.hgignore b/.hgignore
--- a/.hgignore
+++ b/.hgignore
@@ -91,3 +91,8 @@ NT/sql
NT/testing
NT/tools
NT/unistd.h
+
+syntax: regexp
+^java/build$
+syntax: regexp
+^java/target$
diff --git a/clients/Tests/exports.stable.out b/clients/Tests/exports.stable.out
--- a/clients/Tests/exports.stable.out
+++ b/clients/Tests/exports.stable.out
@@ -248,8 +248,10 @@ int GDKdebug;
void GDKerror(_In_z_ _Printf_format_string_ const char *format, ...)
__attribute__((__format__(__printf__, 1, 2)));
void GDKexit(int status) __attribute__((__noreturn__));
int GDKexiting(void);
+int GDKextractParentAndLastDirFromPath(const char *path, char
*last_dir_parent, char *last_dir);
void GDKfatal(_In_z_ _Printf_format_string_ const char *format, ...)
__attribute__((__format__(__printf__, 1, 2))) __attribute__((__noreturn__));
char *GDKfilepath(int farmid, const char *dir, const char *nme, const char
*ext);
+char *GDKfilepath_long(int farmid, const char *dir, const char *ext);
void GDKfree(void *blk);
char *GDKgetenv(const char *name);
int GDKgetenv_int(const char *name, int def);
@@ -417,12 +419,15 @@ int log_tend(logger *lg);
int log_tstart(logger *lg);
log_bid logger_add_bat(logger *lg, BAT *b, const char *name);
lng logger_changes(logger *lg);
-int logger_cleanup(logger *lg);
-logger *logger_create(int debug, const char *fn, const char *logdir, int
version, preversionfix_fptr prefuncp, postversionfix_fptr postfuncp);
+int logger_cleanup(logger *lg, int keep_persisted_log_files);
+logger *logger_create(int debug, const char *fn, const char *logdir, int
version, preversionfix_fptr prefuncp, postversionfix_fptr postfuncp, int
keep_persisted_log_files);
+logger *logger_create_shared(int debug, const char *fn, const char *logdir,
const char *slave_logdir, int version, preversionfix_fptr prefuncp,
postversionfix_fptr postfuncp);
void logger_del_bat(logger *lg, log_bid bid);
void logger_destroy(logger *lg);
int logger_exit(logger *lg);
log_bid logger_find_bat(logger *lg, const char *name);
+lng logger_read_last_transaction_id(logger *lg, char *dir, char *logger_file,
int role);
+int logger_reload(logger *lg);
int logger_restart(logger *lg);
int logger_sequence(logger *lg, int seq, lng *id);
void *mdlopen(const char *library, int mode);
diff --git a/gdk/gdk.h b/gdk/gdk.h
--- a/gdk/gdk.h
+++ b/gdk/gdk.h
@@ -1487,6 +1487,8 @@ gdk_export int BATgetaccess(BAT *b);
#define PERSISTENT 0
#define TRANSIENT 1
+#define LOG_DIR 2
+#define SHARED_LOG_DIR 3
#define BAT_WRITE 0 /* all kinds of access allowed */
#define BAT_READ 1 /* only read-access allowed */
@@ -1568,6 +1570,7 @@ gdk_export size_t BATmemsize(BAT *b, int
#define NOFARM (-1) /* indicate to GDKfilepath to create relative path */
gdk_export char *GDKfilepath(int farmid, const char *dir, const char *nme,
const char *ext);
+gdk_export char *GDKfilepath_long(int farmid, const char *dir, const char
*ext);
gdk_export gdk_return GDKcreatedir(const char *nme);
/*
diff --git a/gdk/gdk_logger.c b/gdk/gdk_logger.c
--- a/gdk/gdk_logger.c
+++ b/gdk/gdk_logger.c
@@ -96,8 +96,6 @@ typedef struct logformat_t {
lng nr;
} logformat;
-#define LOGFILE "log"
-
static int bm_commit(logger *lg);
static int tr_grow(trans *tr);
@@ -851,12 +849,60 @@ tr_abort(logger *lg, trans *tr)
static int log_sequence_nrs(logger *lg);
+/* Update the last transaction id written in the catalog file.
+ * Only used by the shared logger. */
+static int
+logger_update_catalog_file(logger *lg, const char *dir, const char *filename,
int role)
+{
+ FILE *fp;
+ int bak_exists;
+ int farmid = BBPselectfarm(role, 0, offheap);
+
+ bak_exists = 0;
+ /* check if an older file exists and move bak it up */
+#if defined(_MSC_VER)
+ if (_access(filename, 0) != -1) {
+#else
+ if (access(filename, 0) != -1) {
+#endif
+ bak_exists = 1;
+ if (GDKmove(farmid, dir, filename, NULL, dir, filename, "bak")
== GDK_FAIL) {
+ fprintf(stderr, "!ERROR: logger_update_catalog_file:
rename %s to %s.bak in %s failed\n", filename, filename, dir);
+ return LOG_ERR;
+ }
+ }
+
+ if ((fp = GDKfileopen(farmid, dir, filename, NULL, "w")) != NULL) {
+ if (fprintf(fp, "%06d\n\n", lg->version) < 0) {
+ fprintf(stderr, "!ERROR: logger_update_catalog_file:
write to %s failed\n", filename);
+ return LOG_ERR;
+ }
+
+ if (fprintf(fp, LLFMT "\n", lg->id) < 0 || fclose(fp) < 0) {
+ fprintf(stderr, "!ERROR: logger_update_catalog_file:
write/flush to %s failed\n", filename);
+ return LOG_ERR;
+ }
+
+ /* cleanup the bak file, if it exists*/
+ if (bak_exists) {
+ GDKunlink(farmid, dir, filename, "bak");
+ }
+ } else {
+ fprintf(stderr, "!ERROR: logger_update_catalog_file: could not
create %s\n", filename);
+ GDKerror("logger_update_catalog_file: could not open %s\n",
filename);
+ return LOG_ERR;
+ }
+ return LOG_OK;
+}
+
static int
logger_open(logger *lg)
{
- char filename[BUFSIZ];
+ char id[BUFSIZ];
+ char *filename;
- snprintf(filename, sizeof(filename), "%s%s." LLFMT, lg->dir, LOGFILE,
lg->id);
+ snprintf(id, sizeof(id), LLFMT, lg->id);
+ filename = GDKfilepath(BBPselectfarm(lg->dbfarm_role, 0, offheap),
lg->dir, LOGFILE, id);
lg->log = open_wstream(filename);
lg->end = 0;
@@ -888,21 +934,28 @@ logger_readlog(logger *lg, char *filenam
time_t t0, t1;
struct stat sb;
lng fpos;
+ char* path = GDKfilepath_long(BBPselectfarm(lg->dbfarm_role, 0,
offheap), filename, NULL);
- lg->log = open_rstream(filename);
+ if (lg->debug & 1) {
+ fprintf(stderr, "#logger_readlog opening %s\n", filename);
+ }
+
+ lg->log = open_rstream(path);
/* if the file doesn't exist, there is nothing to be read back */
if (!lg->log || mnstr_errnr(lg->log)) {
if (lg->log)
mnstr_destroy(lg->log);
lg->log = NULL;
- return 0;
+ return LOG_ERR;
}
if (fstat(fileno(getFile(lg->log)), &sb) < 0) {
fprintf(stderr, "!ERROR: logger_readlog: fstat on opened file
%s failed\n", filename);
mnstr_destroy(lg->log);
lg->log = NULL;
- return 0;
+ /* If we can't read the files, it might simply be empty.
+ * In that case we can't return LOG_ERR, since it's actually
fine */
+ return 1;
}
t0 = time(NULL);
printf("# Start reading the write-ahead log '%s'\n", filename);
@@ -1008,31 +1061,35 @@ logger_readlog(logger *lg, char *filenam
t0 = time(NULL);
printf("# Finished reading the write-ahead log '%s'\n", filename);
fflush(stdout);
- return 0;
+ return LOG_OK;
}
/*
- * The log files are incrementally numbered. They are processed in the
+ * The log files are incrementally numbered, starting from 2. They are
processed in the
* same sequence.
*/
static int
logger_readlogs(logger *lg, FILE *fp, char *filename)
{
- int res = 0;
+ int res = LOG_OK;
char id[BUFSIZ];
- if (lg->debug & 1)
- fprintf(stderr, "#logger_readlogs %s\n", filename);
+ if (lg->debug & 1) {
+ fprintf(stderr, "#logger_readlogs logger id is " LLFMT "\n",
lg->id);
+ }
while (fgets(id, sizeof(id), fp) != NULL) {
- char buf[BUFSIZ];
+ char log_filename[BUFSIZ];
lng lid = strtoll(id, NULL, 10);
- if (lid >= lg->id) {
+ if (lg->debug & 1) {
+ fprintf(stderr, "#logger_readlogs last logger id
written in %s is " LLFMT "\n", filename, lid);
+ }
+
+ if (!lg->shared && lid >= lg->id) {
lg->id = lid;
- snprintf(buf, sizeof(buf), "%s." LLFMT, filename,
lg->id);
-
- if ((res = logger_readlog(lg, buf)) != 0) {
+ snprintf(log_filename, sizeof(log_filename), "%s."
LLFMT, filename, lg->id);
+ if ((res = logger_readlog(lg, log_filename)) != 0) {
/* we cannot distinguish errors from
* incomplete transactions (even if we
* would log aborts in the logs). So
@@ -1040,6 +1097,27 @@ logger_readlogs(logger *lg, FILE *fp, ch
* next log file */
(void) res;
}
+ } else {
+ while (lid >= lg->id && res != LOG_ERR) {
+ snprintf(log_filename, sizeof(log_filename),
"%s." LLFMT, filename, lg->id);
+ if ((logger_readlog(lg, log_filename)) ==
LOG_ERR && lg->shared && lg->id > 1) {
+ /* The only special case is if the
files is missing altogether
+ * and the logger is a shared one,
+ * then we have missing transactions
and we should abort.
+ * Yeah, and we also ignore the 1st
files it most likely never exists. */
+ res = LOG_ERR;
+ fprintf(stderr, "#logger_readlogs
missing shared logger file %s. Aborting\n", log_filename);
+ }
+ /* Increment the id only at the end, since we
want to re-read the last file.
+ * That is because last time we read it, it was
empty, since the logger creates empty files
+ * and fills them in later. */
+ lg->id++;
+ }
+ if (lid < lg->id) {
+ lg->id = lid;
+ }
+ /* if this is a shared logger, write the id in the
shared file */
+ logger_update_catalog_file(lg, lg->local_dir,
LOGFILE_SHARED, lg->local_dbfarm_role);
}
}
return res;
@@ -1214,77 +1292,70 @@ logger_fatal(const char *format, const c
GDKexit(1);
}
-static logger *
-logger_new(int debug, const char *fn, const char *logdir, int version,
preversionfix_fptr prefuncp, postversionfix_fptr postfuncp)
+/* Set the logdir path, add a dbfarm if needed.
+ * Returns the role of the dbfarm containing the logdir.
+ */
+static int
+logger_set_logdir_path(char *filename, const char *fn, const char *logdir, int
shared) {
+ int role = PERSISTENT; /* default role is persistent, i.e. the default
dbfarm */
+
+ if (MT_path_absolute(logdir)) {
+ char logdir_parent_path[BUFSIZ] = "";
+ char logdir_name[BUFSIZ] = "";
+ /* split the logdir string into absolute parent dir path and
(relative) log dir name */
+ if (GDKextractParentAndLastDirFromPath(logdir,
logdir_parent_path, logdir_name)) {
+ /* set the new relative logdir locaiton including the
logger function name subdir */
+ snprintf(filename, BUFSIZ, "%s%c%s%c", logdir_name,
DIR_SEP, fn, DIR_SEP);
+
+ /* add a new dbfarm for the logger directory using the
parent dir path,
+ * assuming it is set, s.t. the logs are stored in a
location other than the default dbfarm,
+ * or at least it appears so to (multi)dbfarm aware
functions */
+ if (!shared) {
+ role = LOG_DIR;
+ } else {
+ role = SHARED_LOG_DIR;
+ }
+ BBPaddfarm(logdir_parent_path, 1 << role);
+ } else {
+ logger_fatal("logger_set_logdir_path: logdir path is
not correct (%s)."
+ "Make sure you specify a valid absolute
or relative path.\n", logdir, 0, 0);
+ }
+ } else {
+ /* just concat the logdir and fn with appropriate separators */
+ snprintf(filename, BUFSIZ, "%s%c%s%c", logdir, DIR_SEP, fn,
DIR_SEP);
+ }
+
+ return role;
+}
+
+/* Load data from the logger logdir
+ * Initialize new directories and catalog files if none are present, unless
running in read-only mode
+ * Load data and persist it in the BATs
+ * Convert 32bit data to 64bit, unless running in read-only mode */
+static int
+logger_load(int debug, const char* fn, char filename[BUFSIZ], logger* lg)
{
int id = LOG_SID;
- logger *lg;
FILE *fp;
- char filename[BUFSIZ];
char bak[BUFSIZ];
- log_bid snapshots_bid = 0;
_______________________________________________
checkin-list mailing list
[email protected]
https://www.monetdb.org/mailman/listinfo/checkin-list
