GH-35838: [C++] Add backpressure test for asof join node #35874

rtpsw · 2023-06-01T16:41:12Z

What changes are included in this PR?

Passing the correct nodes to the backpressure controller, along with better parameter naming/doc. Also added reusable gate-classes (Gate, GatedNodeOptions and GatedNode) that enable holding all input batches until a gate is released, in order to support more robust backpressure testing in this PR.

Are these changes tested?

Yes.

Are there any user-facing changes?

No.

This PR contains a "Critical Fix".

Closes: [C++] Backpressure broken in asof join node #35838

icexelloss · 2023-06-01T17:59:59Z

@westonpace Are you OK with me merging this to unblock internal issue? @rtpsw Is working on adding tests now but it might take a while.

icexelloss · 2023-06-01T18:15:29Z

cpp/src/arrow/acero/asof_join_node.cc

@@ -668,13 +668,13 @@ class InputState {

  static Result<std::unique_ptr<InputState>> Make(
      size_t index, TolType tolerance, bool must_hash, bool may_rehash,
-      KeyHasher* key_hasher, AsofJoinNode* node, ExecNode* output,
+      KeyHasher* key_hasher, AsofJoinNode* node, ExecNode* input,


This still looking confusing:

BackpressureController takes ExecNode* node, ExecNode* output and this one nows takes AsofJoinNode* node, ExecNode* input which is inconsistent

Can we make this consistent between the two?

It is also confusing that

return std::make_unique<InputState>(index, tolerance, must_hash, may_rehash, key_hasher, node, std::move(handler), schema, time_col_index, key_col_index);

On line 681 passes the asof join node to the input state instead of the input node, why is it?

@rtpsw I think it is easier to revert this line as to unblock the internal first:
https://github.com/apache/arrow/pull/34392/files#diff-5493b6ae7ea2a4d5cfb581034c076e9c4be7608382168de6d1301ef67b6c01eeR1410

Then work on cleaning up changes introduced in GH-36391. The code is quite confusing now.

This still looking confusing:

BackpressureController takes ExecNode* node, ExecNode* output and this one nows takes AsofJoinNode* node, ExecNode* input which is inconsistent

Can we make this consistent between the two?

This is actually intended to make things clearer. The two places have a different perspective of what is an input and what is an output (which likely caused confusion in the first place). The ExecNode passed to Make is an input of the as-of-join node while PauseProducing (and similarly ResumeProducing) sees the as-of-join node as an output of the ExecNode.

I think it is easier to revert this line

This reversion alone doesn't compile because inputs[i] is not of type AsofJoinNode*.

It's probably easier for me to do it instead of back and forth, opened:

#35878

I am fine with figure out the cleaner way to do this in follow up to GH-36391. But for now I think it's easier to just revert the change.

The two places have a different perspective of what is an input and what is an output (which likely caused confusion in the first place). The ExecNode passed to Make is an input of the as-of-join node while PauseProducing (and similarly ResumeProducing) sees the as-of-join node as an output of the ExecNode.

We should probably fix the variable naming in the follow PR to GH-36391 how to call these things then. But for now let's just revert to what was before

Maybe use more specific names like asof_input and asof_node?

pitrou · 2023-06-01T18:50:59Z

cpp/src/arrow/acero/asof_join_node.cc

      std::atomic<int32_t>& backpressure_counter,
      const std::shared_ptr<arrow::Schema>& schema, const col_index_t time_col_index,
      const std::vector<col_index_t>& key_col_index) {
    constexpr size_t low_threshold = 4, high_threshold = 8;
    std::unique_ptr<BackpressureControl> backpressure_control =
-        std::make_unique<BackpressureController>(node, output, backpressure_counter);
+        std::make_unique<BackpressureController>(input, node, backpressure_counter);


Since there's been some confusion, can you make parameter names explicit?

Suggested change

std::make_unique<BackpressureController>(input, node, backpressure_counter);

std::make_unique<BackpressureController>(/*xxx=*/ input, /*yyy=*/ node, backpressure_counter);

rtpsw · 2023-06-01T20:04:36Z

@icexelloss, I added test cases. This PR can now be reviewed as a (non-quick) resolution.

icexelloss · 2023-06-01T21:29:51Z

cpp/src/arrow/acero/asof_join_node_test.cc

@@ -1406,6 +1458,9 @@ void TestBackpressure(BatchesMaker maker, int num_batches, int batch_size,
    total_length += batch->num_rows();
  }
  ASSERT_EQ(static_cast<int64_t>(num_batches * batch_size), total_length);
+
+  ASSERT_GT(pause_count, 0);


Can we validate pause resume counter for all sources? i.e the slow table should not have been paused, but the fast tables should?

Can we validate pause resume counter for all sources?

I added validation ...

i.e the slow table should not have been paused, but the fast tables should?

... but this is not the actual behavior - instead only one fast source gets paused and resumed - so this is what the recent commit validates. @westonpace, is this behavior expected?

We may have crossed - the previous commit already has one pause and one resume counter for each source. The issue is that sometimes (due to non-deterministic timing of operations) no pause/resume is requested on one or both of the fast sources. Because of this, in the recent commit I am attempting this test logic: there is a pause request on a fast source if and only if there is a resume one too.

westonpace

I do have a little bit of concern that this test may become flaky on CI machines which have notoriously weird timing. However, we can give it a few runs and see how it goes. If it does become a problem I'd suggest we invent some kind of "gated source" so we can do something like (very rough pseudocode)...

GatedSourceNode gated_left = GatedSourceNode(l_batches);
GatedSourceNode gated_right = GatedSourceNode(r_batches);
gated_right.Ungate();

// Start plan

WaitFor([&] { right_bp_options->pause_count > 0; });
ASSERT_EQ(right_bp_options->resume_count, 0);
ASSERT_EQ(left_bp_options->pause_count, 0);
ASSERT_EQ(left_bp_options->resume_count, 0);

gated_left->Ungate();
AssertPlanFinishes();

I'm off today and Monday but I can sketch up a possible GatedSourceNode implementation when I get back if needed.

rtpsw · 2023-06-02T12:52:53Z

I do have a little bit of concern that this test may become flaky on CI machines

Indeed it is flaky - this and this and this CI jobs have failed on the new check. I'll make a quick attempt to fix this.

I agree relying on timing is not good, though that can be said about the pre-PR backpressure test code too, and this PR won't be worse in this respect.

icexelloss · 2023-06-02T13:19:45Z

Thinking about loud here:

What we want to test is that if the through put of asof join node is slower than the source, then we would pause the source. Two potential ways that I think we can reliably do this:
(1) Add some sort of "debug options" to manipulate the behavior of asof join to make it run slower. (i.e. Sleep a few seconds before actually starting the work in the processing thread)
(2) Add a downstream node to asof join that processes data slowly (similar to a slow data sink), i.e., process one batch per second. This way, the backpressure would be pushed from the slow sink to asof join then to the data sources.

I think I prefer (2) a bit more because this affects represents a real life case of slow sink.

@westonpace I am not sure if the idea of GatedSourceNode is similar or different, but happy to hear

icexelloss · 2023-06-02T13:22:41Z

cpp/src/arrow/acero/asof_join_node_test.cc

+  ASSERT_GT(counters_by_is_fast[true].resume_count, 0);
+  // runs on some slow machines may not see any pause/resume, but if at least one pause is
+  // seen then at least one resume must also be seen
+  ASSERT_EQ(counters_by_is_fast[false].pause_count > 0,


I don't think we really need to check the slow sources, because it is entirely possible that it is not paused at all in the expected behavior (since it is slower than asof join?)

I think we do. My understanding is that the intention of the pre-PR test code was to drive the as-of-join slower using a slower input to it, which should lead to backpressure from the as-of-join node toward a faster source. The additional intention of the post-PR test code is to check that indeed this backpressure happens, and this is observed at the test nodes inserted after each source.

rtpsw · 2023-06-02T14:50:05Z

Thinking about loud here:

What we want to test is that if the through put of asof join node is slower than the source, then we would pause the source. Two potential ways that I think we can reliably do this: (1) Add some sort of "debug options" to manipulate the behavior of asof join to make it run slower. (i.e. Sleep a few seconds before actually starting the work in the processing thread) (2) Add a downstream node to asof join that processes data slowly (similar to a slow data sink), i.e., process one batch per second. This way, the backpressure would be pushed from the slow sink to asof join then to the data sources.

I think I prefer (2) a bit more because this affects represents a real life case of slow sink.

@westonpace I am not sure if the idea of GatedSourceNode is similar or different, but happy to hear

While I'm not sure exactly what Weston has in mind, my understanding is that the GatedSourceNode's goal is to avoid flakiness due to non-deterministic timing. IMO, both (1) and (2) above could be flaky due to non-deterministic timing.

Between (1) and (2) I also wouldn't prefer (1) because the debug-options would change the behavior of the as-of-join node being tested, and I prefer to change the code driving it instead.

cpp/src/arrow/acero/asof_join_node_test.cc

icexelloss

@rtpsw A few comments:
(1) Please separate out threading related changes to asof join - those should be in separate PR (outside of original scope)
(2) Please update the original iGH ssue to better reflect the change in this PR
(3) other inline comments

icexelloss · 2023-06-14T18:15:40Z

cpp/src/arrow/acero/asof_join_node_test.cc

+  Gate* gate;
+};
+
+struct GatedNode : public ExecNode, public TracedNode {


Can you add documentation for this class to describe the purpose of this?

rtpsw · 2023-06-15T18:43:01Z

(2) Please update the original iGH ssue to better reflect the change in this PR

It's authored by Weston - I don't have permission to edit. I edited the description of this PR instead,

rtpsw · 2023-06-15T19:12:49Z

This CI job failure suggests that the tightening condition (in this commit) isn't always true. We should decide whether to revert the tightened condition or to look for the cause for its violation.

icexelloss · 2023-06-16T13:24:20Z

(2) Please update the original iGH ssue to better reflect the change in this PR

It's authored by Weston - I don't have permission to edit. I edited the description of this PR instead,

Ok we can leave it as is then.

cpp/src/arrow/acero/asof_join_node_test.cc

icexelloss

LGTM

conbench-apache-arrow · 2023-06-22T19:49:00Z

Conbench analyzed the 6 benchmark runs on commit 73239526.

There were no benchmark performance regressions. 🎉

The full Conbench report has more details.

github-actions bot added Component: C++ awaiting review Awaiting review labels Jun 1, 2023

icexelloss reviewed Jun 1, 2023

View reviewed changes

github-actions bot added awaiting changes Awaiting changes and removed awaiting review Awaiting review labels Jun 1, 2023

pitrou reviewed Jun 1, 2023

View reviewed changes

github-actions bot added awaiting change review Awaiting change review awaiting changes Awaiting changes and removed awaiting changes Awaiting changes awaiting change review Awaiting change review labels Jun 1, 2023

rtpsw mentioned this pull request Jun 1, 2023

GH-35838: [C++] Fix asof join backpresure #35878

Merged

icexelloss reviewed Jun 1, 2023

View reviewed changes

apacheGH-35838: [C++] Backpressure broken in asof join node

7d40e92

rtpsw force-pushed the GH-35838 branch from caa5b5d to 7d40e92 Compare June 2, 2023 12:28

github-actions bot added awaiting changes Awaiting changes and removed awaiting change review Awaiting change review labels Jun 2, 2023

westonpace reviewed Jun 2, 2023

View reviewed changes

improve parameter names, attempt to fix flaky test

3e5fdf9

github-actions bot added awaiting change review Awaiting change review awaiting changes Awaiting changes and removed awaiting changes Awaiting changes awaiting change review Awaiting change review labels Jun 2, 2023

icexelloss reviewed Jun 2, 2023

View reviewed changes

icexelloss reviewed Jun 14, 2023

View reviewed changes

cpp/src/arrow/acero/asof_join_node_test.cc Outdated Show resolved Hide resolved

icexelloss reviewed Jun 14, 2023

View reviewed changes

cpp/src/arrow/acero/asof_join_node_test.cc Outdated Show resolved Hide resolved

icexelloss requested changes Jun 14, 2023

View reviewed changes

icexelloss reviewed Jun 14, 2023

View reviewed changes

rtpsw added 5 commits June 15, 2023 10:28

revert concurrency change

ae5c840

doc for key hasher invalidation

29e329e

refactor gate classes

1e3652f

tighten pause counts check

c458c2e

doc GatedNodeOptions

2a1bac1

github-actions bot added awaiting change review Awaiting change review awaiting changes Awaiting changes and removed awaiting changes Awaiting changes awaiting change review Awaiting change review labels Jun 15, 2023

github-actions bot added awaiting change review Awaiting change review and removed awaiting changes Awaiting changes labels Jun 16, 2023

stronger test condition for resume

05ea6c8

rtpsw force-pushed the GH-35838 branch from c0a7496 to 05ea6c8 Compare June 16, 2023 12:53

github-actions bot added awaiting changes Awaiting changes and removed awaiting change review Awaiting change review labels Jun 16, 2023

rtpsw requested a review from icexelloss June 17, 2023 10:36

icexelloss reviewed Jun 17, 2023

View reviewed changes

cpp/src/arrow/acero/asof_join_node_test.cc Show resolved Hide resolved

icexelloss approved these changes Jun 21, 2023

View reviewed changes

icexelloss merged commit 7323952 into apache:main Jun 21, 2023
36 of 39 checks passed

github-actions bot added awaiting merge Awaiting merge and removed awaiting changes Awaiting changes labels Jun 21, 2023

rtpsw deleted the GH-35838 branch June 22, 2023 09:15

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

GH-35838: [C++] Add backpressure test for asof join node #35874

GH-35838: [C++] Add backpressure test for asof join node #35874

rtpsw commented Jun 1, 2023 •

edited

Loading

icexelloss commented Jun 1, 2023

icexelloss Jun 1, 2023 •

edited

Loading

icexelloss Jun 1, 2023

icexelloss Jun 1, 2023

rtpsw Jun 1, 2023 •

edited

Loading

rtpsw Jun 1, 2023

icexelloss Jun 1, 2023

icexelloss Jun 1, 2023

icexelloss Jun 1, 2023 •

edited

Loading

westonpace Jun 2, 2023

rtpsw Jun 2, 2023

pitrou Jun 1, 2023

rtpsw Jun 1, 2023

rtpsw commented Jun 1, 2023

icexelloss Jun 1, 2023

rtpsw Jun 2, 2023

rtpsw Jun 2, 2023

westonpace left a comment •

edited

Loading

rtpsw commented Jun 2, 2023

icexelloss commented Jun 2, 2023

icexelloss Jun 2, 2023

rtpsw Jun 2, 2023 •

edited

Loading

rtpsw commented Jun 2, 2023

icexelloss left a comment

icexelloss Jun 14, 2023

rtpsw Jun 15, 2023

rtpsw commented Jun 15, 2023

rtpsw commented Jun 15, 2023 •

edited

Loading

icexelloss commented Jun 16, 2023

icexelloss left a comment

conbench-apache-arrow bot commented Jun 22, 2023

	std::make_unique<BackpressureController>(input, node, backpressure_counter);
	std::make_unique<BackpressureController>(/xxx=/ input, /yyy=/ node, backpressure_counter);

GH-35838: [C++] Add backpressure test for asof join node #35874

GH-35838: [C++] Add backpressure test for asof join node #35874

Conversation

rtpsw commented Jun 1, 2023 • edited Loading

What changes are included in this PR?

Are these changes tested?

Are there any user-facing changes?

icexelloss commented Jun 1, 2023

icexelloss Jun 1, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rtpsw Jun 1, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

icexelloss Jun 1, 2023 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rtpsw commented Jun 1, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

westonpace left a comment • edited Loading

Choose a reason for hiding this comment

rtpsw commented Jun 2, 2023

icexelloss commented Jun 2, 2023

Choose a reason for hiding this comment

rtpsw Jun 2, 2023 • edited Loading

Choose a reason for hiding this comment

rtpsw commented Jun 2, 2023

icexelloss left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rtpsw commented Jun 15, 2023

rtpsw commented Jun 15, 2023 • edited Loading

icexelloss commented Jun 16, 2023

icexelloss left a comment

Choose a reason for hiding this comment

conbench-apache-arrow bot commented Jun 22, 2023

rtpsw commented Jun 1, 2023 •

edited

Loading

icexelloss Jun 1, 2023 •

edited

Loading

rtpsw Jun 1, 2023 •

edited

Loading

icexelloss Jun 1, 2023 •

edited

Loading

westonpace left a comment •

edited

Loading

rtpsw Jun 2, 2023 •

edited

Loading

rtpsw commented Jun 15, 2023 •

edited

Loading