Changeset: fbea53aed4a1 for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/fbea53aed4a1
Modified Files:
        monetdb5/ChangeLog.Jul2021
        monetdb5/modules/mal/tablet.c
        tools/merovingian/client/monetdb.1
        tools/merovingian/daemon/forkmserver.c
        tools/merovingian/daemon/merovingian.c
        tools/merovingian/utils/properties.c
Branch: Jul2021
Log Message:

Introduced a --set tablet_threads=N option with monetdb support.


diffs (117 lines):

diff --git a/monetdb5/ChangeLog.Jul2021 b/monetdb5/ChangeLog.Jul2021
--- a/monetdb5/ChangeLog.Jul2021
+++ b/monetdb5/ChangeLog.Jul2021
@@ -1,3 +1,9 @@
 # ChangeLog file for MonetDB5
 # This file is updated with Maddlog
 
+* Tue May 23 2023 Sjoerd Mullender <[email protected]>
+- There is now a new option --set tablet_threads=N to limit the number
+  of threads used for a COPY INTO from CSV file query.  This option can
+  also be set for a specific database using the monetdb command using
+  the ncopyintothreads property.
+
diff --git a/monetdb5/modules/mal/tablet.c b/monetdb5/modules/mal/tablet.c
--- a/monetdb5/modules/mal/tablet.c
+++ b/monetdb5/modules/mal/tablet.c
@@ -1549,10 +1549,18 @@ SQLload_file(Client cntxt, Tablet *as, b
        BUN i, attr;
        READERtask task;
        READERtask ptask[MAXWORKERS];
-       int threads = (maxrow< 0 || maxrow > (1 << 16)) && GDKnr_threads > 1 ? 
(GDKnr_threads < MAXWORKERS ? GDKnr_threads - 1 : MAXWORKERS - 1) : 1;
+       int threads = 1;
        lng tio, t1 = 0;
        char name[MT_NAME_LEN];
 
+       if (maxrow < 0 || maxrow > (LL_CONSTANT(1) << 16)) {
+               threads = GDKgetenv_int("tablet_threads", GDKnr_threads);
+               if (threads > 1)
+                       threads = threads < MAXWORKERS ? threads - 1 : 
MAXWORKERS - 1;
+               else
+                       threads = 1;
+       }
+
 /*     TRC_DEBUG(MAL_SERVER, "Prepare copy work for '%d' threads col '%s' rec 
'%s' quot '%c'\n", threads, csep, rsep, quote);*/
 
        memset(ptask, 0, sizeof(ptask));
diff --git a/tools/merovingian/client/monetdb.1 
b/tools/merovingian/client/monetdb.1
--- a/tools/merovingian/client/monetdb.1
+++ b/tools/merovingian/client/monetdb.1
@@ -419,6 +419,17 @@ cores in the system.  Reducing this numb
 parallelism when executing queries, or none at all if set to
 .BR 1 .
 .TP
+\fBncopyintothreads=\fP\fInumber\fP 
+Defines the maximum number of worker threads the server should use to
+perform COPY INTO from a CSV file.  The actual number of threads used is
+never higher than the number of columns, and is \fB1\fP if the number of
+rows is small.  Normally, this number is equal to the value of the
+\fBnthreads\fP property.  Using this number forces the server to use
+more or less parallelism when executing COPY INTO.  Note, COPY INTO
+threads are created in addition to normal worker threads for each COPY
+INTO query that is being executed and therefore contend for the CPU
+with other queries.
+.TP
 \fBoptpipe=\fP\fIstring\fP
 Each server operates with a given optimizer pipeline.  While the default
 usually is the best setting, for some experimental uses the pipeline can
diff --git a/tools/merovingian/daemon/forkmserver.c 
b/tools/merovingian/daemon/forkmserver.c
--- a/tools/merovingian/daemon/forkmserver.c
+++ b/tools/merovingian/daemon/forkmserver.c
@@ -209,6 +209,7 @@ forkMserver(const char *database, sabdb*
        char usock[512];
        bool mydoproxy;
        char nthreads[32];
+       char tabthreads[32];
        char nclients[32];
        char pipeline[512];
        char memmaxsize[64];
@@ -487,6 +488,15 @@ forkMserver(const char *database, sabdb*
                nthreads[0] = '\0';
        }
 
+       kv = findConfKey(ckv, "ncopyintothreads");
+       if (kv->val == NULL)
+               kv = findConfKey(_mero_db_props, "ncopyintothreads");
+       if (kv->val != NULL) {
+               snprintf(tabthreads, sizeof(tabthreads), "tablet_threads=%s", 
kv->val);
+       } else {
+               tabthreads[0] = '\0';
+       }
+
        kv = findConfKey(ckv, "nclients");
        if (kv->val == NULL)
                kv = findConfKey(_mero_db_props, "nclients");
@@ -639,6 +649,9 @@ forkMserver(const char *database, sabdb*
        if (nthreads[0] != '\0') {
                argv[c++] = set; argv[c++] = nthreads;
        }
+       if (tabthreads[0] != '\0') {
+               argv[c++] = set; argv[c++] = tabthreads;
+       }
        if (nclients[0] != '\0') {
                argv[c++] = set; argv[c++] = nclients;
        }
diff --git a/tools/merovingian/daemon/merovingian.c 
b/tools/merovingian/daemon/merovingian.c
--- a/tools/merovingian/daemon/merovingian.c
+++ b/tools/merovingian/daemon/merovingian.c
@@ -500,6 +500,8 @@ main(int argc, char *argv[])
                        snprintf(cnt, sizeof(cnt), "%d", ncpus);
                        kv = findConfKey(_mero_db_props, "nthreads");
                        kv->val = strdup(cnt);
+                       kv = findConfKey(_mero_db_props, "ncopyintothreads");
+                       kv->val = strdup(cnt);
                }
        }
 
diff --git a/tools/merovingian/utils/properties.c 
b/tools/merovingian/utils/properties.c
--- a/tools/merovingian/utils/properties.c
+++ b/tools/merovingian/utils/properties.c
@@ -28,6 +28,7 @@ static const confkeyval _internal_prop_k
        {"type",        NULL, 0, STR},
        {"shared",      NULL, 0, STR},
        {"nthreads",    NULL, 0, INT},
+       {"ncopyintothreads", NULL, 0, INT},
        {"optpipe",     NULL, 0, STR},
        {"readonly",    NULL, 0, BOOLEAN},
        {"embedr",      NULL, 0, BOOLEAN},
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]

Reply via email to