Add DataFusion solution (#18)

Add Datafusion solution

.gitignore

@@ -31,3 +31,5 @@ run.out
 clickhouse/etc_sudoers.bak
 workdir/
 timeout-exit-codes.out
+*/target
+*.lock

_benchplot/benchplot-dict.R

@@ -43,7 +43,8 @@ solution.dict = {list(
  "polars" = list(name=c(short="polars", long="Polars"), color=c(strong="deepskyblue4", light="deepskyblue3")),
  "arrow" = list(name=c(short="arrow", long="Arrow"), color=c(strong="aquamarine3", light="aquamarine1")),
  "duckdb" = list(name=c(short="duckdb", long="DuckDB"), color=c(strong="#ddcd07", light="#fff100")),
- "duckdb-latest" = list(name=c(short="duckdb-latest", long="duckdb-latest"), color=c(strong="#ddcd07", light="#fff100"))
+ "duckdb-latest" = list(name=c(short="duckdb-latest", long="duckdb-latest"), color=c(strong="#ddcd07", light="#fff100")),
+ "datafusion" = list(name=c(short="datafusion", long="Datafusion"), color=c(strong="deepskyblue4", light="deepskyblue3"))
 )}
 #barplot(rep(c(0L,1L,1L), length(solution.dict)),
 # col=rev(c(rbind(sapply(solution.dict, `[[`, "color"), "black"))),
@@ -220,6 +221,18 @@ groupby.syntax.dict = {list(
  "largest two v3 by id6" = "SELECT id6, unnest(list_sort(list(v3), 'desc')[1:2]) AS largest2_v3 FROM (SELECT id6, v3 FROM x WHERE v3 IS NOT NULL) AS subq GROUP BY id6;",
  "regression v1 v2 by id2 id4" = "SELECT id2, id4, pow(corr(v1, v2), 2) AS r2 FROM tbl GROUP BY id2, id4",
  "sum v3 count by id1:id6" = "SELECT id1, id2, id3, id4, id5, id6, sum(v3) AS v3, count(*) AS count FROM tbl GROUP BY id1, id2, id3, id4, id5, id6"
+ )},
+ "datafusion" = {c(
+ "sum v1 by id1" = "SELECT id1, SUM(v1) AS v1 FROM x GROUP BY id1",
+ "sum v1 by id1:id2" = "SELECT id1, id2, SUM(v1) AS v1 FROM x GROUP BY id1, id2",
+ "sum v1 mean v3 by id3" = "SELECT id3, SUM(v1) AS v1, AVG(v3) AS v3 FROM x GROUP BY id3",
+ "mean v1:v3 by id4" = "SELECT id4, AVG(v1) AS v1, AVG(v2) AS v2, AVG(v3) AS v3 FROM x GROUP BY id4",
+ "sum v1:v3 by id6" = "SELECT id6, SUM(v1) AS v1, SUM(v2) AS v2, SUM(v3) AS v3 FROM x GROUP BY id6",
+ "median v3 sd v3 by id4 id5" = "SELECT id4, id5, MEDIAN(v3) AS median_v3, STDDEV(v3) AS sd_v3 FROM tbl GROUP BY id4, id5",
+ "max v1 - min v2 by id3" = "SELECT id3, MAX(v1) - MIN(v2) AS range_v1_v2 FROM x GROUP BY id3",
+ "largest two v3 by id6" = "SELECT id6, v3 from (SELECT id6, v3, row_number() OVER (PARTITION BY id6 ORDER BY v3 DESC) AS row FROM x) t WHERE row <= 2",
+ "regression v1 v2 by id2 id4" = "SELECT id2, id4, POW(CORR(v1, v2), 2) AS r2 FROM tbl GROUP BY id2, id4",
+ "sum v3 count by id1:id6" = "SELECT id1, id2, id3, id4, id5, id6, SUM(v3) as v3, COUNT(*) AS cnt FROM x GROUP BY id1, id2, id3, id4, id5, id6"
  )}
 )}
 groupby.query.exceptions = {list(
@@ -235,7 +248,8 @@ groupby.query.exceptions = {list(
  "polars" = list(),
  "arrow" = list("Expression row_number() <= 2L not supported in Arrow; pulling data into R" = "max v1 - min v2 by id3", "Expression cor(v1, v2, ... is not supported in arrow; pulling data into R" = "regression v1 v2 by id2 id4"),
  "duckdb" = list(),
- "duckdb-latest" = list()
+ "duckdb-latest" = list(),
+ "datafusion" = list(),
 )}
 groupby.data.exceptions = {list( # exceptions as of run 1575727624
  "data.table" = {list(
@@ -292,7 +306,8 @@ groupby.data.exceptions = {list(
  "duckdb-latest" = {list(
  # "out of memory" = c("G1_1e9_1e2_0_0","G1_1e9_1e1_0_0","G1_1e9_2e0_0_0","G1_1e9_1e2_0_1","G1_1e9_1e2_5_0"),
  # "incorrect: duckdb#1737" = c("G1_1e7_1e2_5_0","G1_1e8_1e2_5_0")
- )}
+ )},
+ "datafusion" = {list()}
 )}
 groupby.exceptions = task.exceptions(groupby.query.exceptions, groupby.data.exceptions)
 
@@ -395,6 +410,13 @@ join.syntax.dict = {list(
  "medium outer on int" = "SELECT x.*, medium.id1 AS medium_id1, medium.id4 AS medium_id4, medium.id5 AS medium_id5, v2 FROM x LEFT JOIN medium USING (id2)",
  "medium inner on factor" = "SELECT x.*, medium.id1 AS medium_id1, medium.id2 AS medium_id2, medium.id4 AS medium_id4, v2 FROM x JOIN medium USING (id5)",
  "big inner on int" = "SELECT x.*, big.id1 AS big_id1, big.id2 AS big_id2, big.id4 AS big_id4, big.id5 AS big_id5, big.id6 AS big_id6, v2 FROM x JOIN big USING (id3)"
+ )},
+ "datafusion" = {c(
+ "small inner on int" = "SELECT x.id1, x.id2, x.id3, x.id4 as xid4, small.id4 as smallid4, x.id5, x.id6, x.v1, small.v2 FROM x INNER JOIN small ON x.id1 = small.id1",
+ "medium inner on int" = "SELECT x.id1 as xid1, medium.id1 as mediumid1, x.id2, x.id3, x.id4 as xid4, medium.id4 as mediumid4, x.id5 as xid5, medium.id5 as mediumid5, x.id6, x.v1, medium.v2 FROM x INNER JOIN medium ON x.id2 = medium.id2",
+ "medium outer on int" = "SELECT x.id1 as xid1, medium.id1 as mediumid1, x.id2, x.id3, x.id4 as xid4, medium.id4 as mediumid4, x.id5 as xid5, medium.id5 as mediumid5, x.id6, x.v1, medium.v2 FROM x LEFT JOIN medium ON x.id2 = medium.id2",
+ "medium inner on factor" = "SELECT x.id1 as xid1, medium.id1 as mediumid1, x.id2, x.id3, x.id4 as xid4, medium.id4 as mediumid4, x.id5 as xid5, medium.id5 as mediumid5, x.id6, x.v1, medium.v2 FROM x LEFT JOIN medium ON x.id5 = medium.id5",
+ "big inner on int" = "SELECT x.id1 as xid1, large.id1 as largeid1, x.id2 as xid2, large.id2 as largeid2, x.id3, x.id4 as xid4, large.id4 as largeid4, x.id5 as xid5, large.id5 as largeid5, x.id6 as xid6, large.id6 as largeid6, x.v1, large.v2 FROM x LEFT JOIN large ON x.id3 = large.id3"
  )}
 )}
 join.query.exceptions = {list(
@@ -410,7 +432,8 @@ join.query.exceptions = {list(
  "polars" = list(),
  "arrow" = list(),
  "duckdb" = list(),
- "duckdb-latest" = list()
+ "duckdb-latest" = list(),
+ "datafusion" = list()
 )}
 join.data.exceptions = {list( # exceptions as of run 1575727624
  "data.table" = {list(
@@ -445,7 +468,7 @@ join.data.exceptions = {list(
  "J1_1e9_NA_5_0","J1_1e9_NA_0_1") # q1 r1
  )},
  "polars" = {list(
- "out of memory" = c("J1_1e9_NA_0_0","J1_1e9_NA_5_0","J1_1e9_NA_0_1")
+ "out of memory" = c("J1_1e9_NA_0_0","J1_1e9_NA_5_0","J1_1e9_NA_0_1"),
  )},
  "arrow" = {list(
  "out of memory" = c("J1_1e9_NA_0_0","J1_1e9_NA_5_0","J1_1e9_NA_0_1", "J1_1e8_NA_0_0", "J1_1e8_NA_5_0", "J1_1e8_NA_0_1" )#,
@@ -460,7 +483,8 @@ join.data.exceptions = {list(
  # "internal error: duckdb#1739" = c("J1_1e7_NA_0_0","J1_1e7_NA_5_0","J1_1e7_NA_0_1","J1_1e8_NA_0_0","J1_1e8_NA_5_0","J1_1e8_NA_0_1"),
  "out of memory" = c("J1_1e9_NA_0_0","J1_1e9_NA_5_0","J1_1e9_NA_0_1")#,
  #"incorrect: duckdb#1737" = c("J1_1e7_NA_5_0","J1_1e8_NA_5_0")
- )}
+ )},
+ "datafusion" = {list()}
 )}
 join.exceptions = task.exceptions(join.query.exceptions, join.data.exceptions)
 

_control/solutions.csv

@@ -28,3 +28,5 @@ duckdb,groupby
 duckdb,join
 duckdb-latest,groupby
 duckdb-latest,join
+datafusion,groupby
+datafusion,join

_launcher/launcher.R

@@ -15,9 +15,9 @@ file.ext = function(x) {
  ans = switch(
  x,
  "data.table"=, "dplyr"=, "h2o"=, "arrow"=, "duckdb"="R", "duckdb-latest"="R",
- "pandas"=, "spark"=, "pydatatable"=, "modin"=, "dask"=, "polars"="py",
+ "pandas"=, "spark"=, "pydatatable"=, "modin"=, "dask"=, "datafusion"=, "polars"="py",
  "clickhouse"="sql",
- "juliadf"="jl", "juliads"="jl"
+ "juliadf"="jl", "juliads"="jl",
  )
  if (is.null(ans)) stop(sprintf("solution %s does not have file extension defined in file.ext helper function", x))
  ans

_launcher/solution.R

@@ -111,7 +111,7 @@ file.ext = function(x) {
  ans = switch(
  x,
  "data.table"=, "dplyr"=, "h2o"=, "arrow"=, "duckdb"="R", "duckdb-latest"="R",
- "pandas"="py", "spark"=, "pydatatable"=, "modin"=, "dask"=, "polars"="py",
+ "pandas"="py", "spark"=, "pydatatable"=, "modin"=, "dask"=, "datafusion"=, "polars"="py",
  "clickhouse"="sql",
  "juliadf"="jl", "juliads"="jl"
  )

_report/history.Rmd

@@ -728,3 +728,45 @@ Report was generated on: `r format(Sys.time(), usetz=TRUE)`.
 ```{r status_set_success}
 cat("history\n", file=get_report_status_file(), append=TRUE)
 ```
+
+### datafusion {.tabset .tabset-fade .tabset-pills}
+
+#### groupby {.tabset .tabset-fade .tabset-pills}
+
+##### 0.5 GB
+
+```{r datafusion.groupby.1e7}
+plot(d, "datafusion", 1e7, "groupby")
+```
+
+##### 5 GB
+
+```{r datafusion.groupby.1e8}
+plot(d, "datafusion", 1e8, "groupby")
+```
+
+##### 50 GB {.active}
+
+```{r datafusion.groupby.1e9}
+plot(d, "datafusion", 1e9, "groupby")
+```
+
+#### join {.tabset .tabset-fade .tabset-pills}
+
+##### 0.5 GB
+
+```{r datafusion.join.1e7}
+plot(d, "datafusion", 1e7, "join")
+```
+
+##### 5 GB {.active}
+
+```{r datafusion.join.1e8}
+plot(d, "datafusion", 1e8, "join")
+```
+
+##### 50 GB
+
+```{r datafusion.join.1e9}
+plot(d, "datafusion", 1e9, "join")
+```

_report/report.R

@@ -6,7 +6,7 @@ get_report_status_file = function(path=getwd()) {
  file.path(path, "report-done")
 }
 get_report_solutions = function() {
- c("data.table", "dplyr", "pandas", "pydatatable", "spark", "dask", "juliadf", "juliads", "clickhouse", "cudf", "polars","arrow","duckdb", "duckdb-latest")
+ c("data.table", "dplyr", "pandas", "pydatatable", "spark", "dask", "juliadf", "juliads", "clickhouse", "cudf", "polars","arrow","duckdb", "duckdb-latest", "datafusion")
 }
 get_data_levels = function() {
  ## groupby