Changeset: 992922985ddf for MonetDB URL: http://dev.monetdb.org/hg/MonetDB?cmd=changeset;node=992922985ddf Removed Files: monetdb5/modules/mal/tablet_sql.c Branch: default Log Message:
Single tablet file The tablet_sql part is combined in the tablet.c file. diffs (truncated from 875 to 300 lines): diff --git a/monetdb5/modules/mal/tablet_sql.c b/monetdb5/modules/mal/tablet_sql.c deleted file mode 100644 --- a/monetdb5/modules/mal/tablet_sql.c +++ /dev/null @@ -1,870 +0,0 @@ -/* - * The contents of this file are subject to the MonetDB Public License - * Version 1.1 (the "License"); you may not use this file except in - * compliance with the License. You may obtain a copy of the License at - * http://www.monetdb.org/Legal/MonetDBLicense - * - * Software distributed under the License is distributed on an "AS IS" - * basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the - * License for the specific language governing rights and limitations - * under the License. - * - * The Original Code is the MonetDB Database System. - * - * The Initial Developer of the Original Code is CWI. - * Portions created by CWI are Copyright (C) 1997-July 2008 CWI. - * Copyright August 2008-2012 MonetDB B.V. - * All Rights Reserved. - */ - -/* - * Niels Nes, Martin Kersten - * - * Parallel bulk load for SQL - * The COPY INTO command for SQL is heavily CPU bound, which means - * that ideally we would like to exploit the multi-cores to do that - * work in parallel. - * Complicating factors are the initial record offset, the - * possible variable length of the input, and the original sort order - * that should preferable be maintained. - * - * The code below consists of a file reader, which breaks up the - * file into chunks of distinct lines. Then multiple parallel threads - * grab them, and break them on the field boundaries. - * After all fields are identified this way, the columns are converted - * and stored in the BATs. - * - * The threads get a reference to a private copy of the READERtask. - * It includes a list of columns they should handle. This is a basis - * to distributed cheap and expensive columns over threads. - * - * The file reader overlaps IO with updates of the BAT. - * Also the buffer size of the block stream might be a little small for - * this task (1MB). It has been increased to 8MB, which indeed improved. - * - * The work divider allocates subtasks to threads based on the - * observed time spending so far. - */ -#include "monetdb_config.h" -#include "tablet.h" -#include "algebra.h" - -#include <string.h> -#include <ctype.h> - -/* #define _DEBUG_TABLET_*/ -/* #define MLOCK_TST did not make a difference on sf10 */ - -#define BREAKLINE 1 -#define UPDATEBAT 2 - -typedef struct { - int id; /* for self reference */ - int state; /* line break=1 , 2 = update bat */ - int workers; /* how many concurrent ones */ - int error; /* error during line break */ - int next; - int limit; - lng *time, wtime; /* time per col + time per thread */ - int rounds; /* how often did we divide the work */ - MT_Id tid; - MT_Sema producer; /* reader waits for call */ - MT_Sema consumer; /* data available */ - int ateof; /* io control */ - bstream *b; - stream *out; - MT_Sema sema; /* threads wait for work , negative next implies exit */ - MT_Sema reply; /* let reader continue */ - Tablet *as; - char *errbuf; - char *csep, *rsep; - size_t seplen, rseplen; - char quote; - char *base, *input; /* area for tokenizer */ - size_t basesize; - int *cols; /* columns to handle */ - char ***fields; -} READERtask; - -/* - * @- - * The line is broken into pieces directly on their field separators. It assumes that we have - * the record in the cache already, so we can do most work quickly. - * Furthermore, it assume a uniform (SQL) pattern, without whitespace skipping, but with quote and separator. - */ - -static str -SQLload_error(READERtask *task, int idx) -{ - str line; - size_t sz = 0; - unsigned int i; - - for (i = 0; i < task->as->nr_attrs; i++) - if (task->fields[i][idx]) - sz += strlen(task->fields[i][idx]) + task->seplen; - else - sz += task->seplen; - - line = (str) GDKzalloc(sz + task->rseplen + 1); - if (line == 0) { - task->as->error = M5OutOfMemory; - return 0; - } - for (i = 0; i < task->as->nr_attrs; i++) { - if (task->fields[i][idx]) - strcat(line, task->fields[i][idx]); - if (i < task->as->nr_attrs - 1) - strcat(line, task->csep); - } - strcat(line, task->rsep); - return line; -} - -/* - * @- - * The parsing of the individual values is straightforward. If the value represents - * the null-replacement string then we grab the underlying nil. - * If the string starts with the quote identified from SQL, we locate the tail - * and interpret the body. - */ -static inline int -SQLinsert_val(Column *fmt, char *s, char quote, ptr key, str *err, int col) -{ - ptr *adt; - char buf[BUFSIZ]; - char *e, *t; - int ret = 0; - - /* include testing on the terminating null byte !! */ - if (fmt->nullstr && strncasecmp(s, fmt->nullstr, fmt->null_length + 1) == 0) { -#ifdef _DEBUG_TABLET_ - mnstr_printf(GDKout, "nil value '%s' (%d) found in :%s\n", fmt->nullstr, fmt->nillen, (s ? s : "")); -#endif - adt = fmt->nildata; - fmt->c[0]->T->nonil = 0; - } else if (quote && *s == quote) { - /* strip the quotes when present */ - s++; - for (t = e = s; *t; t++) - if (*t == quote) - e = t; - *e = 0; - adt = fmt->frstr(fmt, fmt->adt, s, e, 0); - /* The user might have specified a null string escape - * e.g. NULL as '', which should be tested */ - if (adt == NULL && s == e && fmt->nullstr && - strncasecmp(s, fmt->nullstr, fmt->null_length + 1) == 0) { - adt = fmt->nildata; - fmt->c[0]->T->nonil = 0; - } - } else { - for (e = s; *e; e++) ; - adt = fmt->frstr(fmt, fmt->adt, s, e, 0); - } - - if (!adt) { - char *val; - val = *s ? GDKstrdup(s) : GDKstrdup(""); - if (*err == NULL) { - if (snprintf(buf, BUFSIZ, - "value '%.*s%s' from line " BUNFMT - " field %d not inserted, expecting type %s\n", - BUFSIZ - 200, val, - strlen(val) > (size_t) BUFSIZ - 200 ? "..." : "", - BATcount(fmt->c[0]) + 1, col, fmt->type) < 0) - snprintf(buf, BUFSIZ, - "value from line " BUNFMT - " field %d not inserted, expecting type %s\n", - BATcount(fmt->c[0]) + 1, col, fmt->type); - *err = GDKstrdup(buf); - } - GDKfree(val); - /* replace it with a nil */ - adt = fmt->nildata; - fmt->c[0]->T->nonil = 0; - ret = -1; - } - /* key may be NULL but that's not a problem, as long as we have void */ - if (fmt->raw) { - mnstr_write(fmt->raw, adt, ATOMsize(fmt->adt), 1); - } else { - bunfastins(fmt->c[0], key, adt); - } - return ret; - bunins_failed: - if (*err == NULL) { - snprintf(buf, BUFSIZ, - "parsing error from line " BUNFMT " field %d not inserted\n", - BATcount(fmt->c[0]) + 1, col); - *err = GDKstrdup(buf); - } - return -1; -} - -static int -SQLworker_column(READERtask *task, int col) -{ - int i; - Column *fmt = task->as->format; - str err = 0; - - /* watch out for concurrent threads */ - mal_set_lock(mal_copyLock, "tablet insert value"); - if (BATcapacity(fmt[col].c[0]) < BATcount(fmt[col].c[0]) + task->next) { - if ((fmt[col].c[0] = BATextend(fmt[col].c[0], BATgrows(fmt[col].c[0]) + task->limit)) == NULL) { - if (task->as->error == NULL) - task->as->error = GDKstrdup("Failed to extend the BAT, perhaps disk full"); - mal_unset_lock(mal_copyLock, "tablet insert value"); - mnstr_printf(GDKout, "Failed to extend the BAT, perhaps disk full"); - return -1; - } - } - mal_unset_lock(mal_copyLock, "tablet insert value"); - - for (i = 0; i < task->next; i++) - if (task->fields[col][i]) { /* no errors */ - if (SQLinsert_val(&fmt[col], task->fields[col][i], task->quote, NULL, &err, col + 1)) { - assert(err != NULL); - mal_set_lock(mal_copyLock, "tablet insert value"); - if (!task->as->tryall) { - /* watch out for concurrent threads */ - if (task->as->error == NULL) - task->as->error = err; /* restore for upper layers */ - } else - BUNins(task->as->complaints, NULL, err, TRUE); - mal_unset_lock(mal_copyLock, "tablet insert value"); - } - } - - if (err) { - /* watch out for concurrent threads */ - mal_set_lock(mal_copyLock, "tablet insert value"); - if (task->as->error == NULL) - task->as->error = err; /* restore for upper layers */ - mal_unset_lock(mal_copyLock, "tablet insert value"); - } - return err ? -1 : 0; -} - -/* - * @- - * The lines are broken on the column separator. Any error is shown and reflected with - * setting the reference of the offending row fields to NULL. - * This allows the loading to continue, skipping the minimal number of rows. - */ -static int -SQLload_file_line(READERtask *task, int idx) -{ - BUN i; - char errmsg[BUFSIZ]; - char ch = *task->csep; - char *line = task->fields[0][idx]; - Tablet *as = task->as; - Column *fmt = as->format; - - errmsg[0] = 0; -#ifdef _DEBUG_TABLET_ - mnstr_printf(GDKout, "SQL break line id %d state %d\n%s", task->id, idx, task->fields[0][idx]); -#endif - - for (i = 0; i < as->nr_attrs; i++) { - task->fields[i][idx] = line; - /* recognize fields starting with a quote, keep them */ - if (task->quote && *line == task->quote) { - line = tablet_skip_string(line + 1, task->quote); - if (!line) { - str errline = SQLload_error(task, task->next); - snprintf(errmsg, BUFSIZ, - "End of string (%c) missing in \"%s\" at line " BUNFMT - " field " BUNFMT "\n", - task->quote, (errline ? errline : ""), - BATcount(as->format->c[0]) + task->next + 1, i); - if (errline) - GDKerror("%s", errmsg); - GDKfree(errline); - goto errors; - } - } - - /* eat away the column separator */ - for (; *line; line++) - if (*line == '\\') { - if (line[1]) - line++; - } else if (*line == ch && (task->seplen == 1 || strncmp(line, task->csep, task->seplen) == 0)) { _______________________________________________ Checkin-list mailing list [email protected] http://mail.monetdb.org/mailman/listinfo/checkin-list
