{"action":"created","comment":{"url":"https://api.github.com/repos/apache/beam/pulls/comments/196598266","pull_request_review_id":130178385,"id":196598266,"node_id":"MDI0OlB1bGxSZXF1ZXN0UmV2aWV3Q29tbWVudDE5NjU5ODI2Ng==","diff_hunk":"@@ -125,6 +162,55 @@ private static PipelineResult run(\n     return result;\n   }\n \n+  private static PipelineResult limitRun(\n+      PipelineOptions options,\n+      BeamRelNode node,\n+      DoFn<Row, KV<String, Row>> collectDoFn,\n+      DoFn<KV<String, Row>, Void> limitCounterDoFn,\n+      LimitStateVar limitStateVar) {\n+    ExecutorService pool = Executors.newFixedThreadPool(1);\n+\n+    options.as(DirectOptions.class).setBlockOnRun(false);\n+    Pipeline pipeline = Pipeline.create(options);\n+    PCollectionTuple.empty(pipeline)\n+        .apply(node.toPTransform())\n+        .apply(ParDo.of(collectDoFn))\n+        .apply(ParDo.of(limitCounterDoFn));\n+\n+    PipelineResult result = pipeline.run();\n+\n+    pool.execute(\n+        new Runnable() {\n+          @Override\n+          public void run() {\n+            // wait every one second to check state of PipelineResult. If returned value\n+            // is null, or return state is non termination state, it means pipeline is still\n+            // running and within loop limit counter state will be checked.\n+            State state;\n+            while (true) {\n+              state = result.waitUntilFinish(Duration.standardSeconds(1));\n+              if (state != null && state.isTerminal()) {\n+                break;\n+              }\n+              // If state is null, or state is not null and indicates pipeline is not terminated\n+              // yet, check continue checking the limit var.\n+              try {\n+                if (limitStateVar.isReached()) {","path":"sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamEnumerableConverter.java","position":122,"original_position":122,"commit_id":"231bde4052d89893aefbcfe7ae69141ebf2cde3f","original_commit_id":"231bde4052d89893aefbcfe7ae69141ebf2cde3f","user":{"login":"apilloud","id":5972627,"node_id":"MDQ6VXNlcjU5NzI2Mjc=","avatar_url":"https://avatars2.githubusercontent.com/u/5972627?v=4","gravatar_id":"","url":"https://api.github.com/users/apilloud","html_url":"https://github.com/apilloud","followers_url":"https://api.github.com/users/apilloud/followers","following_url":"https://api.github.com/users/apilloud/following{/other_user}","gists_url":"https://api.github.com/users/apilloud/gists{/gist_id}","starred_url":"https://api.github.com/users/apilloud/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apilloud/subscriptions","organizations_url":"https://api.github.com/users/apilloud/orgs","repos_url":"https://api.github.com/users/apilloud/repos","events_url":"https://api.github.com/users/apilloud/events{/privacy}","received_events_url":"https://api.github.com/users/apilloud/received_events","type":"User","site_admin":false},"body":"If you did something like `values.size() >= limitCount` here you could replace `LimitCanceller` and `LimitCollector` with `Collector`.","created_at":"2018-06-19T22:25:43Z","updated_at":"2018-06-19T22:27:58Z","html_url":"https://github.com/apache/beam/pull/5682#discussion_r196598266","pull_request_url":"https://api.github.com/repos/apache/beam/pulls/5682","author_association":"CONTRIBUTOR","_links":{"self":{"href":"https://api.github.com/repos/apache/beam/pulls/comments/196598266"},"html":{"href":"https://github.com/apache/beam/pull/5682#discussion_r196598266"},"pull_request":{"href":"https://api.github.com/repos/apache/beam/pulls/5682"}}},"pull_request":{"url":"https://api.github.com/repos/apache/beam/pulls/5682","id":195702929,"node_id":"MDExOlB1bGxSZXF1ZXN0MTk1NzAyOTI5","html_url":"https://github.com/apache/beam/pull/5682","diff_url":"https://github.com/apache/beam/pull/5682.diff","patch_url":"https://github.com/apache/beam/pull/5682.patch","issue_url":"https://api.github.com/repos/apache/beam/issues/5682","number":5682,"state":"open","locked":false,"title":"[BEAM-4194] support unbounded limit","user":{"login":"amaliujia","id":1938382,"node_id":"MDQ6VXNlcjE5MzgzODI=","avatar_url":"https://avatars1.githubusercontent.com/u/1938382?v=4","gravatar_id":"","url":"https://api.github.com/users/amaliujia","html_url":"https://github.com/amaliujia","followers_url":"https://api.github.com/users/amaliujia/followers","following_url":"https://api.github.com/users/amaliujia/following{/other_user}","gists_url":"https://api.github.com/users/amaliujia/gists{/gist_id}","starred_url":"https://api.github.com/users/amaliujia/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/amaliujia/subscriptions","organizations_url":"https://api.github.com/users/amaliujia/orgs","repos_url":"https://api.github.com/users/amaliujia/repos","events_url":"https://api.github.com/users/amaliujia/events{/privacy}","received_events_url":"https://api.github.com/users/amaliujia/received_events","type":"User","site_admin":false},"body":"##### What?\r\nSupport 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.\r\n\r\nThis 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.\r\n\r\n##### Testing\r\n - [x] Unit test.\r\n \r\nAdding two unit tests to mock bounded and unbounded input tables to test LIMIT functionality.\r\n\r\n - [x] Integration test.\r\n\r\nAdding one e2e integration test, which utilizes Google Cloud Pub/Sub to test the unbounded limit on auto-generated Pub/Sub messages.\r\n\r\n - [x] Other test.\r\n\r\nManually tested unbounded limit in Beam SQL shell on a Google Cloud Pub/Sub table. See the screenshot:\r\n<img width=\"1041\" alt=\"screen shot 2018-06-18 at 3 48 07 pm\" src=\"https://user-images.githubusercontent.com/1938382/41574436-106ca752-7336-11e8-91a7-33200513d4b8.png\">\r\n\r\n\r\n------------------------\r\n\r\nFollow this checklist to help us incorporate your contribution quickly and easily:\r\n\r\n - [x] 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.\r\n - [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).\r\n\r\nIt will help us expedite review of your Pull Request if you tag someone (e.g. `@username`) to look at it.\r\n","created_at":"2018-06-19T03:29:30Z","updated_at":"2018-06-19T22:27:58Z","closed_at":null,"merged_at":null,"merge_commit_sha":"fc0db37cb59f7d2e3d7c4568276006b4674cbdbe","assignee":null,"assignees":[],"requested_reviewers":[],"requested_teams":[],"labels":[],"milestone":null,"commits_url":"https://api.github.com/repos/apache/beam/pulls/5682/commits","review_comments_url":"https://api.github.com/repos/apache/beam/pulls/5682/comments","review_comment_url":"https://api.github.com/repos/apache/beam/pulls/comments{/number}","comments_url":"https://api.github.com/repos/apache/beam/issues/5682/comments","statuses_url":"https://api.github.com/repos/apache/beam/statuses/231bde4052d89893aefbcfe7ae69141ebf2cde3f","head":{"label":"amaliujia:rui_wang-test_branch","ref":"rui_wang-test_branch","sha":"231bde4052d89893aefbcfe7ae69141ebf2cde3f","user":{"login":"amaliujia","id":1938382,"node_id":"MDQ6VXNlcjE5MzgzODI=","avatar_url":"https://avatars1.githubusercontent.com/u/1938382?v=4","gravatar_id":"","url":"https://api.github.com/users/amaliujia","html_url":"https://github.com/amaliujia","followers_url":"https://api.github.com/users/amaliujia/followers","following_url":"https://api.github.com/users/amaliujia/following{/other_user}","gists_url":"https://api.github.com/users/amaliujia/gists{/gist_id}","starred_url":"https://api.github.com/users/amaliujia/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/amaliujia/subscriptions","organizations_url":"https://api.github.com/users/amaliujia/orgs","repos_url":"https://api.github.com/users/amaliujia/repos","events_url":"https://api.github.com/users/amaliujia/events{/privacy}","received_events_url":"https://api.github.com/users/amaliujia/received_events","type":"User","site_admin":false},"repo":{"id":80804775,"node_id":"MDEwOlJlcG9zaXRvcnk4MDgwNDc3NQ==","name":"beam","full_name":"amaliujia/beam","owner":{"login":"amaliujia","id":1938382,"node_id":"MDQ6VXNlcjE5MzgzODI=","avatar_url":"https://avatars1.githubusercontent.com/u/1938382?v=4","gravatar_id":"","url":"https://api.github.com/users/amaliujia","html_url":"https://github.com/amaliujia","followers_url":"https://api.github.com/users/amaliujia/followers","following_url":"https://api.github.com/users/amaliujia/following{/other_user}","gists_url":"https://api.github.com/users/amaliujia/gists{/gist_id}","starred_url":"https://api.github.com/users/amaliujia/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/amaliujia/subscriptions","organizations_url":"https://api.github.com/users/amaliujia/orgs","repos_url":"https://api.github.com/users/amaliujia/repos","events_url":"https://api.github.com/users/amaliujia/events{/privacy}","received_events_url":"https://api.github.com/users/amaliujia/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/amaliujia/beam","description":"Mirror of Apache Beam","fork":true,"url":"https://api.github.com/repos/amaliujia/beam","forks_url":"https://api.github.com/repos/amaliujia/beam/forks","keys_url":"https://api.github.com/repos/amaliujia/beam/keys{/key_id}","collaborators_url":"https://api.github.com/repos/amaliujia/beam/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/amaliujia/beam/teams","hooks_url":"https://api.github.com/repos/amaliujia/beam/hooks","issue_events_url":"https://api.github.com/repos/amaliujia/beam/issues/events{/number}","events_url":"https://api.github.com/repos/amaliujia/beam/events","assignees_url":"https://api.github.com/repos/amaliujia/beam/assignees{/user}","branches_url":"https://api.github.com/repos/amaliujia/beam/branches{/branch}","tags_url":"https://api.github.com/repos/amaliujia/beam/tags","blobs_url":"https://api.github.com/repos/amaliujia/beam/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/amaliujia/beam/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/amaliujia/beam/git/refs{/sha}","trees_url":"https://api.github.com/repos/amaliujia/beam/git/trees{/sha}","statuses_url":"https://api.github.com/repos/amaliujia/beam/statuses/{sha}","languages_url":"https://api.github.com/repos/amaliujia/beam/languages","stargazers_url":"https://api.github.com/repos/amaliujia/beam/stargazers","contributors_url":"https://api.github.com/repos/amaliujia/beam/contributors","subscribers_url":"https://api.github.com/repos/amaliujia/beam/subscribers","subscription_url":"https://api.github.com/repos/amaliujia/beam/subscription","commits_url":"https://api.github.com/repos/amaliujia/beam/commits{/sha}","git_commits_url":"https://api.github.com/repos/amaliujia/beam/git/commits{/sha}","comments_url":"https://api.github.com/repos/amaliujia/beam/comments{/number}","issue_comment_url":"https://api.github.com/repos/amaliujia/beam/issues/comments{/number}","contents_url":"https://api.github.com/repos/amaliujia/beam/contents/{+path}","compare_url":"https://api.github.com/repos/amaliujia/beam/compare/{base}...{head}","merges_url":"https://api.github.com/repos/amaliujia/beam/merges","archive_url":"https://api.github.com/repos/amaliujia/beam/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/amaliujia/beam/downloads","issues_url":"https://api.github.com/repos/amaliujia/beam/issues{/number}","pulls_url":"https://api.github.com/repos/amaliujia/beam/pulls{/number}","milestones_url":"https://api.github.com/repos/amaliujia/beam/milestones{/number}","notifications_url":"https://api.github.com/repos/amaliujia/beam/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/amaliujia/beam/labels{/name}","releases_url":"https://api.github.com/repos/amaliujia/beam/releases{/id}","deployments_url":"https://api.github.com/repos/amaliujia/beam/deployments","created_at":"2017-02-03T06:53:41Z","updated_at":"2018-06-05T20:01:25Z","pushed_at":"2018-06-19T22:12:59Z","git_url":"git://github.com/amaliujia/beam.git","ssh_url":"git@github.com:amaliujia/beam.git","clone_url":"https://github.com/amaliujia/beam.git","svn_url":"https://github.com/amaliujia/beam","homepage":"","size":59996,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"archived":false,"open_issues_count":0,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"apache:master","ref":"master","sha":"fcb297ce3146a7d09bbfafde60d0d61261e65844","user":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"repo":{"id":50904245,"node_id":"MDEwOlJlcG9zaXRvcnk1MDkwNDI0NQ==","name":"beam","full_name":"apache/beam","owner":{"login":"apache","id":47359,"node_id":"MDEyOk9yZ2FuaXphdGlvbjQ3MzU5","avatar_url":"https://avatars0.githubusercontent.com/u/47359?v=4","gravatar_id":"","url":"https://api.github.com/users/apache","html_url":"https://github.com/apache","followers_url":"https://api.github.com/users/apache/followers","following_url":"https://api.github.com/users/apache/following{/other_user}","gists_url":"https://api.github.com/users/apache/gists{/gist_id}","starred_url":"https://api.github.com/users/apache/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/apache/subscriptions","organizations_url":"https://api.github.com/users/apache/orgs","repos_url":"https://api.github.com/users/apache/repos","events_url":"https://api.github.com/users/apache/events{/privacy}","received_events_url":"https://api.github.com/users/apache/received_events","type":"Organization","site_admin":false},"private":false,"html_url":"https://github.com/apache/beam","description":"Apache Beam","fork":false,"url":"https://api.github.com/repos/apache/beam","forks_url":"https://api.github.com/repos/apache/beam/forks","keys_url":"https://api.github.com/repos/apache/beam/keys{/key_id}","collaborators_url":"https://api.github.com/repos/apache/beam/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/apache/beam/teams","hooks_url":"https://api.github.com/repos/apache/beam/hooks","issue_events_url":"https://api.github.com/repos/apache/beam/issues/events{/number}","events_url":"https://api.github.com/repos/apache/beam/events","assignees_url":"https://api.github.com/repos/apache/beam/assignees{/user}","branches_url":"https://api.github.com/repos/apache/beam/branches{/branch}","tags_url":"https://api.github.com/repos/apache/beam/tags","blobs_url":"https://api.github.com/repos/apache/beam/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/apache/beam/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/apache/beam/git/refs{/sha}","trees_url":"https://api.github.com/repos/apache/beam/git/trees{/sha}","statuses_url":"https://api.github.com/repos/apache/beam/statuses/{sha}","languages_url":"https://api.github.com/repos/apache/beam/languages","stargazers_url":"https://api.github.com/repos/apache/beam/stargazers","contributors_url":"https://api.github.com/repos/apache/beam/contributors","subscribers_url":"https://api.github.com/repos/apache/beam/subscribers","subscription_url":"https://api.github.com/repos/apache/beam/subscription","commits_url":"https://api.github.com/repos/apache/beam/commits{/sha}","git_commits_url":"https://api.github.com/repos/apache/beam/git/commits{/sha}","comments_url":"https://api.github.com/repos/apache/beam/comments{/number}","issue_comment_url":"https://api.github.com/repos/apache/beam/issues/comments{/number}","contents_url":"https://api.github.com/repos/apache/beam/contents/{+path}","compare_url":"https://api.github.com/repos/apache/beam/compare/{base}...{head}","merges_url":"https://api.github.com/repos/apache/beam/merges","archive_url":"https://api.github.com/repos/apache/beam/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/apache/beam/downloads","issues_url":"https://api.github.com/repos/apache/beam/issues{/number}","pulls_url":"https://api.github.com/repos/apache/beam/pulls{/number}","milestones_url":"https://api.github.com/repos/apache/beam/milestones{/number}","notifications_url":"https://api.github.com/repos/apache/beam/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/apache/beam/labels{/name}","releases_url":"https://api.github.com/repos/apache/beam/releases{/id}","deployments_url":"https://api.github.com/repos/apache/beam/deployments","created_at":"2016-02-02T08:00:06Z","updated_at":"2018-06-19T21:38:48Z","pushed_at":"2018-06-19T22:27:58Z","git_url":"git://github.com/apache/beam.git","ssh_url":"git@github.com:apache/beam.git","clone_url":"https://github.com/apache/beam.git","svn_url":"https://github.com/apache/beam","homepage":"","size":65458,"stargazers_count":1924,"watchers_count":1924,"language":"Java","has_issues":false,"has_projects":true,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":1191,"mirror_url":null,"archived":false,"open_issues_count":93,"license":{"key":"apache-2.0","name":"Apache License 2.0","spdx_id":"Apache-2.0","url":"https://api.github.com/licenses/apache-2.0","node_id":"MDc6TGljZW5zZTI="},"forks":1191,"open_issues":93,"watchers":1924,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/apache/beam/pulls/5682"},"html":{"href":"https://github.com/apache/beam/pull/5682"},"issue":{"href":"https://api.github.com/repos/apache/beam/issues/5682"},"comments":{"href":"https://api.github.com/repos/apache/beam/issues/5682/comments"},"review_comments":{"href":"https://api.github.com/repos/apache/beam/pulls/5682/comments"},"review_comment":{"href":"https://api.github.com/repos/apache/beam/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/apache/beam/pulls/5682/commits"},"statuses":{"href":"https://api.github.com/repos/apache/beam/statuses/231bde4052d89893aefbcfe7ae69141ebf2cde3f"}},"author_association":"CONTRIBUTOR"}}
{"actor":{"display_login":"apilloud"}}