package org.embulk.filter.split_column;

import org.apache.commons.lang3.StringUtils;
import org.embulk.config.Config;
import org.embulk.config.ConfigDefault;
import org.embulk.config.ConfigDiff;
import org.embulk.config.ConfigSource;
import org.embulk.config.Task;
import org.embulk.config.TaskSource;
import org.embulk.spi.Column;
import org.embulk.spi.ColumnConfig;
import org.embulk.spi.Exec;
import org.embulk.spi.FilterPlugin;
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.SchemaConfig;
import org.embulk.spi.time.TimestampParser;
import org.embulk.spi.type.Type;
import org.embulk.spi.type.Types;
import org.embulk.spi.util.Timestamps;

import org.embulk.spi.DataException;
import org.embulk.spi.time.TimestampParseException;

import com.google.common.base.Optional;
import com.google.common.collect.ImmutableList;

public class SplitColumnFilterPlugin
        implements FilterPlugin
{
    public interface PluginTask
            extends Task, TimestampParser.Task
    {
        @Config("delimiter")
        @ConfigDefault("\",\"")
        public String getDelimiter();

        @Config("target_key")
        public String getTargetKey();

        @Config("output_columns")
        public SchemaConfig getOutputColumns();
    }

    @Override
    public void transaction(ConfigSource config, Schema inputSchema,
            FilterPlugin.Control control)
    {
        PluginTask task = config.loadConfig(PluginTask.class);

		ImmutableList.Builder<Column> builder = ImmutableList.builder();
        String targetColumnName = task.getTargetKey();
        int i = 0;
		for (Column inputColumn: inputSchema.getColumns()) {
            String columnName = inputColumn.getName();
            if (columnName.equals(targetColumnName)) {
                continue;
            }
		    Column outputColumn = new Column(i++, inputColumn.getName(), inputColumn.getType());
            builder.add(outputColumn);
        }
        for (ColumnConfig outputColumnConfig : task.getOutputColumns().getColumns()) {
            Column outputColumn = outputColumnConfig.toColumn(i++);
            builder.add(outputColumn);
        }
        Schema outputSchema = new Schema(builder.build());
        control.run(task.dump(), outputSchema);
    }

    @Override
    public PageOutput open(TaskSource taskSource, final Schema inputSchema,
            final Schema outputSchema, final PageOutput output)
    {
        final PluginTask task = taskSource.loadTask(PluginTask.class);
        final Column targetColumn = inputSchema.lookupColumn(task.getTargetKey());
        final TimestampParser[] timestampParsers = Timestamps.newTimestampColumnParsers(task, task.getOutputColumns());

        return new PageOutput() {
            private PageReader reader = new PageReader(inputSchema);
            private PageBuilder builder = new PageBuilder(Exec.getBufferAllocator(), outputSchema, output);

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

            @Override
            public void close() {
                builder.close();
            }

            @Override
            public void add(Page page) {
                reader.setPage(page);
                while (reader.nextRecord()) {
                    String[] words = StringUtils.split(reader.getString(targetColumn),task.getDelimiter());
					int i = 0;
					SchemaConfig outputSchemaConfig = task.getOutputColumns();
                    if (outputSchemaConfig.size() != words.length) {
                        String message = String.format("outputColum has %d columns but value was separated in %d",
                                outputSchemaConfig.size(),
                                words.length
                                );
                        throw new SplitColumnValidateException(message);
                    }
					for (ColumnConfig outputColumnConfig: outputSchemaConfig.getColumns()) {
						Column outputColumn = outputSchema.lookupColumn(outputColumnConfig.getName());
                        Type outputColumnType = outputColumn.getType();
                        if (Types.STRING.equals(outputColumnType)) {
                            builder.setString(outputColumn, words[i++]);
                        } else if (Types.BOOLEAN.equals(outputColumnType)) {
                            builder.setBoolean(outputColumn, Boolean.parseBoolean(words[i++]));
                        } else if (Types.DOUBLE.equals(outputColumnType)) {
                            builder.setDouble(outputColumn, Double.parseDouble(words[i++]));
                        } else if (Types.LONG.equals(outputColumnType)) {
                            builder.setLong(outputColumn, Long.parseLong(words[i++]));
                        } else if (Types.TIMESTAMP.equals(outputColumnType)) {
                            builder.setTimestamp(outputColumn, timestampParsers[i].parse(words[i]));
                            i++;
                        }
                    }
                    for (Column column: inputSchema.getColumns()) {
                        if (column.getName().equals(targetColumn.getName())) {
                            continue;
                        }
                        if (reader.isNull(column)) {
                            builder.setNull(column);
                            continue;
                        }
                        add_builder(column);
                    }
                    builder.addRecord();
                }
            }
            private void add_builder(Column column) {
                if (Types.STRING.equals(column.getType())) {
                    builder.setString(column, reader.getString(column));
                } else if (Types.BOOLEAN.equals(column.getType())) {
                    builder.setBoolean(column, reader.getBoolean(column));
                } else if (Types.DOUBLE.equals(column.getType())) {
                    builder.setDouble(column, reader.getDouble(column));
                } else if (Types.LONG.equals(column.getType())) {
                    builder.setLong(column, reader.getLong(column));
                } else if (Types.TIMESTAMP.equals(column.getType())) {
                    builder.setTimestamp(column, reader.getTimestamp(column));
                }
            }

		};
    }

	static class SplitColumnValidateException
            extends DataException
    {
        SplitColumnValidateException(String message)
        {
            super(message);
        }
    }
}

@@ -3545,24 +3545,74 @@
 tColumns();%0A
+                    // TODO: support skipping row%0A
             
@@ -4017,16 +4017,112 @@
       %7D%0A
+                    // TODO: support default value%0A                    // TODO: throw exception%0A
 %09%09%09%09%09for
@@ -5767,32 +5767,76 @@
 %7D%0A            %7D%0A
+            // TODO: use embulk-core system%0A
             priv

add TODO