Github user davies commented on the pull request:
https://github.com/apache/spark/pull/11248#issuecomment-185583137
Generated code for
```
val df1 = sqlContext.range(N).selectExpr(s"id as k1")
val df2 = sqlContext.range(N).selectExpr(s"id as k2")
df1.join(df2, col("k1") === col("k2")).count()
```
```
/* 001 */
/* 002 */ public Object generate(Object[] references) {
/* 003 */ return new GeneratedIterator(references);
/* 004 */ }
/* 005 */
/* 006 */ /** Codegened pipeline for:
/* 007 */ * TungstenAggregate(key=[],
functions=[(count(1),mode=Partial,isDistinct=false)], output=[count#191L])
/* 008 */ +- Project
/* 009 */ +- SortMergeJoin [k1#185L], [k2#187L], None
/* 010 */ :- INPUT
/* 011 */ +- INPUT
/* 012 */ */
/* 013 */ class GeneratedIterator extends
org.apache.spark.sql.execution.BufferedRowIterator {
/* 014 */ private Object[] references;
/* 015 */ private boolean agg_initAgg;
/* 016 */ private boolean agg_bufIsNull;
/* 017 */ private long agg_bufValue;
/* 018 */ private scala.collection.Iterator smj_leftInput;
/* 019 */ private scala.collection.Iterator smj_rightInput;
/* 020 */ private InternalRow smj_leftRow;
/* 021 */ private InternalRow smj_rightRow;
/* 022 */ private java.util.LinkedList smj_matches;
/* 023 */ private boolean smj_isNull;
/* 024 */ private long smj_value;
/* 025 */ private boolean smj_isNull2;
/* 026 */ private long smj_value2;
/* 027 */ private org.apache.spark.sql.execution.metric.LongSQLMetric
smj_numOutputRows;
/* 028 */ private
org.apache.spark.sql.execution.metric.LongSQLMetricValue smj_metricValue;
/* 029 */ private org.apache.spark.sql.execution.metric.LongSQLMetric
agg_numOutputRows;
/* 030 */ private
org.apache.spark.sql.execution.metric.LongSQLMetricValue agg_metricValue;
/* 031 */ private UnsafeRow agg_result;
/* 032 */ private
org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder agg_holder;
/* 033 */ private
org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter agg_rowWriter;
/* 034 */
/* 035 */ public GeneratedIterator(Object[] references) {
/* 036 */ this.references = references;
/* 037 */ }
/* 038 */
/* 039 */ public void init(scala.collection.Iterator inputs[]) {
/* 040 */ agg_initAgg = false;
/* 041 */
/* 042 */ smj_leftInput = inputs[0];
/* 043 */ smj_rightInput = inputs[1];
/* 044 */
/* 045 */ smj_rightRow = null;
/* 046 */ smj_matches = new java.util.LinkedList();
/* 047 */
/* 048 */ this.smj_numOutputRows =
(org.apache.spark.sql.execution.metric.LongSQLMetric) references[0];
/* 049 */ smj_metricValue =
(org.apache.spark.sql.execution.metric.LongSQLMetricValue)
smj_numOutputRows.localValue();
/* 050 */ this.agg_numOutputRows =
(org.apache.spark.sql.execution.metric.LongSQLMetric) references[1];
/* 051 */ agg_metricValue =
(org.apache.spark.sql.execution.metric.LongSQLMetricValue)
agg_numOutputRows.localValue();
/* 052 */ agg_result = new UnsafeRow(1);
/* 053 */ this.agg_holder = new
org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(agg_result, 0);
/* 054 */ this.agg_rowWriter = new
org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(agg_holder,
1);
/* 055 */ }
/* 056 */
/* 057 */ private void agg_doAggregateWithoutKey() throws
java.io.IOException {
/* 058 */ // initialize aggregation buffer
/* 059 */
/* 060 */ agg_bufIsNull = false;
/* 061 */ agg_bufValue = 0L;
/* 062 */
/* 063 */ while (findNextInnerJoinRows(smj_leftInput, smj_rightInput)) {
/* 064 */ /* input[0, bigint] */
/* 065 */ boolean smj_isNull4 = smj_leftRow.isNullAt(0);
/* 066 */ long smj_value4 = smj_isNull4 ? -1L :
(smj_leftRow.getLong(0));
/* 067 */ while (!smj_matches.isEmpty()) {
/* 068 */ InternalRow smj_rightRow1 = (InternalRow)
smj_matches.remove();
/* 069 */ /* input[0, bigint] */
/* 070 */ boolean smj_isNull5 = smj_rightRow1.isNullAt(0);
/* 071 */ long smj_value5 = smj_isNull5 ? -1L :
(smj_rightRow1.getLong(0));
/* 072 */ smj_metricValue.add(1);
/* 073 */
/* 074 */ // do aggregate
/* 075 */ /* (input[0, bigint] + 1) */
/* 076 */ long agg_value1 = -1L;
/* 077 */ agg_value1 = agg_bufValue + 1L;
/* 078 */ // update aggregation buffer
/* 079 */ agg_bufIsNull = false;
/* 080 */ agg_bufValue = agg_value1;
/* 081 */
/* 082 */ }
/* 083 */ }
/* 084 */
/* 085 */ }
/* 086 */
/* 087 */ private boolean findNextInnerJoinRows(
/* 088 */ scala.collection.Iterator leftIter,
/* 089 */ scala.collection.Iterator rightIter) {
/* 090 */ smj_leftRow = null;
/* 091 */ smj_matches.clear();
/* 092 */ while (smj_leftRow == null) {
/* 093 */ if (!leftIter.hasNext()) return false;
/* 094 */ smj_leftRow = (InternalRow) leftIter.next();
/* 095 */
/* 096 */ /* input[0, bigint] */
/* 097 */ boolean smj_isNull1 = smj_leftRow.isNullAt(0);
/* 098 */ long smj_value1 = smj_isNull1 ? -1L :
(smj_leftRow.getLong(0));
/* 099 */ smj_isNull = smj_isNull1;
/* 100 */ smj_value = smj_value1;
/* 101 */
/* 102 */ if (smj_isNull) {
/* 103 */ smj_leftRow = null;
/* 104 */ continue;
/* 105 */ }
/* 106 */
/* 107 */ do {
/* 108 */ if (smj_rightRow == null) {
/* 109 */ if (!rightIter.hasNext()) {
/* 110 */ return !smj_matches.isEmpty();
/* 111 */ }
/* 112 */ smj_rightRow = (InternalRow) rightIter.next();
/* 113 */ }
/* 114 */
/* 115 */ /* input[0, bigint] */
/* 116 */ boolean smj_isNull3 = smj_rightRow.isNullAt(0);
/* 117 */ long smj_value3 = smj_isNull3 ? -1L :
(smj_rightRow.getLong(0));
/* 118 */ smj_isNull2 = smj_isNull3;
/* 119 */ smj_value2 = smj_value3;
/* 120 */
/* 121 */ if (smj_isNull2) {
/* 122 */ smj_rightRow = null;
/* 123 */ continue;
/* 124 */ }
/* 125 */
/* 126 */ int smj_comp = 0;
/* 127 */ if (smj_comp == 0) {
/* 128 */ smj_comp = (smj_value > smj_value2 ? 1 : smj_value <
smj_value2 ? -1 : 0);
/* 129 */ }
/* 130 */
/* 131 */ if (smj_comp > 0) {
/* 132 */ smj_rightRow = null;
/* 133 */ } else if (smj_comp < 0) {
/* 134 */ if (!smj_matches.isEmpty()) return true;
/* 135 */ smj_leftRow = null;
/* 136 */ } else {
/* 137 */ smj_matches.add(smj_rightRow.copy());
/* 138 */ smj_rightRow = null;;
/* 139 */ }
/* 140 */ } while (smj_leftRow != null);
/* 141 */ }
/* 142 */ return false; // unreachable
/* 143 */ }
/* 144 */
/* 145 */ protected void processNext() throws java.io.IOException {
/* 146 */ if (!agg_initAgg) {
/* 147 */ agg_initAgg = true;
/* 148 */ agg_doAggregateWithoutKey();
/* 149 */
/* 150 */ // output the result
/* 151 */
/* 152 */ agg_metricValue.add(1);
/* 153 */ agg_rowWriter.zeroOutNullBytes();
/* 154 */
/* 155 */ if (agg_bufIsNull) {
/* 156 */ agg_rowWriter.setNullAt(0);
/* 157 */ } else {
/* 158 */ agg_rowWriter.write(0, agg_bufValue);
/* 159 */ }
/* 160 */ append(agg_result.copy());
/* 161 */ }
/* 162 */ }
/* 163 */ }
```
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at [email protected] or file a JIRA ticket
with INFRA.
---
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
