[jira] [Assigned] (SPARK-23676) Support left join codegen in SortMergeJoinExec
[ https://issues.apache.org/jira/browse/SPARK-23676?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Apache Spark reassigned SPARK-23676: Assignee: Apache Spark > Support left join codegen in SortMergeJoinExec > -- > > Key: SPARK-23676 > URL: https://issues.apache.org/jira/browse/SPARK-23676 > Project: Spark > Issue Type: Improvement > Components: SQL >Affects Versions: 2.4.0 >Reporter: caoxuewen >Assignee: Apache Spark >Priority: Major > > This PR generates java code to directly complete the function of LeftOuter in > `SortMergeJoinExec` without using an iterator. > This PR improves runtime performance by this generates java code. > joinBenchmark result: **1.3x** > ``` > Java HotSpot(TM) 64-Bit Server VM 1.8.0_60-b27 on Windows 7 6.1 > Intel(R) Core(TM) i5-6500 CPU @ 3.20GHz > left sort merge join: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative > -- > left merge join wholestage=off 2439 / 2575 0.9 1163.0 1.0X > left merge join wholestage=on 1890 / 1904 1.1 901.1 1.3X > ``` > Benchmark program > ``` > val N = 2 << 20 > runBenchmark("left sort merge join", N) { > val df1 = sparkSession.range(N) > .selectExpr(s"(id * 15485863) % ${N*10} as k1") > val df2 = sparkSession.range(N) > .selectExpr(s"(id * 15485867) % ${N*10} as k2") > val df = df1.join(df2, col("k1") === col("k2"), "left") > > assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[SortMergeJoinExec]).isDefined) > df.count() > ``` > code example > ``` > val df1 = spark.range(2 << 20).selectExpr("id as k1", "id * 2 as v1") > val df2 = spark.range(2 << 20).selectExpr("id as k2", "id * 3 as v2") > df1.join(df2, col("k1") === col("k2") && col("v1") < col("v2"), > "left").collect > ``` > Generated code > ``` > /* 001 */ public Object generate(Object[] references) { > /* 002 */ return new GeneratedIteratorForCodegenStage5(references); > /* 003 */ } > /* 004 */ > /* 005 */ // codegenStageId=5 > /* 006 */ final class GeneratedIteratorForCodegenStage5 extends > org.apache.spark.sql.execution.BufferedRowIterator { > /* 007 */ private Object[] references; > /* 008 */ private scala.collection.Iterator[] inputs; > /* 009 */ private scala.collection.Iterator smj_leftInput; > /* 010 */ private scala.collection.Iterator smj_rightInput; > /* 011 */ private InternalRow smj_leftRow; > /* 012 */ private InternalRow smj_rightRow; > /* 013 */ private long smj_value2; > /* 014 */ private > org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray smj_matches; > /* 015 */ private long smj_value3; > /* 016 */ private long smj_value4; > /* 017 */ private long smj_value5; > /* 018 */ private long smj_value6; > /* 019 */ private boolean smj_isNull2; > /* 020 */ private long smj_value7; > /* 021 */ private boolean smj_isNull3; > /* 022 */ private > org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder[] > smj_mutableStateArray1 = new > org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder[1]; > /* 023 */ private > org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[] > smj_mutableStateArray2 = new > org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[1]; > /* 024 */ private UnsafeRow[] smj_mutableStateArray = new UnsafeRow[1]; > /* 025 */ > /* 026 */ public GeneratedIteratorForCodegenStage5(Object[] references) { > /* 027 */ this.references = references; > /* 028 */ } > /* 029 */ > /* 030 */ public void init(int index, scala.collection.Iterator[] inputs) { > /* 031 */ partitionIndex = index; > /* 032 */ this.inputs = inputs; > /* 033 */ smj_leftInput = inputs[0]; > /* 034 */ smj_rightInput = inputs[1]; > /* 035 */ > /* 036 */ smj_matches = new > org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray(2147483647, > 2147483647); > /* 037 */ smj_mutableStateArray[0] = new UnsafeRow(4); > /* 038 */ smj_mutableStateArray1[0] = new > org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(smj_mutableStateArray[0], > 0); > /* 039 */ smj_mutableStateArray2[0] = new > org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(smj_mutableStateArray1[0], > 4); > /* 040 */ > /* 041 */ } > /* 042 */ > /* 043 */ private void writeJoinRows() throws java.io.IOException { > /* 044 */ smj_mutableStateArray2[0].zeroOutNullBytes(); > /* 045 */ > /* 046 */ smj_mutableStateArray2[0].write(0, smj_value4); > /* 047 */ > /* 048 */ smj_mutableStateArray2[0].write(1, smj_value5); > /* 049 */ > /* 050 */ if (smj_isNull2) { > /* 051 */ smj_mutableStateArray2[0].setNullAt(2); > /* 052 */ } else { > /* 053 */ smj_mutableStateArray2[0].write(2, smj_value6); > /* 054 */ } > /* 055 */ > /* 056 */ if (smj_isNull3) { > /* 057 */ smj_mutableStateArray2[0].setNullAt(3); > /* 058 */ } else { > /* 059 */ smj_mutable
[jira] [Assigned] (SPARK-23676) Support left join codegen in SortMergeJoinExec
[ https://issues.apache.org/jira/browse/SPARK-23676?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ] Apache Spark reassigned SPARK-23676: Assignee: (was: Apache Spark) > Support left join codegen in SortMergeJoinExec > -- > > Key: SPARK-23676 > URL: https://issues.apache.org/jira/browse/SPARK-23676 > Project: Spark > Issue Type: Improvement > Components: SQL >Affects Versions: 2.4.0 >Reporter: caoxuewen >Priority: Major > > This PR generates java code to directly complete the function of LeftOuter in > `SortMergeJoinExec` without using an iterator. > This PR improves runtime performance by this generates java code. > joinBenchmark result: **1.3x** > ``` > Java HotSpot(TM) 64-Bit Server VM 1.8.0_60-b27 on Windows 7 6.1 > Intel(R) Core(TM) i5-6500 CPU @ 3.20GHz > left sort merge join: Best/Avg Time(ms) Rate(M/s) Per Row(ns) Relative > -- > left merge join wholestage=off 2439 / 2575 0.9 1163.0 1.0X > left merge join wholestage=on 1890 / 1904 1.1 901.1 1.3X > ``` > Benchmark program > ``` > val N = 2 << 20 > runBenchmark("left sort merge join", N) { > val df1 = sparkSession.range(N) > .selectExpr(s"(id * 15485863) % ${N*10} as k1") > val df2 = sparkSession.range(N) > .selectExpr(s"(id * 15485867) % ${N*10} as k2") > val df = df1.join(df2, col("k1") === col("k2"), "left") > > assert(df.queryExecution.sparkPlan.find(_.isInstanceOf[SortMergeJoinExec]).isDefined) > df.count() > ``` > code example > ``` > val df1 = spark.range(2 << 20).selectExpr("id as k1", "id * 2 as v1") > val df2 = spark.range(2 << 20).selectExpr("id as k2", "id * 3 as v2") > df1.join(df2, col("k1") === col("k2") && col("v1") < col("v2"), > "left").collect > ``` > Generated code > ``` > /* 001 */ public Object generate(Object[] references) { > /* 002 */ return new GeneratedIteratorForCodegenStage5(references); > /* 003 */ } > /* 004 */ > /* 005 */ // codegenStageId=5 > /* 006 */ final class GeneratedIteratorForCodegenStage5 extends > org.apache.spark.sql.execution.BufferedRowIterator { > /* 007 */ private Object[] references; > /* 008 */ private scala.collection.Iterator[] inputs; > /* 009 */ private scala.collection.Iterator smj_leftInput; > /* 010 */ private scala.collection.Iterator smj_rightInput; > /* 011 */ private InternalRow smj_leftRow; > /* 012 */ private InternalRow smj_rightRow; > /* 013 */ private long smj_value2; > /* 014 */ private > org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray smj_matches; > /* 015 */ private long smj_value3; > /* 016 */ private long smj_value4; > /* 017 */ private long smj_value5; > /* 018 */ private long smj_value6; > /* 019 */ private boolean smj_isNull2; > /* 020 */ private long smj_value7; > /* 021 */ private boolean smj_isNull3; > /* 022 */ private > org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder[] > smj_mutableStateArray1 = new > org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder[1]; > /* 023 */ private > org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[] > smj_mutableStateArray2 = new > org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter[1]; > /* 024 */ private UnsafeRow[] smj_mutableStateArray = new UnsafeRow[1]; > /* 025 */ > /* 026 */ public GeneratedIteratorForCodegenStage5(Object[] references) { > /* 027 */ this.references = references; > /* 028 */ } > /* 029 */ > /* 030 */ public void init(int index, scala.collection.Iterator[] inputs) { > /* 031 */ partitionIndex = index; > /* 032 */ this.inputs = inputs; > /* 033 */ smj_leftInput = inputs[0]; > /* 034 */ smj_rightInput = inputs[1]; > /* 035 */ > /* 036 */ smj_matches = new > org.apache.spark.sql.execution.ExternalAppendOnlyUnsafeRowArray(2147483647, > 2147483647); > /* 037 */ smj_mutableStateArray[0] = new UnsafeRow(4); > /* 038 */ smj_mutableStateArray1[0] = new > org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(smj_mutableStateArray[0], > 0); > /* 039 */ smj_mutableStateArray2[0] = new > org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(smj_mutableStateArray1[0], > 4); > /* 040 */ > /* 041 */ } > /* 042 */ > /* 043 */ private void writeJoinRows() throws java.io.IOException { > /* 044 */ smj_mutableStateArray2[0].zeroOutNullBytes(); > /* 045 */ > /* 046 */ smj_mutableStateArray2[0].write(0, smj_value4); > /* 047 */ > /* 048 */ smj_mutableStateArray2[0].write(1, smj_value5); > /* 049 */ > /* 050 */ if (smj_isNull2) { > /* 051 */ smj_mutableStateArray2[0].setNullAt(2); > /* 052 */ } else { > /* 053 */ smj_mutableStateArray2[0].write(2, smj_value6); > /* 054 */ } > /* 055 */ > /* 056 */ if (smj_isNull3) { > /* 057 */ smj_mutableStateArray2[0].setNullAt(3); > /* 058 */ } else { > /* 059 */ smj_mutableStateArray2[0].write(3, s