[BEAM-4194] support unbounded limit #5682

amaliujia · 2018-06-19T03:29:30Z

What?

Support unbounded limit in Beam SQL shell. In the past, due to default global window and default trigger, queries like "SELECT col_name FROM unbounded_table LIMIT 1" will not return in SQL Shell.

This PR tries to support unbounded limit by starting a daemon thread to monitoring return value collection in BeanEnumerableConverter, and stopping pipeline when collected values reach limit count. More detailed description can be found here: https://docs.google.com/document/d/13zeTewHH9nfwhSlcE4x77WQwr1U2Z4sTiNRjOXUj2aw/edit?usp=sharing.

Testing

Unit test.

Adding two unit tests to mock bounded and unbounded input tables to test LIMIT functionality.

Integration test.

Adding one e2e integration test, which utilizes Google Cloud Pub/Sub to test the unbounded limit on auto-generated Pub/Sub messages.

Other test.

Manually tested unbounded limit in Beam SQL shell on a Google Cloud Pub/Sub table. See the screenshot:

Follow this checklist to help us incorporate your contribution quickly and easily:

Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

It will help us expedite review of your Pull Request if you tag someone (e.g. @username) to look at it.

amaliujia · 2018-06-19T06:46:43Z

run java precommit

amaliujia · 2018-06-19T07:59:30Z

run java postcommit

amaliujia · 2018-06-19T16:55:57Z

R: @kennknowles
cc: @apilloud @akedin

kennknowles · 2018-06-19T17:16:59Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

+          public void run() {
+            while (!result.getState().isTerminal()) {
+              try {
+                Thread.sleep(DaemonThreadSleepIntervalMillis);


I think result.waitUntilFinish(duration) is good here: https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/PipelineResult.java#L58

Great suggestion!

kennknowles · 2018-06-19T17:19:40Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

+    NOT_REACHED
+  }
+
+  private static class LimitStateWrapper implements Serializable {


Why is there a wrapper? Is this to make it clearly something like an LVar?

Rename it to LimitStateVar

kennknowles · 2018-06-19T17:20:40Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

+        options
+            .getRunner()
+            .getCanonicalName()
+            .equals("org.apache.beam.runners.direct.DirectRunner"));


Can you add a message to this check? That way when it fails it will explain what is going on.

Added something to say only DirectRunner is supported and please check option setting.

kennknowles · 2018-06-19T17:25:07Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

+  }
+
+  private static class LimitCounter extends DoFn<Row, Void> {
+    private static final Map<Long, Integer> globalLimitArguments =


In the DirectRunner this is OK. I think you could pretty easily make it a bit more general by using stateful ParDo(DoFn) for counting. Then the only thing left to make it work on a distributed runner is a communication path back to the shell. If you leave it like this, I think the DoFn also is not ready, because in a distributed run there will be lots of copies and separate JVMs on different workers, not gathered to one key, etc.

Use stateful ParDo to implement counter/cancelling now.

amaliujia · 2018-06-19T20:56:52Z

run java precommit

amaliujia · 2018-06-19T21:30:47Z

run java precommit

amaliujia · 2018-06-19T21:30:57Z

run java postcommit

amaliujia · 2018-06-19T22:25:28Z

run java postcommit

apilloud

This is really cool. It can probably be simplified significantly without any loss of functionality though.

apilloud · 2018-06-19T22:07:33Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

      return collect(options, node);
    } finally {
      Thread.currentThread().setContextClassLoader(originalClassLoader);
    }
  }

+  enum LimitState {


nit: Isn't this just a Boolean?

Yep. Using enum to have descriptive state than true/false.

apilloud · 2018-06-19T22:08:59Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

@@ -111,6 +147,55 @@ private static PipelineResult run(
    return result;
  }

+  private static PipelineResult limitRun(


The point of run was to have exactly one for everything. Seeing as that isn't going to work, you should probably just inline this.

I want to use this function to encapsulate related logic together.

To avoid a long limitCollect function.

apilloud · 2018-06-19T22:11:21Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

+
+    PipelineResult result = pipeline.run();
+
+    pool.execute(


The extra thread seems unnecessary. You can do this in the main thread.

apilloud · 2018-06-19T22:13:24Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

@@ -133,14 +219,108 @@ private static PipelineResult run(
        options
            .getRunner()
            .getCanonicalName()
-            .equals("org.apache.beam.runners.direct.DirectRunner"));
+            .equals("org.apache.beam.runners.direct.DirectRunner"),
+        "Only DirectRunner is supported in SQL Shell. " + "Please check your Runner setting.");


This isn't exactly true. Should be something like "SELECT without INSERT is only supported in DirectRunner in SQL Shell."

apilloud · 2018-06-19T22:25:43Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

+              // If state is null, or state is not null and indicates pipeline is not terminated
+              // yet, check continue checking the limit var.
+              try {
+                if (limitStateVar.isReached()) {


If you did something like values.size() >= limitCount here you could replace LimitCanceller and LimitCollector with Collector.

I could replace LimitCollector with Collector with the cost of removing extra collected values after the pipeline finishes, or I modify Collector and add something like values.size() >= limitCount to make Collector general. To make code more readable, I wrote a dedicated LimitCollector.

Also, I'd like to separate count and cancel and collect steps to support unbounded limit. Right now Collector seems like only work in DirectRunner but I can make count and cancel more general to fit into other runners by using stateful ParDo (that's what LimitCanceller does). So in the future if we start to support unbounded limit in other runners, only collect step needs modification.

I think we have a very different model as to how this will work in other runners. The runners shouldn't need to be aware of the limit, as that would be applied client side.

apilloud

Thanks for the cleanup. I still think there is a significant opportunity for simplification here around Collector. LimitCollector, and LimitCanceller, but I'm happy to do it as a followup. (I'm a big fan of minimalist programming.)

apilloud · 2018-06-20T00:19:14Z

...s/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java

+    @ProcessElement
+    public void processElement(ProcessContext context) {
+      Object[] input = context.element().getValues().toArray();
+      if (values.size() < count) {


Now that I think about it, you have a race between checking this count and adding to the queue. The directRunner has a minimum of 3 worker threads so it is possible to hit. I still think the simple, safe, and portable implementation is to use Collector, check the size of the queue outside of the pipeline, and truncate after terminating the pipeline.

That's a great catch! Yes I agree to use Collector here and drop extra values after terminating the pipeline.

Because I still need to keep LimitCanceller(as a general "count and cancel" solution for all runners), and LimitCanceller needs KV as its input type(a requirement of stateful ParDo), I will have to keep this LimitCollector to emit KV elements to LimitCanceller.

I will take care of the race condition.

amaliujia · 2018-06-20T06:08:22Z

run java postcommit

kennknowles · 2018-06-21T17:09:19Z

LGTM, just squash commits and I'll merge

amaliujia · 2018-06-21T18:40:02Z

run java postcommit

amaliujia · 2018-06-21T20:02:44Z

ElasticSearchIO

amaliujia · 2018-06-21T20:02:49Z

run java postcommit

amaliujia · 2018-06-21T20:59:15Z

requesting more quota failed a bunch of PostCommit tests.

amaliujia · 2018-06-21T21:00:06Z

run java postcommit

kennknowles · 2018-06-21T22:32:28Z

The tests actually part of this PR passed. We have some other quota issues but just for Dataflow.

kennknowles reviewed Jun 19, 2018

View reviewed changes

apilloud requested changes Jun 19, 2018

View reviewed changes

apilloud approved these changes Jun 19, 2018

View reviewed changes

apilloud reviewed Jun 20, 2018

View reviewed changes

support unbounded limit.

60c8baa

amaliujia force-pushed the rui_wang-test_branch branch from e88d8f5 to 60c8baa Compare June 21, 2018 18:21

kennknowles merged commit 0a92663 into apache:master Jun 21, 2018

amaliujia deleted the rui_wang-test_branch branch June 25, 2018 17:59

charlesccychen pushed a commit to charlesccychen/beam that referenced this pull request Jul 26, 2018

Merge pull request apache#5682: [BEAM-4194] support unbounded limit

a919bbc

[BEAM-4194] support unbounded limit #5682

[BEAM-4194] support unbounded limit #5682

Conversation

amaliujia commented Jun 19, 2018 • edited

What?

Testing

amaliujia commented Jun 19, 2018

amaliujia commented Jun 19, 2018

amaliujia commented Jun 19, 2018

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amaliujia commented Jun 19, 2018

amaliujia commented Jun 19, 2018

amaliujia commented Jun 19, 2018

amaliujia commented Jun 19, 2018

apilloud left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

amaliujia Jun 19, 2018 • edited

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

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

amaliujia Jun 19, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

apilloud left a 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

amaliujia commented Jun 20, 2018

kennknowles commented Jun 21, 2018

amaliujia commented Jun 21, 2018

amaliujia commented Jun 21, 2018

amaliujia commented Jun 21, 2018

amaliujia commented Jun 21, 2018 • edited

amaliujia commented Jun 21, 2018

kennknowles commented Jun 21, 2018

amaliujia commented Jun 19, 2018 •

edited

amaliujia Jun 19, 2018 •

edited

amaliujia Jun 19, 2018 •

edited

amaliujia commented Jun 21, 2018 •

edited