/*
 * Copyright 2004-present Facebook. All Rights Reserved.
 */
package com.facebook.presto;

import com.facebook.presto.UncompressedBlockSerde.UncompressedColumnWriter;
import com.google.common.base.Charsets;
import com.google.common.base.Preconditions;
import com.google.common.collect.ImmutableList;
import com.google.common.io.Files;
import com.google.common.io.InputSupplier;
import com.google.common.io.OutputSupplier;
import io.airlift.command.Arguments;
import io.airlift.command.Cli;
import io.airlift.command.Cli.CliBuilder;
import io.airlift.command.Command;
import io.airlift.command.Help;
import io.airlift.command.Option;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.util.List;
import java.util.concurrent.Callable;

import static com.facebook.presto.TupleInfo.Type.FIXED_INT_64;
import static com.facebook.presto.TupleInfo.Type.VARIABLE_BINARY;

public class Main
{
    public static void main(String[] args)
            throws Exception
    {
        CliBuilder<Callable<Void>> builder = (CliBuilder<Callable<Void>>) (Object) Cli.buildCli("presto", Callable.class)
                .withDefaultCommand(Help.class)
                .withCommands(Help.class);

        builder.withGroup("convert")
                .withDescription("convert file formats")
                .withDefaultCommand(Help.class)
                .withCommand(ConvertCsv.class);

        Cli<Callable<Void>> cli = builder.build();

        cli.parse(args).call();
    }

    public static class BaseCommand implements Callable<Void>
    {
        @Override
        public Void call()
                throws Exception
        {
            run();
            return null;
        }

        public void run()
                throws Exception
        {
            System.out.println(getClass().getSimpleName());
        }
    }

    @Command(name = "csv", description = "Convert CSB to columns")
    public static class ConvertCsv extends BaseCommand
    {
        @Option(name = {"--column-separator"}, description = "Column separator")
        public char columnSeparator = ',';

        @Option(name = {"-o", "--output-dir"}, description = "Output dir")
        public String outputDir = "data";

        @Option(name = {"-t", "--type"}, description = "Column type")
        public List<String> types;

        @Arguments(description = "CSV file to convert")
        public String csvFile;

        @Override
        public void run()
                throws Exception
        {
            Preconditions.checkArgument(types != null && !types.isEmpty(), "Type is required");

            File dir = new File(outputDir);

            ImmutableList.Builder<ColumnProcessor> processors = ImmutableList.builder();
            int index = 0;
            for (String type : types) {
                File file = new File(dir, "column" + index++ + ".data");
                switch (type) {
                    case "long":
                        processors.add(new UncompressedColumnWriter(newCreateDirectoryOutputStreamSupplier(file), FIXED_INT_64));
                        break;
                    case "string":
                        processors.add(new UncompressedColumnWriter(newCreateDirectoryOutputStreamSupplier(file), VARIABLE_BINARY));
                        break;
                    default:
                        throw new IllegalArgumentException("Unsupported type " + type);
                }
            }

            InputSupplier<InputStreamReader> inputSupplier;
            if (csvFile != null) {
                inputSupplier = Files.newReaderSupplier(new File(csvFile), Charsets.UTF_8);
            }
            else {
                inputSupplier = new InputSupplier<InputStreamReader>()
                {
                    public InputStreamReader getInput()
                    {
                        return new InputStreamReader(System.in, Charsets.UTF_8);
                    }
                };
            }
            Csv.processCsv(inputSupplier, columnSeparator, processors.build());
        }

        private OutputSupplier<FileOutputStream> newCreateDirectoryOutputStreamSupplier(final File file)
        {
            return new OutputSupplier<FileOutputStream>()
            {
                public FileOutputStream getOutput()
                        throws IOException
                {
                    file.getParentFile().mkdirs();
                    return new FileOutputStream(file);
                }
            };
        }
    }
}

@@ -2041,16 +2041,22 @@
 name = %7B
+%22-d%22, 
 %22--colum
@@ -2057,24 +2057,24 @@
 -column-
-separato
+delimite
 r%22%7D, des
@@ -2096,16 +2096,26 @@
 umn 
-separato
+delimiter characte
 r%22)%0A
@@ -2129,20 +2129,22 @@
  public 
-char
+String
  columnS
@@ -2158,11 +2158,11 @@
 r = 
-','
+%22,%22
 ;%0A%0A 
@@ -4070,16 +4070,23 @@
 pplier, 
+toChar(
 columnSe
@@ -4092,16 +4092,17 @@
 eparator
+)
 , proces
@@ -4119,32 +4119,565 @@
 ());%0A        %7D%0A%0A
+        private char toChar(String string)%0A        %7B%0A            Preconditions.checkArgument(!string.isEmpty(), %22String is empty%22);%0A            if (string.length() == 1) %7B%0A                return string.charAt(0);%0A            %7D%0A            if (string.length() == 6 && string.startsWith(%22%5C%5Cu%22)) %7B%0A                int value = Integer.parseInt(string.substring(2), 16);%0A                return (char) value;%0A            %7D%0A            throw new IllegalArgumentException(String.format(%22Can not convert '%25s' to a char%22, string));%0A        %7D%0A%0A
         private 

Add support for '\u0001' column delimiter