{"action":"created","issue":{"url":"https://api.github.com/repos/apache/spark/issues/21109","repository_url":"https://api.github.com/repos/apache/spark","labels_url":"https://api.github.com/repos/apache/spark/issues/21109/labels{/name}","comments_url":"https://api.github.com/repos/apache/spark/issues/21109/comments","events_url":"https://api.github.com/repos/apache/spark/issues/21109/events","html_url":"https://github.com/apache/spark/pull/21109","id":316043961,"node_id":"MDExOlB1bGxSZXF1ZXN0MTgyOTAxMDQz","number":21109,"title":"[SPARK-24020][SQL] Sort-merge join inner range optimization","user":{"login":"zecevicp","id":9744763,"node_id":"MDQ6VXNlcjk3NDQ3NjM=","avatar_url":"https://avatars2.githubusercontent.com/u/9744763?v=4","gravatar_id":"","url":"https://api.github.com/users/zecevicp","html_url":"https://github.com/zecevicp","followers_url":"https://api.github.com/users/zecevicp/followers","following_url":"https://api.github.com/users/zecevicp/following{/other_user}","gists_url":"https://api.github.com/users/zecevicp/gists{/gist_id}","starred_url":"https://api.github.com/users/zecevicp/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/zecevicp/subscriptions","organizations_url":"https://api.github.com/users/zecevicp/orgs","repos_url":"https://api.github.com/users/zecevicp/repos","events_url":"https://api.github.com/users/zecevicp/events{/privacy}","received_events_url":"https://api.github.com/users/zecevicp/received_events","type":"User","site_admin":false},"labels":[],"state":"open","locked":false,"assignee":null,"assignees":[],"milestone":null,"comments":33,"created_at":"2018-04-19T21:16:19Z","updated_at":"2018-06-09T00:06:47Z","closed_at":null,"author_association":"NONE","pull_request":{"url":"https://api.github.com/repos/apache/spark/pulls/21109","html_url":"https://github.com/apache/spark/pull/21109","diff_url":"https://github.com/apache/spark/pull/21109.diff","patch_url":"https://github.com/apache/spark/pull/21109.patch"},"body":"## JIRA description\r\n\r\nThe problem we are solving is the case where you have two big tables partitioned by X column, but also sorted by Y column (within partitions) and you need to calculate an expensive function on the joined rows. During a sort-merge join, Spark will do cross-joins of all rows that have the same X values and calculate the function's value on all of them. If the two tables have a large number of rows per X, this can result in a huge number of calculations.\r\n\r\nWe hereby propose an optimization that would allow you to reduce the number of matching rows per X using a range condition on Y columns of the two tables. Something like:\r\n\r\n`... WHERE t1.X = t2.X AND t1.Y BETWEEN t2.Y - d AND t2.Y + d`\r\n\r\nThe way SMJ is currently implemented, these extra conditions have no influence on the number of rows (per X) being checked because these extra conditions are put in the same block with the function being calculated.\r\n\r\nHere we propose a change to the sort-merge join so that, when these extra conditions are specified, a queue is used instead of the ExternalAppendOnlyUnsafeRowArray class. This queue would then used as a moving window across the values from the right relation as the left row changes. You could call this a combination of an equi-join and a theta join (we call it \"sort-merge inner range join\").\r\n\r\nPotential use-cases for this are joins based on spatial or temporal distance calculations.\r\n\r\nThe optimization should be triggered automatically when an equi-join expression is present AND lower and upper range conditions on a secondary column are specified. If the tables aren't sorted by both columns, appropriate sorts should be added.\r\n\r\nTo limit the impact of this change we also propose adding a new parameter (tentatively named \"spark.sql.join.smj.useInnerRangeOptimization\") which could be used to switch off the optimization entirely.\r\n\r\n## What changes were proposed in this pull request?\r\n\r\nThe main changes are to `ExtractEquiJoinKeys`, `SortMergeJoinExec` and the new `InMemoryUnsafeRowQueue` class.\r\nChanges to equi-join key extraction (`ExtractEquiJoinKeys` class) to recognize the situation where the join condition contains an equi-join and a range condition on a second column (e.g. `t1.x = t2.x AND t1.y BETWEEN t2.y - d1 AND t2.y + d2`).\r\nChanges to `SortMergeJoinExec` with implementation of the inner range optimization. Range condition (upper and lower limits) is used to iterate through the rows of the right relation using a moving window, which moves as the values of the current left row change. \r\nThe moving window is implemented as the `InMemoryUnsafeRowQueue` class (new class added as part of the PR). \r\n\r\n## How was this patch tested?\r\nUnit tests (`InnerRangeSuite.scala`)\r\nPerformance tests in `JoinBenchmark`. The tests show 8x improvement over non-optimized code. Although, it should be noted that the results depend on the exact range conditions and the calculations performed on each matched row. \r\nIn our case, we were not able to cross-match two rather large datasets (1.2 billion rows x 800 million rows) without this optimization. With the optimization, the cross-match finishes in less than 2 minutes.\r\n"},"comment":{"url":"https://api.github.com/repos/apache/spark/issues/comments/395922852","html_url":"https://github.com/apache/spark/pull/21109#issuecomment-395922852","issue_url":"https://api.github.com/repos/apache/spark/issues/21109","id":395922852,"node_id":"MDEyOklzc3VlQ29tbWVudDM5NTkyMjg1Mg==","user":{"login":"AmplabJenkins","id":4031291,"node_id":"MDQ6VXNlcjQwMzEyOTE=","avatar_url":"https://avatars3.githubusercontent.com/u/4031291?v=4","gravatar_id":"","url":"https://api.github.com/users/AmplabJenkins","html_url":"https://github.com/AmplabJenkins","followers_url":"https://api.github.com/users/AmplabJenkins/followers","following_url":"https://api.github.com/users/AmplabJenkins/following{/other_user}","gists_url":"https://api.github.com/users/AmplabJenkins/gists{/gist_id}","starred_url":"https://api.github.com/users/AmplabJenkins/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/AmplabJenkins/subscriptions","organizations_url":"https://api.github.com/users/AmplabJenkins/orgs","repos_url":"https://api.github.com/users/AmplabJenkins/repos","events_url":"https://api.github.com/users/AmplabJenkins/events{/privacy}","received_events_url":"https://api.github.com/users/AmplabJenkins/received_events","type":"User","site_admin":false},"created_at":"2018-06-09T00:06:47Z","updated_at":"2018-06-09T00:06:47Z","author_association":"NONE","body":"Can one of the admins verify this patch?"}}
{"actor":{"display_login":"AmplabJenkins"}}