package org.embulk.filter.left_outer_join_json_table;

import org.embulk.spi.Column;
import org.embulk.spi.Exec;
import org.embulk.spi.Page;
import org.embulk.spi.PageBuilder;
import org.embulk.spi.PageOutput;
import org.embulk.spi.PageReader;
import org.embulk.spi.Schema;
import org.embulk.spi.time.TimestampParser;
import org.embulk.spi.type.Types;

import java.util.HashMap;
import java.util.List;

/**
 * Created by Civitaspo on 10/10/15.
 */
public class LeftOuterJoinJsonTableFilterFilteredPageOutput
    implements PageOutput
{
    private final PageReader pageReader;
    private final PageOutput pageOutput;
    private final Schema inputSchema;
    private final Schema outputSchema;
    private final Column joinBaseColumn;
    private final HashMap<String, HashMap<String, String>> jsonTable;
    private final List<Column> jsonColumns;
    private final HashMap<String, TimestampParser> timestampParserMap;

    LeftOuterJoinJsonTableFilterFilteredPageOutput(
            Schema inputSchema,
            Schema outputSchema,
            Column joinBaseColumn,
            HashMap<String, HashMap<String, String>> jsonTable,
            List<Column> jsonColumns,
            HashMap<String, TimestampParser> timestampParserMap,
            PageOutput pageOutput
            )
    {
        this.pageReader = new PageReader(inputSchema);
        this.pageOutput = pageOutput;
        this.inputSchema = inputSchema;
        this.outputSchema = outputSchema;
        this.joinBaseColumn = joinBaseColumn;
        this.jsonTable = jsonTable;
        this.jsonColumns = jsonColumns;
        this.timestampParserMap = timestampParserMap;
    }

    @Override
    public void add(Page page)
    {
        try (final PageBuilder pageBuilder = new PageBuilder(Exec.getBufferAllocator(), outputSchema, pageOutput)) {
            pageReader.setPage(page);

            while (pageReader.nextRecord()) {
                setInputValue(pageBuilder);
                setJoinedJsonValue(pageBuilder);
                pageBuilder.addRecord();
            }
            pageBuilder.finish();
        }

    }

    @Override
    public void finish()
    {
        pageOutput.finish();
    }

    @Override
    public void close()
    {
        pageReader.close();
        pageOutput.close();
    }

    private void setInputValue(PageBuilder pageBuilder) {
        for (Column inputColumn: inputSchema.getColumns()) {
            if (pageReader.isNull(inputColumn)) {
                pageBuilder.setNull(inputColumn);
                continue;
            }

            if (Types.STRING.equals(inputColumn.getType())) {
                pageBuilder.setString(inputColumn, pageReader.getString(inputColumn));
            }
            else if (Types.BOOLEAN.equals(inputColumn.getType())) {
                pageBuilder.setBoolean(inputColumn, pageReader.getBoolean(inputColumn));
            }
            else if (Types.DOUBLE.equals(inputColumn.getType())) {
                pageBuilder.setDouble(inputColumn, pageReader.getDouble(inputColumn));
            }
            else if (Types.LONG.equals(inputColumn.getType())) {
                pageBuilder.setLong(inputColumn, pageReader.getLong(inputColumn));
            }
            else if (Types.TIMESTAMP.equals(inputColumn.getType())) {
                pageBuilder.setTimestamp(inputColumn, pageReader.getTimestamp(inputColumn));
            }
        }
    }

    private void setJoinedJsonValue(PageBuilder pageBuilder) {
        for (Column jsonColumn: jsonColumns) {
            // get value from JSON Table
            String value = jsonTable.get(getCurrentJoinBaseColumnValue(pageReader, joinBaseColumn)).get(jsonColumn.getName());

            if (value == null) {
                pageBuilder.setNull(jsonColumn);
                continue;
            }

            if (Types.STRING.equals(jsonColumn.getType())) {
                pageBuilder.setString(jsonColumn, value);
            }
            else if (Types.BOOLEAN.equals(jsonColumn.getType())) {
                pageBuilder.setBoolean(jsonColumn, Boolean.parseBoolean(value));
            }
            else if (Types.DOUBLE.equals(jsonColumn.getType())) {
                pageBuilder.setDouble(jsonColumn, Double.parseDouble(value));
            }
            else if (Types.LONG.equals(jsonColumn.getType())) {
                pageBuilder.setLong(jsonColumn, Long.parseLong(value));
            }
            else if (Types.TIMESTAMP.equals(jsonColumn.getType())) {
                TimestampParser parser = timestampParserMap.get(jsonColumn.getName());
                pageBuilder.setTimestamp(jsonColumn, parser.parse(value));
            }
        }
    }

    private static String getCurrentJoinBaseColumnValue(PageReader pageReader, Column joinBaseColumn) {
        if (Types.STRING.equals(joinBaseColumn.getType())) {
            return pageReader.getString(joinBaseColumn);
        }
        else if (Types.BOOLEAN.equals(joinBaseColumn.getType())) {
            return String.valueOf(pageReader.getBoolean(joinBaseColumn));
        }
        else if (Types.DOUBLE.equals(joinBaseColumn.getType())) {
            return String.valueOf(pageReader.getDouble(joinBaseColumn));
        }
        else if (Types.LONG.equals(joinBaseColumn.getType())) {
            return String.valueOf(pageReader.getLong(joinBaseColumn));
        }
        else if (Types.TIMESTAMP.equals(joinBaseColumn.getType())) {
            return String.valueOf(pageReader.getTimestamp(joinBaseColumn));
        }

        return null;
    }
}

@@ -48,16 +48,58 @@
 table;%0A%0A
+import com.google.common.base.Throwables;%0A
 import o
@@ -571,16 +571,130 @@
 utput%0A%7B%0A
+    private final org.slf4j.Logger logger = Exec.getLogger(LeftOuterJoinJsonTableFilterFilteredPageOutput.class);%0A
     priv
@@ -3735,30 +3735,17 @@
 ing 
-value = jsonTable.get(
+rowKey = 
 getC
@@ -3797,16 +3797,189 @@
 eColumn)
+;%0A            if (rowKey == null) %7B%0A                pageBuilder.setNull(jsonColumn);%0A                continue;%0A            %7D%0A%0A            String value = jsonTable.get(rowKey
 ).get(js
@@ -3999,17 +3999,16 @@
 ame());%0A
-%0A
         
@@ -5088,18 +5088,107 @@
 eColumn)
- %7B
+%0A    %7B%0A        if (pageReader.isNull(joinBaseColumn)) %7B%0A            return null;%0A        %7D%0A
 %0A       
@@ -5913,33 +5913,119 @@
  %7D%0A%0A        
-return null
+throw Throwables.propagate(new Throwable(%22Unsupported Column Type: %22 + joinBaseColumn.getType()))
 ;%0A    %7D%0A%7D%0A

fix a bug: join null