@Override
  public JdbcRecordIterator
    getRecordIterator(Configuration conf, String partitionColumn, String lowerBound, String upperBound, int limit, int
          offset) throws
          HiveJdbcDatabaseAccessException {

    Connection conn = null;
    PreparedStatement ps = null;
    ResultSet rs = null;

    try {
      initializeDatabaseConnection(conf);
      String tableName = getQualifiedTableName(conf);
      // Always use JDBC_QUERY if available both for correctness and performance. JDBC_QUERY can be set by the user
      // or the CBO including pushdown optimizations. SELECT all query should be used only when JDBC_QUERY is null.
      String sql = firstNonNull(conf.get(Constants.JDBC_QUERY), selectAllFromTable(tableName));
      String partitionQuery;
      if (partitionColumn != null) {
        partitionQuery = addBoundaryToQuery(tableName, sql, partitionColumn, lowerBound, upperBound);
      } else {
        partitionQuery = addLimitAndOffsetToQuery(sql, limit, offset);
      }
      LOGGER.info("Query to execute is [{}]", partitionQuery);

      conn = dbcpDataSource.getConnection();
      ps = conn.prepareStatement(partitionQuery, ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY);
      ps.setFetchSize(getFetchSize(conf));
      rs = ps.executeQuery();

      return new JdbcRecordIterator(this, conn, ps, rs, conf);
    }
    catch (Exception e) {
      LOGGER.error("Caught exception while trying to execute query", e);
      cleanupResources(conn, ps, rs);
      throw new HiveJdbcDatabaseAccessException("Caught exception while trying to execute query: " + e.getMessage(), e);
    }
  }
@Test
  public void testGetRecordIterator() throws HiveJdbcDatabaseAccessException {
    Configuration conf = buildConfiguration();
    DatabaseAccessor accessor = DatabaseAccessorFactory.getAccessor(conf);
    JdbcRecordIterator iterator = accessor.getRecordIterator(conf, null, null, null,2, 0);

    assertThat(iterator, is(notNullValue()));

    int count = 0;
    while (iterator.hasNext()) {
      Map<String, Object> record = iterator.next();
      count++;

      assertThat(record, is(notNullValue()));
      assertThat(record.size(), is(equalTo(7)));
      assertThat(record.get("strategy_id"), is(equalTo(count)));
    }

    assertThat(count, is(equalTo(2)));
    iterator.close();
  }