[SPARK-13123][SQL] Implement whole state codegen for sort.

[nongli]

This does the simplest thing of just assembly a row on consume and driving the
underlying external sorter object.

[nongli]

/* 001 */ public Object generate(Object[] references) {
/* 002 */   return new GeneratedIterator(references);
/* 003 */ }
/* 004 */ 
/* 005 */ class GeneratedIterator extends org.apache.spark.sql.execution.BufferedRowIterator {
/* 006 */   
/* 007 */   private Object[] references;
/* 008 */   private boolean Sort_needToSort0;
/* 009 */   private org.apache.spark.sql.execution.Sort Sort_plan1;
/* 010 */   private org.apache.spark.sql.execution.UnsafeExternalRowSorter Sort_sorter2;
/* 011 */   private scala.collection.Iterator<UnsafeRow> Sort_sortedIter3;
/* 012 */   private UnsafeRow Sort_result24;
/* 013 */   private org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder Sort_holder25;
/* 014 */   private org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter Sort_rowWriter26;
/* 015 */   
/* 016 */   private void Sort_addToSorter4() throws java.io.IOException {
/* 017 */     
/* 018 */     while (input.hasNext()) {
/* 019 */       InternalRow InputAdapter_row5 = (InternalRow) input.next();
/* 020 */       /* input[0, int] */
/* 021 */       boolean InputAdapter_isNull6 = InputAdapter_row5.isNullAt(0);
/* 022 */       int InputAdapter_value7 = InputAdapter_isNull6 ? -1 : (InputAdapter_row5.getInt(0));
/* 023 */       /* input[1, string] */
/* 024 */       boolean InputAdapter_isNull8 = InputAdapter_row5.isNullAt(1);
/* 025 */       UTF8String InputAdapter_value9 = InputAdapter_isNull8 ? null : (InputAdapter_row5.getUTF8String(1));
/* 026 */       
/* 027 */       /* (input[0, int] < 20) */
/* 028 */       /* input[0, int] */
/* 029 */       
/* 030 */       /* 20 */
/* 031 */       
/* 032 */       boolean Filter_value11 = false;
/* 033 */       Filter_value11 = InputAdapter_value7 < 20;
/* 034 */       if (!false && Filter_value11) {
/* 035 */         
/* 036 */         /* input[0, int] */
/* 037 */         
/* 038 */         /* input[1, string] */
/* 039 */         
/* 040 */         
/* 041 */         // Convert the input attributes to an UnsafeRow and add it to the sorter
/* 042 */         
/* 043 */         Sort_holder25.reset();
/* 044 */         
/* 045 */         Sort_rowWriter26.zeroOutNullBytes();
/* 046 */         
/* 047 */         /* input[0, int] */
/* 048 */         
/* 049 */         if (InputAdapter_isNull6) {
/* 050 */           Sort_rowWriter26.setNullAt(0);
/* 051 */         } else {
/* 052 */           Sort_rowWriter26.write(0, InputAdapter_value7);
/* 053 */         }
/* 054 */         
/* 055 */         /* input[1, string] */
/* 056 */         
/* 057 */         if (InputAdapter_isNull8) {
/* 058 */           Sort_rowWriter26.setNullAt(1);
/* 059 */         } else {
/* 060 */           Sort_rowWriter26.write(1, InputAdapter_value9);
/* 061 */         }
/* 062 */         Sort_result24.setTotalSize(Sort_holder25.totalSize());
/* 063 */         
/* 064 */         Sort_sorter2.insertRow(Sort_result24);
/* 065 */         
/* 066 */       }
/* 067 */       
/* 068 */     }
/* 069 */     
/* 070 */   }
/* 071 */   
/* 072 */   public GeneratedIterator(Object[] references) {
/* 073 */     this.references = references;
/* 074 */     Sort_needToSort0 = true;
/* 075 */     this.Sort_plan1 = (org.apache.spark.sql.execution.Sort) references[0];
/* 076 */     Sort_sorter2 = Sort_plan1.createSorter();
/* 077 */     
/* 078 */     Sort_result24 = new UnsafeRow(2);
/* 079 */     this.Sort_holder25 = new org.apache.spark.sql.catalyst.expressions.codegen.BufferHolder(Sort_result24, 32);
/* 080 */     this.Sort_rowWriter26 = new org.apache.spark.sql.catalyst.expressions.codegen.UnsafeRowWriter(Sort_holder25, 2);
/* 081 */   }
/* 082 */   
/* 083 */   protected void processNext() throws java.io.IOException {
/* 084 */     if (Sort_needToSort0) {
/* 085 */       Sort_addToSorter4();
/* 086 */       Sort_sortedIter3 = Sort_sorter2.sort();
/* 087 */       Sort_needToSort0 = false;
/* 088 */     }
/* 089 */     
/* 090 */     while (Sort_sortedIter3.hasNext()) {
/* 091 */       UnsafeRow Sort_outputRow29 = (UnsafeRow)Sort_sortedIter3.next();
/* 092 */       System.out.println(Sort_outputRow29);
/* 093 */       currentRow = Sort_outputRow29;
/* 094 */       return;
/* 095 */     }
/* 096 */   }
/* 097 */ }

[SparkQA]

Test build #50511 has finished for PR 11008 at commit 11e26c9.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • public final class UnsafeExternalRowSorter

[rxin]

this is pretty ghetto... (although i understand maybe it's the simplest way to implement this)

[nongli]

Why? This is the state that needs to be kept between the two member functions in this class.

[rxin]

it's ok here as discussed offline. i just found mutable state in here as a way to pass variable names through pretty brittle. maybe good to have a more general abstraction for this in codegen, but not that big of a deal right now/

[SparkQA]

Test build #50584 has finished for PR 11008 at commit 564a5b3.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[davies]

If the child can produce UnsafeRow (for example, Exchange), we should have a way to avoid this unpack and pack again, or we will see regression (generated version slower than non-generated).

I think we can pass the variable for input row into doCosume, could be null. It's better to do this after #11274 , then we don't need to worry about whether should we create variables for input or not.

[davies]

@nongli Can you close this PR?