[C++] Hashjoin + datasets hanging

[asfimport]

I’m getting a hang on the TPC-H query 4 pretty reliably (though it’s not every time). The query is:

l <- input_table("lineitem") %>%
    select(l_orderkey, l_commitdate, l_receiptdate) %>%
    filter(l_commitdate < l_receiptdate) %>%
    select(l_orderkey)

  o <- input_table("orders") %>%
    select(o_orderkey, o_orderdate, o_orderpriority) %>%
    # kludge: filter(o_orderdate >= "1993-07-01", o_orderdate < "1993-07-01" + interval '3' month) %>%
    filter(o_orderdate >= as.Date("1993-07-01"), o_orderdate < as.Date("1993-10-01")) %>%
    select(o_orderkey, o_orderpriority)

  # distinct after join, tested and indeed faster
  lo <- inner_join(l, o, by = c("l_orderkey" = "o_orderkey")) %>%
    distinct() %>%
    select(o_orderpriority)

  aggr <- lo %>%
    group_by(o_orderpriority) %>%
    summarise(order_count = n()) %>%
    arrange(o_orderpriority) %>% 
    collect()

Basically, filtered lineitems, filtered orders, join those together, group_by, summarise, arrange.

This happens pretty reliably when the input_table is a dataset backed by parquet or feather fiels (e.g. input_table returns something like arrow::open_dataset("path/to/{filename}.feather", format = "feather")

One can replicate this by installing an arrowbench branch (voltrondata-labs/arrowbench#37) with, in R: remotes::install_github("ursacomputing/arrowbench@moar-tpch" and then running the following:

library(arrowbench)

results <- run_benchmark(
  tpc_h,
  scale_factor = 1,
  cpu_count = 8,
  query_id = 4,
  lib_path = "remote-apache/arrow@HEAD", # remove this line if you have a recent install of the arrow r package that supports hash joins and want to avoid building a separate copy.
  format = "feather",
  n_iter = 20
)

Note this sometimes will finish, but frequently it will not and be stuck.

Reporter: Jonathan Keane / @jonkeane
Assignee: Michal Nowakiewicz / @michalursa

Original Issue Attachments:

• gdb.2.log
• gdb.log
• sample-while-hung.out.txt
• tpch_repro.cc

PRs and other links:

• GitHub Pull Request #11335
• GitHub Pull Request #11350

_{Note: This issue was originally created as ARROW-14197. Please see the migration documentation for further details.}

[asfimport]

Jonathan Keane / @jonkeane:
I've attached a sample of the R process while it's hung from my machine, in case that's helpful

[asfimport]

Weston Pace / @westonpace:
I should've just looked at your sample as it may have saved me some time. But, I do have a bit more info. The query itself resolves to the following plan (note, we need to do some work on our ToString as it is difficult to make out which node connects to which):

ExecPlan with 14 nodes:
SourceNode{"", outputs=[""]}
FilterNode{"", inputs=[target: ""], outputs=[""], filter=(l_commitdate < l_receiptdate)}
SourceNode{"", outputs=[""]}
FilterNode{"", inputs=[target: ""], outputs=[""], filter=((o_orderdate >= 1993-07-01) and (o_orderdate < 1993-10-01))}
ProjectNode{"", inputs=[target: ""], outputs=[""], projection=[o_orderkey, o_orderpriority]}
HashJoinNode{"", inputs=[left: "", right: ""], outputs=[""]}
ProjectNode{"", inputs=[target: ""], outputs=[""], projection=[l_orderkey, o_orderpriority]}
GroupByNode{"", inputs=[groupby: ""], outputs=[""], keys=["l_orderkey", "o_orderpriority"], aggregates=[
]}
ProjectNode{"", inputs=[target: ""], outputs=[""], projection=[l_orderkey, o_orderpriority]}
ProjectNode{"", inputs=[target: ""], outputs=[""], projection=["order_count": 1, o_orderpriority]}
GroupByNode{"", inputs=[groupby: ""], outputs=[""], keys=["o_orderpriority"], aggregates=[
	hash_sum(order_count, {skip_nulls=true, min_count=1}),
]}
ProjectNode{"", inputs=[target: ""], outputs=[""], projection=[o_orderpriority, order_count]}
ProjectNode{"", inputs=[target: ""], outputs=[""], projection=[o_orderpriority, order_count]}
OrderBySinkNode{"", inputs=[collected: ""], by={sort_keys=[o_orderpriority ASC], null_placement=AtEnd}}

The hang is a never-ending busy-loop (i.e. 100% CPU usage) in arrow::compute::SwissTable::find (this function never returns).

I'll attach a GDB stack trace of a sample hang. gdb.log

[asfimport]

Weston Pace / @westonpace:
Here is an annotated version of the exec plan.

SourceNode{"0:", outputs=["1:"], FileSystemDataset:
/home/pace/dev/arrow/r/data/temp/lineitem_1.uncompressed.feather}
FilterNode{"1:", inputs=[target: "0:"], outputs=["5:"], filter=(l_commitdate < l_receiptdate)}
SourceNode{"2:", outputs=["3:"], FileSystemDataset:
/home/pace/dev/arrow/r/data/temp/orders_1.uncompressed.feather}
FilterNode{"3:", inputs=[target: "2:"], outputs=["4:"], filter=((o_orderdate >= 1993-07-01) and (o_orderdate < 1993-10-01))}
ProjectNode{"4:", inputs=[target: "3:"], outputs=["5:"], projection=[o_orderkey, o_orderpriority]}
HashJoinNode{"5:", inputs=[left: "1:", right: "4:"], outputs=["6:"]}
ProjectNode{"6:", inputs=[target: "5:"], outputs=["7:"], projection=[l_orderkey, o_orderpriority]}
GroupByNode{"7:", inputs=[groupby: "6:"], outputs=["8:"], keys=["l_orderkey", "o_orderpriority"], aggregates=[
]}
ProjectNode{"8:", inputs=[target: "7:"], outputs=["9:"], projection=[l_orderkey, o_orderpriority]}
ProjectNode{"9:", inputs=[target: "8:"], outputs=["10:"], projection=["order_count": 1, o_orderpriority]}
GroupByNode{"10:", inputs=[groupby: "9:"], outputs=["11:"], keys=["o_orderpriority"], aggregates=[
	hash_sum(order_count, {skip_nulls=true, min_count=1}),
]}
ProjectNode{"11:", inputs=[target: "10:"], outputs=["12:"], projection=[o_orderpriority, order_count]}
ProjectNode{"12:", inputs=[target: "11:"], outputs=["13:"], projection=[o_orderpriority, order_count]}
OrderBySinkNode{"13:", inputs=[collected: "12:"], by={sort_keys=[o_orderpriority ASC], null_placement=AtEnd}}

[asfimport]

David Li / @lidavidm:
Tangential, but the exec plan string repr being hard to read is unfortunate. It would help if R assigned names to the nodes, or perhaps we could have the ExecPlan auto-assign names if the client bindings don't give any. (Also, maybe GraphViz output would be nice after all…)

I filed ARROW-14233.

[asfimport]

Weston Pace / @westonpace:
I'm attaching a C++ reproduction. It will require some input data. I can host a copy of the input data for a while on Google Drive (only ~200MB): https://drive.google.com/file/d/1kg7Xts6w8Ktzau4Go7euOVSQKdh2qJuG/view?usp=sharing

[asfimport]

Neal Richardson / @nealrichardson:
Issue resolved by pull request 11335
#11335