diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcBatchCreator.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcBatchCreator.java
deleted file mode 100755
index f5db0b7..0000000
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcBatchCreator.java
+++ /dev/null
@@ -1,42 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.exec.store.jdbc;
-
-import java.util.Collections;
-import java.util.List;
-
-import org.apache.drill.common.exceptions.ExecutionSetupException;
-import org.apache.drill.exec.ops.ExecutorFragmentContext;
-import org.apache.drill.exec.physical.impl.BatchCreator;
-import org.apache.drill.exec.physical.impl.ScanBatch;
-import org.apache.drill.exec.record.RecordBatch;
-import org.apache.drill.exec.store.RecordReader;
-
-import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
-
-public class JdbcBatchCreator implements BatchCreator<JdbcSubScan> {
-  @Override
-  public ScanBatch getBatch(ExecutorFragmentContext context, JdbcSubScan config,
-      List<RecordBatch> children) throws ExecutionSetupException {
-    Preconditions.checkArgument(children.isEmpty());
-    JdbcStoragePlugin plugin = config.getPlugin();
-    RecordReader reader = new JdbcRecordReader(plugin.getDataSource(),
-        config.getSql(), plugin.getName(), config.getColumns());
-    return new ScanBatch(config, context, Collections.singletonList(reader));
-  }
-}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcBatchReader.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcBatchReader.java
new file mode 100644
index 0000000..8a86215
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcBatchReader.java
@@ -0,0 +1,315 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc;
+
+import org.apache.drill.common.AutoCloseables;
+import org.apache.drill.common.exceptions.CustomErrorContext;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.expression.SchemaPath;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.exec.physical.impl.scan.framework.ManagedReader;
+import org.apache.drill.exec.physical.impl.scan.framework.SchemaNegotiator;
+import org.apache.drill.exec.physical.resultSet.ResultSetLoader;
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+import org.apache.drill.exec.record.metadata.SchemaBuilder;
+import org.apache.drill.exec.record.metadata.TupleMetadata;
+import org.apache.drill.exec.store.jdbc.writers.JdbcBigintWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcBitWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcColumnWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcDateWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcDoubleWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcFloatWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcIntWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcTimeWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcTimestampWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcVarbinaryWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcVarcharWriter;
+import org.apache.drill.exec.store.jdbc.writers.JdbcVardecimalWriter;
+import org.apache.drill.shaded.guava.com.google.common.collect.ImmutableMap;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.sql.DataSource;
+import java.lang.reflect.Field;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.drill.exec.planner.types.DrillRelDataTypeSystem.DRILL_REL_DATATYPE_SYSTEM;
+
+public class JdbcBatchReader implements ManagedReader<SchemaNegotiator> {
+
+  private static final Logger logger = LoggerFactory.getLogger(JdbcBatchReader.class);
+  private static final ImmutableMap<Integer, MinorType> JDBC_TYPE_MAPPINGS;
+  private final DataSource source;
+  private final String sql;
+  private final List<SchemaPath> columns;
+  private Connection connection;
+  private PreparedStatement statement;
+  private ResultSet resultSet;
+  private RowSetLoader rowWriter;
+  private CustomErrorContext errorContext;
+  private List<JdbcColumnWriter> columnWriters;
+  private List<JdbcColumn> jdbcColumns;
+
+
+  public JdbcBatchReader(DataSource source, String sql, List<SchemaPath> columns) {
+    this.source = source;
+    this.sql = sql;
+    this.columns = columns;
+  }
+
+  /*
+   * This map maps JDBC data types to their Drill equivalents.  The basic strategy is that if there
+   * is a Drill equivalent, then do the mapping as expected.  All flavors of INT (SMALLINT, TINYINT etc)
+   * are mapped to INT in Drill, with the exception of BIGINT.
+   *
+   * All flavors of character fields are mapped to VARCHAR in Drill. All versions of binary fields are
+   * mapped to VARBINARY.
+   *
+   */
+  static {
+    JDBC_TYPE_MAPPINGS = ImmutableMap.<Integer, MinorType>builder()
+      .put(java.sql.Types.DOUBLE, MinorType.FLOAT8)
+      .put(java.sql.Types.FLOAT, MinorType.FLOAT4)
+      .put(java.sql.Types.TINYINT, MinorType.INT)
+      .put(java.sql.Types.SMALLINT, MinorType.INT)
+      .put(java.sql.Types.INTEGER, MinorType.INT)
+      .put(java.sql.Types.BIGINT, MinorType.BIGINT)
+
+      .put(java.sql.Types.CHAR, MinorType.VARCHAR)
+      .put(java.sql.Types.VARCHAR, MinorType.VARCHAR)
+      .put(java.sql.Types.LONGVARCHAR, MinorType.VARCHAR)
+      .put(java.sql.Types.CLOB, MinorType.VARCHAR)
+
+      .put(java.sql.Types.NCHAR, MinorType.VARCHAR)
+      .put(java.sql.Types.NVARCHAR, MinorType.VARCHAR)
+      .put(java.sql.Types.LONGNVARCHAR, MinorType.VARCHAR)
+
+      .put(java.sql.Types.VARBINARY, MinorType.VARBINARY)
+      .put(java.sql.Types.LONGVARBINARY, MinorType.VARBINARY)
+      .put(java.sql.Types.BLOB, MinorType.VARBINARY)
+
+      .put(java.sql.Types.NUMERIC, MinorType.FLOAT8)
+      .put(java.sql.Types.DECIMAL, MinorType.VARDECIMAL)
+      .put(java.sql.Types.REAL, MinorType.FLOAT8)
+
+      .put(java.sql.Types.DATE, MinorType.DATE)
+      .put(java.sql.Types.TIME, MinorType.TIME)
+      .put(java.sql.Types.TIMESTAMP, MinorType.TIMESTAMP)
+
+      .put(java.sql.Types.BOOLEAN, MinorType.BIT)
+      .put(java.sql.Types.BIT, MinorType.BIT)
+
+      .build();
+  }
+
+  @Override
+  public boolean open(SchemaNegotiator negotiator) {
+
+    this.errorContext = negotiator.parentErrorContext();
+    try {
+      connection = source.getConnection();
+      statement = connection.prepareStatement(sql);
+      resultSet = statement.executeQuery();
+
+      TupleMetadata drillSchema = buildSchema();
+      negotiator.tableSchema(drillSchema, true);
+      ResultSetLoader resultSetLoader = negotiator.build();
+
+      // Create ScalarWriters
+      rowWriter = resultSetLoader.writer();
+      populateWriterArray();
+
+    } catch (SQLException e) {
+      throw UserException.dataReadError(e)
+        .message("The JDBC storage plugin failed while trying setup the SQL query. ")
+        .addContext("Sql", sql)
+        .addContext(errorContext)
+        .build(logger);
+    }
+
+    return true;
+  }
+
+  @Override
+  public boolean next() {
+    while (!rowWriter.isFull()) {
+      if (!processRow()) {
+        return false;
+      }
+    }
+    return true;
+  }
+
+  private boolean processRow() {
+    try {
+      if (!resultSet.next()) {
+        return false;
+      }
+      rowWriter.start();
+      // Process results
+      for (JdbcColumnWriter writer : columnWriters) {
+        writer.load(resultSet);
+      }
+      rowWriter.save();
+    } catch (SQLException e) {
+      throw UserException
+        .dataReadError(e)
+        .message("Failure while attempting to read from database.")
+        .addContext("Sql", sql)
+        .addContext(errorContext)
+        .build(logger);
+    }
+
+    return true;
+  }
+
+  @Override
+  public void close() {
+    AutoCloseables.closeSilently(resultSet, statement, connection);
+  }
+
+  private TupleMetadata buildSchema() throws SQLException {
+    SchemaBuilder builder = new SchemaBuilder();
+    ResultSetMetaData meta = resultSet.getMetaData();
+    jdbcColumns = new ArrayList<>();
+
+    int columnsCount = meta.getColumnCount();
+
+    if (columns.size() != columnsCount) {
+      throw UserException
+        .validationError()
+        .message(
+          "Expected columns count differs from the returned one.\n" +
+            "Expected columns: %s\n" +
+            "Returned columns count: %s",
+          columns, columnsCount)
+        .addContext("Sql", sql)
+        .addContext(errorContext)
+        .build(logger);
+    }
+
+    for (int i = 1; i <= columnsCount; i++) {
+      String name = columns.get(i - 1).getRootSegmentPath();
+      // column index in ResultSetMetaData starts from 1
+      int jdbcType = meta.getColumnType(i);
+      int width = Math.min(meta.getPrecision(i), DRILL_REL_DATATYPE_SYSTEM.getMaxNumericPrecision());
+      int scale = Math.min(meta.getScale(i), DRILL_REL_DATATYPE_SYSTEM.getMaxNumericScale());
+
+      MinorType minorType = JDBC_TYPE_MAPPINGS.get(jdbcType);
+      if (minorType == null) {
+        logger.warn("Ignoring column that is unsupported.", UserException
+          .unsupportedError()
+          .message(
+            "A column you queried has a data type that is not currently supported by the JDBC storage plugin. "
+              + "The column's name was %s and its JDBC data type was %s. ",
+            name,
+            nameFromType(jdbcType))
+          .addContext("Sql", sql)
+          .addContext("Column Name", name)
+          .addContext(errorContext)
+          .build(logger));
+        continue;
+      }
+
+      jdbcColumns.add(new JdbcColumn(name, minorType, i));
+      // Precision and scale are passed for all readers whether they are needed or not.
+      builder.addNullable(name, minorType, width, scale);
+    }
+
+    return builder.buildSchema();
+  }
+
+  private void populateWriterArray() {
+    columnWriters = new ArrayList<>();
+
+    for (JdbcColumn col : jdbcColumns) {
+      switch (col.type) {
+        case VARCHAR:
+          columnWriters.add(new JdbcVarcharWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case FLOAT4:
+          columnWriters.add(new JdbcFloatWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case FLOAT8:
+          columnWriters.add(new JdbcDoubleWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case INT:
+          columnWriters.add(new JdbcIntWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case BIGINT:
+          columnWriters.add(new JdbcBigintWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case DATE:
+          columnWriters.add(new JdbcDateWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case TIME:
+          columnWriters.add(new JdbcTimeWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case TIMESTAMP:
+          columnWriters.add(new JdbcTimestampWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case VARBINARY:
+          columnWriters.add(new JdbcVarbinaryWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case BIT:
+          columnWriters.add(new JdbcBitWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        case VARDECIMAL:
+          columnWriters.add(new JdbcVardecimalWriter(col.colName, rowWriter, col.colPosition));
+          break;
+        default:
+          logger.warn("Unsupported data type {} found at column {}", col.type.getDescriptorForType(), col.colName);
+      }
+    }
+  }
+
+  private static String nameFromType(int javaSqlType) {
+    try {
+      for (Field f : java.sql.Types.class.getFields()) {
+        if (java.lang.reflect.Modifier.isStatic(f.getModifiers()) &&
+          f.getType() == int.class &&
+          f.getInt(null) == javaSqlType) {
+          return f.getName();
+        }
+      }
+    } catch (IllegalArgumentException | IllegalAccessException e) {
+      logger.debug("Unable to SQL type {} into String: {}", javaSqlType, e.getMessage());
+    }
+
+    return Integer.toString(javaSqlType);
+  }
+
+  public static class JdbcColumn {
+    final String colName;
+    final MinorType type;
+    final int colPosition;
+
+    public JdbcColumn (String colName, MinorType type, int colPosition) {
+      this.colName = colName;
+      this.type = type;
+      this.colPosition = colPosition;
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java
deleted file mode 100644
index a77de8d..0000000
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordReader.java
+++ /dev/null
@@ -1,495 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements.  See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership.  The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License.  You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.drill.exec.store.jdbc;
-
-import java.lang.reflect.Field;
-import java.math.BigDecimal;
-import java.sql.Connection;
-import java.sql.Date;
-import java.sql.PreparedStatement;
-import java.sql.ResultSet;
-import java.sql.ResultSetMetaData;
-import java.sql.SQLException;
-import java.sql.Time;
-import java.sql.Timestamp;
-import java.util.Calendar;
-import java.util.List;
-import java.util.TimeZone;
-
-import javax.sql.DataSource;
-
-import org.apache.drill.common.AutoCloseables;
-import org.apache.drill.common.exceptions.UserException;
-import org.apache.drill.common.expression.SchemaPath;
-import org.apache.drill.common.types.TypeProtos;
-import org.apache.drill.common.types.TypeProtos.MajorType;
-import org.apache.drill.common.types.TypeProtos.MinorType;
-import org.apache.drill.exec.exception.SchemaChangeException;
-import org.apache.drill.exec.expr.TypeHelper;
-import org.apache.drill.exec.ops.OperatorContext;
-import org.apache.drill.exec.physical.impl.OutputMutator;
-import org.apache.drill.exec.record.MaterializedField;
-import org.apache.drill.exec.store.AbstractRecordReader;
-import org.apache.drill.exec.vector.NullableBigIntVector;
-import org.apache.drill.exec.vector.NullableBitVector;
-import org.apache.drill.exec.vector.NullableDateVector;
-import org.apache.drill.exec.vector.NullableFloat4Vector;
-import org.apache.drill.exec.vector.NullableFloat8Vector;
-import org.apache.drill.exec.vector.NullableIntVector;
-import org.apache.drill.exec.vector.NullableTimeStampVector;
-import org.apache.drill.exec.vector.NullableTimeVector;
-import org.apache.drill.exec.vector.NullableVarBinaryVector;
-import org.apache.drill.exec.vector.NullableVarCharVector;
-import org.apache.drill.exec.vector.NullableVarDecimalVector;
-import org.apache.drill.exec.vector.ValueVector;
-
-import org.apache.drill.shaded.guava.com.google.common.base.Charsets;
-import org.apache.drill.shaded.guava.com.google.common.collect.ImmutableList;
-import org.apache.drill.shaded.guava.com.google.common.collect.ImmutableMap;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.apache.drill.exec.planner.types.DrillRelDataTypeSystem.DRILL_REL_DATATYPE_SYSTEM;
-
-class JdbcRecordReader extends AbstractRecordReader {
-
-  private static final Logger logger = LoggerFactory.getLogger(JdbcRecordReader.class);
-
-  private static final ImmutableMap<Integer, MinorType> JDBC_TYPE_MAPPINGS;
-  private final DataSource source;
-  private ResultSet resultSet;
-  private final String storagePluginName;
-  private Connection connection;
-  private PreparedStatement statement;
-  private final String sql;
-  private ImmutableList<ValueVector> vectors;
-  private ImmutableList<Copier<?>> copiers;
-  private final List<SchemaPath> columns;
-
-  public JdbcRecordReader(DataSource source, String sql, String storagePluginName, List<SchemaPath> columns) {
-    this.source = source;
-    this.sql = sql;
-    this.storagePluginName = storagePluginName;
-    this.columns = columns;
-  }
-
-  static {
-    JDBC_TYPE_MAPPINGS = ImmutableMap.<Integer, MinorType>builder()
-        .put(java.sql.Types.DOUBLE, MinorType.FLOAT8)
-        .put(java.sql.Types.FLOAT, MinorType.FLOAT4)
-        .put(java.sql.Types.TINYINT, MinorType.INT)
-        .put(java.sql.Types.SMALLINT, MinorType.INT)
-        .put(java.sql.Types.INTEGER, MinorType.INT)
-        .put(java.sql.Types.BIGINT, MinorType.BIGINT)
-
-        .put(java.sql.Types.CHAR, MinorType.VARCHAR)
-        .put(java.sql.Types.VARCHAR, MinorType.VARCHAR)
-        .put(java.sql.Types.LONGVARCHAR, MinorType.VARCHAR)
-        .put(java.sql.Types.CLOB, MinorType.VARCHAR)
-
-        .put(java.sql.Types.NCHAR, MinorType.VARCHAR)
-        .put(java.sql.Types.NVARCHAR, MinorType.VARCHAR)
-        .put(java.sql.Types.LONGNVARCHAR, MinorType.VARCHAR)
-
-        .put(java.sql.Types.VARBINARY, MinorType.VARBINARY)
-        .put(java.sql.Types.LONGVARBINARY, MinorType.VARBINARY)
-        .put(java.sql.Types.BLOB, MinorType.VARBINARY)
-
-        .put(java.sql.Types.NUMERIC, MinorType.FLOAT8)
-        .put(java.sql.Types.DECIMAL, MinorType.VARDECIMAL)
-        .put(java.sql.Types.REAL, MinorType.FLOAT8)
-
-        .put(java.sql.Types.DATE, MinorType.DATE)
-        .put(java.sql.Types.TIME, MinorType.TIME)
-        .put(java.sql.Types.TIMESTAMP, MinorType.TIMESTAMP)
-
-        .put(java.sql.Types.BOOLEAN, MinorType.BIT)
-
-        .put(java.sql.Types.BIT, MinorType.BIT)
-
-        .build();
-  }
-
-  private static String nameFromType(int javaSqlType) {
-    try {
-      for (Field f : java.sql.Types.class.getFields()) {
-        if (java.lang.reflect.Modifier.isStatic(f.getModifiers()) &&
-            f.getType() == int.class &&
-            f.getInt(null) == javaSqlType) {
-          return f.getName();
-        }
-      }
-    } catch (IllegalArgumentException | IllegalAccessException e) {
-      logger.trace("Unable to SQL type {} into String: {}", javaSqlType, e.getMessage());
-    }
-
-    return Integer.toString(javaSqlType);
-
-  }
-
-  private Copier<?> getCopier(int jdbcType, int offset, ResultSet result, ValueVector v) {
-    switch (jdbcType) {
-      case java.sql.Types.BIGINT:
-        return new BigIntCopier(offset, result, (NullableBigIntVector.Mutator) v.getMutator());
-      case java.sql.Types.FLOAT:
-        return new Float4Copier(offset, result, (NullableFloat4Vector.Mutator) v.getMutator());
-      case java.sql.Types.DOUBLE:
-      case java.sql.Types.NUMERIC:
-      case java.sql.Types.REAL:
-        return new Float8Copier(offset, result, (NullableFloat8Vector.Mutator) v.getMutator());
-      case java.sql.Types.TINYINT:
-      case java.sql.Types.SMALLINT:
-      case java.sql.Types.INTEGER:
-        return new IntCopier(offset, result, (NullableIntVector.Mutator) v.getMutator());
-      case java.sql.Types.CHAR:
-      case java.sql.Types.VARCHAR:
-      case java.sql.Types.LONGVARCHAR:
-      case java.sql.Types.CLOB:
-      case java.sql.Types.NCHAR:
-      case java.sql.Types.NVARCHAR:
-      case java.sql.Types.LONGNVARCHAR:
-        return new VarCharCopier(offset, result, (NullableVarCharVector.Mutator) v.getMutator());
-      case java.sql.Types.VARBINARY:
-      case java.sql.Types.LONGVARBINARY:
-      case java.sql.Types.BLOB:
-        return new VarBinaryCopier(offset, result, (NullableVarBinaryVector.Mutator) v.getMutator());
-      case java.sql.Types.DATE:
-        return new DateCopier(offset, result, (NullableDateVector.Mutator) v.getMutator());
-      case java.sql.Types.TIME:
-        return new TimeCopier(offset, result, (NullableTimeVector.Mutator) v.getMutator());
-      case java.sql.Types.TIMESTAMP:
-        return new TimeStampCopier(offset, result, (NullableTimeStampVector.Mutator) v.getMutator());
-      case java.sql.Types.BOOLEAN:
-      case java.sql.Types.BIT:
-        return new BitCopier(offset, result, (NullableBitVector.Mutator) v.getMutator());
-      case java.sql.Types.DECIMAL:
-        return new DecimalCopier(offset, result, (NullableVarDecimalVector.Mutator) v.getMutator());
-      default:
-        throw new IllegalArgumentException("Unknown how to handle vector.");
-    }
-  }
-
-  @Override
-  public void setup(OperatorContext operatorContext, OutputMutator output) {
-    try {
-      connection = source.getConnection();
-      statement = connection.prepareStatement(sql);
-      resultSet = statement.executeQuery();
-
-      ResultSetMetaData meta = resultSet.getMetaData();
-      int columnsCount = meta.getColumnCount();
-      if (columns.size() != columnsCount) {
-        throw UserException
-            .validationError()
-            .message(
-              "Expected columns count differs from the returned one.\n" +
-                  "Expected columns: %s\n" +
-                  "Returned columns count: %s",
-              columns, columnsCount)
-            .addContext("Sql", sql)
-            .addContext("Plugin", storagePluginName)
-            .build(logger);
-      }
-      ImmutableList.Builder<ValueVector> vectorBuilder = ImmutableList.builder();
-      ImmutableList.Builder<Copier<?>> copierBuilder = ImmutableList.builder();
-
-      for (int i = 1; i <= columnsCount; i++) {
-        String name = columns.get(i - 1).getRootSegmentPath();
-        // column index in ResultSetMetaData starts from 1
-        int jdbcType = meta.getColumnType(i);
-        int width = Math.min(meta.getPrecision(i), DRILL_REL_DATATYPE_SYSTEM.getMaxNumericPrecision());
-        int scale = Math.min(meta.getScale(i), DRILL_REL_DATATYPE_SYSTEM.getMaxNumericScale());
-        MinorType minorType = JDBC_TYPE_MAPPINGS.get(jdbcType);
-        if (minorType == null) {
-          logger.warn("Ignoring column that is unsupported.", UserException
-              .unsupportedError()
-              .message(
-                  "A column you queried has a data type that is not currently supported by the JDBC storage plugin. "
-                      + "The column's name was %s and its JDBC data type was %s. ",
-                  name,
-                  nameFromType(jdbcType))
-              .addContext("Sql", sql)
-              .addContext("Column Name", name)
-              .addContext("Plugin", storagePluginName)
-              .build(logger));
-          continue;
-        }
-
-        MajorType type = MajorType.newBuilder()
-            .setMode(TypeProtos.DataMode.OPTIONAL)
-            .setMinorType(minorType)
-            .setScale(scale)
-            .setPrecision(width)
-            .build();
-        MaterializedField field = MaterializedField.create(name, type);
-        Class<? extends ValueVector> clazz = TypeHelper.getValueVectorClass(
-            minorType, type.getMode());
-        ValueVector vector = output.addField(field, clazz);
-        vectorBuilder.add(vector);
-        copierBuilder.add(getCopier(jdbcType, i, resultSet, vector));
-      }
-
-      vectors = vectorBuilder.build();
-      copiers = copierBuilder.build();
-
-    } catch (SQLException | SchemaChangeException e) {
-      throw UserException.dataReadError(e)
-          .message("The JDBC storage plugin failed while trying setup the SQL query. ")
-          .addContext("Sql", sql)
-          .addContext("Plugin", storagePluginName)
-          .build(logger);
-    }
-  }
-
-  @Override
-  public int next() {
-    int counter = 0;
-    try {
-      while (counter < 4095) { // loop at 4095 since nullables use one more than record count and we
-                                            // allocate on powers of two.
-        if (!resultSet.next()) {
-          break;
-        }
-        for (Copier<?> c : copiers) {
-          c.copy(counter);
-        }
-        counter++;
-      }
-    } catch (SQLException e) {
-      throw UserException
-          .dataReadError(e)
-          .message("Failure while attempting to read from database.")
-          .addContext("Sql", sql)
-          .addContext("Plugin", storagePluginName)
-          .build(logger);
-    }
-
-    int valueCount = Math.max(counter, 0);
-    for (ValueVector vv : vectors) {
-      vv.getMutator().setValueCount(valueCount);
-    }
-
-    return valueCount;
-  }
-
-  @Override
-  public void close() {
-    AutoCloseables.closeSilently(resultSet, statement, connection);
-  }
-
-  @Override
-  public String toString() {
-    return "JdbcRecordReader[sql=" + sql
-        + ", Plugin=" + storagePluginName
-        + "]";
-  }
-
-  private abstract static class Copier<T extends ValueVector.Mutator> {
-    final int columnIndex;
-    final ResultSet result;
-    final T mutator;
-
-    Copier(int columnIndex, ResultSet result, T mutator) {
-      this.columnIndex = columnIndex;
-      this.result = result;
-      this.mutator = mutator;
-    }
-
-    abstract void copy(int index) throws SQLException;
-  }
-
-  private static class IntCopier extends Copier<NullableIntVector.Mutator> {
-
-    IntCopier(int offset, ResultSet set, NullableIntVector.Mutator mutator) {
-      super(offset, set, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      mutator.setSafe(index, result.getInt(columnIndex));
-      if (result.wasNull()) {
-        mutator.setNull(index);
-      }
-    }
-  }
-
-  private static class BigIntCopier extends Copier<NullableBigIntVector.Mutator> {
-
-    BigIntCopier(int offset, ResultSet set, NullableBigIntVector.Mutator mutator) {
-      super(offset, set, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      mutator.setSafe(index, result.getLong(columnIndex));
-      if (result.wasNull()) {
-        mutator.setNull(index);
-      }
-    }
-  }
-
-  private static class Float4Copier extends Copier<NullableFloat4Vector.Mutator> {
-
-    Float4Copier(int columnIndex, ResultSet result, NullableFloat4Vector.Mutator mutator) {
-      super(columnIndex, result, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      mutator.setSafe(index, result.getFloat(columnIndex));
-      if (result.wasNull()) {
-        mutator.setNull(index);
-      }
-    }
-  }
-
-  private static class Float8Copier extends Copier<NullableFloat8Vector.Mutator> {
-
-    Float8Copier(int columnIndex, ResultSet result, NullableFloat8Vector.Mutator mutator) {
-      super(columnIndex, result, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      mutator.setSafe(index, result.getDouble(columnIndex));
-      if (result.wasNull()) {
-        mutator.setNull(index);
-      }
-    }
-  }
-
-  private static class DecimalCopier extends Copier<NullableVarDecimalVector.Mutator> {
-
-    DecimalCopier(int columnIndex, ResultSet result, NullableVarDecimalVector.Mutator mutator) {
-      super(columnIndex, result, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      BigDecimal decimal = result.getBigDecimal(columnIndex);
-      if (decimal != null) {
-        if (decimal.precision() > DRILL_REL_DATATYPE_SYSTEM.getMaxNumericPrecision()
-            || decimal.scale() > DRILL_REL_DATATYPE_SYSTEM.getMaxNumericScale()) {
-          throw UserException.unsupportedError()
-              .message("Drill doesn't support reading values with precision or scale larger than 38.\n" +
-                  "Please use round() UDF to obtain results with supported precision")
-              .addContext("Column index", index)
-              .build(logger);
-        }
-        mutator.setSafe(index, decimal);
-      }
-    }
-  }
-
-  private static class VarCharCopier extends Copier<NullableVarCharVector.Mutator> {
-
-    VarCharCopier(int columnIndex, ResultSet result, NullableVarCharVector.Mutator mutator) {
-      super(columnIndex, result, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      String val = result.getString(columnIndex);
-      if (val != null) {
-        byte[] record = val.getBytes(Charsets.UTF_8);
-        mutator.setSafe(index, record, 0, record.length);
-      }
-    }
-  }
-
-  private static class VarBinaryCopier extends Copier<NullableVarBinaryVector.Mutator> {
-
-    VarBinaryCopier(int columnIndex, ResultSet result, NullableVarBinaryVector.Mutator mutator) {
-      super(columnIndex, result, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      byte[] record = result.getBytes(columnIndex);
-      if (record != null) {
-        mutator.setSafe(index, record, 0, record.length);
-      }
-    }
-  }
-
-  private static class DateCopier extends Copier<NullableDateVector.Mutator> {
-
-    private final Calendar calendar = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
-
-    DateCopier(int columnIndex, ResultSet result, NullableDateVector.Mutator mutator) {
-      super(columnIndex, result, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      Date date = result.getDate(columnIndex, calendar);
-      if (date != null) {
-        mutator.setSafe(index, date.getTime());
-      }
-    }
-  }
-
-  private static class TimeCopier extends Copier<NullableTimeVector.Mutator> {
-
-    private final Calendar calendar = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
-
-    TimeCopier(int columnIndex, ResultSet result, NullableTimeVector.Mutator mutator) {
-      super(columnIndex, result, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      Time time = result.getTime(columnIndex, calendar);
-      if (time != null) {
-        mutator.setSafe(index, (int) time.getTime());
-      }
-    }
-  }
-
-  private static class TimeStampCopier extends Copier<NullableTimeStampVector.Mutator> {
-
-    private final Calendar calendar = Calendar.getInstance(TimeZone.getTimeZone("UTC"));
-
-    TimeStampCopier(int columnIndex, ResultSet result, NullableTimeStampVector.Mutator mutator) {
-      super(columnIndex, result, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      Timestamp stamp = result.getTimestamp(columnIndex, calendar);
-      if (stamp != null) {
-        mutator.setSafe(index, stamp.getTime());
-      }
-    }
-  }
-
-  private static class BitCopier extends Copier<NullableBitVector.Mutator> {
-
-    BitCopier(int columnIndex, ResultSet result, NullableBitVector.Mutator mutator) {
-      super(columnIndex, result, mutator);
-    }
-
-    @Override
-    void copy(int index) throws SQLException {
-      mutator.setSafe(index, result.getBoolean(columnIndex) ? 1 : 0);
-      if (result.wasNull()) {
-        mutator.setNull(index);
-      }
-    }
-  }
-}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcScanBatchCreator.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcScanBatchCreator.java
new file mode 100644
index 0000000..6a727579
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcScanBatchCreator.java
@@ -0,0 +1,72 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.drill.exec.store.jdbc;
+
+import org.apache.drill.common.exceptions.ExecutionSetupException;
+import org.apache.drill.common.exceptions.UserException;
+import org.apache.drill.common.types.TypeProtos.MinorType;
+import org.apache.drill.common.types.Types;
+import org.apache.drill.exec.ops.ExecutorFragmentContext;
+import org.apache.drill.exec.physical.impl.BatchCreator;
+import org.apache.drill.exec.physical.impl.scan.framework.BasicScanFactory;
+import org.apache.drill.exec.physical.impl.scan.framework.ManagedReader;
+import org.apache.drill.exec.physical.impl.scan.framework.ManagedScanFramework;
+import org.apache.drill.exec.physical.impl.scan.framework.ManagedScanFramework.ScanFrameworkBuilder;
+import org.apache.drill.exec.physical.impl.scan.framework.SchemaNegotiator;
+import org.apache.drill.exec.record.CloseableRecordBatch;
+import org.apache.drill.exec.record.RecordBatch;
+import org.apache.drill.exec.server.options.OptionManager;
+import org.apache.drill.shaded.guava.com.google.common.base.Preconditions;
+
+import java.util.Collections;
+import java.util.List;
+
+public class JdbcScanBatchCreator implements BatchCreator<JdbcSubScan> {
+
+  @Override
+  public CloseableRecordBatch getBatch(ExecutorFragmentContext context,
+                                       JdbcSubScan subScan, List<RecordBatch> children) throws ExecutionSetupException {
+    Preconditions.checkArgument(children.isEmpty());
+
+    try {
+      ScanFrameworkBuilder builder = createBuilder(context.getOptions(), subScan);
+      return builder.buildScanOperator(context, subScan);
+    } catch (UserException e) {
+      // Rethrow user exceptions directly
+      throw e;
+    } catch (Throwable e) {
+      // Wrap all others
+      throw new ExecutionSetupException(e);
+    }
+  }
+
+  private ScanFrameworkBuilder createBuilder(OptionManager options, JdbcSubScan subScan) {
+    JdbcStorageConfig config = subScan.getConfig();
+    ScanFrameworkBuilder builder = new ScanFrameworkBuilder();
+    builder.projection(subScan.getColumns());
+    builder.setUserName(subScan.getUserName());
+    JdbcStoragePlugin plugin = subScan.getPlugin();
+    List<ManagedReader<SchemaNegotiator>> readers =
+      Collections.singletonList(new JdbcBatchReader(plugin.getDataSource(), subScan.getSql(), subScan.getColumns()));
+
+    ManagedScanFramework.ReaderFactory readerFactory = new BasicScanFactory(readers.iterator());
+    builder.setReaderFactory(readerFactory);
+    builder.nullType(Types.optional(MinorType.VARCHAR));
+    return builder;
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcSubScan.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcSubScan.java
index def31fa..f08f4c1 100644
--- a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcSubScan.java
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcSubScan.java
@@ -17,6 +17,7 @@
  */
 package org.apache.drill.exec.store.jdbc;
 
+import org.apache.drill.common.PlanStringBuilder;
 import org.apache.drill.common.exceptions.ExecutionSetupException;
 import org.apache.drill.common.expression.SchemaPath;
 import org.apache.drill.common.logical.StoragePluginConfig;
@@ -72,7 +73,7 @@
     return columns;
   }
 
-  public StoragePluginConfig getConfig() {
+  public JdbcStorageConfig getConfig() {
     return plugin.getConfig();
   }
 
@@ -80,4 +81,12 @@
   public JdbcStoragePlugin getPlugin() {
     return plugin;
   }
+
+  @Override
+  public String toString() {
+    return new PlanStringBuilder(this)
+      .field("sql", sql)
+      .field("columns", columns)
+      .toString();
+  }
 }
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcBigintWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcBigintWriter.java
new file mode 100644
index 0000000..09b7d12
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcBigintWriter.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class JdbcBigintWriter extends JdbcColumnWriter {
+
+  public JdbcBigintWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    boolean b = results.wasNull();
+    if (! results.wasNull()) {
+      long value = results.getLong(columnIndex);
+      columnWriter.setLong(value);
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcBitWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcBitWriter.java
new file mode 100644
index 0000000..29d688f
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcBitWriter.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class JdbcBitWriter extends JdbcColumnWriter {
+
+  public JdbcBitWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    if (! results.wasNull()) {
+      boolean value = results.getBoolean(columnIndex);
+      columnWriter.setBoolean(value);
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcColumnWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcColumnWriter.java
new file mode 100644
index 0000000..730828b
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcColumnWriter.java
@@ -0,0 +1,42 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+import org.apache.drill.exec.vector.accessor.ScalarWriter;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public abstract class JdbcColumnWriter {
+
+  final String colName;
+  ScalarWriter columnWriter;
+  RowSetLoader rowWriter;
+  int columnIndex;
+
+  public JdbcColumnWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    this.colName = colName;
+    this.rowWriter = rowWriter;
+    this.columnWriter = rowWriter.scalar(colName);
+    this.columnIndex = columnIndex;
+  }
+
+  public void load(ResultSet resultSet) throws SQLException {}
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcDateWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcDateWriter.java
new file mode 100644
index 0000000..b050864
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcDateWriter.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.sql.Date;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class JdbcDateWriter extends JdbcColumnWriter {
+
+  public JdbcDateWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    Date value = results.getDate(columnIndex);
+    if (value != null) {
+      columnWriter.setDate(value.toLocalDate());
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcDoubleWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcDoubleWriter.java
new file mode 100644
index 0000000..95d9402
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcDoubleWriter.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class JdbcDoubleWriter extends JdbcColumnWriter {
+
+  public JdbcDoubleWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    if (!results.wasNull()) {
+      double value = results.getDouble(columnIndex);
+      columnWriter.setDouble(value);
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcFloatWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcFloatWriter.java
new file mode 100644
index 0000000..fd2188d
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcFloatWriter.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class JdbcFloatWriter extends JdbcColumnWriter {
+
+  public JdbcFloatWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    if (!results.wasNull()) {
+      float value = results.getFloat(columnIndex);
+      columnWriter.setFloat(value);
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcIntWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcIntWriter.java
new file mode 100644
index 0000000..5b4704c
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcIntWriter.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class JdbcIntWriter extends JdbcColumnWriter {
+
+  public JdbcIntWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    if (!results.wasNull()) {
+      int value = results.getInt(columnIndex);
+      columnWriter.setInt(value);
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcTimeWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcTimeWriter.java
new file mode 100644
index 0000000..a989995
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcTimeWriter.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Time;
+
+public class JdbcTimeWriter extends JdbcColumnWriter {
+
+  public JdbcTimeWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    Time value = results.getTime(columnIndex);
+    if (value != null) {
+      columnWriter.setTime(value.toLocalTime());
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcTimestampWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcTimestampWriter.java
new file mode 100644
index 0000000..885b0c8
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcTimestampWriter.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+
+public class JdbcTimestampWriter extends JdbcColumnWriter {
+
+  public JdbcTimestampWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    Timestamp value = results.getTimestamp(columnIndex);
+    if (value != null) {
+      columnWriter.setTimestamp(value.toInstant());
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcVarbinaryWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcVarbinaryWriter.java
new file mode 100644
index 0000000..24d1c59
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcVarbinaryWriter.java
@@ -0,0 +1,39 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class JdbcVarbinaryWriter extends JdbcColumnWriter {
+
+  public JdbcVarbinaryWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    byte[] value = results.getBytes(columnIndex);
+    if (value != null) {
+      columnWriter.setBytes(value, value.length);
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcVarcharWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcVarcharWriter.java
new file mode 100644
index 0000000..95e2cd2
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcVarcharWriter.java
@@ -0,0 +1,43 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+import org.apache.drill.shaded.guava.com.google.common.base.Strings;
+
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class JdbcVarcharWriter extends JdbcColumnWriter {
+
+  public JdbcVarcharWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    String value = results.getString(columnIndex);
+
+    if (Strings.isNullOrEmpty(value)) {
+      columnWriter.setNull();
+    }  else {
+      columnWriter.setString(value);
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcVardecimalWriter.java b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcVardecimalWriter.java
new file mode 100644
index 0000000..ecdbaab
--- /dev/null
+++ b/contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/writers/JdbcVardecimalWriter.java
@@ -0,0 +1,40 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.drill.exec.store.jdbc.writers;
+
+import org.apache.drill.exec.physical.resultSet.RowSetLoader;
+
+import java.math.BigDecimal;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+
+public class JdbcVardecimalWriter extends JdbcColumnWriter {
+
+  public JdbcVardecimalWriter(String colName, RowSetLoader rowWriter, int columnIndex) {
+    super(colName, rowWriter, columnIndex);
+  }
+
+  @Override
+  public void load(ResultSet results) throws SQLException {
+    BigDecimal value = results.getBigDecimal(columnIndex);
+    if (value != null) {
+      columnWriter.setDecimal(value);
+    }
+  }
+}
diff --git a/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithH2IT.java b/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithH2IT.java
index 07b142c..94b60cf 100644
--- a/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithH2IT.java
+++ b/contrib/storage-jdbc/src/test/java/org/apache/drill/exec/store/jdbc/TestJdbcPluginWithH2IT.java
@@ -27,6 +27,7 @@
 import org.apache.drill.test.ClusterFixture;
 import org.apache.drill.test.ClusterTest;
 import org.h2.tools.RunScript;
+import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
 import org.junit.experimental.categories.Category;
@@ -37,6 +38,7 @@
 import java.nio.file.Paths;
 import java.sql.Connection;
 import java.sql.DriverManager;
+import java.util.TimeZone;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertNotNull;
@@ -49,10 +51,15 @@
 
   private static final String TABLE_PATH = "jdbcmulti/";
   private static final String TABLE_NAME = String.format("%s.`%s`", StoragePluginTestUtils.DFS_PLUGIN_NAME, TABLE_PATH);
+  private static TimeZone defaultTimeZone;
 
   @BeforeClass
   public static void init() throws Exception {
     startCluster(ClusterFixture.builder(dirTestWatcher));
+    // Force timezone to UTC for these tests.
+    defaultTimeZone = TimeZone.getDefault();
+    TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
+
     dirTestWatcher.copyResourceToRoot(Paths.get(TABLE_PATH));
     Class.forName("org.h2.Driver");
     String connString = "jdbc:h2:" + dirTestWatcher.getTmpDir().getCanonicalPath();
@@ -71,6 +78,11 @@
     cluster.defineStoragePlugin("h2o", jdbcStorageConfig);
   }
 
+  @AfterClass
+  public static void cleanUp() {
+    TimeZone.setDefault(defaultTimeZone);
+  }
+
   @Test
   public void testCrossSourceMultiFragmentJoin() throws Exception {
     try {
diff --git a/pom.xml b/pom.xml
index caf3a20..e567330 100644
--- a/pom.xml
+++ b/pom.xml
@@ -114,7 +114,7 @@
     <surefire.version>3.0.0-M5</surefire.version>
     <jna.version>5.8.0</jna.version>
     <commons.compress.version>1.20</commons.compress.version>
-    <hikari.version>3.4.2</hikari.version>
+    <hikari.version>4.0.3</hikari.version>
     <netty.version>4.1.59.Final</netty.version>
     <httpclient.version>4.5.13</httpclient.version>
     <libthrift.version>0.14.0</libthrift.version>