package io.ddf.spark.etl;


import io.ddf.DDF;
import io.ddf.content.Schema;
import io.ddf.content.Schema.Column;
import io.ddf.etl.IHandleJoins;
import io.ddf.etl.Types.JoinType;
import io.ddf.exception.DDFException;
import io.ddf.misc.ADDFFunctionalGroupHandler;
import io.ddf.spark.util.SparkUtils;
import org.apache.spark.sql.DataFrame;

import java.util.HashSet;
import java.util.List;

public class JoinHandler extends ADDFFunctionalGroupHandler implements IHandleJoins {


  public JoinHandler(DDF theDDF) {
    super(theDDF);
    // TODO Auto-generated constructor stub
  }

  @Override
  public DDF join(DDF anotherDDF, JoinType joinType, List<String> byColumns, List<String> byLeftColumns,
      List<String> byRightColumns) throws DDFException {

    String leftTableName = getDDF().getTableName();
    String rightTableName = anotherDDF.getTableName();
    List<Column> rightColumns = anotherDDF.getSchema().getColumns();
    HashSet<String> rightColumNameSet = new HashSet<String>();
    for (Column m : rightColumns) {
      rightColumNameSet.add(m.getName());
    }

    String joinSqlCommand = "SELECT lt.*,%s FROM %s lt %s JOIN %s rt ON (%s)";
    String joinLeftSemiCommand = "SELECT lt.* FROM %s lt %s JOIN %s rt ON (%s)";
    String columnString = "";

    if (byColumns != null && !byColumns.isEmpty()) {
      for (int i = 0; i < byColumns.size(); i++) {
        columnString += String.format("lt.%s = rt.%s AND ", byColumns.get(i), byColumns.get(i));
        rightColumNameSet.remove(byColumns.get(i));
      }
    } else {
      if (byLeftColumns != null && byRightColumns != null && byLeftColumns.size() == byRightColumns.size()
          && !byLeftColumns.isEmpty()) {
        for (int i = 0; i < byLeftColumns.size(); i++) {
          columnString += String.format("lt.%s = rt.%s AND ", byLeftColumns.get(i), byRightColumns.get(i));
          rightColumNameSet.remove(byRightColumns.get(i));
        }
      } else {
        throw new DDFException(String.format("Left and right column specifications are missing or not compatible"),
            null);
      }
    }
    columnString = columnString.substring(0, columnString.length() - 5);

    // we will not select column that is already in left table
    String rightSelectColumns = "";
    for (String name : rightColumNameSet) {
      rightSelectColumns += String.format("rt.%s AS r_%s,", name, name);
    }
    rightSelectColumns = rightSelectColumns.substring(0, rightSelectColumns.length() - 1);

    try {
      if (joinType == JoinType.LEFTSEMI) {
        joinSqlCommand = String.format(joinLeftSemiCommand, leftTableName, joinType.getStringRepr(), rightTableName,
            columnString);
      } else {
        joinSqlCommand = String.format(joinSqlCommand, rightSelectColumns, leftTableName, joinType.getStringRepr(),
            rightTableName, columnString);
      }
    } catch (Exception ex) {
      throw new DDFException(String.format("Error while joinType.getStringRepr()"), null);
    }

    try {
      DDF resultDDF = this.getManager().sql2ddf(joinSqlCommand, "SparkSQL");
      return resultDDF;
    } catch (Exception e) {
      e.printStackTrace();
      throw new DDFException(String.format("Error while executing query QueryExecutionException"), e);
    }

  }
  @Override
  public DDF merge(DDF anotherDDF) throws DDFException {
    DataFrame rdd1 = ((DataFrame) this.getDDF().getRepresentationHandler().get(DataFrame.class));
    DataFrame rdd2 = ((DataFrame) anotherDDF.getRepresentationHandler().get(DataFrame.class));
    DataFrame newRDD = rdd1.unionAll(rdd2);
    Schema schema = SparkUtils.schemaFromDataFrame(newRDD);
    return this.getManager().newDDF(newRDD, new Class<?>[]{DataFrame.class},
            null, null, null, schema);
  }
}

@@ -923,24 +923,95 @@
 tColumns();%0A
+    List%3CColumn%3E leftColumns = getDDF().getSchema().getColumns();%0A    %0A
     HashSet%3C
@@ -1024,24 +1024,25 @@
 %3E rightColum
+n
 NameSet = ne
@@ -1106,32 +1106,33 @@
       rightColum
+n
 NameSet.add(m.ge
@@ -1319,21 +1319,28 @@
  String 
-colum
+joinConditio
 nString 
@@ -1450,37 +1450,44 @@
  i++) %7B%0A        
-colum
+joinConditio
 nString += Strin
@@ -1564,32 +1564,33 @@
       rightColum
+n
 NameSet.remove(b
@@ -1839,29 +1839,36 @@
 %7B%0A          
-colum
+joinConditio
 nString += S
@@ -1964,24 +1964,25 @@
   rightColum
+n
 NameSet.remo
@@ -2190,28 +2190,42 @@
     
-columnString = colum
+joinConditionString = joinConditio
 nStr
@@ -2241,21 +2241,28 @@
 ring(0, 
-colum
+joinConditio
 nString.
@@ -2275,16 +2275,44 @@
 () - 5);
+ //remove %22 AND %22 at the end
 %0A%0A    //
@@ -2404,16 +2404,240 @@
 s = %22%22;%0A
+    %0A    for (Column m : leftColumns) %7B%0A      if (rightColumnNameSet.contains(m.getName())) %7B%0A        rightColumnNameSet.remove(m.getName());%0A        rightColumnNameSet.add(String.format(%22r_%25s%22, m.getName()));%0A      %7D%0A    %7D%0A
     for 
@@ -2661,16 +2661,17 @@
 ghtColum
+n
 NameSet)
@@ -2725,24 +2725,10 @@
 t.%25s
- AS r_%25s,%22, name
+,%22
 , na
@@ -2828,16 +2828,41 @@
 () - 1);
+ // remove %22,%22 at the end
 %0A%0A    tr
@@ -3037,21 +3037,28 @@
         
-colum
+joinConditio
 nString)
@@ -3218,21 +3218,28 @@
 leName, 
-colum
+joinConditio
 nString)

remove 'r' prefix in resulting join column names