Fixes to benchmark script

.github/workflows/RunBenchmark.yml

@@ -100,12 +100,14 @@ jobs:
           fi
 
       - name: name new branch
+        if: always()
         shell: bash
         run: |
           echo "new_branch_name=results-`date +%Y-%m-%d-%Hh%Mm`" >> $GITHUB_ENV
           echo ${{ env.new_branch_name }}
 
       - name: Commit updates
+        if: always()
         shell: bash
         working-directory: /var/lib/mount/db-benchmark-metal
         run: |

_launcher/launcher.R

@@ -203,7 +203,21 @@ launch = function(dt, mockup, out_dir="out") {
           if (file.exists(ret_file)) file.remove(ret_file)
         }
         cmd = sprintf("%s > %s 2> %s", solution.cmd(s, t, d), out_file, err_file) # ./_launcher/solution.R ... > out 2> err
-        shcmd = sprintf("/bin/bash -c \"%s%s\"", venv, cmd) # this is needed to source python venv
+        inner = sprintf("/bin/bash -c \"%s%s\"", venv, cmd) # this is needed to source python venv
+        # Run each solution in its own transient systemd scope so the OOM killer
+        # targets only that scope and cannot propagate to the R launcher process.
+        # MemoryMax is set to 90% of total physical RAM.
+        mem_max = tryCatch({
+          meminfo = readLines("/proc/meminfo")
+          total_kb = as.integer(sub(".*:\\s*(\\d+)\\s*kB", "\\1", meminfo[grepl("^MemTotal", meminfo)]))
+          as.integer(total_kb * 1024 * 0.9)
+        }, error = function(e) NA_integer_)
+        cat(sprintf("systemd-run MemoryMax: %s\n", if (is.na(mem_max)) "NA (/proc/meminfo unavailable, scope wrapping disabled)" else sprintf("%d bytes (%.1f GB)", mem_max, mem_max/1024^3)))
+        if (!is.na(mem_max) && nzchar(Sys.which("systemd-run"))) {
+          shcmd = sprintf("systemd-run --scope -p MemoryMax=%d %s", mem_max, inner)
+        } else {
+          shcmd = inner
+        }
         if (mockup) {
           cat(cmd)
           cat(shcmd)

clickhouse/exec.sh

@@ -52,7 +52,9 @@ if [ $1 == 'groupby' ]; then
   fi
   clickhouse-client --user db_benchmark --query "INSERT INTO $SRC_DATANAME FROM INFILE 'data/${SRC_DATANAME}.csv'"
   # confirm all data loaded
-  echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $SRC_DATANAME'\n$(echo $SRC_DATANAME | cut -d'_' -f2)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
+  if [[ $TEST_RUN != "true" ]]; then
+    echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $SRC_DATANAME'\n$(echo $SRC_DATANAME | cut -d'_' -f2)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
+  fi
   export THREADS
 elif [ $1 == 'join' ]; then
   RHS=$(join_to_tbls $SRC_DATANAME)
@@ -106,10 +108,12 @@ elif [ $1 == 'join' ]; then
   tail -n+2 data/$RHS3.csv | clickhouse-client --user db_benchmark --query "INSERT INTO $RHS3 SELECT * FROM input('id1 Nullable(Int32), id2 Nullable(Int32), id3 Nullable(Int32), id4 Nullable(String), id5 Nullable(String), id6 Nullable(String), v2 Nullable(Float64)') FORMAT CSV"
 
   # validate
-  echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $SRC_DATANAME'\n$(echo $SRC_DATANAME | cut -d'_' -f2)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
-  echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS1'\n$(echo $RHS1 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
-  echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS2'\n$(echo $RHS2 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
-  echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS3'\n$(echo $RHS3 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
+  if [[ $TEST_RUN != "true" ]]; then
+    echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $SRC_DATANAME'\n$(echo $SRC_DATANAME | cut -d'_' -f2)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
+    echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS1'\n$(echo $RHS1 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
+    echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS2'\n$(echo $RHS2 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
+    echo -e "clickhouse-client --user db_benchmark --query 'SELECT count(*) FROM $RHS3'\n$(echo $RHS3 | cut -d'_' -f3)" | Rscript -e 'stdin=readLines(file("stdin")); if ((loaded<-as.numeric(system(stdin[1L], intern=TRUE)))!=as.numeric(stdin[2L])) stop("incomplete data load, expected: ", stdin[2L],", loaded: ", loaded)'
+  fi
 
   export RHS1
   export RHS2

dask/groupby_dask.py

@@ -63,7 +63,6 @@ def load_dataset(
     x = dd.read_csv(
         src_grp,
         dtype={"id1":"category","id2":"category","id3":"category","id4":"Int32","id5":"Int32","id6":"Int32","v1":"Int32","v2":"Int32","v3":"float64"},
-        engine="pyarrow",
         **kwargs
     )
     x = x.persist()
@@ -139,7 +138,7 @@ class QuerySix(Query):
 
     @staticmethod
     def query(x: dd.DataFrame) -> dd.DataFrame:
-        ans = x.groupby(['id4','id5'], dropna=False, observed=True).agg({'v3': ['median','std']}, shuffle='p2p').compute()
+        ans = x.groupby(['id4','id5'], dropna=False, observed=True).agg({'v3': ['median','std']}, shuffle='tasks').compute()
         ans.reset_index(inplace=True) # #68
         return ans
 
@@ -166,7 +165,7 @@ class QueryEight(Query):
 
     @staticmethod
     def query(x: dd.DataFrame) -> dd.DataFrame:
-        ans = x[~x['v3'].isna()][['id6','v3']].groupby('id6', dropna=False, observed=True).apply(lambda x: x.nlargest(2, columns='v3'), meta={'id6':'Int64', 'v3':'float64'})[['v3']].compute()
+        ans = x[~x['v3'].isna()][['id6','v3']].groupby('id6', dropna=False, observed=True).apply(lambda x: x.nlargest(2, columns='v3'), meta={'v3':'float64'}, include_groups=False)[['v3']].compute()
         ans.reset_index(level='id6', inplace=True)
         ans.reset_index(drop=True, inplace=True) # drop because nlargest creates some extra new index field
         return ans
@@ -180,7 +179,7 @@ class QueryNine(Query):
 
     @staticmethod
     def query(x: dd.DataFrame) -> dd.DataFrame:
-        ans = x[['id2','id4','v1','v2']].groupby(['id2','id4'], dropna=False, observed=True).apply(lambda x: pd.Series({'r2': x.corr()['v1']['v2']**2}), meta={'r2':'float64'}).compute()
+        ans = x[['id2','id4','v1','v2']].groupby(['id2','id4'], dropna=False, observed=True).apply(lambda x: pd.Series({'r2': x.corr()['v1']['v2']**2}), meta={'r2':'float64'}, include_groups=False).compute()
         ans.reset_index(inplace=True)
         return ans
 

dask/join_dask.py

@@ -153,14 +153,14 @@ def load_datasets(
     logging.info("Loading dataset: %s" % data_name)
 
     logging.info("Reading source: %s" % src_jn_x)
-    x = dd.read_csv(src_jn_x, engine="pyarrow", **kwargs).persist()
+    x = dd.read_csv(src_jn_x, **kwargs).persist()
 
     logging.info("Reading source: %s" % src_jn_y[0])
-    small = dd.read_csv(src_jn_y[0], engine="pyarrow", **kwargs).persist()
+    small = dd.read_csv(src_jn_y[0], **kwargs).persist()
     logging.info("Reading source: %s" % src_jn_y[1])
-    medium = dd.read_csv(src_jn_y[1], engine="pyarrow", **kwargs).persist()
+    medium = dd.read_csv(src_jn_y[1], **kwargs).persist()
     logging.info("Reading source: %s" % src_jn_y[2])
-    big = dd.read_csv(src_jn_y[2], engine="pyarrow", **kwargs).persist()
+    big = dd.read_csv(src_jn_y[2], **kwargs).persist()
 
     return [
         x,

dask/ver-dask.sh

@@ -2,4 +2,4 @@
 set -e
 
 source ./dask/py-dask/bin/activate
-python3 -c 'import dask as dk; open("dask/VERSION","w").write(dk.__version__); open("dask/REVISION","w").write(dk.__git_revision__);' > /dev/null
+python3 -c 'import dask as dk; open("dask/VERSION","w").write(dk.__version__); open("dask/REVISION","w").write(dk.__git_revision__ or "");' > /dev/null

path.env

@@ -1,4 +1,8 @@
-export JULIA_HOME=/opt/julia-1.9.2
+if [ -d "$HOME/.juliaup/bin" ]; then
+  export JULIA_HOME=$HOME/.juliaup
+else
+  export JULIA_HOME=/opt/julia-1.9.2
+fi
 export JAVA_HOME=/usr/lib/jvm/java-8-openjdk-amd64
 export PATH=$PATH:$JULIA_HOME/bin
 export MOUNT_POINT=/var/lib/mount

run.sh

@@ -98,7 +98,13 @@ if [[ "$RUN_SOLUTIONS" =~ "haskell" ]]; then ./haskell/ver-haskell.sh; fi;
 # run
 if [[ -f ./stop ]]; then echo "# Benchmark run $BATCH has been interrupted after $(($(date +%s)-$BATCH))s due to 'stop' file" && rm -f ./stop && rm -f ./run.lock && exit; fi;
 echo "# Running benchmark scripts launcher"
+set +e
 Rscript ./_launcher/launch.R
+LAUNCH_EXIT=$?
+set -e
+if [ $LAUNCH_EXIT -ne 0 ]; then
+  echo "# WARNING: launcher exited with code $LAUNCH_EXIT (possible OOM or crash), partial results may have been recorded"
+fi
 if [[ -f ./stop ]]; then echo "# Benchmark run $BATCH has been interrupted after $(($(date +%s)-$BATCH))s due to 'stop' file" && rm -f ./stop && rm -f ./run.lock && exit; fi;
 
 # publish report for all tasks

spark/groupby-spark.py

@@ -25,10 +25,10 @@
 print("loading dataset %s" % data_name, flush=True)
 
 mem_usage = "240g"
-if "TEST_RUN" in os.environ:
-     mem_usage = "2g"
 if machine_type == 'c6id.4xlarge':
      mem_usage = "30g"
+if "TEST_RUN" in os.environ:
+     mem_usage = "2g"
 
 
 from pyspark.conf import SparkConf

spark/join-spark.py

@@ -27,10 +27,10 @@
     raise Exception("Something went wrong in preparing files used for join")
 
 mem_usage = "220g"
-if "TEST_RUN" in os.environ:
-     mem_usage = "2g"
 if machine_type == 'c6id.4xlarge':
      mem_usage = "30g"
+if "TEST_RUN" in os.environ:
+     mem_usage = "2g"
 
 from pyspark.conf import SparkConf
 spark = SparkSession.builder \