src/main/java/org/embulk/input/jdbc/getter/ColumnGetterFactory.java in embulk-input-jdbc-0.6.4 vs src/main/java/org/embulk/input/jdbc/getter/ColumnGetterFactory.java in embulk-input-jdbc-0.7.0

@@ -1,8 +1,11 @@
 package org.embulk.input.jdbc.getter;
 
+import java.lang.reflect.Field;
 import java.sql.Types;
+import java.util.HashMap;
+import java.util.Map;
 
 import org.embulk.config.ConfigException;
 import org.embulk.input.jdbc.JdbcColumn;
 import org.embulk.input.jdbc.JdbcColumnOption;
 import org.embulk.spi.PageBuilder;
@@ -11,12 +14,13 @@
 import org.embulk.spi.type.Type;
 import org.joda.time.DateTimeZone;
 
 public class ColumnGetterFactory
 {
-    private final PageBuilder to;
+    protected final PageBuilder to;
     private final DateTimeZone defaultTimeZone;
+    private final Map<Integer, String> jdbcTypes = getAllJDBCTypes();
 
     public ColumnGetterFactory(PageBuilder to, DateTimeZone defaultTimeZone)
     {
         this.to = to;
         this.defaultTimeZone = defaultTimeZone;
@@ -41,10 +45,12 @@
             return new DoubleColumnGetter(to, toType);
         case "boolean":
             return new BooleanColumnGetter(to, toType);
         case "string":
             return new StringColumnGetter(to, toType);
+        case "json":
+            return new JsonColumnGetter(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":
@@ -54,12 +60,33 @@
         default:
             throw new ConfigException(String.format("Unknown value_type '%s' for column '%s'", option.getValueType(), column.getName()));
         }
     }
 
-    private String sqlTypeToValueType(JdbcColumn column, int sqlType)
+    protected Map<Integer,String> getAllJDBCTypes() {
+        Map<Integer,String> map = new HashMap<Integer, String>();
+        for(Field f: Types.class.getFields()){
+            try {
+                map.put((Integer) f.get(null), f.getName());
+            } catch(IllegalAccessException iea){
+            }
+        }
+        return map;
+    }
+
+    public String getJdbcType(int sqlType)
     {
+        String sqlTypeName = jdbcTypes.get(sqlType);
+        if(sqlTypeName == null){
+            throw new UnsupportedOperationException(String.format("Unknown SQL type value '%d'", sqlType));
+        }
+        return sqlTypeName;
+    }
+
+
+    protected String sqlTypeToValueType(JdbcColumn column, int sqlType)
+    {
         switch(sqlType) {
         // getLong
         case Types.TINYINT:
         case Types.SMALLINT:
         case Types.INTEGER:
@@ -133,10 +160,10 @@
         default:
             throw unsupportedOperationException(column);
         }
     }
 
-    private Type getToType(JdbcColumnOption option)
+    protected Type getToType(JdbcColumnOption option)
     {
         if (!option.getType().isPresent()) {
             return null;
         }
         Type toType = option.getType().get();