Changeset: fa129fba76bd for MonetDB
URL: https://dev.monetdb.org/hg/MonetDB/rev/fa129fba76bd
Added Files:
sql/scripts/28_copy.sql
Modified Files:
sql/backends/monet5/CMakeLists.txt
sql/backends/monet5/copy.c
sql/backends/monet5/rel_bin.c
sql/backends/monet5/rel_copy.c
sql/backends/monet5/rel_copy.h
sql/scripts/CMakeLists.txt
Branch: copyparpipe
Log Message:
Enable/disable parallel copy and configure block size from command line and
through sql
diffs (267 lines):
diff --git a/sql/backends/monet5/CMakeLists.txt
b/sql/backends/monet5/CMakeLists.txt
--- a/sql/backends/monet5/CMakeLists.txt
+++ b/sql/backends/monet5/CMakeLists.txt
@@ -94,6 +94,7 @@ set(include_sql_files
25_debug
26_sysmon
27_rejects
+ 28_copy
39_analytics
40_json
42_md5sum
diff --git a/sql/backends/monet5/copy.c b/sql/backends/monet5/copy.c
--- a/sql/backends/monet5/copy.c
+++ b/sql/backends/monet5/copy.c
@@ -17,6 +17,8 @@
#include "mal_exception.h"
#include "mal_interpreter.h"
+#include "rel_copy.h"
+
#define bailout(f, ...) do { \
msg = createException(SQL, f, __VA_ARGS__); \
@@ -802,7 +804,7 @@ COPYpair_assign(Client cntxt, MalBlkPtr
return MAL_SUCCEED;
}
-static inline str
+static str
COPYpair_assign_bats(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
(void)cntxt;
@@ -824,6 +826,40 @@ COPYpair_assign_bats(Client cntxt, MalBl
return MAL_SUCCEED;
}
+static str
+COPYset_blocksize(int *dummy, int *blocksize)
+{
+ (void)dummy;
+ char buffer[128];
+ sprintf(buffer, "%d", *blocksize);
+ GDKsetenv(COPY_BLOCKSIZE_SETTING, buffer);
+ return MAL_SUCCEED;
+}
+
+static str
+COPYget_blocksize(int *blocksize)
+{
+ int size = GDKgetenv_int(COPY_BLOCKSIZE_SETTING, -1);
+ *blocksize = size > 0 ? size : DEFAULT_COPY_BLOCKSIZE;
+ return MAL_SUCCEED;
+}
+
+static str
+COPYset_parallel(bit *dummy, bit *parallel)
+{
+ (void)dummy;
+ char *value = *parallel ? "true" : "false";
+ GDKsetenv(COPY_PARALLEL_SETTING, value);
+ return MAL_SUCCEED;
+}
+
+static str
+COPYget_parallel(bit *parallel)
+{
+ *parallel = GDKgetenv_istrue(COPY_PARALLEL_SETTING);
+ return MAL_SUCCEED;
+}
+
static mel_func copy_init_funcs[] = {
command("copy", "read", COPYread, true, "Clear the BAT and read 'block_size'
bytes into it from 's'",
args(1, 4,
@@ -845,6 +881,20 @@ static mel_func copy_init_funcs[] = {
batarg("block", bte), batarg("offsets", int), argany("type", 1)
)),
+ command("copy", "set_blocksize", COPYset_blocksize, true, "set the COPY block
size", args(1, 2,
+ arg("blocksize", int)
+ )),
+ command("copy", "get_blocksize", COPYget_blocksize, true, "set the COPY block
size", args(1, 1,
+ arg("", int)
+ )),
+
+ command("copy", "set_parallel", COPYset_parallel, true, "set the COPY block
size", args(1, 2,
+ arg("blocksize", bit)
+ )),
+ command("copy", "get_parallel", COPYget_parallel, true, "set the COPY block
size", args(1, 1,
+ arg("", bit)
+ )),
+
// temporary
pattern("copy", "send", COPYpair_assign, false, "dummy", args(2, 4,
argany("", 1), argany("", 1),
diff --git a/sql/backends/monet5/rel_bin.c b/sql/backends/monet5/rel_bin.c
--- a/sql/backends/monet5/rel_bin.c
+++ b/sql/backends/monet5/rel_bin.c
@@ -4451,19 +4451,15 @@ rel2bin_directappend(backend *be, sql_re
mvc *mvc = be->mvc;
MalBlkPtr mb = be->mb;
- list *args = copyfrom->l;
- // X_27:bat[:oid] := sql.append_from(0x7f727c15e310:ptr, "|":str,
"\n":str, nil:str, "null":str, "/tmp/joeri":str, -1:lng, 0:lng, 0:int, nil:str,
0:int, 1:int); |
- node *fname_node = args->h->next->next->next->next->next;
- sql_exp *fname_exp = fname_node->data;
- atom *fname_atom = fname_exp->l;
- const char *fname = fname_atom->data.val.sval;
- if (strstr(fname, "banana") != NULL)
+
+ if (parallel_copy_enabled())
return rel2bin_copyparpipe(be, rel, refs, copyfrom);
// We're about to emit a custom sql.copy_from invocation.
// Temporarily, until we learn how to do that properly.
// First emit statements for all copyfrom's arguments.
+ list *args = copyfrom->l;
list *l = sa_list(mvc->sa);
for (node *n = args->h; n; n = n->next) {
sql_exp *arg = n->data;
diff --git a/sql/backends/monet5/rel_copy.c b/sql/backends/monet5/rel_copy.c
--- a/sql/backends/monet5/rel_copy.c
+++ b/sql/backends/monet5/rel_copy.c
@@ -14,6 +14,35 @@
#include "opt_prelude.h"
+bool
+parallel_copy_enabled(void)
+{
+ return GDKgetenv_istrue(COPY_PARALLEL_SETTING);
+}
+
+static int
+get_copy_blocksize(void) {
+ int size = GDKgetenv_int(COPY_BLOCKSIZE_SETTING, -1);
+ return size > 0 ? size : DEFAULT_COPY_BLOCKSIZE;
+}
+
+static int
+allocation_size(int blocksize)
+{
+ int alt;
+ int size;
+
+ size = blocksize + blocksize / 8;
+
+ alt = blocksize + 4096;
+ size = size < alt ? alt : size;
+
+ alt = 8192;
+ size = size < alt ? alt : size;
+
+ return size;
+}
+
static int
extract_parameter(backend *be, list *stmts, sql_exp *copyfrom, int argno)
{
@@ -58,13 +87,13 @@ struct loop_vars {
static void
emit_onserver(
MalBlkPtr mb, struct loop_vars *loop_vars,
- int var_fname, int block_size, int margin,
+ int var_fname, int block_size,
int var_line_sep, int var_quote_char, int var_escape)
{
InstrPtr q;
-
int streams_type = ATOMindex("streams");
int bte_bat_type = newBatType(TYPE_bte);
+ int alloc = allocation_size(block_size);
q = newStmt(mb, "streams", "openRead");
q = pushArgument(mb, q, var_fname);
@@ -72,7 +101,7 @@ emit_onserver(
q = newStmt(mb, "bat", "new");
q = pushNil(mb, q, TYPE_bte);
- q = pushLng(mb, q, block_size + margin);
+ q = pushLng(mb, q, alloc);
int var_block_channel = getDestVar(q);
q = newAssignment(mb);
@@ -90,7 +119,7 @@ emit_onserver(
q = newStmt(mb, "bat", "new");
q = pushNil(mb, q, TYPE_bte);
- q = pushLng(mb, q, 300);
+ q = pushLng(mb, q, alloc);
int var_next_block = getDestVar(q);
// START READ BLOCK
@@ -168,8 +197,7 @@ rel2bin_copyparpipe(backend *be, sql_rel
{
(void)rel;
(void)refs;
- const int block_size = 1024 * 1024;
- const int margin = 8 * 1024;
+ const int block_size = get_copy_blocksize();
struct loop_vars loop_vars;
InstrPtr q;
@@ -228,7 +256,7 @@ rel2bin_copyparpipe(backend *be, sql_rel
int var_claim_channel = getDestVar(q);
- emit_onserver(mb, &loop_vars, var_fname, block_size, margin,
var_line_sep, var_quote_char, var_escape);
+ emit_onserver(mb, &loop_vars, var_fname, block_size, var_line_sep,
var_quote_char, var_escape);
int var_claim_token = emit_receive(mb, var_claim_channel, TYPE_bit);
diff --git a/sql/backends/monet5/rel_copy.h b/sql/backends/monet5/rel_copy.h
--- a/sql/backends/monet5/rel_copy.h
+++ b/sql/backends/monet5/rel_copy.h
@@ -13,6 +13,14 @@
#include "sql_statement.h"
#include "mal_backend.h"
+#define COPY_BLOCKSIZE_SETTING "copy_blocksize"
+#define COPY_PARALLEL_SETTING "copy_parallel"
+
+#define DEFAULT_COPY_BLOCKSIZE (1024 * 1024)
+
+
extern stmt *rel2bin_copyparpipe(backend *be, sql_rel *rel, list *refs,
sql_exp *copyfrom);
+extern bool parallel_copy_enabled(void);
+
#endif /*_REL_COPY_H_*/
diff --git a/sql/scripts/28_copy.sql b/sql/scripts/28_copy.sql
new file mode 100644
--- /dev/null
+++ b/sql/scripts/28_copy.sql
@@ -0,0 +1,21 @@
+-- This Source Code Form is subject to the terms of the Mozilla Public
+-- License, v. 2.0. If a copy of the MPL was not distributed with this
+-- file, You can obtain one at http://mozilla.org/MPL/2.0/.
+--
+-- Copyright 1997 - July 2008 CWI, August 2008 - 2022 MonetDB B.V.
+
+
+
+create function sys.copy_blocksize()
+returns int
+external name "copy".get_blocksize;
+
+create procedure sys.copy_blocksize(size int)
+external name "copy".set_blocksize;
+
+create function sys.copy_parallel()
+returns bool
+external name "copy".get_parallel;
+
+create procedure sys.copy_parallel(parallel bool)
+external name "copy".set_parallel;
diff --git a/sql/scripts/CMakeLists.txt b/sql/scripts/CMakeLists.txt
--- a/sql/scripts/CMakeLists.txt
+++ b/sql/scripts/CMakeLists.txt
@@ -22,6 +22,7 @@ install(FILES
25_debug.sql
26_sysmon.sql
27_rejects.sql
+ 28_copy.sql
39_analytics.sql
40_json.sql
42_md5sum.sql
_______________________________________________
checkin-list mailing list -- [email protected]
To unsubscribe send an email to [email protected]
