jdbc-protolib/src/main/java/com/streamsets/pipeline/lib/jdbc/parser/sql/DateTimeColumnHandler.java
public Field getDateTimeStampField(
      String column,
      String columnValue,
      int columnType,
      String actualType
  ) throws StageException {
    Field.Type type;
    if (DATE.equalsIgnoreCase(actualType)) {
      type = Field.Type.DATE;
    } else if (TIME.equalsIgnoreCase(actualType)) {
      type = Field.Type.TIME;
    } else if (TIMESTAMP.equalsIgnoreCase(actualType)) {
      type = Field.Type.DATETIME;
    } else {
      throw new StageException(JDBC_37, columnType, column);
    }
    if (columnValue == null) {
      return Field.create(type, null);
    } else {
      Optional<String> ts = matchDateTimeString(toTimestampPattern.matcher(columnValue));
      if (ts.isPresent()) {
        if (timestampAsString) {
          return Field.create(Field.Type.STRING, ts.get());
        }
        Timestamp timestamp = Timestamp.valueOf(ts.get());
        Field field = Field.create(type, timestamp);
        JdbcUtil.setNanosecondsinAttribute(timestamp.getNanos(), field);
        return field;
      }
      // We did not find TO_TIMESTAMP, so try TO_DATE
      Optional<String> dt = matchDateTimeString(toDatePattern.matcher(columnValue));
      return Field.create(Field.Type.DATE,
          dt.map(s -> Date.from(getDate(s).atZone(zoneId).toInstant())).orElse(null));
    }
  }
public Field getDateTimeStampField(
      String column,
      String columnValue,
      int columnType,
      String actualType
  ) throws StageException {
    Field.Type type;
    if (DATE.equalsIgnoreCase(actualType)) {
      type = Field.Type.DATE;
    } else if (TIME.equalsIgnoreCase(actualType)) {
      type = Field.Type.TIME;
    } else if (TIMESTAMP.equalsIgnoreCase(actualType)) {
      type = Field.Type.DATETIME;
    } else {
      throw new StageException(JDBC_37, columnType, column);
    }
    if (columnValue == null) {
      return Field.create(type, null);
    } else {
      Optional<String> ts = matchDateTimeString(toTimestampPattern.matcher(columnValue));
      if (ts.isPresent()) {
        if (timestampAsString) {
          return Field.create(Field.Type.STRING, ts.get());
        }
        Timestamp timestamp = Timestamp.valueOf(ts.get());
        Field field = Field.create(type, timestamp);
        JdbcUtil.setNanosecondsinAttribute(timestamp.getNanos(), field);
        return field;
      }
      // We did not find TO_TIMESTAMP, so try TO_DATE
      Optional<String> dt = matchDateTimeString(toDatePattern.matcher(columnValue));
      return Field.create(Field.Type.DATE,
          dt.map(s -> Date.from(getDate(s).atZone(zoneId).toInstant())).orElse(null));
    }
  }
This method returns an {@linkplain Field} that represents a DATE, TIME or TIMESTAMP. It is possible for user to upgrade
a field from DATE to TIMESTAMP, and if we read the table schema on startup after this upgrade, we would assume the field
should be returned as DATETIME field. But it is possible that the first change we read was made before the upgrade from
DATE to TIMESTAMP. So we check whether the returned SQL has TO_TIMESTAMP - if it does we return it as DATETIME, else we
return it as DATE.