package org.embulk.filter;

import java.text.Normalizer;
import java.util.Set;

import com.google.common.collect.ImmutableSet;
import org.embulk.config.Config;
import org.embulk.config.ConfigDefault;
import org.embulk.config.ConfigException;
import org.embulk.config.ConfigSource;
import org.embulk.config.Task;
import org.embulk.config.TaskSource;
import org.embulk.spi.Column;
import org.embulk.spi.ColumnVisitor;
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;

public class NormalizeFilterPlugin
        implements FilterPlugin
{
    public interface PluginTask
            extends Task
    {
        @Config("columns")
        Set<String> getColumns();

        @Config("form")
        @ConfigDefault("\"NFKC\"")
        Normalizer.Form getForm();

        @Config("trim")
        @ConfigDefault("true")
        boolean getTrim();
    }

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

        ImmutableSet.Builder<String> builder = ImmutableSet.builder();
        for (Column inputColumn : inputSchema.getColumns()) {
            builder.add(inputColumn.getName());
        }
        Set<String> inputColumnNames = builder.build();
        for (String name : task.getColumns()) {
            if (!inputColumnNames.contains(name)) {
                throw new ConfigException("Column " + name + " is not included in input columns");
            }
        }

        control.run(task.dump(), inputSchema);
    }

    @Override
    public PageOutput open(TaskSource taskSource, final Schema inputSchema,
            final Schema outputSchema, final PageOutput output)
    {
        final PluginTask task = taskSource.loadTask(PluginTask.class);
        final Set<String> normalizeColumns = task.getColumns();
        final Normalizer.Form form = task.getForm();
        final boolean trim = task.getTrim();

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

            @Override
            public void add(Page page)
            {
                reader.setPage(page);
                ColumnVisitor visitor = new NormalizeColumnVisitor(builder);
                while (reader.nextRecord()) {
                    inputSchema.visitColumns(visitor);
                    builder.addRecord();
                }
            }

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

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

            class NormalizeColumnVisitor
                    implements ColumnVisitor
            {
                private final PageBuilder builder;

                public NormalizeColumnVisitor(PageBuilder builder)
                {
                    this.builder = builder;
                }

                @Override
                public void booleanColumn(Column column)
                {
                    if (reader.isNull(column)) {
                        builder.setNull(column);
                    } else {
                        builder.setBoolean(column, reader.getBoolean(column));
                    }
                }

                @Override
                public void longColumn(Column column)
                {
                    if (reader.isNull(column)) {
                        builder.setNull(column);
                    } else {
                        builder.setLong(column, reader.getLong(column));
                    }
                }

                @Override
                public void doubleColumn(Column column)
                {
                    if (reader.isNull(column)) {
                        builder.setNull(column);
                    } else {
                        builder.setDouble(column, reader.getDouble(column));
                    }
                }

                @Override
                public void stringColumn(Column column)
                {
                    if (reader.isNull(column)) {
                        builder.setNull(column);
                    } else {
                        if (normalizeColumns.contains(column.getName())) {
                            String normalized = Normalizer.normalize(reader.getString(column), form);
                            if (trim) {
                                normalized = normalized.trim();
                            }
                            if (normalized.isEmpty()) {
                                builder.setNull(column);
                            } else {
                                builder.setString(column, normalized);
                            }
                        } else {
                            builder.setString(column, reader.getString(column));
                        }
                    }
                }

                @Override
                public void timestampColumn(Column column)
                {
                    if (reader.isNull(column)) {
                        builder.setNull(column);
                    } else {
                        builder.setTimestamp(column, reader.getTimestamp(column));
                    }
                }
            }
        };
    }
}

@@ -46,24 +46,46 @@
 Normalizer;%0A
+import java.util.Map;%0A
 import java.
@@ -137,19 +137,19 @@
 mmutable
-Set
+Map
 ;%0Aimport
@@ -655,16 +655,88 @@
 .Schema;
+%0Aimport org.embulk.spi.type.StringType;%0Aimport org.embulk.spi.type.Type;
 %0A%0Apublic
@@ -1307,19 +1307,19 @@
 mmutable
-Set
+Map
 .Builder
@@ -1325,16 +1325,22 @@
 r%3CString
+, Type
 %3E builde
@@ -1352,19 +1352,19 @@
 mmutable
-Set
+Map
 .builder
@@ -1449,19 +1449,19 @@
 builder.
-add
+put
 (inputCo
@@ -1474,16 +1474,39 @@
 etName()
+, inputColumn.getType()
 );%0A     
@@ -1518,26 +1518,32 @@
         
-Set
+Map
 %3CString
+, Type
 %3E inputC
@@ -1551,17 +1551,22 @@
 lumnName
-s
+ToType
  = build
@@ -1641,144 +1641,435 @@
     
-if (!inputColumnNames.contains(name)) %7B%0A                throw new ConfigException(%22Column %22 + name + %22 is not included in input columns%22
+Type inputColumnType = inputColumnNameToType.get(name);%0A            if (inputColumnType == null) %7B%0A                throw new ConfigException(String.format(%0A                        %22Column %25s is not included in input schema.%22, name));%0A            %7D%0A            if (!(inputColumnType instanceof StringType)) %7B%0A                throw new ConfigException(String.format(%0A                        %22Column %25s is not string type.%22, name)
 );%0A 
@@ -3315,32 +3315,64 @@
 )%0A            %7B%0A
+                reader.close();%0A
                 

Add string type check