@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(expected = HiveJdbcDatabaseAccessException.class)
  public void testGetRecordIterator_invalidQuery() throws HiveJdbcDatabaseAccessException {
    Configuration conf = buildConfiguration();
    conf.set(JdbcStorageConfig.QUERY.getPropertyName(), "select * from strategyx");
    DatabaseAccessor accessor = DatabaseAccessorFactory.getAccessor(conf);
    @SuppressWarnings("unused")
      JdbcRecordIterator iterator = accessor.getRecordIterator(conf, null, null, null, 0, 2);
  }