Add Redshift timestamp/timestamptz incremental support

Author: sakama

embulk-input-redshift/src/main/java/org/embulk/input/RedshiftInputPlugin.java

@@ -9,7 +9,11 @@
 import org.embulk.config.Config;
 import org.embulk.config.ConfigDefault;
 import org.embulk.input.jdbc.AbstractJdbcInputPlugin;
+import org.embulk.input.jdbc.getter.ColumnGetterFactory;
 import org.embulk.input.postgresql.PostgreSQLInputConnection;
+import org.embulk.input.redshift.getter.RedshiftColumnGetterFactory;
+import org.embulk.spi.PageBuilder;
+import org.joda.time.DateTimeZone;
 
 public class RedshiftInputPlugin
         extends AbstractJdbcInputPlugin
@@ -90,4 +94,10 @@ protected PostgreSQLInputConnection newConnection(PluginTask task) throws SQLExc
             }
         }
     }
+
+    @Override
+    protected ColumnGetterFactory newColumnGetterFactory(PageBuilder pageBuilder, DateTimeZone dateTimeZone)
+    {
+        return new RedshiftColumnGetterFactory(pageBuilder, dateTimeZone);
+    }
 }

embulk-input-redshift/src/main/java/org/embulk/input/redshift/getter/RedshiftColumnGetterFactory.java

@@ -0,0 +1,36 @@
+package org.embulk.input.redshift.getter;
+
+import org.embulk.input.jdbc.AbstractJdbcInputPlugin.PluginTask;
+import org.embulk.input.jdbc.JdbcColumn;
+import org.embulk.input.jdbc.JdbcColumnOption;
+import org.embulk.input.jdbc.JdbcInputConnection;
+import org.embulk.input.jdbc.getter.ColumnGetter;
+import org.embulk.input.jdbc.getter.ColumnGetterFactory;
+import org.embulk.input.jdbc.getter.TimestampWithTimeZoneIncrementalHandler;
+import org.embulk.input.jdbc.getter.TimestampWithoutTimeZoneIncrementalHandler;
+import org.embulk.spi.PageBuilder;
+import org.joda.time.DateTimeZone;
+
+public class RedshiftColumnGetterFactory extends ColumnGetterFactory
+{
+    public RedshiftColumnGetterFactory(PageBuilder to, DateTimeZone defaultTimeZone)
+    {
+        super(to, defaultTimeZone);
+    }
+
+    @Override
+    public ColumnGetter newColumnGetter(JdbcInputConnection con, PluginTask task, JdbcColumn column, JdbcColumnOption option)
+    {
+        ColumnGetter getter = super.newColumnGetter(con, task, column, option);
+
+        // incremental loading wrapper
+        switch (column.getTypeName()) {
+        case "timestamptz":
+            return new TimestampWithTimeZoneIncrementalHandler(getter);
+        case "timestamp":
+            return new TimestampWithoutTimeZoneIncrementalHandler(getter);
+        default:
+            return getter;
+        }
+    }
+}