package tech.tablesaw.join;

import java.time.LocalDate;

import com.google.common.collect.Streams;

import tech.tablesaw.api.DateColumn;
import tech.tablesaw.api.Table;
import tech.tablesaw.columns.Column;
import tech.tablesaw.index.DateIndex;

public class DataFrameJoiner {

  private final Table table;
  private final Column column;
  
  public DataFrameJoiner(Table table, String column) {
    this.table = table;
    this.column = table.column(column);
  }

  public Table inner(Table table2, String col2Name) {
    DateIndex index;
    if (column instanceof DateColumn) {
      index = new DateIndex(table2.dateColumn(col2Name));
    } else {
      throw new IllegalArgumentException(
          "Only joining on date columns is supported so far");
    }

    Table result = emptyTableFromColumns(table, table2, col2Name);

    DateColumn col1 = (DateColumn) column;
    for (int i = 0; i < col1.size(); i++) {
      LocalDate date = col1.get(i);
      Table table1Rows = table.selectRow(i);
      Table table2Rows = table2.selectWhere(index.get(date));
      table2Rows.removeColumns(col2Name);
      crossProduct(result, table1Rows, table2Rows);
    }

    return result;
  }

  private Table emptyTableFromColumns(Table table1, Table table2, String col2Name) {
    Column[] cols = Streams.concat(
            table1.columns().stream(),
            table2.columns().stream().filter(c -> !c.name().equals(col2Name))
       ).map(col -> col.columnMetadata().createColumn()).toArray(size -> new Column[size]);
    return Table.create(table1.name(), cols);
  }

  private void crossProduct(Table destination, Table table1, Table table2) {
    for (int c = 0; c < table1.columnCount() + table2.columnCount(); c++) {
      for (int r = 0; r < table1.rowCount() * table2.rowCount(); r++) {
        if (c < table1.columnCount()) {
          destination.column(c).appendCell(table1.get(r, c));          
        } else {
          destination.column(c).appendCell(table2.get(r, c - table1.columnCount()));
        }
      }    
    }
  }

}

@@ -49,16 +49,76 @@
 calDate;
+%0Aimport java.time.LocalDateTime;%0Aimport java.time.LocalTime;
 %0A%0Aimport
@@ -220,82 +220,240 @@
 api.
-Table;%0Aimport tech.tablesaw.columns.Column;%0Aimport tech.tablesaw.index.Dat
+DateTimeColumn;%0Aimport tech.tablesaw.api.Table;%0Aimport tech.tablesaw.api.TimeColumn;%0Aimport tech.tablesaw.columns.Column;%0Aimport tech.tablesaw.index.DateIndex;%0Aimport tech.tablesaw.index.DateTimeIndex;%0Aimport tech.tablesaw.index.Tim
 eInd
@@ -738,23 +738,69 @@
     
-DateIndex index
+Table result = emptyTableFromColumns(table, table2, col2Name)
 ;%0A  
@@ -843,16 +843,26 @@
 %7B%0A      
+DateIndex 
 index = 
@@ -913,129 +913,629 @@
     
-%7D else %7B%0A      throw new IllegalArgumentException(
+  DateColumn col1 = (DateColumn) column;%0A      for (int i = 0; i %3C col1.size(); i++) %7B%0A        LocalDate value = col1.get(i);%0A        Table table1Rows = table.selectRow(i);%0A        Table table2Rows = table2.selectWhere(index.get(value));%0A        table2Rows.removeColumns(col2Name);%0A        crossProduct(result, table1Rows, table2Rows);
 %0A      
+%7D%0A
     
-%22Only joining on date columns is supported so far%22
+%7D else if (column instanceof DateTimeColumn) %7B%0A      DateTimeIndex index = new DateTimeIndex(table2.dateTimeColumn(col2Name));%0A      DateTimeColumn col1 = (DateTimeColumn) column;%0A      for (int i = 0; i %3C col1.size(); i++) %7B%0A        LocalDateTime value = col1.get(i
 );%0A    
-%7D%0A%0A
     
@@ -1544,54 +1544,315 @@
 ble 
-result = emptyTableFromColumns(table, table2, 
+table1Rows = table.selectRow(i);%0A        Table table2Rows = table2.selectWhere(index.get(value));%0A        table2Rows.removeColumns(col2Name);%0A        crossProduct(result, table1Rows, table2Rows);%0A      %7D%0A    %7D else if (column instanceof TimeColumn) %7B%0A      TimeIndex index = new TimeIndex(table2.timeColumn(
 col2
@@ -1860,18 +1860,20 @@
 ame)
+)
 ;
-%0A
 %0A    
-Dat
+  Tim
 eCol
@@ -1876,35 +1876,35 @@
 eColumn col1 = (
-Dat
+Tim
 eColumn) column;
@@ -1900,24 +1900,26 @@
 mn) column;%0A
+  
     for (int
@@ -1960,21 +1960,24 @@
     
+  
 Local
-Date dat
+Time valu
 e = 
@@ -1985,24 +1985,26 @@
 ol1.get(i);%0A
+  
       Table 
@@ -2038,24 +2038,26 @@
 w(i);%0A      
+ 
+ 
 Table table2
@@ -2096,16 +2096,19 @@
 get(
-dat
+valu
 e));%0A
+  
     
@@ -2147,24 +2147,26 @@
 ame);%0A      
+ 
+ 
 crossProduct
@@ -2195,24 +2195,155 @@
 able2Rows);%0A
+      %7D%0A    %7D else %7B%0A      throw new IllegalArgumentException(%0A          %22Only joining on date-like columns is supported so far%22);%0A
     %7D%0A%0A    r

Implement joins on LocalTime and LocalDateTime