package io.ddf.jdbc.etl;


import io.ddf.DDF;
import io.ddf.content.Schema;
import io.ddf.content.SqlResult;
import io.ddf.content.SqlTypedResult;
import io.ddf.datasource.DataFormat;
import io.ddf.datasource.DataSourceDescriptor;
import io.ddf.etl.ASqlHandler;
import io.ddf.exception.DDFException;
import io.ddf.jdbc.JDBCDDF;
import io.ddf.jdbc.JDBCDDFManager;
import io.ddf.jdbc.JDBCUtils;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;

/**
 * Created by freeman on 7/15/15.
 */
public class SqlHandler extends ASqlHandler {
  public SqlHandler(DDF theDDF) {
    super(theDDF);
  }

  @Override public DDF sql2ddf(String command) throws DDFException {
    return this.sql2ddf(command, null, null, null);
  }

  @Override public DDF sql2ddf(String command, Schema schema) throws DDFException {
    return this.sql2ddf(command, schema, null, null);
  }

  @Override public DDF sql2ddf(String command, DataFormat dataFormat) throws DDFException {
    return this.sql2ddf(command, null, null, dataFormat);
  }

  @Override public DDF sql2ddf(String command, Schema schema, DataSourceDescriptor dataSource) throws DDFException {
    return this.sql2ddf(command, schema, dataSource, null);
  }

  @Override public DDF sql2ddf(String command, Schema schema, DataFormat dataFormat) throws DDFException {
    return this.sql2ddf(command, schema, null, dataFormat);
  }

  @Override public DDF sql2ddf(String command, Schema schema, DataSourceDescriptor dataSource, DataFormat dataFormat)
      throws DDFException {
    // TODO: We can easily run sql, but we should generate ddf from the SqlResult.
    this.getManager().log("sql2ddf in jdbc");
    Connection conn = this.getConn();
    try {
      Statement statement = conn.createStatement();
      statement.execute("create table tmptable as (" + command + ")");
      DDF ddf = new JDBCDDF((JDBCDDFManager)this.getManager(), null, null, null,
              "tmptable");
      this.getManager().addDDF(ddf);
      this.getManager().log("add ddf");
      return ddf;

    } catch (SQLException e) {
      e.printStackTrace();
      throw new DDFException("Unable to generate jdbc ddf");
    }
  }

  /**
   * @brief Get connection.
   * @return The connection.
   */
  private Connection  getConn() {
    return ((JDBCDDFManager)this.getManager()).getConn();
  }

  @Override
  public SqlResult sql(String command) throws DDFException {
    return this.sql(command, null);
  }

  @Override public SqlResult sql(String command, Integer maxRows) throws DDFException {
    return this.sql(command, null, null);
  }

  //TODO: handle dataSource
  @Override public SqlResult sql(String command, Integer maxRows, DataSourceDescriptor dataSource) throws DDFException {
    Connection conn = this.getConn();
    Statement statement;
    try {
      statement = conn.createStatement();
      ResultSet rs = statement.executeQuery(command);
      ResultSetMetaData rsmd = rs.getMetaData();
      int colSize = rsmd.getColumnCount();
      if (colSize == 0) {
        // TODO: How to return empty results?
        return new SqlResult(null, null);
      }
      List<Schema.Column> columnList = new ArrayList<Schema.Column>();
      for (int colIdx = 1; colIdx <= colSize; ++colIdx) {
        columnList.add(new Schema.Column(rsmd.getColumnName(colIdx),
            JDBCUtils.getDDFType(rsmd.getColumnType(colIdx))));
      }

      // TODO: check the table name please.
      // Generate the schema;
      Schema schema = new Schema(rsmd.getTableName(1), columnList);
      List<String> result = new ArrayList<String>();
      StringBuilder sb = new StringBuilder();

      while (rs.next()) {
        //sb.append(rs.getObject(1).toString());
        for (int colIdx = 1; colIdx <= colSize; ++colIdx) {
          Object obj = rs.getObject(colIdx);
          if (obj != null) {
            sb.append(obj.toString());
          } else {
            sb.append("null"); //TODO: how to append NULL
          }
          if (colIdx < colSize) {
            sb.append("\t");
          }
        }
        result.add(sb.toString());
        sb.delete(0, sb.length());
      }

      return new SqlResult(schema, result);
    } catch (SQLException e) {
      mLog.debug(e.getMessage());
      e.printStackTrace();
      throw new DDFException("Encouter error when running sql query using jdbc");
    }
  }

  @Override public SqlTypedResult sqlTyped(String command) throws DDFException {
    return null;
  }

  @Override public SqlTypedResult sqlTyped(String command, Integer maxRows) throws DDFException {
    return null;
  }

  @Override public SqlTypedResult sqlTyped(String command, Integer maxRows, DataSourceDescriptor dataSource)
      throws DDFException {
    return null;
  }

}

@@ -456,16 +456,88 @@
 il.List;
+%0Aimport java.util.Random;%0Aimport java.util.concurrent.ThreadLocalRandom;
 %0A%0A/**%0A *
@@ -618,16 +618,321 @@
 ndler %7B%0A
+%0A  static final String AB = %22ABCDEFGHIJKLMNOPQRSTUVWXYZ%22;%0A  static Random rnd = new Random();%0A%0A  String randomString( int len )%0A  %7B%0A    StringBuilder sb = new StringBuilder( len );%0A    for( int i = 0; i %3C len; i++ )%0A      sb.append( AB.charAt( rnd.nextInt(AB.length()) ) );%0A    return sb.toString();%0A  %7D%0A%0A
   public
@@ -2130,24 +2130,136 @@
 tatement();%0A
+      String randomTbName = this.randomString(24);%0A      this.getManager().log(%22create table %22 + randomTbName);%0A
       statem
@@ -2284,24 +2284,35 @@
 e table 
-tmptable
+%22+ randomTbName + %22
  as (%22 +
@@ -2321,16 +2321,30 @@
 ommand +
+%0A             
  %22)%22);%0A 
@@ -2440,19 +2440,20 @@
     
- %22tmptable%22
+randomTbName
 );%0A 
@@ -2546,17 +2546,16 @@
 rn ddf;%0A
-%0A
     %7D ca
@@ -2657,25 +2657,43 @@
 ate jdbc ddf
-%22
+ %22 + e.getMessage()
 );%0A    %7D%0A  %7D

fix bug