LCOV - code coverage report
Current view: top level - monetdb5/modules/mal - tablet.c (source / functions) Hit Total Coverage
Test: coverage.info Lines: 703 930 75.6 %
Date: 2021-01-13 20:07:21 Functions: 27 28 96.4 %

          Line data    Source code
       1             : /*
       2             :  * This Source Code Form is subject to the terms of the Mozilla Public
       3             :  * License, v. 2.0.  If a copy of the MPL was not distributed with this
       4             :  * file, You can obtain one at http://mozilla.org/MPL/2.0/.
       5             :  *
       6             :  * Copyright 1997 - July 2008 CWI, August 2008 - 2021 MonetDB B.V.
       7             :  */
       8             : 
       9             : /*
      10             :  *  Niels Nes, Martin Kersten
      11             :  *
      12             :  * Parallel bulk load for SQL
      13             :  * The COPY INTO command for SQL is heavily CPU bound, which means
      14             :  * that ideally we would like to exploit the multi-cores to do that
      15             :  * work in parallel.
      16             :  * Complicating factors are the initial record offset, the
      17             :  * possible variable length of the input, and the original sort order
      18             :  * that should preferable be maintained.
      19             :  *
      20             :  * The code below consists of a file reader, which breaks up the
      21             :  * file into chunks of distinct lines. Then multiple parallel threads
      22             :  * grab them, and break them on the field boundaries.
      23             :  * After all fields are identified this way, the columns are converted
      24             :  * and stored in the BATs.
      25             :  *
      26             :  * The threads get a reference to a private copy of the READERtask.
      27             :  * It includes a list of columns they should handle. This is a basis
      28             :  * to distributed cheap and expensive columns over threads.
      29             :  *
      30             :  * The file reader overlaps IO with updates of the BAT.
      31             :  * Also the buffer size of the block stream might be a little small for
      32             :  * this task (1MB). It has been increased to 8MB, which indeed improved.
      33             :  *
      34             :  * The work divider allocates subtasks to threads based on the
      35             :  * observed time spending so far.
      36             :  */
      37             : 
      38             : #include "monetdb_config.h"
      39             : #include "tablet.h"
      40             : #include "mapi_prompt.h"
      41             : 
      42             : #include <string.h>
      43             : #include <ctype.h>
      44             : 
      45             : #define MAXWORKERS      64
      46             : #define MAXBUFFERS 2
      47             : /* We restrict the row length to be 32MB for the time being */
      48             : #define MAXROWSIZE(X) (X > 32*1024*1024 ? X : 32*1024*1024)
      49             : 
      50             : static MT_Lock errorlock = MT_LOCK_INITIALIZER("errorlock");
      51             : 
      52             : static BAT *
      53        9394 : void_bat_create(int adt, BUN nr)
      54             : {
      55        9394 :         BAT *b = COLnew(0, adt, BATTINY, PERSISTENT);
      56             : 
      57             :         /* check for correct structures */
      58        9394 :         if (b == NULL)
      59             :                 return NULL;
      60        9394 :         if (BATsetaccess(b, BAT_APPEND) != GDK_SUCCEED) {
      61           0 :                 BBPunfix(b->batCacheid);
      62           0 :                 return NULL;
      63             :         }
      64        9394 :         if (nr > BATTINY && adt && BATextend(b, nr) != GDK_SUCCEED) {
      65           0 :                 BBPunfix(b->batCacheid);
      66           0 :                 return NULL;
      67             :         }
      68             : 
      69             :         /* disable all properties here */
      70        9394 :         b->tsorted = false;
      71        9394 :         b->trevsorted = false;
      72        9394 :         b->tnosorted = 0;
      73        9394 :         b->tnorevsorted = 0;
      74        9394 :         b->tseqbase = oid_nil;
      75        9394 :         b->tkey = false;
      76        9394 :         b->tnokey[0] = 0;
      77        9394 :         b->tnokey[1] = 0;
      78        9394 :         return b;
      79             : }
      80             : 
      81             : void
      82       61738 : TABLETdestroy_format(Tablet *as)
      83             : {
      84             :         BUN p;
      85       61738 :         Column *fmt = as->format;
      86             : 
      87      340354 :         for (p = 0; p < as->nr_attrs; p++) {
      88      278616 :                 if (fmt[p].c)
      89      217776 :                         BBPunfix(fmt[p].c->batCacheid);
      90      278616 :                 if (fmt[p].data)
      91        9461 :                         GDKfree(fmt[p].data);
      92             :         }
      93       61738 :         GDKfree(fmt);
      94       61738 : }
      95             : 
      96             : static oid
      97       60833 : check_BATs(Tablet *as)
      98             : {
      99       60833 :         Column *fmt = as->format;
     100             :         BUN i = 0;
     101             :         BUN cnt;
     102             :         oid base;
     103             : 
     104       60833 :         if (fmt[i].c == NULL)
     105             :                 i++;
     106       60833 :         cnt = BATcount(fmt[i].c);
     107       60833 :         base = fmt[i].c->hseqbase;
     108             : 
     109       60833 :         if (as->nr != cnt)
     110        7603 :                 return oid_nil;
     111             : 
     112      266289 :         for (i = 0; i < as->nr_attrs; i++) {
     113             :                 BAT *b;
     114             :                 BUN offset;
     115             : 
     116      213060 :                 b = fmt[i].c;
     117      213060 :                 if (b == NULL)
     118       53230 :                         continue;
     119      159830 :                 offset = as->offset;
     120             : 
     121      159830 :                 if (BATcount(b) != cnt || b->hseqbase != base)
     122           1 :                         return oid_nil;
     123             : 
     124      159829 :                 fmt[i].p = offset;
     125             :         }
     126             :         return base;
     127             : }
     128             : 
     129             : str
     130         895 : TABLETcreate_bats(Tablet *as, BUN est)
     131             : {
     132         895 :         Column *fmt = as->format;
     133             :         BUN i, nr = 0;
     134             : 
     135       10295 :         for (i = 0; i < as->nr_attrs; i++) {
     136        9400 :                 if (fmt[i].skip)
     137           6 :                         continue;
     138        9394 :                 fmt[i].c = void_bat_create(fmt[i].adt, est);
     139        9394 :                 if (!fmt[i].c) {
     140           0 :                         while (i > 0) {
     141           0 :                                 if (!fmt[--i].skip)
     142           0 :                                         BBPreclaim(fmt[i].c);
     143             :                         }
     144           0 :                         throw(SQL, "copy", "Failed to create bat of size " BUNFMT "\n", as->nr);
     145             :                 }
     146        9394 :                 fmt[i].ci = bat_iterator(fmt[i].c);
     147        9394 :                 nr++;
     148             :         }
     149         895 :         if (!nr)
     150           0 :                 throw(SQL, "copy", "At least one column should be read from the input\n");
     151             :         return MAL_SUCCEED;
     152             : }
     153             : 
     154             : str
     155         871 : TABLETcollect(BAT **bats, Tablet *as)
     156             : {
     157         871 :         Column *fmt = as->format;
     158             :         BUN i, j;
     159             :         BUN cnt = 0;
     160             : 
     161         871 :         if (bats == NULL)
     162           0 :                 throw(SQL, "copy", "Missing container");
     163        2146 :         for (i = 0; i < as->nr_attrs && !cnt; i++)
     164        1275 :                 if (!fmt[i].skip)
     165        1272 :                         cnt = BATcount(fmt[i].c);
     166       10179 :         for (i = 0, j = 0; i < as->nr_attrs; i++) {
     167        9308 :                 if (fmt[i].skip)
     168           6 :                         continue;
     169        9302 :                 bats[j] = fmt[i].c;
     170        9302 :                 BBPfix(bats[j]->batCacheid);
     171        9302 :                 if (BATsetaccess(fmt[i].c, BAT_READ) != GDK_SUCCEED)
     172           0 :                         throw(SQL, "copy", "Failed to set access at tablet part " BUNFMT "\n", cnt);
     173        9302 :                 fmt[i].c->tsorted = fmt[i].c->trevsorted = false;
     174        9302 :                 fmt[i].c->tkey = false;
     175        9302 :                 BATsettrivprop(fmt[i].c);
     176             : 
     177        9302 :                 if (cnt != BATcount(fmt[i].c))
     178           0 :                         throw(SQL, "copy", "Count " BUNFMT " differs from " BUNFMT "\n", BATcount(fmt[i].c), cnt);
     179        9302 :                 j++;
     180             :         }
     181             :         return MAL_SUCCEED;
     182             : }
     183             : 
     184             : str
     185           9 : TABLETcollect_parts(BAT **bats, Tablet *as, BUN offset)
     186             : {
     187           9 :         Column *fmt = as->format;
     188             :         BUN i, j;
     189             :         BUN cnt = 0;
     190             : 
     191          18 :         for (i = 0; i < as->nr_attrs && !cnt; i++)
     192           9 :                 if (!fmt[i].skip)
     193           9 :                         cnt = BATcount(fmt[i].c);
     194          71 :         for (i = 0, j = 0; i < as->nr_attrs; i++) {
     195             :                 BAT *b, *bv = NULL;
     196          62 :                 if (fmt[i].skip)
     197           0 :                         continue;
     198          62 :                 b = fmt[i].c;
     199          62 :                 b->tsorted = b->trevsorted = false;
     200          62 :                 b->tkey = false;
     201          62 :                 BATsettrivprop(b);
     202          62 :                 if (BATsetaccess(b, BAT_READ) != GDK_SUCCEED)
     203           0 :                         throw(SQL, "copy", "Failed to set access at tablet part " BUNFMT "\n", cnt);
     204          62 :                 bv = BATslice(b, (offset > 0) ? offset - 1 : 0, BATcount(b));
     205          62 :                 bats[j] = bv;
     206             : 
     207          62 :                 b->tkey = (offset > 0) ? FALSE : bv->tkey;
     208          62 :                 b->tnonil &= bv->tnonil;
     209          62 :                 if (b->tsorted != bv->tsorted)
     210           0 :                         b->tsorted = false;
     211          62 :                 if (b->trevsorted != bv->trevsorted)
     212           0 :                         b->trevsorted = false;
     213          62 :                 if (BATtdense(b))
     214           0 :                         b->tkey = true;
     215          62 :                 b->batDirtydesc = true;
     216             : 
     217          62 :                 if (offset > 0) {
     218           0 :                         BBPunfix(bv->batCacheid);
     219           0 :                         bats[j] = BATslice(b, offset, BATcount(b));
     220             :                 }
     221          62 :                 if (cnt != BATcount(b))
     222           0 :                         throw(SQL, "copy", "Count " BUNFMT " differs from " BUNFMT "\n", BATcount(b), cnt);
     223          62 :                 j++;
     224             :         }
     225             :         return MAL_SUCCEED;
     226             : }
     227             : 
     228             : // the starting quote character has already been skipped
     229             : 
     230             : static char *
     231     4583040 : tablet_skip_string(char *s, char quote, bool escape)
     232             : {
     233             :         size_t i = 0, j = 0;
     234   100493000 :         while (s[i]) {
     235   100487000 :                 if (escape && s[i] == '\\' && s[i + 1] != '\0')
     236      493285 :                         s[j++] = s[i++];
     237    99993500 :                 else if (s[i] == quote) {
     238     3515970 :                         if (s[i + 1] != quote)
     239             :                                 break;
     240             :                         i++;                            /* skip the first quote */
     241             :                 }
     242    95910300 :                 s[j++] = s[i++];
     243             :         }
     244             :         assert(s[i] == quote || s[i] == '\0');
     245     4583040 :         if (s[i] == 0)
     246             :                 return NULL;
     247     4583040 :         s[j] = 0;
     248     4583040 :         return s + i;
     249             : }
     250             : 
     251             : static int
     252             : TABLET_error(stream *s)
     253             : {
     254           0 :         char *err = mnstr_error(s);
     255             :         /* use free as stream allocates outside GDK */
     256           0 :         if (err)
     257           0 :                 free(err);
     258             :         return -1;
     259             : }
     260             : 
     261             : /* The output line is first built before being sent. It solves a problem
     262             :    with UDP, where you may loose most of the information using short writes
     263             : */
     264             : static inline int
     265      731705 : output_line(char **buf, size_t *len, char **localbuf, size_t *locallen, Column *fmt, stream *fd, BUN nr_attrs, oid id)
     266             : {
     267             :         BUN i;
     268             :         ssize_t fill = 0;
     269             : 
     270     6151430 :         for (i = 0; i < nr_attrs; i++) {
     271     5419720 :                 if (fmt[i].c == NULL)
     272      731705 :                         continue;
     273     4688020 :                 if (id < fmt[i].c->hseqbase || id >= fmt[i].c->hseqbase + BATcount(fmt[i].c))
     274             :                         break;
     275     4688020 :                 fmt[i].p = id - fmt[i].c->hseqbase;
     276             :         }
     277      731705 :         if (i == nr_attrs) {
     278     6151430 :                 for (i = 0; i < nr_attrs; i++) {
     279     5419720 :                         Column *f = fmt + i;
     280             :                         const char *p;
     281             :                         ssize_t l;
     282             : 
     283     5419720 :                         if (f->c) {
     284     4688020 :                                 p = BUNtail(f->ci, f->p);
     285             : 
     286     4688020 :                                 if (!p || ATOMcmp(f->adt, ATOMnilptr(f->adt), p) == 0) {
     287        4325 :                                         p = f->nullstr;
     288        4325 :                                         l = (ssize_t) strlen(f->nullstr);
     289             :                                 } else {
     290     4683690 :                                         l = f->tostr(f->extra, localbuf, locallen, f->adt, p);
     291     4683690 :                                         if (l < 0)
     292             :                                                 return -1;
     293     4683690 :                                         p = *localbuf;
     294             :                                 }
     295     4688020 :                                 if (fill + l + f->seplen >= (ssize_t) *len) {
     296             :                                         /* extend the buffer */
     297             :                                         char *nbuf;
     298           0 :                                         nbuf = GDKrealloc(*buf, fill + l + f->seplen + BUFSIZ);
     299           0 :                                         if( nbuf == NULL)
     300             :                                                 return -1; /* *buf freed by caller */
     301           0 :                                         *buf = nbuf;
     302           0 :                                         *len = fill + l + f->seplen + BUFSIZ;
     303             :                                 }
     304     4688020 :                                 strncpy(*buf + fill, p, l);
     305             :                                 fill += l;
     306             :                         }
     307     5419720 :                         strncpy(*buf + fill, f->sep, f->seplen);
     308     5419720 :                         fill += f->seplen;
     309             :                 }
     310             :         }
     311      731705 :         if (fd && mnstr_write(fd, *buf, 1, fill) != fill)
     312           0 :                 return TABLET_error(fd);
     313             :         return 0;
     314             : }
     315             : 
     316             : static inline int
     317     1033590 : output_line_dense(char **buf, size_t *len, char **localbuf, size_t *locallen, Column *fmt, stream *fd, BUN nr_attrs)
     318             : {
     319             :         BUN i;
     320             :         ssize_t fill = 0;
     321             : 
     322     6044510 :         for (i = 0; i < nr_attrs; i++) {
     323     5010930 :                 Column *f = fmt + i;
     324             :                 const char *p;
     325             :                 ssize_t l;
     326             : 
     327     5010930 :                 if (f->c) {
     328     3977340 :                         p = BUNtail(f->ci, f->p);
     329             : 
     330     3977340 :                         if (!p || ATOMcmp(f->adt, ATOMnilptr(f->adt), p) == 0) {
     331      581121 :                                 p = f->nullstr;
     332      581121 :                                 l = (ssize_t) strlen(p);
     333             :                         } else {
     334     3396220 :                                 l = f->tostr(f->extra, localbuf, locallen, f->adt, p);
     335     3396220 :                                 if (l < 0)
     336             :                                         return -1;
     337     3396220 :                                 p = *localbuf;
     338             :                         }
     339     3977340 :                         if (fill + l + f->seplen >= (ssize_t) *len) {
     340             :                                 /* extend the buffer */
     341             :                                 char *nbuf;
     342          65 :                                 nbuf = GDKrealloc(*buf, fill + l + f->seplen + BUFSIZ);
     343          65 :                                 if( nbuf == NULL)
     344             :                                         return -1;      /* *buf freed by caller */
     345          65 :                                 *buf = nbuf;
     346          65 :                                 *len = fill + l + f->seplen + BUFSIZ;
     347             :                         }
     348     3977340 :                         strncpy(*buf + fill, p, l);
     349             :                         fill += l;
     350     3977340 :                         f->p++;
     351             :                 }
     352     5010930 :                 strncpy(*buf + fill, f->sep, f->seplen);
     353     5010930 :                 fill += f->seplen;
     354             :         }
     355     1033590 :         if (fd && mnstr_write(fd, *buf, 1, fill) != fill)
     356           0 :                 return TABLET_error(fd);
     357             :         return 0;
     358             : }
     359             : 
     360             : static inline int
     361           0 : output_line_lookup(char **buf, size_t *len, Column *fmt, stream *fd, BUN nr_attrs, oid id)
     362             : {
     363             :         BUN i;
     364             : 
     365           0 :         for (i = 0; i < nr_attrs; i++) {
     366           0 :                 Column *f = fmt + i;
     367             : 
     368           0 :                 if (f->c) {
     369           0 :                         const void *p = BUNtail(f->ci, id - f->c->hseqbase);
     370             : 
     371           0 :                         if (!p || ATOMcmp(f->adt, ATOMnilptr(f->adt), p) == 0) {
     372           0 :                                 size_t l = strlen(f->nullstr);
     373           0 :                                 if (mnstr_write(fd, f->nullstr, 1, l) != (ssize_t) l)
     374           0 :                                         return TABLET_error(fd);
     375             :                         } else {
     376           0 :                                 ssize_t l = f->tostr(f->extra, buf, len, f->adt, p);
     377             : 
     378           0 :                                 if (l < 0 || mnstr_write(fd, *buf, 1, l) != l)
     379           0 :                                         return TABLET_error(fd);
     380             :                         }
     381             :                 }
     382           0 :                 if (mnstr_write(fd, f->sep, 1, f->seplen) != f->seplen)
     383           0 :                         return TABLET_error(fd);
     384             :         }
     385             :         return 0;
     386             : }
     387             : 
     388             : /* returns TRUE if there is/might be more */
     389             : static bool
     390      607672 : tablet_read_more(bstream *in, stream *out, size_t n)
     391             : {
     392      607672 :         if (out) {
     393             :                 do {
     394             :                         /* query is not finished ask for more */
     395             :                         /* we need more query text */
     396      606306 :                         if (bstream_next(in) < 0)
     397             :                                 return false;
     398      606306 :                         if (in->eof) {
     399      606288 :                                 if (mnstr_write(out, PROMPT2, sizeof(PROMPT2) - 1, 1) == 1)
     400      606288 :                                         mnstr_flush(out, MNSTR_FLUSH_DATA);
     401      606288 :                                 in->eof = false;
     402             :                                 /* we need more query text */
     403      606288 :                                 if (bstream_next(in) <= 0)
     404             :                                         return false;
     405             :                         }
     406      606301 :                 } while (in->len <= in->pos);
     407        1366 :         } else if (bstream_read(in, n) <= 0) {
     408         117 :                 return false;
     409             :         }
     410             :         return true;
     411             : }
     412             : 
     413             : /*
     414             :  * Fast Load
     415             :  * To speedup the CPU intensive loading of files we have to break
     416             :  * the file into pieces and perform parallel analysis. Experimentation
     417             :  * against lineitem SF1 showed that half of the time goes into very
     418             :  * basis atom analysis (41 out of 102 B instructions).
     419             :  * Furthermore, the actual insertion into the BATs takes only
     420             :  * about 10% of the total. With multi-core processors around
     421             :  * it seems we can gain here significantly.
     422             :  *
     423             :  * The approach taken is to fork a parallel scan over the text file.
     424             :  * We assume that the blocked stream is already
     425             :  * positioned correctly at the reading position. The start and limit
     426             :  * indicates the byte range to search for tuples.
     427             :  * If start> 0 then we first skip to the next record separator.
     428             :  * If necessary we read more than 'limit' bytes to ensure parsing a complete
     429             :  * record and stop at the record boundary.
     430             :  * Beware, we should allocate Tablet descriptors for each file segment,
     431             :  * otherwise we end up with a gross concurrency control problem.
     432             :  * The resulting BATs should be glued at the final phase.
     433             :  *
     434             :  * Raw Load
     435             :  * Front-ends can bypass most of the overhead in loading the BATs
     436             :  * by preparing the corresponding files directly and replace those
     437             :  * created by e.g. the SQL frontend.
     438             :  * This strategy is only advisable for cases where we have very
     439             :  * large files >200GB and/or are created by a well debugged code.
     440             :  *
     441             :  * To experiment with this approach, the code base responds
     442             :  * on negative number of cores by dumping the data directly in BAT
     443             :  * storage format into a collections of files on disk.
     444             :  * It reports on the actions to be taken to replace BATs.
     445             :  * This technique is initially only supported for fixed-sized columns.
     446             :  * The rawmode() indicator acts as the internal switch.
     447             :  */
     448             : 
     449             : /*
     450             :  * To speed up loading ascii files we have to determine the number of blocks.
     451             :  * This depends on the number of cores available.
     452             :  * For the time being we hardwire this decision based on our own
     453             :  * platforms.
     454             :  * Furthermore, we only consider parallel load for file-based requests.
     455             :  *
     456             :  * To simplify our world, we assume a single producer process.
     457             :  */
     458             : 
     459             : static int
     460             : output_file_default(Tablet *as, BAT *order, stream *fd)
     461             : {
     462             :         size_t len = BUFSIZ, locallen = BUFSIZ;
     463             :         int res = 0;
     464             :         char *buf = GDKmalloc(len);
     465             :         char *localbuf = GDKmalloc(len);
     466             :         BUN p, q;
     467             :         oid id;
     468             :         BUN i = 0;
     469             :         BUN offset = as->offset;
     470             : 
     471             :         if (buf == NULL || localbuf == NULL) {
     472             :                 GDKfree(buf);
     473             :                 GDKfree(localbuf);
     474             :                 return -1;
     475             :         }
     476             :         for (q = offset + as->nr, p = offset, id = order->hseqbase + offset; p < q; p++, id++) {
     477             :                 if ((res = output_line(&buf, &len, &localbuf, &locallen, as->format, fd, as->nr_attrs, id)) < 0) {
     478             :                         GDKfree(buf);
     479             :                         GDKfree(localbuf);
     480             :                         return res;
     481             :                 }
     482             :                 i++;
     483             :         }
     484             :         GDKfree(localbuf);
     485             :         GDKfree(buf);
     486             :         return res;
     487             : }
     488             : 
     489             : static int
     490       53230 : output_file_dense(Tablet *as, stream *fd)
     491             : {
     492       53230 :         size_t len = BUFSIZ, locallen = BUFSIZ;
     493             :         int res = 0;
     494       53230 :         char *buf = GDKmalloc(len);
     495       53229 :         char *localbuf = GDKmalloc(len);
     496             :         BUN i = 0;
     497             : 
     498       53230 :         if (buf == NULL || localbuf == NULL) {
     499           0 :                 GDKfree(buf);
     500           0 :                 GDKfree(localbuf);
     501           0 :                 return -1;
     502             :         }
     503     1086820 :         for (i = 0; i < as->nr; i++) {
     504     1033590 :                 if ((res = output_line_dense(&buf, &len, &localbuf, &locallen, as->format, fd, as->nr_attrs)) < 0) {
     505           0 :                         GDKfree(buf);
     506           0 :                         GDKfree(localbuf);
     507           0 :                         return res;
     508             :                 }
     509             :         }
     510       53230 :         GDKfree(localbuf);
     511       53230 :         GDKfree(buf);
     512       53230 :         return res;
     513             : }
     514             : 
     515             : static int
     516             : output_file_ordered(Tablet *as, BAT *order, stream *fd)
     517             : {
     518             :         size_t len = BUFSIZ;
     519             :         int res = 0;
     520             :         char *buf = GDKmalloc(len);
     521             :         BUN p, q;
     522             :         BUN i = 0;
     523             :         BUN offset = as->offset;
     524             : 
     525             :         if (buf == NULL)
     526             :                 return -1;
     527             :         for (q = offset + as->nr, p = offset; p < q; p++, i++) {
     528             :                 oid h = order->hseqbase + p;
     529             : 
     530             :                 if ((res = output_line_lookup(&buf, &len, as->format, fd, as->nr_attrs, h)) < 0) {
     531             :                         GDKfree(buf);
     532             :                         return res;
     533             :                 }
     534             :         }
     535             :         GDKfree(buf);
     536             :         return res;
     537             : }
     538             : 
     539             : int
     540       60834 : TABLEToutput_file(Tablet *as, BAT *order, stream *s)
     541             : {
     542       60834 :         oid base = oid_nil;
     543       60834 :         BUN maxnr = BATcount(order);
     544             :         int ret = 0;
     545             : 
     546             :         /* only set nr if it is zero or lower (bogus) to the maximum value
     547             :          * possible (BATcount), if already set within BATcount range,
     548             :          * preserve value such that for instance SQL's reply_size still
     549             :          * works
     550             :          */
     551       60834 :         if (as->nr == BUN_NONE || as->nr > maxnr)
     552           0 :                 as->nr = maxnr;
     553             : 
     554       60834 :         base = check_BATs(as);
     555       60834 :         if (!is_oid_nil(base)) {
     556       53230 :                 if (order->hseqbase == base)
     557       53230 :                         ret = output_file_dense(as, s);
     558             :                 else
     559           0 :                         ret = output_file_ordered(as, order, s);
     560             :         } else {
     561        7604 :                 ret = output_file_default(as, order, s);
     562             :         }
     563       60834 :         return ret;
     564             : }
     565             : 
     566             : /*
     567             :  *  Niels Nes, Martin Kersten
     568             :  *
     569             :  * Parallel bulk load for SQL
     570             :  * The COPY INTO command for SQL is heavily CPU bound, which means
     571             :  * that ideally we would like to exploit the multi-cores to do that
     572             :  * work in parallel.
     573             :  * Complicating factors are the initial record offset, the
     574             :  * possible variable length of the input, and the original sort order
     575             :  * that should preferable be maintained.
     576             :  *
     577             :  * The code below consists of a file reader, which breaks up the
     578             :  * file into chunks of distinct lines. Then multiple parallel threads
     579             :  * grab them, and break them on the field boundaries.
     580             :  * After all fields are identified this way, the columns are converted
     581             :  * and stored in the BATs.
     582             :  *
     583             :  * The threads get a reference to a private copy of the READERtask.
     584             :  * It includes a list of columns they should handle. This is a basis
     585             :  * to distributed cheap and expensive columns over threads.
     586             :  *
     587             :  * The file reader overlaps IO with updates of the BAT.
     588             :  * Also the buffer size of the block stream might be a little small for
     589             :  * this task (1MB). It has been increased to 8MB, which indeed improved.
     590             :  *
     591             :  * The work divider allocates subtasks to threads based on the
     592             :  * observed time spending so far.
     593             :  */
     594             : 
     595             : /* #define MLOCK_TST did not make a difference on sf10 */
     596             : 
     597             : #define BREAKLINE 1
     598             : #define UPDATEBAT 2
     599             : #define SYNCBAT 3
     600             : #define ENDOFCOPY 4
     601             : 
     602             : typedef struct {
     603             :         Client cntxt;
     604             :         int id;                                         /* for self reference */
     605             :         int state;                                      /* line break=1 , 2 = update bat */
     606             :         int workers;                            /* how many concurrent ones */
     607             :         int error;                                      /* error during line break */
     608             :         int next;
     609             :         int limit;
     610             :         BUN cnt, maxrow;                        /* first row in file chunk. */
     611             :         lng skip;                                       /* number of lines to be skipped */
     612             :         lng *time, wtime;                       /* time per col + time per thread */
     613             :         int rounds;                                     /* how often did we divide the work */
     614             :         bool ateof;                                     /* io control */
     615             :         bool from_stdin;
     616             :         bool escape;                            /* whether to handle \ escapes */
     617             :         bstream *b;
     618             :         stream *out;
     619             :         MT_Id tid;
     620             :         MT_Sema producer;                       /* reader waits for call */
     621             :         MT_Sema consumer;                       /* reader waits for call */
     622             :         MT_Sema sema; /* threads wait for work , negative next implies exit */
     623             :         MT_Sema reply;                          /* let reader continue */
     624             :         Tablet *as;
     625             :         char *errbuf;
     626             :         const char *csep, *rsep;
     627             :         size_t seplen, rseplen;
     628             :         char quote;
     629             : 
     630             :         char *base[MAXBUFFERS], *input[MAXBUFFERS];     /* buffers for line splitter and tokenizer */
     631             :         size_t rowlimit[MAXBUFFERS]; /* determines maximal record length buffer */
     632             :         char **lines[MAXBUFFERS];
     633             :         int top[MAXBUFFERS];            /* number of lines in this buffer */
     634             :         int cur;  /* current buffer used by splitter and update threads */
     635             : 
     636             :         int *cols;                                      /* columns to handle */
     637             :         char ***fields;
     638             :         int besteffort;
     639             :         bte *rowerror;
     640             :         int errorcnt;
     641             : } READERtask;
     642             : 
     643             : static void
     644           7 : tablet_error(READERtask *task, lng row, int col, const char *msg, const char *fcn)
     645             : {
     646           7 :         MT_lock_set(&errorlock);
     647           7 :         if (task->cntxt->error_row != NULL) {
     648          14 :                 if (BUNappend(task->cntxt->error_row, &row, false) != GDK_SUCCEED ||
     649          14 :                         BUNappend(task->cntxt->error_fld, &col, false) != GDK_SUCCEED ||
     650          14 :                         BUNappend(task->cntxt->error_msg, msg, false) != GDK_SUCCEED ||
     651           7 :                         BUNappend(task->cntxt->error_input, fcn, false) != GDK_SUCCEED)
     652           0 :                         task->besteffort = 0;
     653           7 :                 if (!is_lng_nil(row) && task->rowerror && row < task->limit)
     654           7 :                         task->rowerror[row]++;
     655             :         }
     656           7 :         if (task->as->error == NULL) {
     657           5 :                 if (msg == NULL)
     658           0 :                         task->besteffort = 0;
     659           5 :                 else if (!is_lng_nil(row)) {
     660           5 :                         if (!is_int_nil(col))
     661           3 :                                 task->as->error = createException(MAL, "sql.copy_from", "line " LLFMT ": column %d: %s", row + 1, col + 1, msg);
     662             :                         else
     663           2 :                                 task->as->error = createException(MAL, "sql.copy_from", "line " LLFMT ": %s", row + 1, msg);
     664             :                 } else
     665           0 :                         task->as->error = createException(MAL, "sql.copy_from", "%s", msg);
     666             :         }
     667           7 :         task->errorcnt++;
     668           7 :         MT_lock_unset(&errorlock);
     669           7 : }
     670             : 
     671             : /*
     672             :  * The line is broken into pieces directly on their field separators. It assumes that we have
     673             :  * the record in the cache already, so we can do most work quickly.
     674             :  * Furthermore, it assume a uniform (SQL) pattern, without whitespace skipping, but with quote and separator.
     675             :  */
     676             : 
     677             : static size_t
     678         113 : mystrlen(const char *s)
     679             : {
     680             :         /* Calculate and return the space that is needed for the function
     681             :          * mycpstr below to do its work. */
     682             :         size_t len = 0;
     683             :         const char *s0 = s;
     684             : 
     685       30528 :         while (*s) {
     686       30415 :                 if ((*s & 0x80) == 0) {
     687             :                         ;
     688           6 :                 } else if ((*s & 0xC0) == 0x80) {
     689             :                         /* continuation byte */
     690           0 :                         len += 3;
     691           6 :                 } else if ((*s & 0xE0) == 0xC0) {
     692             :                         /* two-byte sequence */
     693           6 :                         if ((s[1] & 0xC0) != 0x80)
     694           0 :                                 len += 3;
     695             :                         else
     696           6 :                                 s += 2;
     697           0 :                 } else if ((*s & 0xF0) == 0xE0) {
     698             :                         /* three-byte sequence */
     699           0 :                         if ((s[1] & 0xC0) != 0x80 || (s[2] & 0xC0) != 0x80)
     700           0 :                                 len += 3;
     701             :                         else
     702           0 :                                 s += 3;
     703           0 :                 } else if ((*s & 0xF8) == 0xF0) {
     704             :                         /* four-byte sequence */
     705           0 :                         if ((s[1] & 0xC0) != 0x80 || (s[2] & 0xC0) != 0x80 || (s[3] & 0xC0) != 0x80)
     706           0 :                                 len += 3;
     707             :                         else
     708           0 :                                 s += 4;
     709             :                 } else {
     710             :                         /* not a valid start byte */
     711           0 :                         len += 3;
     712             :                 }
     713       30415 :                 s++;
     714             :         }
     715         113 :         len += s - s0;
     716         113 :         return len;
     717             : }
     718             : 
     719             : static char *
     720         179 : mycpstr(char *t, const char *s)
     721             : {
     722             :         /* Copy the string pointed to by s into the buffer pointed to by
     723             :          * t, and return a pointer to the NULL byte at the end.  During
     724             :          * the copy we translate incorrect UTF-8 sequences to escapes
     725             :          * looking like <XX> where XX is the hexadecimal representation of
     726             :          * the incorrect byte.  The buffer t needs to be large enough to
     727             :          * hold the result, but the correct length can be calculated by
     728             :          * the function mystrlen above.*/
     729       30666 :         while (*s) {
     730       30487 :                 if ((*s & 0x80) == 0) {
     731       30481 :                         *t++ = *s++;
     732           6 :                 } else if ((*s & 0xC0) == 0x80) {
     733           0 :                         t += sprintf(t, "<%02X>", (uint8_t) *s++);
     734           6 :                 } else if ((*s & 0xE0) == 0xC0) {
     735             :                         /* two-byte sequence */
     736           6 :                         if ((s[1] & 0xC0) != 0x80)
     737           0 :                                 t += sprintf(t, "<%02X>", (uint8_t) *s++);
     738             :                         else {
     739           6 :                                 *t++ = *s++;
     740           6 :                                 *t++ = *s++;
     741             :                         }
     742           0 :                 } else if ((*s & 0xF0) == 0xE0) {
     743             :                         /* three-byte sequence */
     744           0 :                         if ((s[1] & 0xC0) != 0x80 || (s[2] & 0xC0) != 0x80)
     745           0 :                                 t += sprintf(t, "<%02X>", (uint8_t) *s++);
     746             :                         else {
     747           0 :                                 *t++ = *s++;
     748           0 :                                 *t++ = *s++;
     749           0 :                                 *t++ = *s++;
     750             :                         }
     751           0 :                 } else if ((*s & 0xF8) == 0xF0) {
     752             :                         /* four-byte sequence */
     753           0 :                         if ((s[1] & 0xC0) != 0x80 || (s[2] & 0xC0) != 0x80 || (s[3] & 0xC0) != 0x80)
     754           0 :                                 t += sprintf(t, "<%02X>", (uint8_t) *s++);
     755             :                         else {
     756           0 :                                 *t++ = *s++;
     757           0 :                                 *t++ = *s++;
     758           0 :                                 *t++ = *s++;
     759           0 :                                 *t++ = *s++;
     760             :                         }
     761             :                 } else {
     762             :                         /* not a valid start byte */
     763           0 :                         t += sprintf(t, "<%02X>", (uint8_t) *s++);
     764             :                 }
     765             :         }
     766         179 :         *t = 0;
     767         179 :         return t;
     768             : }
     769             : 
     770             : static str
     771          31 : SQLload_error(READERtask *task, lng idx, BUN attrs)
     772             : {
     773             :         str line;
     774             :         char *s;
     775             :         size_t sz = 0;
     776             :         BUN i;
     777             : 
     778         128 :         for (i = 0; i < attrs; i++) {
     779          97 :                 if (task->fields[i][idx])
     780          87 :                         sz += mystrlen(task->fields[i][idx]);
     781          97 :                 sz += task->seplen;
     782             :         }
     783             : 
     784          31 :         s = line = GDKmalloc(sz + task->rseplen + 1);
     785          31 :         if (line == 0) {
     786           0 :                 tablet_error(task, idx, int_nil, "SQLload malloc error", "SQLload_error");
     787           0 :                 return 0;
     788             :         }
     789         128 :         for (i = 0; i < attrs; i++) {
     790          97 :                 if (task->fields[i][idx])
     791          87 :                         s = mycpstr(s, task->fields[i][idx]);
     792          97 :                 if (i < attrs - 1)
     793          66 :                         s = mycpstr(s, task->csep);
     794             :         }
     795          31 :         strcpy(s, task->rsep);
     796          31 :         return line;
     797             : }
     798             : 
     799             : /*
     800             :  * The parsing of the individual values is straightforward. If the value represents
     801             :  * the null-replacement string then we grab the underlying nil.
     802             :  * If the string starts with the quote identified from SQL, we locate the tail
     803             :  * and interpret the body.
     804             :  *
     805             :  * If inserting fails, we return -1; if the value cannot be parsed, we
     806             :  * return -1 if besteffort is not set, otherwise we return 0, but in
     807             :  * either case an entry is added to the error table.
     808             :  */
     809             : static inline int
     810   351434000 : SQLinsert_val(READERtask *task, int col, int idx)
     811             : {
     812   351434000 :         Column *fmt = task->as->format + col;
     813             :         const void *adt;
     814             :         char buf[BUFSIZ];
     815   351434000 :         char *s = task->fields[col][idx];
     816             :         char *err = NULL;
     817             :         int ret = 0;
     818             : 
     819             :         /* include testing on the terminating null byte !! */
     820   351434000 :         if (s == 0) {
     821     6607280 :                 adt = fmt->nildata;
     822     6607280 :                 fmt->c->tnonil = false;
     823             :         } else {
     824   344827000 :                 if (task->escape) {
     825   344627000 :                         char *data = GDKmalloc(strlen(s) + 1);
     826   703200000 :                         if (data == NULL ||
     827   349535000 :                                 GDKstrFromStr((unsigned char *) data, (unsigned char *) s, strlen(s)) < 0)
     828             :                                 adt = NULL;
     829             :                         else
     830   353665000 :                                 adt = fmt->frstr(fmt, fmt->adt, data);
     831   348602000 :                         GDKfree(data);
     832             :                 } else
     833      200020 :                         adt = fmt->frstr(fmt, fmt->adt, s);
     834             :         }
     835             : 
     836             :         /* col is zero-based, but for error messages it needs to be
     837             :          * one-based, and from here on, we only use col anymore to produce
     838             :          * error messages */
     839   361083000 :         col++;
     840             : 
     841   361083000 :         if (adt == NULL) {
     842          26 :                 lng row = task->cnt + idx + 1;
     843          26 :                 snprintf(buf, sizeof(buf), "'%s' expected", fmt->type);
     844          26 :                 err = SQLload_error(task, idx, task->as->nr_attrs);
     845          26 :                 if (task->rowerror) {
     846          26 :                         if (s) {
     847          26 :                                 size_t slen = mystrlen(s);
     848          26 :                                 char *scpy = GDKmalloc(slen + 1);
     849          26 :                                 if ( scpy == NULL){
     850           0 :                                         task->rowerror[idx]++;
     851           0 :                                         task->errorcnt++;
     852           0 :                                         task->besteffort = 0; /* no longer best effort */
     853           0 :                                         if (task->cntxt->error_row == NULL ||
     854           0 :                                                 BUNappend(task->cntxt->error_row, &row, false) != GDK_SUCCEED ||
     855           0 :                                                 BUNappend(task->cntxt->error_fld, &col, false) != GDK_SUCCEED ||
     856           0 :                                                 BUNappend(task->cntxt->error_msg, SQLSTATE(HY013) MAL_MALLOC_FAIL, false) != GDK_SUCCEED ||
     857           0 :                                                 BUNappend(task->cntxt->error_input, err, false) != GDK_SUCCEED) {
     858             :                                                 ;               /* ignore error here: we're already not best effort */
     859             :                                         }
     860           0 :                                         GDKfree(err);
     861           0 :                                         return -1;
     862             :                                 }
     863          26 :                                 mycpstr(scpy, s);
     864             :                                 s = scpy;
     865             :                         }
     866          26 :                         MT_lock_set(&errorlock);
     867          26 :                         snprintf(buf, sizeof(buf),
     868             :                                          "line " LLFMT " field %s '%s' expected%s%s%s",
     869          26 :                                          row, fmt->name ? fmt->name : "", fmt->type,
     870             :                                          s ? " in '" : "", s ? s : "", s ? "'" : "");
     871          26 :                         GDKfree(s);
     872          26 :                         if (task->as->error == NULL && (task->as->error = GDKstrdup(buf)) == NULL)
     873           0 :                                 task->as->error = createException(MAL, "sql.copy_from", SQLSTATE(HY013) MAL_MALLOC_FAIL);
     874          26 :                         task->rowerror[idx]++;
     875          26 :                         task->errorcnt++;
     876          52 :                         if (task->cntxt->error_row == NULL ||
     877          52 :                                 BUNappend(task->cntxt->error_row, &row, false) != GDK_SUCCEED ||
     878          52 :                                 BUNappend(task->cntxt->error_fld, &col, false) != GDK_SUCCEED ||
     879          52 :                                 BUNappend(task->cntxt->error_msg, buf, false) != GDK_SUCCEED ||
     880          26 :                                 BUNappend(task->cntxt->error_input, err, false) != GDK_SUCCEED) {
     881           0 :                                 GDKfree(err);
     882           0 :                                 task->besteffort = 0; /* no longer best effort */
     883           0 :                                 MT_lock_unset(&errorlock);
     884           0 :                                 return -1;
     885             :                         }
     886          26 :                         MT_lock_unset(&errorlock);
     887             :                 }
     888          26 :                 ret = -!task->besteffort; /* yep, two unary operators ;-) */
     889          26 :                 GDKfree(err);
     890             :                 err = NULL;
     891             :                 /* replace it with a nil */
     892          26 :                 adt = fmt->nildata;
     893          26 :                 fmt->c->tnonil = false;
     894             :         }
     895   361083000 :         if (bunfastapp(fmt->c, adt) == GDK_SUCCEED)
     896             :                 return ret;
     897             : 
     898             :         /* failure */
     899           0 :         if (task->rowerror) {
     900           0 :                 lng row = BATcount(fmt->c);
     901           0 :                 MT_lock_set(&errorlock);
     902           0 :                 if (task->cntxt->error_row == NULL ||
     903           0 :                         BUNappend(task->cntxt->error_row, &row, false) != GDK_SUCCEED ||
     904           0 :                         BUNappend(task->cntxt->error_fld, &col, false) != GDK_SUCCEED ||
     905           0 :                         BUNappend(task->cntxt->error_msg, "insert failed", false) != GDK_SUCCEED ||
     906           0 :                         (err = SQLload_error(task, idx,task->as->nr_attrs)) == NULL ||
     907           0 :                         BUNappend(task->cntxt->error_input, err, false) != GDK_SUCCEED)
     908           0 :                         task->besteffort = 0;
     909           0 :                 GDKfree(err);
     910           0 :                 task->rowerror[idx]++;
     911           0 :                 task->errorcnt++;
     912           0 :                 MT_lock_unset(&errorlock);
     913             :         }
     914           0 :         task->besteffort = 0;                /* no longer best effort */
     915           0 :         return -1;
     916             : }
     917             : 
     918             : static int
     919     2757300 : SQLworker_column(READERtask *task, int col)
     920             : {
     921             :         int i;
     922     2757300 :         Column *fmt = task->as->format;
     923             : 
     924     2757300 :         if (fmt[col].c == NULL)
     925             :                 return 0;
     926             : 
     927             :         /* watch out for concurrent threads */
     928     2757290 :         MT_lock_set(&mal_copyLock);
     929     2760830 :         if (!fmt[col].skip && BATcapacity(fmt[col].c) < BATcount(fmt[col].c) + task->next) {
     930         252 :                 if (BATextend(fmt[col].c, BATgrows(fmt[col].c) + task->limit) != GDK_SUCCEED) {
     931           0 :                         tablet_error(task, lng_nil, col, "Failed to extend the BAT, perhaps disk full\n", "SQLworker_column");
     932           0 :                         MT_lock_unset(&mal_copyLock);
     933           0 :                         return -1;
     934             :                 }
     935             :         }
     936     2760830 :         MT_lock_unset(&mal_copyLock);
     937             : 
     938   354654000 :         for (i = 0; i < task->top[task->cur]; i++) {
     939   351901000 :                 if (!fmt[col].skip && SQLinsert_val(task, col, i) < 0) {
     940          20 :                         BATsetcount(fmt[col].c, BATcount(fmt[col].c));
     941          20 :                         return -1;
     942             :                 }
     943             :         }
     944     2753540 :         BATsetcount(fmt[col].c, BATcount(fmt[col].c));
     945     2756260 :         fmt[col].c->theap.dirty |= BATcount(fmt[col].c) > 0;
     946             : 
     947     2756260 :         return 0;
     948             : }
     949             : 
     950             : /*
     951             :  * The lines are broken on the column separator. Any error is shown and reflected with
     952             :  * setting the reference of the offending row fields to NULL.
     953             :  * This allows the loading to continue, skipping the minimal number of rows.
     954             :  * The details about the locations can be inspected from the error table.
     955             :  * We also trim the quotes around strings.
     956             :  */
     957             : static int
     958   135165000 : SQLload_parse_line(READERtask *task, int idx)
     959             : {
     960             :         BUN i;
     961             :         char errmsg[BUFSIZ];
     962   135165000 :         char ch = *task->csep;
     963   135165000 :         char *line = task->lines[task->cur][idx];
     964   135165000 :         Tablet *as = task->as;
     965   135165000 :         Column *fmt = as->format;
     966             :         bool error = false;
     967             :         str errline = 0;
     968             : 
     969             :         assert(idx < task->top[task->cur]);
     970             :         assert(line);
     971   135165000 :         errmsg[0] = 0;
     972             : 
     973   135165000 :         if (task->quote || task->seplen != 1) {
     974    12434100 :                 for (i = 0; i < as->nr_attrs; i++) {
     975             :                         bool quote = false;
     976    10212100 :                         task->fields[i][idx] = line;
     977             :                         /* recognize fields starting with a quote, keep them */
     978    10212100 :                         if (*line && *line == task->quote) {
     979             :                                 quote = true;
     980     4973450 :                                 task->fields[i][idx] = line + 1;
     981     4973450 :                                 line = tablet_skip_string(line + 1, task->quote, task->escape);
     982             : 
     983     4611040 :                                 if (!line) {
     984           0 :                                         errline = SQLload_error(task, idx, i+1);
     985           0 :                                         snprintf(errmsg, BUFSIZ, "Quote (%c) missing", task->quote);
     986           0 :                                         tablet_error(task, idx, (int) i, errmsg, errline);
     987           0 :                                         GDKfree(errline);
     988             :                                         error = true;
     989           0 :                                         goto errors1;
     990             :                                 } else
     991     4611040 :                                         *line++ = 0;
     992             :                         }
     993             : 
     994             :                         /* eat away the column separator */
     995    37451500 :                         for (; *line; line++)
     996    35951800 :                                 if (*line == '\\') {
     997           2 :                                         if (line[1])
     998           2 :                                                 line++;
     999    35951800 :                                 } else if (*line == ch && (task->seplen == 1 || strncmp(line, task->csep, task->seplen) == 0)) {
    1000     8349930 :                                         *line = 0;
    1001     8349930 :                                         line += task->seplen;
    1002     8349930 :                                         goto endoffieldcheck;
    1003             :                                 }
    1004             : 
    1005             :                         /* not enough fields */
    1006     1499720 :                         if (i < as->nr_attrs - 1) {
    1007           0 :                                 errline = SQLload_error(task, idx, i+1);
    1008           0 :                                 tablet_error(task, idx, (int) i, "Column value missing", errline);
    1009           0 :                                 GDKfree(errline);
    1010             :                                 error = true;
    1011           0 :                           errors1:
    1012             :                                 /* we save all errors detected  as NULL values */
    1013           0 :                                 for (; i < as->nr_attrs; i++)
    1014           0 :                                         task->fields[i][idx] = NULL;
    1015           0 :                                 i--;
    1016             :                         }
    1017     1499720 :                   endoffieldcheck:
    1018             :                         ;
    1019             :                         /* check for user defined NULL string */
    1020     9849650 :                         if ((!quote || !fmt->null_length) && fmt->nullstr && task->fields[i][idx] && strncasecmp(task->fields[i][idx], fmt->nullstr, fmt->null_length + 1) == 0)
    1021     2298470 :                                 task->fields[i][idx] = 0;
    1022             :                 }
    1023             :         } else {
    1024             :                 assert(!task->quote);
    1025             :                 assert(task->seplen == 1);
    1026   361551000 :                 for (i = 0; i < as->nr_attrs; i++) {
    1027   228970000 :                         task->fields[i][idx] = line;
    1028             : 
    1029             :                         /* eat away the column separator */
    1030  1871180000 :                         for (; *line; line++)
    1031  1739040000 :                                 if (*line == '\\') {
    1032    16757600 :                                         if (line[1])
    1033    16757600 :                                                 line++;
    1034  1722280000 :                                 } else if (*line == ch) {
    1035    96831200 :                                         *line = 0;
    1036    96831200 :                                         line++;
    1037    96831200 :                                         goto endoffield2;
    1038             :                                 }
    1039             : 
    1040             :                         /* not enough fields */
    1041   132139000 :                         if (i < as->nr_attrs - 1) {
    1042           4 :                                 errline = SQLload_error(task, idx,i+1);
    1043           4 :                                 tablet_error(task, idx, (int) i, "Column value missing", errline);
    1044           4 :                                 GDKfree(errline);
    1045             :                                 error = true;
    1046             :                                 /* we save all errors detected */
    1047          12 :                                 for (; i < as->nr_attrs; i++)
    1048           8 :                                         task->fields[i][idx] = NULL;
    1049           4 :                                 i--;
    1050             :                         }
    1051   132139000 :                   endoffield2:
    1052             :                         ;
    1053             :                         /* check for user defined NULL string */
    1054   228970000 :                         if (fmt->nullstr && task->fields[i][idx] && strncasecmp(task->fields[i][idx], fmt->nullstr, fmt->null_length + 1) == 0) {
    1055     3869310 :                                 task->fields[i][idx] = 0;
    1056             :                         }
    1057             :                 }
    1058             :         }
    1059             :         /* check for too many values as well*/
    1060   134802000 :         if (line && *line && i == as->nr_attrs) {
    1061           1 :                 errline = SQLload_error(task, idx, task->as->nr_attrs);
    1062           1 :                 snprintf(errmsg, BUFSIZ, "Leftover data '%s'",line);
    1063           1 :                 tablet_error(task, idx, (int) i, errmsg, errline);
    1064           1 :                 GDKfree(errline);
    1065             :                 error = true;
    1066             :         }
    1067   134802000 :         return error ? -1 : 0;
    1068             : }
    1069             : 
    1070             : static void
    1071        1032 : SQLworker(void *arg)
    1072             : {
    1073             :         READERtask *task = (READERtask *) arg;
    1074             :         unsigned int i;
    1075             :         int j, piece;
    1076             :         lng t0;
    1077             : 
    1078        1032 :         GDKsetbuf(GDKmalloc(GDKMAXERRLEN));     /* where to leave errors */
    1079        1032 :         GDKclrerr();
    1080        1032 :         task->errbuf = GDKerrbuf;
    1081             : 
    1082     3057410 :         while (task->top[task->cur] >= 0) {
    1083     3057410 :                 MT_sema_down(&task->sema);
    1084             : 
    1085             :                 /* stage one, break the lines spread the worker over the workers */
    1086     3059880 :                 switch (task->state) {
    1087     1528910 :                 case BREAKLINE:
    1088     1528910 :                         t0 = GDKusec();
    1089     1528760 :                         piece = (task->top[task->cur] + task->workers) / task->workers;
    1090             : 
    1091   136178000 :                         for (j = piece * task->id; j < task->top[task->cur] && j < piece * (task->id +1); j++)
    1092   134649000 :                                 if (task->lines[task->cur][j]) {
    1093   134649000 :                                         if (SQLload_parse_line(task, j) < 0) {
    1094           5 :                                                 task->errorcnt++;
    1095             :                                                 // early break unless best effort
    1096           5 :                                                 if (!task->besteffort)
    1097             :                                                         break;
    1098             :                                         }
    1099             :                                 }
    1100     1528740 :                         task->wtime = GDKusec() - t0;
    1101     1528620 :                         break;
    1102     1528940 :                 case UPDATEBAT:
    1103     1528940 :                         if (!task->besteffort && task->errorcnt)
    1104             :                                 break;
    1105             :                         /* stage two, updating the BATs */
    1106     7977290 :                         for (i = 0; i < task->as->nr_attrs; i++)
    1107     6452260 :                                 if (task->cols[i]) {
    1108     2758760 :                                         t0 = GDKusec();
    1109     2758350 :                                         if (SQLworker_column(task, task->cols[i] - 1) < 0)
    1110             :                                                 break;
    1111     2755370 :                                         t0 = GDKusec() - t0;
    1112     2754850 :                                         task->time[i] += t0;
    1113     2754850 :                                         task->wtime += t0;
    1114             :                                 }
    1115             :                         break;
    1116        1010 :                 case SYNCBAT:
    1117        1010 :                         if (!task->besteffort && task->errorcnt)
    1118             :                                 break;
    1119       12535 :                         for (i = 0; i < task->as->nr_attrs; i++)
    1120       11526 :                                 if (task->cols[i]) {
    1121        8899 :                                         BAT *b = task->as->format[task->cols[i] - 1].c;
    1122        8899 :                                         if (b == NULL)
    1123           6 :                                                 continue;
    1124        8893 :                                         t0 = GDKusec();
    1125        8892 :                                         if (b->batTransient)
    1126        8830 :                                                 continue;
    1127          62 :                                         BATmsync(b);
    1128          62 :                                         t0 = GDKusec() - t0;
    1129          62 :                                         task->time[i] += t0;
    1130          62 :                                         task->wtime += t0;
    1131             :                                 }
    1132             :                         break;
    1133        1032 :                 case ENDOFCOPY:
    1134        1032 :                         MT_sema_up(&task->reply);
    1135        1032 :                         goto do_return;
    1136             :                 }
    1137     3054680 :                 MT_sema_up(&task->reply);
    1138             :         }
    1139           0 :         MT_sema_up(&task->reply);
    1140             : 
    1141        1032 :   do_return:
    1142        1032 :         GDKfree(GDKerrbuf);
    1143        1032 :         GDKsetbuf(0);
    1144        1032 : }
    1145             : 
    1146             : static void
    1147      607688 : SQLworkdivider(READERtask *task, READERtask *ptask, int nr_attrs, int threads)
    1148             : {
    1149             :         int i, j, mi;
    1150             :         lng loc[MAXWORKERS];
    1151             : 
    1152             :         /* after a few rounds we stick to the work assignment */
    1153      607688 :         if (task->rounds > 8)
    1154      607542 :                 return;
    1155             :         /* simple round robin the first time */
    1156      143827 :         if (threads == 1 || task->rounds++ == 0) {
    1157     1037600 :                 for (i = j = 0; i < nr_attrs; i++, j++)
    1158      893920 :                         ptask[j % threads].cols[i] = task->cols[i];
    1159             :                 return;
    1160             :         }
    1161         146 :         memset((char *) loc, 0, sizeof(lng) * MAXWORKERS);
    1162             :         /* use of load directives */
    1163        2145 :         for (i = 0; i < nr_attrs; i++)
    1164        7972 :                 for (j = 0; j < threads; j++)
    1165        5973 :                         ptask[j].cols[i] = 0;
    1166             : 
    1167             :         /* now allocate the work to the threads */
    1168        2145 :         for (i = 0; i < nr_attrs; i++, j++) {
    1169             :                 mi = 0;
    1170        5973 :                 for (j = 1; j < threads; j++)
    1171        3974 :                         if (loc[j] < loc[mi])
    1172             :                                 mi = j;
    1173             : 
    1174        1999 :                 ptask[mi].cols[i] = task->cols[i];
    1175        1999 :                 loc[mi] += task->time[i];
    1176             :         }
    1177             :         /* reset the timer */
    1178        2145 :         for (i = 0; i < nr_attrs; i++, j++)
    1179        1999 :                 task->time[i] = 0;
    1180             : }
    1181             : 
    1182             : /*
    1183             :  * Reading is handled by a separate task as a preparation for more parallelism.
    1184             :  * A buffer is filled with proper lines.
    1185             :  * If we are reading from a file then a double buffering scheme ia activated.
    1186             :  * Reading from the console (stdin) remains single buffered only.
    1187             :  * If we end up with unfinished records, then the rowlimit will terminate the process.
    1188             :  */
    1189             : 
    1190             : typedef unsigned char (*dfa_t)[256];
    1191             : 
    1192             : static dfa_t
    1193         904 : mkdfa(const unsigned char *sep, size_t seplen)
    1194             : {
    1195             :         dfa_t dfa;
    1196             :         size_t i, j, k;
    1197             : 
    1198         904 :         dfa = GDKzalloc(seplen * sizeof(*dfa));
    1199         904 :         if (dfa == NULL)
    1200             :                 return NULL;
    1201             :         /* Each character in the separator string advances the state by
    1202             :          * one.  If state reaches seplen, the separator was recognized.
    1203             :          *
    1204             :          * The first loop and the nested loop make sure that if in any
    1205             :          * state we encounter an invalid character, but part of what we've
    1206             :          * matched so far is a prefix of the separator, we go to the
    1207             :          * appropriate state. */
    1208        1840 :         for (i = 0; i < seplen; i++)
    1209         936 :                 dfa[i][sep[0]] = 1;
    1210        1840 :         for (j = 0; j < seplen; j++) {
    1211         936 :                 dfa[j][sep[j]] = (unsigned char) (j + 1);
    1212         968 :                 for (k = 0; k < j; k++) {
    1213          64 :                         for (i = 0; i < j - k; i++)
    1214          32 :                                 if (sep[k + i] != sep[i])
    1215             :                                         break;
    1216          32 :                         if (i == j - k && dfa[j][sep[i]] <= i)
    1217           0 :                                 dfa[j][sep[i]] = (unsigned char) (i + 1);
    1218             :                 }
    1219             :         }
    1220             :         return dfa;
    1221             : }
    1222             : 
    1223             : static void
    1224         904 : SQLproducer(void *p)
    1225             : {
    1226             :         READERtask *task = (READERtask *) p;
    1227             :         bool consoleinput = false;
    1228             :         int cur = 0;            // buffer being filled
    1229         904 :         bool blocked[MAXBUFFERS] = { false };
    1230         904 :         bool ateof[MAXBUFFERS] = { false };
    1231         904 :         BUN cnt = 0, bufcnt[MAXBUFFERS] = { 0 };
    1232             :         char *end, *e, *s = NULL, *base;
    1233         904 :         const char *rsep = task->rsep;
    1234         904 :         size_t rseplen = strlen(rsep), partial = 0;
    1235         904 :         char quote = task->quote;
    1236             :         dfa_t rdfa;
    1237             :         lng rowno = 0;
    1238             : 
    1239         904 :         MT_sema_down(&task->producer);
    1240         904 :         if (task->id < 0) {
    1241             :                 return;
    1242             :         }
    1243             : 
    1244         904 :         rdfa = mkdfa((const unsigned char *) rsep, rseplen);
    1245         904 :         if (rdfa == NULL) {
    1246           0 :                 tablet_error(task, lng_nil, int_nil, "cannot allocate memory", "");
    1247           0 :                 ateof[cur] = true;
    1248           0 :                 goto reportlackofinput;
    1249             :         }
    1250             : 
    1251             : /*      TRC_DEBUG(MAL_SERVER, "SQLproducer started size '%zu' and len '%zu'\n", task->b->size, task->b->len);*/
    1252             : 
    1253         904 :         base = end = s = task->input[cur];
    1254         904 :         *s = 0;
    1255         904 :         task->cur = cur;
    1256         904 :         if (task->as->filename == NULL) {
    1257             :                 consoleinput = true;
    1258         658 :                 goto parseSTDIN;
    1259             :         }
    1260             :         for (;;) {
    1261      607672 :                 ateof[cur] = !tablet_read_more(task->b, task->out, task->b->size);
    1262             : 
    1263             :                 // we may be reading from standard input and may be out of input
    1264             :                 // warn the consumers
    1265      607672 :                 if (ateof[cur] && partial) {
    1266             :                         if (partial) {
    1267           1 :                                 tablet_error(task, rowno, int_nil, "incomplete record at end of file", s);
    1268           1 :                                 task->b->pos += partial;
    1269             :                         }
    1270           1 :                         goto reportlackofinput;
    1271             :                 }
    1272             : 
    1273      607671 :                 if (task->errbuf && task->errbuf[0]) {
    1274           0 :                         if (GDKerrbuf && GDKerrbuf[0]) {
    1275           0 :                                 tablet_error(task, rowno, int_nil, GDKerrbuf, "SQLload_file");
    1276             : /*                              TRC_DEBUG(MAL_SERVER, "Bailout on SQLload\n");*/
    1277             :                                 ateof[cur] = true;
    1278             :                                 break;
    1279             :                         }
    1280             :                 }
    1281             : 
    1282      607671 :           parseSTDIN:
    1283             : 
    1284             :                 /* copy the stream buffer into the input buffer, which is guaranteed larger, but still limited */
    1285             :                 partial = 0;
    1286      608329 :                 task->top[cur] = 0;
    1287      608329 :                 s = task->input[cur];
    1288             :                 base = end;
    1289             :                 /* avoid too long records */
    1290      608329 :                 if (end - s + task->b->len - task->b->pos >= task->rowlimit[cur]) {
    1291             :                         /* the input buffer should be extended, but 'base' is not shared
    1292             :                            between the threads, which we can not now update.
    1293             :                            Mimick an ateof instead; */
    1294           0 :                         tablet_error(task, rowno, int_nil, "record too long", "");
    1295           0 :                         ateof[cur] = true;
    1296             : /*                      TRC_DEBUG(MAL_SERVER, "Bailout on SQLload confronted with too large record\n");*/
    1297           0 :                         goto reportlackofinput;
    1298             :                 }
    1299      608329 :                 memcpy(end, task->b->buf + task->b->pos, task->b->len - task->b->pos);
    1300      608329 :                 end = end + task->b->len - task->b->pos;
    1301      608329 :                 *end = '\0';    /* this is safe, as the stream ensures an extra byte */
    1302             :                 /* Note that we rescan from the start of a record (the last
    1303             :                  * partial buffer from the previous iteration), even if in the
    1304             :                  * previous iteration we have already established that there
    1305             :                  * is no record separator in the first, perhaps significant,
    1306             :                  * part of the buffer. This is because if the record separator
    1307             :                  * is longer than one byte, it is too complex (i.e. would
    1308             :                  * require more state) to be sure what the state of the quote
    1309             :                  * status is when we back off a few bytes from where the last
    1310             :                  * scan ended (we need to back off some since we could be in
    1311             :                  * the middle of the record separator).  If this is too
    1312             :                  * costly, we have to rethink the matter. */
    1313      608329 :                 if (task->from_stdin && *s == '\n' && task->maxrow == BUN_MAX) {
    1314           3 :                         ateof[cur] = true;
    1315           3 :                         goto reportlackofinput;
    1316             :                 }
    1317   141133000 :                 for (e = s; *e && e < end && cnt < task->maxrow;) {
    1318             :                         /* tokenize the record completely
    1319             :                          *
    1320             :                          * The format of the input should comply to the following
    1321             :                          * grammar rule [ [[quote][[esc]char]*[quote]csep]*rsep]*
    1322             :                          * where quote is a single user-defined character.
    1323             :                          * Within the quoted fields a character may be escaped
    1324             :                          * with a backslash.  The correct number of fields should
    1325             :                          * be supplied.  In the first phase we simply break the
    1326             :                          * lines at the record boundary. */
    1327             :                         int nutf = 0;
    1328             :                         int m = 0;
    1329             :                         bool bs = false;
    1330             :                         char q = 0;
    1331             :                         size_t i = 0;
    1332  2872880000 :                         while (*e) {
    1333  2872870000 :                                 if (task->skip > 0) {
    1334             :                                         /* no interpretation of data we're skipping, just
    1335             :                                          * look for newline */
    1336        1709 :                                         if (*e == '\n')
    1337             :                                                 break;
    1338             :                                 } else {
    1339             :                                         /* check for correctly encoded UTF-8 */
    1340  2872870000 :                                         if (nutf > 0) {
    1341        1729 :                                                 if ((*e & 0xC0) != 0x80)
    1342           1 :                                                         goto badutf8;
    1343        1728 :                                                 if (m != 0 && (*e & m) == 0)
    1344           0 :                                                         goto badutf8;
    1345             :                                                 m = 0;
    1346        1728 :                                                 nutf--;
    1347  2872870000 :                                         } else if ((*e & 0xE0) == 0xC0) {
    1348             :                                                 nutf = 1;
    1349        1576 :                                                 if ((e[0] & 0x1E) == 0)
    1350           0 :                                                         goto badutf8;
    1351  2872860000 :                                         } else if ((*e & 0xF0) == 0xE0) {
    1352             :                                                 nutf = 2;
    1353          72 :                                                 if ((e[0] & 0x0F) == 0)
    1354             :                                                         m = 0x20;
    1355  2872860000 :                                         } else if ((*e & 0xF8) == 0xF0) {
    1356             :                                                 nutf = 3;
    1357           3 :                                                 if ((e[0] & 0x07) == 0)
    1358             :                                                         m = 0x30;
    1359  2872860000 :                                         } else if ((*e & 0x80) != 0) {
    1360           0 :                                                 goto badutf8;
    1361             :                                         }
    1362             :                                         /* check for quoting and the row separator */
    1363  2872870000 :                                         if (bs) {
    1364             :                                                 bs = false;
    1365  2872330000 :                                         } else if (task->escape && *e == '\\') {
    1366             :                                                 bs = true;
    1367             :                                                 i = 0;
    1368  2871790000 :                                         } else if (*e == q) {
    1369             :                                                 q = 0;
    1370  2866850000 :                                         } else if (*e == quote) {
    1371             :                                                 q = quote;
    1372             :                                                 i = 0;
    1373  2861900000 :                                         } else if (q == 0) {
    1374  2721980000 :                                                 i = rdfa[i][(unsigned char) *e];
    1375  2721980000 :                                                 if (i == rseplen)
    1376             :                                                         break;
    1377             :                                         }
    1378             :                                 }
    1379  2732350000 :                                 e++;
    1380             :                         }
    1381   140532000 :                         if (*e == 0) {
    1382        7596 :                                 partial = e - s;
    1383             :                                 /* found an incomplete record, saved for next round */
    1384        7596 :                                 if (s+partial < end) {
    1385             :                                         /* found a EOS in the input */
    1386           0 :                                         tablet_error(task, rowno, int_nil, "record too long (EOS found)", "");
    1387           0 :                                         ateof[cur] = true;
    1388           0 :                                         goto reportlackofinput;
    1389             :                                 }
    1390             :                                 break;
    1391             :                         } else {
    1392   140524000 :                                 rowno++;
    1393   140524000 :                                 if (task->skip > 0) {
    1394          13 :                                         task->skip--;
    1395             :                                 } else {
    1396             :                                         if (cnt < task->maxrow) {
    1397   140524000 :                                                 task->lines[cur][task->top[cur]++] = s;
    1398   140524000 :                                                 cnt++;
    1399             :                                         }
    1400   140524000 :                                         *(e + 1 - rseplen) = 0;
    1401             :                                 }
    1402   140524000 :                                 s = ++e;
    1403   140524000 :                                 task->b->pos += (size_t) (e - base);
    1404             :                                 base = e;
    1405   140524000 :                                 if (task->top[cur] == task->limit)
    1406             :                                         break;
    1407             :                         }
    1408             :                 }
    1409             : 
    1410      600729 :           reportlackofinput:
    1411             : /*        TRC_DEBUG(MAL_SERVER, "SQL producer got buffer '%d' filled with '%d' records\n", cur, task->top[cur]);*/
    1412             : 
    1413      608330 :                 if (consoleinput) {
    1414      606930 :                         task->cur = cur;
    1415      606930 :                         task->ateof = ateof[cur];
    1416      606930 :                         task->cnt = bufcnt[cur];
    1417             :                         /* tell consumer to go ahead */
    1418      606930 :                         MT_sema_up(&task->consumer);
    1419             :                         /* then wait until it is done */
    1420      606930 :                         MT_sema_down(&task->producer);
    1421      606930 :                         if (cnt == task->maxrow) {
    1422         651 :                                 GDKfree(rdfa);
    1423         651 :                                 return;
    1424             :                         }
    1425             :                 } else {
    1426             :                         assert(!blocked[cur]);
    1427        1400 :                         if (blocked[(cur + 1) % MAXBUFFERS]) {
    1428             :                                 /* first wait until other buffer is done */
    1429             : /*                              TRC_DEBUG(MAL_SERVER, "Wait for consumers to finish buffer: %d\n", (cur + 1) % MAXBUFFERS);*/
    1430             : 
    1431        1154 :                                 MT_sema_down(&task->producer);
    1432        1154 :                                 blocked[(cur + 1) % MAXBUFFERS] = false;
    1433        1154 :                                 if (task->state == ENDOFCOPY) {
    1434           0 :                                         GDKfree(rdfa);
    1435           0 :                                         return;
    1436             :                                 }
    1437             :                         }
    1438             :                         /* other buffer is done, proceed with current buffer */
    1439             :                         assert(!blocked[(cur + 1) % MAXBUFFERS]);
    1440        1400 :                         blocked[cur] = true;
    1441        1400 :                         task->cur = cur;
    1442        1400 :                         task->ateof = ateof[cur];
    1443        1400 :                         task->cnt = bufcnt[cur];
    1444             : /*                      TRC_DEBUG(MAL_SERVER, "SQL producer got buffer '%d' filled with '%d' records\n", cur, task->top[cur]);*/
    1445             : 
    1446        1400 :                         MT_sema_up(&task->consumer);
    1447             : 
    1448             :                         cur = (cur + 1) % MAXBUFFERS;
    1449             : /*                      TRC_DEBUG(MAL_SERVER, "May continue with buffer: %d\n", cur);*/
    1450             : 
    1451        1400 :                         if (cnt == task->maxrow) {
    1452         127 :                                 MT_sema_down(&task->producer);
    1453             : /*                              TRC_DEBUG(MAL_SERVER, "Producer delivered all\n");*/
    1454         127 :                                 GDKfree(rdfa);
    1455         127 :                                 return;
    1456             :                         }
    1457             :                 }
    1458             : /*              TRC_DEBUG(MAL_SERVER, "Continue producer buffer: %d\n", cur);*/
    1459             : 
    1460             :                 /* we ran out of input? */
    1461      607552 :                 if (task->ateof) {
    1462             : /*                      TRC_DEBUG(MAL_SERVER, "Producer encountered eof\n");*/
    1463         126 :                         GDKfree(rdfa);
    1464         126 :                         return;
    1465             :                 }
    1466             :                 /* consumers ask us to stop? */
    1467      607426 :                 if (task->state == ENDOFCOPY) {
    1468           0 :                         GDKfree(rdfa);
    1469           0 :                         return;
    1470             :                 }
    1471      607426 :                 bufcnt[cur] = cnt;
    1472             :                 /* move the non-parsed correct row data to the head of the next buffer */
    1473      607426 :                 end = s = task->input[cur];
    1474             :         }
    1475           0 :         if (cnt < task->maxrow && task->maxrow != BUN_NONE) {
    1476             :                 char msg[256];
    1477           0 :                 snprintf(msg, sizeof(msg), "incomplete record at end of file:%s\n", s);
    1478           0 :                 task->as->error = GDKstrdup(msg);
    1479           0 :                 tablet_error(task, rowno, int_nil, "incomplete record at end of file", s);
    1480           0 :                 task->b->pos += partial;
    1481             :         }
    1482           0 :         GDKfree(rdfa);
    1483           0 :         return;
    1484             : 
    1485           1 :   badutf8:
    1486           1 :         tablet_error(task, rowno, int_nil, "input not properly encoded UTF-8", "");
    1487           1 :         ateof[cur] = true;
    1488           1 :         goto reportlackofinput;
    1489             : }
    1490             : 
    1491             : static void
    1492         925 : create_rejects_table(Client cntxt)
    1493             : {
    1494         925 :         MT_lock_set(&mal_contextLock);
    1495         925 :         if (cntxt->error_row == NULL) {
    1496         396 :                 cntxt->error_row = COLnew(0, TYPE_lng, 0, TRANSIENT);
    1497         396 :                 cntxt->error_fld = COLnew(0, TYPE_int, 0, TRANSIENT);
    1498         396 :                 cntxt->error_msg = COLnew(0, TYPE_str, 0, TRANSIENT);
    1499         396 :                 cntxt->error_input = COLnew(0, TYPE_str, 0, TRANSIENT);
    1500         396 :                 if (cntxt->error_row == NULL || cntxt->error_fld == NULL || cntxt->error_msg == NULL || cntxt->error_input == NULL) {
    1501           0 :                         if (cntxt->error_row)
    1502           0 :                                 BBPunfix(cntxt->error_row->batCacheid);
    1503           0 :                         if (cntxt->error_fld)
    1504           0 :                                 BBPunfix(cntxt->error_fld->batCacheid);
    1505           0 :                         if (cntxt->error_msg)
    1506           0 :                                 BBPunfix(cntxt->error_msg->batCacheid);
    1507           0 :                         if (cntxt->error_input)
    1508           0 :                                 BBPunfix(cntxt->error_input->batCacheid);
    1509           0 :                         cntxt->error_row = cntxt->error_fld = cntxt->error_msg = cntxt->error_input = NULL;
    1510             :                 }
    1511             :         }
    1512         925 :         MT_lock_unset(&mal_contextLock);
    1513         925 : }
    1514             : 
    1515             : BUN
    1516         904 : SQLload_file(Client cntxt, Tablet *as, bstream *b, stream *out, const char *csep, const char *rsep, char quote, lng skip, lng maxrow, int best, bool from_stdin, const char *tabnam, bool escape)
    1517             : {
    1518             :         BUN cnt = 0, cntstart = 0, leftover = 0;
    1519             :         int res = 0;            /* < 0: error, > 0: success, == 0: continue processing */
    1520             :         int j;
    1521             :         BUN firstcol;
    1522             :         BUN i, attr;
    1523             :         READERtask task;
    1524             :         READERtask ptask[MAXWORKERS];
    1525         904 :         int threads = (maxrow< 0 || maxrow > (1 << 16)) ? (GDKnr_threads < MAXWORKERS && GDKnr_threads > 1 ? GDKnr_threads - 1 : MAXWORKERS - 1) : 1;
    1526             :         lng lio = 0, tio, t1 = 0, total = 0, iototal = 0;
    1527             :         char name[MT_NAME_LEN];
    1528             : 
    1529             : /*      TRC_DEBUG(MAL_SERVER, "Prepare copy work for '%d' threads col '%s' rec '%s' quot '%c'\n", threads, csep, rsep, quote);*/
    1530             : 
    1531         904 :         memset(ptask, 0, sizeof(ptask));
    1532         904 :         task = (READERtask) {
    1533             :                 .cntxt = cntxt,
    1534             :                 .from_stdin = from_stdin,
    1535             :                 .as = as,
    1536             :                 .escape = escape,               /* TODO: implement feature!!! */
    1537             :         };
    1538             : 
    1539             :         /* create the reject tables */
    1540         904 :         create_rejects_table(task.cntxt);
    1541         904 :         if (task.cntxt->error_row == NULL || task.cntxt->error_fld == NULL || task.cntxt->error_msg == NULL || task.cntxt->error_input == NULL) {
    1542           0 :                 tablet_error(&task, lng_nil, int_nil, "SQLload initialization failed", "");
    1543           0 :                 goto bailout;
    1544             :         }
    1545             : 
    1546             :         assert(rsep);
    1547             :         assert(csep);
    1548             :         assert(maxrow < 0 || maxrow <= (lng) BUN_MAX);
    1549         904 :         task.fields = (char ***) GDKmalloc(as->nr_attrs * sizeof(char **));
    1550         904 :         task.cols = (int *) GDKzalloc(as->nr_attrs * sizeof(int));
    1551         904 :         task.time = (lng *) GDKzalloc(as->nr_attrs * sizeof(lng));
    1552         904 :         if (task.fields == NULL || task.cols == NULL || task.time == NULL) {
    1553           0 :                 tablet_error(&task, lng_nil, int_nil, "memory allocation failed", "SQLload_file");
    1554           0 :                 goto bailout;
    1555             :         }
    1556         904 :         task.cur = 0;
    1557        2712 :         for (i = 0; i < MAXBUFFERS; i++) {
    1558        1808 :                 task.base[i] = GDKmalloc(MAXROWSIZE(2 * b->size) + 2);
    1559        1808 :                 task.rowlimit[i] = MAXROWSIZE(2 * b->size);
    1560        1808 :                 if (task.base[i] == 0) {
    1561           0 :                         tablet_error(&task, lng_nil, int_nil, SQLSTATE(HY013) MAL_MALLOC_FAIL, "SQLload_file");
    1562           0 :                         goto bailout;
    1563             :                 }
    1564        1808 :                 task.base[i][0] = task.base[i][b->size + 1] = 0;
    1565        1808 :                 task.input[i] = task.base[i] + 1;       /* wrap the buffer with null bytes */
    1566             :         }
    1567         904 :         task.besteffort = best;
    1568             : 
    1569         904 :         if (maxrow < 0)
    1570          44 :                 task.maxrow = BUN_MAX;
    1571             :         else
    1572         860 :                 task.maxrow = (BUN) maxrow;
    1573             : 
    1574         904 :         if (task.fields == 0 || task.cols == 0 || task.time == 0) {
    1575           0 :                 tablet_error(&task, lng_nil, int_nil, SQLSTATE(HY013) MAL_MALLOC_FAIL, "SQLload_file");
    1576           0 :                 goto bailout;
    1577             :         }
    1578             : 
    1579         904 :         task.skip = skip;
    1580         904 :         task.quote = quote;
    1581         904 :         task.csep = csep;
    1582         904 :         task.seplen = strlen(csep);
    1583         904 :         task.rsep = rsep;
    1584         904 :         task.rseplen = strlen(rsep);
    1585         904 :         task.errbuf = cntxt->errbuf;
    1586             : 
    1587         904 :         MT_sema_init(&task.producer, 0, "task.producer");
    1588         904 :         MT_sema_init(&task.consumer, 0, "task.consumer");
    1589         904 :         task.ateof = false;
    1590         904 :         task.b = b;
    1591         904 :         task.out = out;
    1592             : 
    1593             : #ifdef MLOCK_TST
    1594             :         mlock(task.fields, as->nr_attrs * sizeof(char *));
    1595             :         mlock(task.cols, as->nr_attrs * sizeof(int));
    1596             :         mlock(task.time, as->nr_attrs * sizeof(lng));
    1597             :         for (i = 0; i < MAXBUFFERS; i++)
    1598             :                 mlock(task.base[i], b->size + 2);
    1599             : #endif
    1600         904 :         as->error = NULL;
    1601             : 
    1602             :         /* there is no point in creating more threads than we have columns */
    1603         904 :         if (as->nr_attrs < (BUN) threads)
    1604          18 :                 threads = (int) as->nr_attrs;
    1605             : 
    1606             :         /* allocate enough space for pointers into the buffer pool.  */
    1607             :         /* the record separator is considered a column */
    1608         904 :         task.limit = (int) (b->size / as->nr_attrs + as->nr_attrs);
    1609       10366 :         for (i = 0; i < as->nr_attrs; i++) {
    1610        9462 :                 task.fields[i] = GDKmalloc(sizeof(char *) * task.limit);
    1611        9462 :                 if (task.fields[i] == 0) {
    1612           0 :                         if (task.as->error == NULL)
    1613           0 :                                 as->error = createException(MAL, "sql.copy_from", SQLSTATE(HY013) MAL_MALLOC_FAIL);
    1614           0 :                         goto bailout;
    1615             :                 }
    1616             : #ifdef MLOCK_TST
    1617             :                 mlock(task.fields[i], sizeof(char *) * task.limit);
    1618             : #endif
    1619        9462 :                 task.cols[i] = (int) (i + 1);   /* to distinguish non initialized later with zero */
    1620             :         }
    1621        2712 :         for (i = 0; i < MAXBUFFERS; i++) {
    1622        1808 :                 task.lines[i] = GDKzalloc(sizeof(char *) * task.limit);
    1623        1808 :                 if (task.lines[i] == NULL) {
    1624           0 :                         tablet_error(&task, lng_nil, int_nil, SQLSTATE(HY013) MAL_MALLOC_FAIL, "SQLload_file:failed to alloc buffers");
    1625           0 :                         goto bailout;
    1626             :                 }
    1627             :         }
    1628         904 :         task.rowerror = (bte *) GDKzalloc(sizeof(bte) * task.limit);
    1629         904 :         if( task.rowerror == NULL){
    1630           0 :                 tablet_error(&task, lng_nil, int_nil, SQLSTATE(HY013) MAL_MALLOC_FAIL, "SQLload_file:failed to alloc rowerror buffer");
    1631           0 :                 goto bailout;
    1632             :         }
    1633             : 
    1634         904 :         task.id = 0;
    1635         904 :         snprintf(name, sizeof(name), "prod-%s", tabnam);
    1636         904 :         if ((task.tid = THRcreate(SQLproducer, (void *) &task, MT_THR_JOINABLE, name)) == 0) {
    1637           0 :                 tablet_error(&task, lng_nil, int_nil, SQLSTATE(42000) "failed to start producer thread", "SQLload_file");
    1638           0 :                 goto bailout;
    1639             :         }
    1640             : /*      TRC_DEBUG(MAL_SERVER, "Parallel bulk load " LLFMT " - " BUNFMT "\n", skip, task.maxrow);*/
    1641             : 
    1642         904 :         task.workers = threads;
    1643        1936 :         for (j = 0; j < threads; j++) {
    1644        1032 :                 ptask[j] = task;
    1645        1032 :                 ptask[j].id = j;
    1646        1032 :                 ptask[j].cols = (int *) GDKzalloc(as->nr_attrs * sizeof(int));
    1647        1032 :                 if (ptask[j].cols == 0) {
    1648           0 :                         tablet_error(&task, lng_nil, int_nil, SQLSTATE(HY013) MAL_MALLOC_FAIL, "SQLload_file");
    1649           0 :                         task.id = -1;
    1650           0 :                         MT_sema_up(&task.producer);
    1651           0 :                         goto bailout;
    1652             :                 }
    1653             : #ifdef MLOCK_TST
    1654             :                 mlock(ptask[j].cols, sizeof(char *) * task.limit);
    1655             : #endif
    1656        1032 :                 snprintf(name, sizeof(name), "ptask%d.sema", j);
    1657        1032 :                 MT_sema_init(&ptask[j].sema, 0, name);
    1658        1032 :                 snprintf(name, sizeof(name), "ptask%d.repl", j);
    1659        1032 :                 MT_sema_init(&ptask[j].reply, 0, name);
    1660        1032 :                 snprintf(name, sizeof(name), "wrkr%d-%s", j, tabnam);
    1661        1032 :                 if ((ptask[j].tid = THRcreate(SQLworker, (void *) &ptask[j], MT_THR_JOINABLE, name)) == 0) {
    1662           0 :                         tablet_error(&task, lng_nil, int_nil, SQLSTATE(42000) "failed to start worker thread", "SQLload_file");
    1663             :                         threads = j;
    1664           0 :                         for (j = 0; j < threads; j++)
    1665           0 :                                 ptask[j].workers = threads;
    1666             :                 }
    1667             :         }
    1668         904 :         if (threads == 0) {
    1669             :                 /* no threads started */
    1670           0 :                 task.id = -1;
    1671           0 :                 MT_sema_up(&task.producer);
    1672           0 :                 goto bailout;
    1673             :         }
    1674         904 :         MT_sema_up(&task.producer);
    1675             : 
    1676         904 :         tio = GDKusec();
    1677         904 :         tio = GDKusec() - tio;
    1678         904 :         t1 = GDKusec();
    1679             : #ifdef MLOCK_TST
    1680             :         mlock(task.b->buf, task.b->size);
    1681             : #endif
    1682         907 :         for (firstcol = 0; firstcol < task.as->nr_attrs; firstcol++)
    1683         907 :                 if (task.as->format[firstcol].c != NULL)
    1684             :                         break;
    1685      609108 :         while (res == 0 && cnt < task.maxrow) {
    1686             : 
    1687             :                 // track how many elements are in the aggregated BATs
    1688      608330 :                 cntstart = BATcount(task.as->format[firstcol].c);
    1689             :                 /* block until the producer has data available */
    1690      608330 :                 MT_sema_down(&task.consumer);
    1691      608330 :                 cnt += task.top[task.cur];
    1692      608330 :                 if (task.ateof && !task.top[task.cur])
    1693             :                         break;
    1694      608204 :                 t1 = GDKusec() - t1;
    1695             :                 total += t1;
    1696             :                 iototal += tio;
    1697             : /*              TRC_DEBUG(MAL_SERVER, "Break: %d lines\n", task.top[task.cur]);*/
    1698             : 
    1699      608204 :                 t1 = GDKusec();
    1700      608204 :                 if (task.top[task.cur]) {
    1701             :                         /* activate the workers to break lines */
    1702     2136860 :                         for (j = 0; j < threads; j++) {
    1703             :                                 /* stage one, break the lines in parallel */
    1704     1529170 :                                 ptask[j].error = 0;
    1705     1529170 :                                 ptask[j].state = BREAKLINE;
    1706     1529170 :                                 ptask[j].next = task.top[task.cur];
    1707     1529170 :                                 ptask[j].fields = task.fields;
    1708     1529170 :                                 ptask[j].limit = task.limit;
    1709     1529170 :                                 ptask[j].cnt = task.cnt;
    1710     1529170 :                                 ptask[j].cur = task.cur;
    1711     1529170 :                                 ptask[j].top[task.cur] = task.top[task.cur];
    1712     1529170 :                                 MT_sema_up(&ptask[j].sema);
    1713             :                         }
    1714             :                 }
    1715      608204 :                 if (task.top[task.cur]) {
    1716             :                         /* await completion of line break phase */
    1717     2136860 :                         for (j = 0; j < threads; j++) {
    1718     1529170 :                                 MT_sema_down(&ptask[j].reply);
    1719     1529170 :                                 if (ptask[j].error) {
    1720             :                                         res = -1;
    1721             : /*                                      TRC_ERROR(MAL_SERVER, "Error in task: %d %d\n", j, ptask[j].error);*/
    1722             :                                 }
    1723             :                         }
    1724             :                 }
    1725             : 
    1726             : /*              TRC_DEBUG(MAL_SERVER,
    1727             :                         "Fill the BATs '%d' " BUNFMT " cap " BUNFMT "\n",
    1728             :                         task.top[task.cur], task.cnt, BATcapacity(as->format[task.cur].c));*/
    1729             : 
    1730      608204 :                 lio += GDKusec() - t1;  /* line break done */
    1731      608204 :                 if (task.top[task.cur]) {
    1732      607688 :                         if (res == 0) {
    1733      607688 :                                 SQLworkdivider(&task, ptask, (int) as->nr_attrs, threads);
    1734             : 
    1735             :                                 /* activate the workers to update the BATs */
    1736     2136860 :                                 for (j = 0; j < threads; j++) {
    1737             :                                         /* stage two, update the BATs */
    1738     1529170 :                                         ptask[j].state = UPDATEBAT;
    1739     1529170 :                                         MT_sema_up(&ptask[j].sema);
    1740             :                                 }
    1741             :                         }
    1742             :                 }
    1743      608204 :                 tio = GDKusec();
    1744             :                 tio = t1 - tio;
    1745             : 
    1746             :                 /* await completion of the BAT updates */
    1747      608204 :                 if (res == 0 && task.top[task.cur]) {
    1748     2136860 :                         for (j = 0; j < threads; j++) {
    1749     1529170 :                                 MT_sema_down(&ptask[j].reply);
    1750     1529170 :                                 if (ptask[j].errorcnt > 0 && !ptask[j].besteffort) {
    1751             :                                         res = -1;
    1752             :                                         best = 0;
    1753             :                                 }
    1754             :                         }
    1755             :                 }
    1756             : 
    1757             :                 /* trim the BATs discarding error tuples */
    1758             : #define trimerrors(TYPE)                                                                                                \
    1759             :                 do {                                                                                                                    \
    1760             :                         TYPE *src, *dst;                                                                                        \
    1761             :                         leftover= BATcount(task.as->format[attr].c);                         \
    1762             :                         limit = leftover - cntstart;                                                            \
    1763             :                         dst =src= (TYPE *) BUNtloc(task.as->format[attr].ci,cntstart); \
    1764             :                         for(j = 0; j < (int) limit; j++, src++){                                     \
    1765             :                                 if ( task.rowerror[j]){                                                                 \
    1766             :                                         leftover--;                                                                                     \
    1767             :                                         continue;                                                                                       \
    1768             :                                 }                                                                                                               \
    1769             :                                 *dst++ = *src;                                                                                  \
    1770             :                         }                                                                                                                       \
    1771             :                         BATsetcount(task.as->format[attr].c, leftover );                     \
    1772             :                 } while (0)
    1773             : 
    1774             : /*              TRC_DEBUG(MAL_SERVER, "Trim bbest '%d' table size " BUNFMT " - rows found so far " BUNFMT "\n",
    1775             :                                          best, BATcount(as->format[firstcol].c), task.cnt); */
    1776             : 
    1777      608204 :                 if (best && BATcount(as->format[firstcol].c)) {
    1778             :                         BUN limit;
    1779             :                         int width;
    1780             : 
    1781          77 :                         for (attr = 0; attr < as->nr_attrs; attr++) {
    1782          50 :                                 if (as->format[attr].skip)
    1783           5 :                                         continue;
    1784          45 :                                 width = as->format[attr].c->twidth;
    1785          45 :                                 switch (width){
    1786          11 :                                 case 1:
    1787          22 :                                         trimerrors(bte);
    1788          11 :                                         break;
    1789           0 :                                 case 2:
    1790           0 :                                         trimerrors(sht);
    1791           0 :                                         break;
    1792          34 :                                 case 4:
    1793          68 :                                         trimerrors(int);
    1794          34 :                                         break;
    1795           0 :                                 case 8:
    1796           0 :                                         trimerrors(lng);
    1797           0 :                                         break;
    1798             : #ifdef HAVE_HGE
    1799           0 :                                 case 16:
    1800           0 :                                         trimerrors(hge);
    1801           0 :                                         break;
    1802             : #endif
    1803           0 :                                 default:
    1804             :                                         {
    1805             :                                                 char *src, *dst;
    1806           0 :                                                 leftover= BATcount(task.as->format[attr].c);
    1807           0 :                                                 limit = leftover - cntstart;
    1808           0 :                                                 dst = src= BUNtloc(task.as->format[attr].ci,cntstart);
    1809           0 :                                                 for(j = 0; j < (int) limit; j++, src += width){
    1810           0 :                                                         if ( task.rowerror[j]){
    1811           0 :                                                                 leftover--;
    1812           0 :                                                                 continue;
    1813             :                                                         }
    1814           0 :                                                         if (dst != src)
    1815           0 :                                                                 memcpy(dst, src, width);
    1816           0 :                                                         dst += width;
    1817             :                                                 }
    1818           0 :                                                 BATsetcount(task.as->format[attr].c, leftover );
    1819             :                                         }
    1820           0 :                                         break;
    1821             :                                 }
    1822             :                         }
    1823             :                         // re-initialize the error vector;
    1824          27 :                         memset(task.rowerror, 0, task.limit);
    1825          27 :                         task.errorcnt = 0;
    1826             :                 }
    1827             : 
    1828      608204 :                 if (res < 0) {
    1829             :                         /* producer should stop */
    1830          22 :                         task.maxrow = cnt;
    1831          22 :                         task.state = ENDOFCOPY;
    1832             :                 }
    1833      608204 :                 MT_sema_up(&task.producer);
    1834             :         }
    1835             : 
    1836             : /*      TRC_DEBUG(MAL_SERVER, "End of block stream eof=%d - res=%d\n", task.ateof, res);*/
    1837             : 
    1838         904 :         cnt = BATcount(task.as->format[firstcol].c);
    1839             : 
    1840         904 :         task.ateof = true;
    1841         904 :         task.state = ENDOFCOPY;
    1842             : /*      TRC_DEBUG(MAL_SERVER, "Activate sync on disk\n");*/
    1843             : 
    1844             :         // activate the workers to sync the BATs to disk
    1845         904 :         if (res == 0) {
    1846        1892 :                 for (j = 0; j < threads; j++) {
    1847             :                         // stage three, update the BATs
    1848        1010 :                         ptask[j].state = SYNCBAT;
    1849        1010 :                         MT_sema_up(&ptask[j].sema);
    1850             :                 }
    1851             :         }
    1852             : 
    1853         904 :         if (!task.ateof || cnt < task.maxrow) {
    1854             : /*              TRC_DEBUG(MAL_SERVER, "Shut down reader\n");*/
    1855         133 :                 MT_sema_up(&task.producer);
    1856             :         }
    1857         904 :         MT_join_thread(task.tid);
    1858         904 :         if (res == 0) {
    1859             :                 // await completion of the BAT syncs
    1860        1892 :                 for (j = 0; j < threads; j++)
    1861        1010 :                         MT_sema_down(&ptask[j].reply);
    1862             :         }
    1863             : 
    1864             : /*      TRC_DEBUG(MAL_SERVER, "Activate endofcopy\n");*/
    1865             : 
    1866        1936 :         for (j = 0; j < threads; j++) {
    1867        1032 :                 ptask[j].state = ENDOFCOPY;
    1868        1032 :                 MT_sema_up(&ptask[j].sema);
    1869             :         }
    1870             :         /* wait for their death */
    1871        1936 :         for (j = 0; j < threads; j++)
    1872        1032 :                 MT_sema_down(&ptask[j].reply);
    1873             : 
    1874             : /*      TRC_DEBUG(MAL_SERVER, "Kill the workers\n");*/
    1875             : 
    1876        1936 :         for (j = 0; j < threads; j++) {
    1877        1032 :                 MT_join_thread(ptask[j].tid);
    1878        1032 :                 GDKfree(ptask[j].cols);
    1879        1032 :                 MT_sema_destroy(&ptask[j].sema);
    1880        1032 :                 MT_sema_destroy(&ptask[j].reply);
    1881             :         }
    1882             : 
    1883             : /*      TRC_DEBUG(MAL_SERVER, "Found " BUNFMT " tuples\n", cnt);*/
    1884             : /*      TRC_DEBUG(MAL_SERVER, "Leftover input: %.63s\n", task.b->buf + task.b->pos);*/
    1885             : 
    1886       10366 :         for (i = 0; i < as->nr_attrs; i++) {
    1887        9462 :                 BAT *b = task.as->format[i].c;
    1888        9462 :                 if (b)
    1889        9456 :                         BATsettrivprop(b);
    1890        9462 :                 GDKfree(task.fields[i]);
    1891             :         }
    1892         904 :         GDKfree(task.fields);
    1893         904 :         GDKfree(task.cols);
    1894         904 :         GDKfree(task.time);
    1895        2712 :         for (i = 0; i < MAXBUFFERS; i++) {
    1896        1808 :                 if (task.base[i])
    1897        1808 :                         GDKfree(task.base[i]);
    1898        1808 :                 if (task.lines[i])
    1899        1808 :                         GDKfree(task.lines[i]);
    1900             :         }
    1901         904 :         if (task.rowerror)
    1902         904 :                 GDKfree(task.rowerror);
    1903         904 :         MT_sema_destroy(&task.producer);
    1904         904 :         MT_sema_destroy(&task.consumer);
    1905             : #ifdef MLOCK_TST
    1906             :         munlockall();
    1907             : #endif
    1908             : 
    1909         904 :         return res < 0 ? BUN_NONE : cnt;
    1910             : 
    1911           0 :   bailout:
    1912           0 :         if (task.fields) {
    1913           0 :                 for (i = 0; i < as->nr_attrs; i++) {
    1914           0 :                         if (task.fields[i])
    1915           0 :                                 GDKfree(task.fields[i]);
    1916             :                 }
    1917           0 :                 GDKfree(task.fields);
    1918             :         }
    1919           0 :         GDKfree(task.time);
    1920           0 :         GDKfree(task.cols);
    1921           0 :         GDKfree(task.base[task.cur]);
    1922           0 :         GDKfree(task.rowerror);
    1923           0 :         for (i = 0; i < MAXWORKERS; i++)
    1924           0 :                 GDKfree(ptask[i].cols);
    1925             : #ifdef MLOCK_TST
    1926             :         munlockall();
    1927             : #endif
    1928             :         return BUN_NONE;
    1929             : }
    1930             : 
    1931             : /* return the latest reject table, to be on the safe side we should
    1932             :  * actually create copies within a critical section. Ignored for now. */
    1933             : str
    1934          21 : COPYrejects(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
    1935             : {
    1936          21 :         bat *row = getArgReference_bat(stk, pci, 0);
    1937          21 :         bat *fld = getArgReference_bat(stk, pci, 1);
    1938          21 :         bat *msg = getArgReference_bat(stk, pci, 2);
    1939          21 :         bat *inp = getArgReference_bat(stk, pci, 3);
    1940             : 
    1941          21 :         create_rejects_table(cntxt);
    1942          21 :         if (cntxt->error_row == NULL)
    1943           0 :                 throw(MAL, "sql.rejects", "No reject table available");
    1944          21 :         BBPretain(*row = cntxt->error_row->batCacheid);
    1945          21 :         BBPretain(*fld = cntxt->error_fld->batCacheid);
    1946          21 :         BBPretain(*msg = cntxt->error_msg->batCacheid);
    1947          21 :         BBPretain(*inp = cntxt->error_input->batCacheid);
    1948             :         (void) mb;
    1949          21 :         return MAL_SUCCEED;
    1950             : }
    1951             : 
    1952             : str
    1953          13 : COPYrejects_clear(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
    1954             : {
    1955          13 :         if (cntxt->error_row) {
    1956          13 :                 MT_lock_set(&errorlock);
    1957          13 :                 BATclear(cntxt->error_row, true);
    1958          13 :                 if(cntxt->error_fld) BATclear(cntxt->error_fld, true);
    1959          13 :                 if(cntxt->error_msg) BATclear(cntxt->error_msg, true);
    1960          13 :                 if(cntxt->error_input) BATclear(cntxt->error_input, true);
    1961          13 :                 MT_lock_unset(&errorlock);
    1962             :         }
    1963             :         (void) mb;
    1964             :         (void) stk;
    1965             :         (void) pci;
    1966          13 :         return MAL_SUCCEED;
    1967             : }

Generated by: LCOV version 1.14