package org.embulk.filter.distinct;

import com.google.common.base.Strings;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.ObjectArrays;
import com.google.common.collect.Sets;
import org.embulk.filter.distinct.DistinctFilterPlugin.PluginTask;
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.type.Types;
import org.slf4j.Logger;

import java.util.List;
import java.util.Set;

/**
 * Created by takahiro.nakayama on 12/6/15.
 */
class FilteredPageOutput
    implements PageOutput
{
    private final static Logger logger = Exec.getLogger(FilteredPageOutput.class);
    private final PageReader pageReader;
    private final PageBuilder pageBuilder;
    private final ColumnVisitorImpl visitor;
    private final Schema outputSchema;
    private final List<Column> distinctColumns;

    private final static Set<List<Object>> filter = Sets.newConcurrentHashSet();

    FilteredPageOutput(PluginTask task, Schema inputSchema,
                       Schema outputSchema, PageOutput pageOutput)
    {
        this.pageReader = new PageReader(inputSchema);
        this.pageBuilder = new PageBuilder(Exec.getBufferAllocator(), outputSchema, pageOutput);
        this.visitor = new ColumnVisitorImpl(pageReader, pageBuilder);
        this.outputSchema = outputSchema;
        this.distinctColumns = task.getDistinctColumns();
    }

    @Override
    public void add(Page page)
    {
        pageReader.setPage(page);

        while (pageReader.nextRecord()) {
            if (filter.add(getCurrentDistinctKey())) {
                outputSchema.visitColumns(visitor);
                pageBuilder.addRecord();
            }
        }
    }

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

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

    private List<Object> getCurrentDistinctKey()
    {
        ImmutableList.Builder<Object> builder = ImmutableList.builder();
        for (Column distinctColumn : distinctColumns) {
            if (!pageReader.isNull(distinctColumn)) {
                if (Types.BOOLEAN.equals(distinctColumn.getType())) {
                    builder.add(pageReader.getBoolean(distinctColumn));
                }
                else if (Types.DOUBLE.equals(distinctColumn.getType())) {
                    builder.add(pageReader.getDouble(distinctColumn));
                }
                else if (Types.LONG.equals(distinctColumn.getType())) {
                    builder.add(pageReader.getLong(distinctColumn));
                }
                else if (Types.STRING.equals(distinctColumn.getType())) {
                    builder.add(pageReader.getString(distinctColumn));
                }
                else if (Types.TIMESTAMP.equals(distinctColumn.getType())) {
                    builder.add(pageReader.getTimestamp(distinctColumn));
                }
                else {
                    throw new RuntimeException("unsupported type: " + distinctColumn.getType());
                }
            }
        }

        return builder.build();
    }
}

@@ -30,16 +30,56 @@
 tinct;%0A%0A
+import com.google.common.base.Optional;%0A
 import c
@@ -2286,9 +2286,8 @@
 if (
-!
 page
@@ -2331,24 +2331,87 @@
             
+builder.add(Optional.absent());%0A            %7D%0A            else 
 if (Types.BO
@@ -2444,36 +2444,32 @@
 n.getType())) %7B%0A
-    
                 
@@ -2524,36 +2524,32 @@
 ));%0A            
-    
 %7D%0A              
@@ -2538,36 +2538,32 @@
   %7D%0A            
- 
-   
 else if (Types.D
@@ -2612,36 +2612,32 @@
                 
-    
 builder.add(page
@@ -2679,34 +2679,26 @@
             
-   
- 
 %7D%0A
-    
             
@@ -2745,36 +2745,32 @@
 n.getType())) %7B%0A
-    
                 
@@ -2826,34 +2826,26 @@
             
-    %7D%0A    
+%7D%0A
             
@@ -2910,36 +2910,32 @@
                 
-    
 builder.add(page
@@ -2977,32 +2977,26 @@
             
-    %7D%0A  
+%7D%0A
             
@@ -2987,34 +2987,32 @@
   %7D%0A            
-  
 else if (Types.T
@@ -3064,36 +3064,32 @@
                 
-    
 builder.add(page
@@ -3134,34 +3134,26 @@
             
-   
- 
 %7D%0A
-    
             
@@ -3159,20 +3159,16 @@
  else %7B%0A
-    
         
@@ -3256,26 +3256,8 @@
 ));%0A
-                %7D%0A
     

Fix a bug: when the value is null