fix other solutions to use machine type

Author: Tmonster

R-arrow/groupby-R-arrow.R

@@ -17,6 +17,7 @@ cache = TRUE
 on_disk = FALSE
 
 data_name = Sys.getenv("SRC_DATANAME")
+machine_type = Sys.getenv("MACHINE_TYPE")
 src_jn_x = file.path("data", paste(data_name, "csv", sep="."))
 y_data_name = join_to_tbls(data_name)
 src_jn_y = setNames(file.path("data", paste(y_data_name, "csv", sep=".")), names(y_data_name))
@@ -46,15 +47,15 @@ t = system.time({
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 rm(ans)
 t = system.time({
   ans<-collect(inner_join(x, small, by="id1"))
   print(dim(ans))
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 ans <- collect(ans)
 print(head(ans, 3))
 print(tail(ans, 3))
@@ -68,15 +69,15 @@ t = system.time({
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 rm(ans)
 t = system.time({
   ans<-collect(inner_join(x, medium, by="id2"))
   print(dim(ans))
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 ans <- collect(ans)
 print(head(ans, 3))
 print(tail(ans, 3))
@@ -90,15 +91,15 @@ t = system.time({
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 rm(ans)
 t = system.time({
   ans<-collect(left_join(x, medium, by="id2"))
   print(dim(ans))
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 ans <- collect(ans)
 print(head(ans, 3))
 print(tail(ans, 3))
@@ -112,15 +113,15 @@ t = system.time({
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 rm(ans)
 t = system.time({
   ans <- collect(inner_join(x, medium, by="id5"))
   print(dim(ans))
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 ans <- collect(ans)
 print(head(ans, 3))
 print(tail(ans, 3))
@@ -134,15 +135,15 @@ t = system.time({
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=1L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 rm(ans)
 t = system.time({
   ans<-collect(inner_join(x, big, by="id3"))
   print(dim(ans))
 })[["elapsed"]]
 m = memory_usage()
 chkt = system.time(chk <- collect(summarise(ans, sum(v1, na.rm=TRUE), sum(v2, na.rm=TRUE))))[["elapsed"]]
-write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk)
+write.log(run=2L, task=task, data=data_name, in_rows=nrow(x), question=question, out_rows=nrow(ans), out_cols=ncol(ans), solution=solution, version=ver, git=git, fun=fun, time_sec=t, mem_gb=m, cache=cache, chk=make_chk(chk), chk_time_sec=chkt, on_disk=on_disk, machine_type=machine_type)
 ans <- collect(ans)
 print(head(ans, 3))
 print(tail(ans, 3))

R-arrow/join-R-arrow.R

@@ -7,7 +7,7 @@ function getpkgmeta(name::AbstractString)
     Pkg.TOML.parse(read(fname, String))["deps"][name][1]
 end;
 
-function write_log(run, task, data, in_rows, question, out_rows, out_cols, solution, version, git, fun, time_sec, mem_gb, cache, chk, chk_time_sec, on_disk)
+function write_log(run, task, data, in_rows, question, out_rows, out_cols, solution, version, git, fun, time_sec, mem_gb, cache, chk, chk_time_sec, on_disk, machine_type)
   file=try
     ENV["CSV_TIME_FILE"]
   catch
@@ -31,7 +31,7 @@ function write_log(run, task, data, in_rows, question, out_rows, out_cols, solut
   chk_time_sec=round(chk_time_sec, digits=3)
   timestamp=@sprintf("%0.6f", time())
   csv_verbose = false # hardcoded for now, TODO ENV["CSV_VERBOSE"] and print
-  log = DataFrame(nodename=nodename, batch=batch, timestamp=timestamp, task=task, data=data, in_rows=in_rows, question=question, out_rows=out_rows, out_cols=out_cols, solution=solution, version=version, git=git, fun=fun, run=run, time_sec=time_sec, mem_gb=mem_gb, cache=uppercase(string(cache)), chk=chk, chk_time_sec=chk_time_sec, comment=comment, on_disk=uppercase(string(on_disk)))
+  log = DataFrame(nodename=nodename, batch=batch, timestamp=timestamp, task=task, data=data, in_rows=in_rows, question=question, out_rows=out_rows, out_cols=out_cols, solution=solution, version=version, git=git, fun=fun, run=run, time_sec=time_sec, mem_gb=mem_gb, cache=uppercase(string(cache)), chk=chk, chk_time_sec=chk_time_sec, comment=comment, on_disk=uppercase(string(on_disk)), machine_type=machine_type)
   CSV.write(file, log, append=isfile(file))
 end;
 

_helpers/helpers.jl

@@ -5,7 +5,7 @@
 import os
 import platform
 
-def write_log(task, data, in_rows, question, out_rows, out_cols, solution, version, git, fun, run, time_sec, mem_gb, cache, chk, chk_time_sec, on_disk):
+def write_log(task, data, in_rows, question, out_rows, out_cols, solution, version, git, fun, run, time_sec, mem_gb, cache, chk, chk_time_sec, on_disk, machine_type='xlarge'):
    batch = os.getenv('BATCH', "") 
    timestamp = time.time()
    csv_file = os.getenv('CSV_TIME_FILE', "time.csv")
@@ -18,7 +18,7 @@ def write_log(task, data, in_rows, question, out_rows, out_cols, solution, versi
       time_sec = ""
    if math.isnan(mem_gb):
       mem_gb = ""
-   log_row = [nodename, batch, timestamp, task, data, in_rows, question, out_rows, out_cols, solution, version, git, fun, run, time_sec, mem_gb, cache, chk, chk_time_sec, comment, on_disk]
+   log_row = [nodename, batch, timestamp, task, data, in_rows, question, out_rows, out_cols, solution, version, git, fun, run, time_sec, mem_gb, cache, chk, chk_time_sec, comment, on_disk, machine_type]
    log_header = ["nodename","batch","timestamp","task","data","in_rows","question","out_rows","out_cols","solution","version","git","fun","run","time_sec","mem_gb","cache","chk","chk_time_sec","comment","on_disk"]
    if os.path.isfile(csv_file) and not(os.path.getsize(csv_file)):
       os.remove(csv_file)

_helpers/helpers.py

@@ -7,6 +7,7 @@ args = commandArgs(TRUE) # args = c("groupby","G1_1e6_1e2_0_0")
 stopifnot(length(args)==2L)
 task = args[1L]
 data_name = args[2L]
+machine_type = Sys.getenv("MACHINE_TYPE", "large")
 
 library(data.table)
 # sort files according to question and run
@@ -34,7 +35,7 @@ stopifnot(all(d$task==task), all(d$data_name==data_name))
 d[,
   write.log(run=as.integer(run), timestamp=as.numeric(timestamp), task=as.character(task), data=as.character(data_name), in_rows=as.numeric(.in_rows), question=as.character(question),
             out_rows=as.numeric(NA), out_cols=as.integer(NA), solution=as.character(solution), version=as.character(version), git=as.character(NA), fun=as.character(fun), 
-            time_sec=as.numeric(time_sec), mem_gb=as.numeric(NA), cache=as.logical(cache), chk=as.character(NA), chk_time_sec=as.numeric(NA), on_disk=as.logical(on_disk)),
+            time_sec=as.numeric(time_sec), mem_gb=as.numeric(NA), cache=as.logical(cache), chk=as.character(NA), chk_time_sec=as.numeric(NA), on_disk=as.logical(on_disk), machine_type=as.character(machine_type)),
   by = seq_len(nrow(d))] -> nul
 
 cat("# clickhouse-parse-log.R: parsing timings to time.csv finished\n")

clickhouse/clickhouse-parse-log.R

@@ -34,7 +34,7 @@ IS_SORTED=$(clickhouse-client --query "SELECT splitByChar('_','$SRC_DATANAME')[5
 ON_DISK=0
 
 if [ $1 == 'groupby' ]; then
-  ON_DISK=$(clickhouse-client --query "SELECT (splitByChar('_','$SRC_DATANAME')[2])::Float32 >= 1e10::Float32 FORMAT TSV")
+  ON_DISK=$(clickhouse-client --query "SELECT (splitByChar('_','$SRC_DATANAME')[2])::Float32 >= 1e10::Float32 FORMAT TSV") || ($(clickhouse-client --query "SELECT (splitByChar('_','$SRC_DATANAME')[2])::Float32 >= 1e9::Float32 || '$MACHINE_TYPE' == 'small') FORMAT TSV"))
   clickhouse-client --query "DROP TABLE IF EXISTS $SRC_DATANAME"
   if [ $HAS_NULL -eq 1 ]; then
     if [ $IS_SORTED -eq 1 ]; then