src/main/java/org/embulk/input/jdbc/getter/ColumnGetterFactory.java in embulk-input-jdbc-0.4.0 vs src/main/java/org/embulk/input/jdbc/getter/ColumnGetterFactory.java in embulk-input-jdbc-0.5.0

@@ -1,50 +1,96 @@
 package org.embulk.input.jdbc.getter;
 
 import java.sql.Types;
+
+import org.embulk.config.ConfigException;
 import org.embulk.input.jdbc.JdbcColumn;
-import org.embulk.input.jdbc.getter.ColumnGetters.BooleanColumnGetter;
-import org.embulk.input.jdbc.getter.ColumnGetters.LongColumnGetter;
-import org.embulk.input.jdbc.getter.ColumnGetters.DoubleColumnGetter;
-import org.embulk.input.jdbc.getter.ColumnGetters.StringColumnGetter;
-import org.embulk.input.jdbc.getter.ColumnGetters.DateColumnGetter;
-import org.embulk.input.jdbc.getter.ColumnGetters.TimeColumnGetter;
-import org.embulk.input.jdbc.getter.ColumnGetters.TimestampColumnGetter;
-import org.embulk.input.jdbc.getter.ColumnGetters.BigDecimalToDoubleColumnGetter;
+import org.embulk.input.jdbc.JdbcColumnOption;
+import org.embulk.spi.PageBuilder;
+import org.embulk.spi.time.TimestampFormatter;
+import org.embulk.spi.type.TimestampType;
+import org.embulk.spi.type.Type;
+import org.joda.time.DateTimeZone;
 
 public class ColumnGetterFactory
 {
-    public ColumnGetter newColumnGetter(JdbcColumn column)
+    private final PageBuilder to;
+    private final DateTimeZone defaultTimeZone;
+
+    public ColumnGetterFactory(PageBuilder to, DateTimeZone defaultTimeZone)
     {
-        switch(column.getSqlType()) {
+        this.to = to;
+        this.defaultTimeZone = defaultTimeZone;
+    }
+
+    public ColumnGetter newColumnGetter(JdbcColumn column, JdbcColumnOption option)
+    {
+        return newColumnGetter(column, option, option.getValueType());
+    }
+
+    private ColumnGetter newColumnGetter(JdbcColumn column, JdbcColumnOption option, String valueType)
+    {
+        Type toType = getToType(option);
+        switch(valueType) {
+        case "coalesce":
+            return newColumnGetter(column, option, sqlTypeToValueType(column, column.getSqlType()));
+        case "long":
+            return new LongColumnGetter(to, toType);
+        case "float":
+            return new FloatColumnGetter(to, toType);
+        case "double":
+            return new DoubleColumnGetter(to, toType);
+        case "boolean":
+            return new BooleanColumnGetter(to, toType);
+        case "string":
+            return new StringColumnGetter(to, toType);
+        case "date":
+            return new DateColumnGetter(to, toType, newTimestampFormatter(option, DateColumnGetter.DEFAULT_FORMAT));
+        case "time":
+            return new TimeColumnGetter(to, toType, newTimestampFormatter(option, DateColumnGetter.DEFAULT_FORMAT));
+        case "timestamp":
+            return new TimestampColumnGetter(to, toType, newTimestampFormatter(option, DateColumnGetter.DEFAULT_FORMAT));
+        case "decimal":
+            return new BigDecimalColumnGetter(to, toType);
+        default:
+            throw new ConfigException(String.format("Unknown value_type '%s' for column '%s'", option.getValueType(), column.getName()));
+        }
+    }
+
+    private String sqlTypeToValueType(JdbcColumn column, int sqlType)
+    {
+        switch(sqlType) {
         // getLong
         case Types.TINYINT:
         case Types.SMALLINT:
         case Types.INTEGER:
         case Types.BIGINT:
-            return new LongColumnGetter();
+            return "long";
 
-        // getDouble
-        case Types.DOUBLE:
+        // getFloat
         case Types.FLOAT:
         case Types.REAL:
-            return new DoubleColumnGetter();
+            return "float";
 
+        // getDouble
+        case Types.DOUBLE:
+            return "double";
+
         // getBool
         case Types.BOOLEAN:
         case Types.BIT:  // JDBC BIT is boolean, unlike SQL-92
-            return new BooleanColumnGetter();
+            return "boolean";
 
         // getString, Clob
         case Types.CHAR:
         case Types.VARCHAR:
         case Types.LONGVARCHAR:
         case Types.CLOB:
         case Types.NCHAR:
         case Types.NVARCHAR:
         case Types.LONGNVARCHAR:
-            return new StringColumnGetter();
+            return "string";
 
         // TODO
         //// getBytes Blob
         //case Types.BINARY:
         //case Types.VARBINARY:
@@ -52,29 +98,29 @@
         //case Types.BLOB:
         //    return new BytesColumnGetter();
 
         // getDate
         case Types.DATE:
-            return new DateColumnGetter(); // TODO
+            return "date";
 
         // getTime
         case Types.TIME:
-            return new TimeColumnGetter(); // TODO
+            return "time";
 
         // getTimestamp
         case Types.TIMESTAMP:
-            return new TimestampColumnGetter();
+            return "timestamp";
 
         // TODO
         //// Null
         //case Types.NULL:
         //    return new NullColumnGetter();
 
         // getBigDecimal
         case Types.NUMERIC:
         case Types.DECIMAL:
-            return new BigDecimalToDoubleColumnGetter();
+            return "decimal";
 
         // others
         case Types.ARRAY:  // array
         case Types.STRUCT: // map
         case Types.REF:
@@ -85,9 +131,29 @@
         case Types.JAVA_OBJECT:
         case Types.OTHER:
         default:
             throw unsupportedOperationException(column);
         }
+    }
+
+    private Type getToType(JdbcColumnOption option)
+    {
+        if (!option.getType().isPresent()) {
+            return null;
+        }
+        Type toType = option.getType().get();
+        if (toType instanceof TimestampType && option.getTimestampFormat().isPresent()) {
+            toType = ((TimestampType)toType).withFormat(option.getTimestampFormat().get().getFormat());
+        }
+        return toType;
+    }
+
+    private TimestampFormatter newTimestampFormatter(JdbcColumnOption option, String defaultTimestampFormat)
+    {
+        return new TimestampFormatter(
+                option.getJRuby(),
+                option.getTimestampFormat().isPresent() ? option.getTimestampFormat().get().getFormat() : defaultTimestampFormat,
+                option.getTimeZone().or(defaultTimeZone));
     }
 
     private static UnsupportedOperationException unsupportedOperationException(JdbcColumn column)
     {
         throw new UnsupportedOperationException(