contrib/storage-jdbc/src/main/java/org/apache/drill/exec/store/jdbc/JdbcRecordWriter.java

/*
 * 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.calcite.sql.SqlDialect;
import org.apache.calcite.sql.SqlDialect.DatabaseProduct;
import org.apache.calcite.sql.util.SqlBuilder;
import org.apache.drill.common.AutoCloseables;
import org.apache.drill.common.exceptions.UserException;
import org.apache.drill.common.types.TypeProtos.DataMode;
import org.apache.drill.common.types.TypeProtos.MinorType;
import org.apache.drill.exec.expr.fn.impl.StringFunctionHelpers;
import org.apache.drill.exec.expr.holders.BigIntHolder;
import org.apache.drill.exec.expr.holders.BitHolder;
import org.apache.drill.exec.expr.holders.DateHolder;
import org.apache.drill.exec.expr.holders.Float4Holder;
import org.apache.drill.exec.expr.holders.Float8Holder;
import org.apache.drill.exec.expr.holders.IntHolder;
import org.apache.drill.exec.expr.holders.NullableBigIntHolder;
import org.apache.drill.exec.expr.holders.NullableBitHolder;
import org.apache.drill.exec.expr.holders.NullableDateHolder;
import org.apache.drill.exec.expr.holders.NullableFloat4Holder;
import org.apache.drill.exec.expr.holders.NullableFloat8Holder;
import org.apache.drill.exec.expr.holders.NullableIntHolder;
import org.apache.drill.exec.expr.holders.NullableSmallIntHolder;
import org.apache.drill.exec.expr.holders.NullableTimeHolder;
import org.apache.drill.exec.expr.holders.NullableTimeStampHolder;
import org.apache.drill.exec.expr.holders.NullableTinyIntHolder;
import org.apache.drill.exec.expr.holders.NullableVarCharHolder;
import org.apache.drill.exec.expr.holders.NullableVarDecimalHolder;
import org.apache.drill.exec.expr.holders.SmallIntHolder;
import org.apache.drill.exec.expr.holders.TimeHolder;
import org.apache.drill.exec.expr.holders.TimeStampHolder;
import org.apache.drill.exec.expr.holders.TinyIntHolder;
import org.apache.drill.exec.expr.holders.VarCharHolder;
import org.apache.drill.exec.expr.holders.VarDecimalHolder;
import org.apache.drill.exec.ops.OperatorContext;
import org.apache.drill.exec.record.BatchSchema;
import org.apache.drill.exec.record.MaterializedField;
import org.apache.drill.exec.record.VectorAccessible;
import org.apache.drill.exec.store.AbstractRecordWriter;
import org.apache.drill.exec.store.EventBasedRecordWriter.FieldConverter;
import org.apache.drill.exec.store.jdbc.utils.JdbcDDLQueryUtils;
import org.apache.drill.exec.store.jdbc.utils.CreateTableStmtBuilder;
import org.apache.drill.exec.util.DecimalUtility;
import org.apache.drill.exec.vector.complex.reader.FieldReader;
import org.apache.drill.shaded.guava.com.google.common.collect.ImmutableMap;
import org.apache.parquet.Strings;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.sql.DataSource;
import java.io.IOException;
import java.math.BigDecimal;
import java.sql.Connection;
import java.sql.SQLException;
import java.sql.Statement;
import java.text.Format;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

public class JdbcRecordWriter extends AbstractRecordWriter {

  private static final Logger logger = LoggerFactory.getLogger(JdbcRecordWriter.class);

  private final String tableName;
  private Connection connection;
  private final SqlDialect dialect;
  private final List<Object> rowList;
  private final List<JdbcWriterField> fields;
  private final String rawTableName;
  private final JdbcWriter config;
  private SqlBuilder insertQueryBuilder;
  private boolean firstRecord;
  private int recordCount;

  /*
   * 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 character fields are mapped to VARCHAR in Drill. All versions of binary fields are
   * mapped to VARBINARY.
   */
  public static final ImmutableMap<MinorType, Integer> JDBC_TYPE_MAPPINGS = ImmutableMap.<MinorType, Integer>builder()
      .put(MinorType.FLOAT8, java.sql.Types.DOUBLE)
      .put(MinorType.FLOAT4, java.sql.Types.FLOAT)
      .put(MinorType.TINYINT, java.sql.Types.TINYINT)
      .put(MinorType.SMALLINT, java.sql.Types.SMALLINT)
      .put(MinorType.INT, java.sql.Types.INTEGER)
      .put(MinorType.BIGINT, java.sql.Types.BIGINT)
      .put(MinorType.VARCHAR, java.sql.Types.VARCHAR)
      .put(MinorType.VARBINARY, java.sql.Types.VARBINARY)
      .put(MinorType.VARDECIMAL, java.sql.Types.DECIMAL)
      .put(MinorType.DATE, java.sql.Types.DATE)
      .put(MinorType.TIME, java.sql.Types.TIME)
      .put(MinorType.TIMESTAMP, java.sql.Types.TIMESTAMP)
      .put(MinorType.BIT, java.sql.Types.BOOLEAN)
      .build();

  public JdbcRecordWriter(DataSource source, OperatorContext context, String name, JdbcWriter config) {
    this.tableName = JdbcDDLQueryUtils.addBackTicksToTable(name);
    this.rowList = new ArrayList<>();
    this.dialect = config.getPlugin().getDialect(source);
    this.config = config;
    this.rawTableName = name;
    this.fields = new ArrayList<>();
    this.firstRecord = true;
    this.recordCount = 0;

    this.insertQueryBuilder = initializeInsertQuery();

    try {
      this.connection = source.getConnection();
    } catch (SQLException e) {
      throw UserException.connectionError()
        .message("Unable to open JDBC connection for writing.")
        .addContext(e.getSQLState())
        .build(logger);
    }
  }

  @Override
  public void init(Map<String, String> writerOptions) {
    // Nothing to see here...
  }

  @Override
  public void updateSchema(VectorAccessible batch) {
    BatchSchema schema = batch.getSchema();
    String columnName;
    MinorType type;
    String sql;
    boolean nullable = false;
    CreateTableStmtBuilder queryBuilder = new CreateTableStmtBuilder(tableName, dialect);

    for (MaterializedField field : schema) {
      columnName = JdbcDDLQueryUtils.addBackTicksToField(field.getName());
      type = field.getType().getMinorType();
      logger.debug("Adding column {} of type {}.", columnName, type);

      if (field.getType().getMode() == DataMode.REPEATED) {
        throw UserException.dataWriteError()
          .message("Drill does not yet support writing arrays to JDBC. " + columnName + " is an array.")
          .build(logger);
      }

      if (field.getType().getMode() == DataMode.OPTIONAL) {
        nullable = true;
      }

      int precision = field.getPrecision();
      int scale = field.getScale();

      queryBuilder.addColumn(columnName, field.getType().getMinorType(), nullable, precision, scale);
    }

    sql = queryBuilder.build().getCreateTableQuery();
    sql = JdbcDDLQueryUtils.cleanDDLQuery(sql, dialect);
    logger.debug("Final query: {}", sql);

    // Execute the query to build the schema
    try (Statement statement = connection.createStatement()) {
      logger.debug("Executing CREATE query: {}", sql);
      statement.execute(sql);
    } catch (SQLException e) {
      throw UserException.dataReadError(e)
        .message("The JDBC storage plugin failed while trying to create the schema. ")
        .addContext("Sql", sql)
        .build(logger);
    }
  }

  @Override
  public void startRecord() {
    rowList.clear();

    if (!firstRecord) {
      insertQueryBuilder.append(",");
    }
    insertQueryBuilder.append("(");
    logger.debug("Start record");
  }

  @Override
  public void endRecord() throws IOException {
    logger.debug("Ending record");

    // Add values to rowString
    for (int i = 0; i < rowList.size(); i++) {
      if (i > 0) {
        insertQueryBuilder.append(",");
      }

      // Add null value to rowstring
      if (rowList.get(i) instanceof String && ((String) rowList.get(i)).equalsIgnoreCase("null")) {
        insertQueryBuilder.append("null");
        continue;
      }

      JdbcWriterField currentField = fields.get(i);
      if (currentField.getDataType() == MinorType.VARCHAR) {
        String value = null;
        // Get the string value
        if (currentField.getMode() == DataMode.REQUIRED) {
          VarCharHolder varCharHolder = (VarCharHolder) rowList.get(i);
          value = StringFunctionHelpers.getStringFromVarCharHolder(varCharHolder);
        } else {
          try {
            NullableVarCharHolder nullableVarCharHolder = (NullableVarCharHolder) rowList.get(i);
            value = StringFunctionHelpers.getStringFromVarCharHolder(nullableVarCharHolder);
          } catch (ClassCastException e) {
            logger.error("Unable to read field: {}",  rowList.get(i));
          }
        }

        // Add to value string
        insertQueryBuilder.literal(value);
      } else if (currentField.getDataType() == MinorType.DATE) {
        String dateString = formatDateForInsertQuery((Long) rowList.get(i));
        insertQueryBuilder.literal(dateString);
      } else if (currentField.getDataType() == MinorType.TIME) {
        String timeString = formatTimeForInsertQuery((Integer) rowList.get(i));
        insertQueryBuilder.literal(timeString);
      } else if (currentField.getDataType() == MinorType.TIMESTAMP) {
        String timeString = formatTimeStampForInsertQuery((Long) rowList.get(i));
        insertQueryBuilder.literal(timeString);
      } else {
        if (Strings.isNullOrEmpty(rowList.get(i).toString())) {
          insertQueryBuilder.append("null");
        } else {
          insertQueryBuilder.append(rowList.get(i).toString());
        }
      }
    }

    recordCount++;
    firstRecord = false;
    insertQueryBuilder.append(")");

    if (recordCount >= config.getPlugin().getConfig().getWriterBatchSize()) {
      // Execute the insert query
      String insertQuery = insertQueryBuilder.toString();
      executeInsert(insertQuery);

      // Reset the batch
      recordCount = 0;
      firstRecord = true;
      insertQueryBuilder = initializeInsertQuery();
    }

    rowList.clear();
  }

  @Override
  public void abort() {
    logger.debug("Abort insert.");
  }

  @Override
  public void cleanup() throws IOException {
    logger.debug("Cleanup record");
    // Execute last query
    String insertQuery = insertQueryBuilder.toString();
    if (recordCount != 0) {
      executeInsert(insertQuery);
    }
    AutoCloseables.closeSilently(connection);
  }

  private void executeInsert(String insertQuery) throws IOException {
    try (Statement stmt = connection.createStatement()) {
      logger.debug("Executing insert query: {}", insertQuery);
      stmt.execute(insertQuery);
      logger.debug("Query complete");
      // Close connection
      AutoCloseables.closeSilently(stmt);
    } catch (SQLException e) {
      logger.error("Error: {} {} {}", e.getMessage(), e.getSQLState(), e.getErrorCode());
      AutoCloseables.closeSilently(connection);
      throw new IOException(e.getMessage() + " " + e.getSQLState() + "\n" + insertQuery);
    }
  }

  private SqlBuilder initializeInsertQuery() {
    SqlBuilder builder = new SqlBuilder(this.dialect);

    // Apache Phoenix does not support INSERT but does support UPSERT using the same syntax
    if (dialect == DatabaseProduct.PHOENIX.getDialect()) {
      builder.append("UPSERT INTO ");
    } else {
      builder.append("INSERT INTO ");
    }

    JdbcDDLQueryUtils.addTableToInsertQuery(builder, rawTableName);
    builder.append (" VALUES ");
    return builder;
  }

  /**
   * Drill returns longs for date values. This function converts longs into dates formatted
   * in YYYY-MM-dd format for insertion into a database.
   * @param dateVal long representing a naive date
   * @return A date string formatted YYYY-MM-dd
   */
  private String formatDateForInsertQuery(Long dateVal) {
    Date date=new Date(dateVal);
    SimpleDateFormat df2 = new SimpleDateFormat("yyyy-MM-dd");
    return df2.format(date);
  }

  /**
   * Drill returns longs for time values. This function converts longs into times formatted
   * in HH:mm:ss format for insertion into a database.
   * @param millis Milliseconds since the epoch.
   * @return A time string formatted for insertion into a database.
   */
  private String formatTimeForInsertQuery(Integer millis) {
    return String.format("%02d:%02d:%02d", TimeUnit.MILLISECONDS.toHours(millis),
      TimeUnit.MILLISECONDS.toMinutes(millis) % TimeUnit.HOURS.toMinutes(1),
      TimeUnit.MILLISECONDS.toSeconds(millis) % TimeUnit.MINUTES.toSeconds(1));
  }

  /**
   * Drill returns longs for date times. This function converts
   * @param time An input long that represents a timestamp
   * @return A ISO formatted timestamp.
   */
  private String formatTimeStampForInsertQuery(Long time) {
    Date date = new Date(time);
    Format format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    return format.format(date);
  }

  @Override
  public FieldConverter getNewNullableIntConverter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableIntJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableIntJDBCConverter extends FieldConverter {
    private final NullableIntHolder holder = new NullableIntHolder();

    public NullableIntJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.INT, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewIntConverter(int fieldId, String fieldName, FieldReader reader) {
    return new IntJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class IntJDBCConverter extends FieldConverter {
    private final IntHolder holder = new IntHolder();

    public IntJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.INT, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewNullableBigIntConverter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableBigIntJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableBigIntJDBCConverter extends FieldConverter {
    private final NullableBigIntHolder holder = new NullableBigIntHolder();

    public NullableBigIntJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.BIGINT, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewBigIntConverter(int fieldId, String fieldName, FieldReader reader) {
    return new BigIntJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class BigIntJDBCConverter extends FieldConverter {
    private final BigIntHolder holder = new BigIntHolder();

    public BigIntJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.BIGINT, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewNullableSmallIntConverter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableSmallIntJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableSmallIntJDBCConverter extends FieldConverter {
    private final NullableSmallIntHolder holder = new NullableSmallIntHolder();

    public NullableSmallIntJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.SMALLINT, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewSmallIntConverter(int fieldId, String fieldName, FieldReader reader) {
    return new SmallIntJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class SmallIntJDBCConverter extends FieldConverter {
    private final SmallIntHolder holder = new SmallIntHolder();

    public SmallIntJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.SMALLINT, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewNullableTinyIntConverter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableTinyIntJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableTinyIntJDBCConverter extends FieldConverter {
    private final NullableTinyIntHolder holder = new NullableTinyIntHolder();

    public NullableTinyIntJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.TINYINT, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewTinyIntConverter(int fieldId, String fieldName, FieldReader reader) {
    return new TinyIntJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class TinyIntJDBCConverter extends FieldConverter {
    private final TinyIntHolder holder = new TinyIntHolder();

    public TinyIntJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.TINYINT, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewNullableFloat4Converter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableFloat4JDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableFloat4JDBCConverter extends FieldConverter {
    private final NullableFloat4Holder holder = new NullableFloat4Holder();

    public NullableFloat4JDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.FLOAT4, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewFloat4Converter(int fieldId, String fieldName, FieldReader reader) {
    return new Float4JDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class Float4JDBCConverter extends FieldConverter {
    private final Float4Holder holder = new Float4Holder();

    public Float4JDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.FLOAT4, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewNullableFloat8Converter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableFloat8JDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableFloat8JDBCConverter extends FieldConverter {
    private final NullableFloat8Holder holder = new NullableFloat8Holder();

    public NullableFloat8JDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.FLOAT8, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewFloat8Converter(int fieldId, String fieldName, FieldReader reader) {
    return new Float8JDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class Float8JDBCConverter extends FieldConverter {
    private final Float8Holder holder = new Float8Holder();

    public Float8JDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.FLOAT8, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewNullableVarDecimalConverter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableVardecimalJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableVardecimalJDBCConverter extends FieldConverter {
    private final NullableVarDecimalHolder holder = new NullableVarDecimalHolder();

    public NullableVardecimalJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.VARDECIMAL, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      BigDecimal value = DecimalUtility.getBigDecimalFromDrillBuf(holder.buffer,
        holder.start, holder.end - holder.start, holder.scale);
      rowList.add(value);
    }
  }

  @Override
  public FieldConverter getNewVarDecimalConverter(int fieldId, String fieldName, FieldReader reader) {
    return new VardecimalJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class VardecimalJDBCConverter extends FieldConverter {
    private final VarDecimalHolder holder = new VarDecimalHolder();

    public VardecimalJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.VARDECIMAL, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      BigDecimal value = DecimalUtility.getBigDecimalFromDrillBuf(holder.buffer,
        holder.start, holder.end - holder.start, holder.scale);
      rowList.add(value);
    }
  }

  @Override
  public FieldConverter getNewNullableVarCharConverter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableVarCharJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableVarCharJDBCConverter extends FieldConverter {
    private final NullableVarCharHolder holder = new NullableVarCharHolder();

    public NullableVarCharJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.VARCHAR, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      if (reader.isSet()) {
        byte[] bytes = new byte[holder.end - holder.start];
        holder.buffer.getBytes(holder.start, bytes);
      }
      rowList.add(holder);
    }
  }

  @Override
  public FieldConverter getNewVarCharConverter(int fieldId, String fieldName, FieldReader reader) {
    return new VarCharJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class VarCharJDBCConverter extends FieldConverter {
    private final VarCharHolder holder = new VarCharHolder();

    public VarCharJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.VARCHAR, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      if (reader.isSet()) {
        byte[] bytes = new byte[holder.end - holder.start];
        holder.buffer.getBytes(holder.start, bytes);
        rowList.add(holder);
      }
    }
  }

  @Override
  public FieldConverter getNewNullableDateConverter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableDateJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableDateJDBCConverter extends FieldConverter {
    private final NullableDateHolder holder = new NullableDateHolder();

    public NullableDateJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.DATE, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewDateConverter(int fieldId, String fieldName, FieldReader reader) {
    return new DateJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class DateJDBCConverter extends FieldConverter {
    private final DateHolder holder = new DateHolder();

    public DateJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.DATE, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewNullableTimeConverter(int fieldId, String fieldName, FieldReader reader) {
      return new NullableTimeJDBCConverter(fieldId, fieldName, reader, fields);
    }

    public class NullableTimeJDBCConverter extends FieldConverter {
    private final NullableTimeHolder holder = new NullableTimeHolder();

    public NullableTimeJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.TIME, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewTimeConverter(int fieldId, String fieldName, FieldReader reader) {
    return new TimeJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class TimeJDBCConverter extends FieldConverter {
    private final TimeHolder holder = new TimeHolder();

    public TimeJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.TIME, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewNullableTimeStampConverter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableTimeStampJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableTimeStampJDBCConverter extends FieldConverter {
    private final NullableTimeStampHolder holder = new NullableTimeStampHolder();

    public NullableTimeStampJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.TIMESTAMP, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewTimeStampConverter(int fieldId, String fieldName, FieldReader reader) {
    return new TimeStampJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class TimeStampJDBCConverter extends FieldConverter {
    private final TimeStampHolder holder = new TimeStampHolder();

    public TimeStampJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.TIMESTAMP, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      rowList.add(holder.value);
    }
  }

  @Override
  public FieldConverter getNewNullableBitConverter(int fieldId, String fieldName, FieldReader reader) {
    return new NullableBitJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class NullableBitJDBCConverter extends FieldConverter {
    private final NullableBitHolder holder = new NullableBitHolder();

    public NullableBitJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.BIT, DataMode.OPTIONAL));
    }

    @Override
    public void writeField() {
      if (!reader.isSet()) {
        rowList.add("null");
        return;
      }
      reader.read(holder);
      String booleanValue = "false";
      if (holder.value == 1) {
        booleanValue = "true";
      }
      rowList.add(booleanValue);
    }
  }
  @Override
  public FieldConverter getNewBitConverter(int fieldId, String fieldName, FieldReader reader) {
    return new BitJDBCConverter(fieldId, fieldName, reader, fields);
  }

  public class BitJDBCConverter extends FieldConverter {
    private final BitHolder holder = new BitHolder();

    public BitJDBCConverter(int fieldID, String fieldName, FieldReader reader, List<JdbcWriterField> fields) {
      super(fieldID, fieldName, reader);
      fields.add(new JdbcWriterField(fieldName, MinorType.BIT, DataMode.REQUIRED));
    }

    @Override
    public void writeField() {
      reader.read(holder);
      String booleanValue = "false";
      if (holder.value == 1) {
        booleanValue = "true";
      }
      rowList.add(booleanValue);
    }
  }
}