@Test(dataProvider = "executeWithPreFetchDP") public void testExecuteWithPreFetch(int rowsToPreFecth, boolean isComplteleyFetched, int rowsPreFetched, boolean createTable, long executeTimeoutMillis) throws Exception { if (createTable) { createTable("execute_prefetch_test"); insertData("execute_prefetch_test"); } final String query = "SELECT * FROM execute_prefetch_test"; Configuration conf = new Configuration(baseConf); conf.setBoolean(LensConfConstants.QUERY_PERSISTENT_RESULT_SET, true); conf.setBoolean(LensConfConstants.QUERY_PERSISTENT_RESULT_INDRIVER, false); conf.setBoolean(LensConfConstants.PREFETCH_INMEMORY_RESULTSET, true); conf.setInt(LensConfConstants.PREFETCH_INMEMORY_RESULTSET_ROWS, rowsToPreFecth); QueryContext context = createQueryContext(query, conf); context.setExecuteTimeoutMillis(executeTimeoutMillis); driver.executeAsync(context); while (!context.getDriverStatus().isFinished()) { Thread.sleep(1000); } LensResultSet resultSet = driver.fetchResultSet(context); assertNotNull(resultSet); if (executeTimeoutMillis > 1000) { assertTrue(resultSet instanceof PartiallyFetchedInMemoryResultSet); } else { assertFalse(resultSet instanceof PartiallyFetchedInMemoryResultSet); return; } PartiallyFetchedInMemoryResultSet prs = (PartiallyFetchedInMemoryResultSet) resultSet; assertEquals(prs.isComplteleyFetched(), isComplteleyFetched); if (isComplteleyFetched) { assertTrue(prs.isComplteleyFetched()); prs.getPreFetchedRows(); assertEquals(prs.size().intValue(), rowsPreFetched); } else { assertFalse(prs.isComplteleyFetched()); assertEquals(prs.getPreFetchedRows().size(), rowsPreFetched); } assertEquals(prs.getMetadata().getColumns().size(), 1); assertEquals(prs.getMetadata().getColumns().get(0).getName(), "ID"); int rowCount = 0; while (prs.hasNext()) { ResultRow row = prs.next(); assertEquals(row.getValues().get(0), rowCount); rowCount++; } assertEquals(rowCount, 10); prs.setFullyAccessed(true); assertEquals(prs.canBePurged(), true); }
@Override public void executeAsync(QueryContext context) throws LensException { checkConfigured(); String rewrittenQuery = rewriteQuery(context); JdbcQueryContext jdbcCtx = new JdbcQueryContext(context); jdbcCtx.setRewrittenQuery(rewrittenQuery); try { Future<QueryResult> future = asyncQueryPool.submit(new QueryCallable(jdbcCtx, logSegregationContext)); jdbcCtx.setResultFuture(future); } catch (RejectedExecutionException e) { log.error("Query execution rejected: {} reason:{}", context.getQueryHandle(), e.getMessage(), e); throw new LensException("Query execution rejected: " + context.getQueryHandle() + " reason:" + e.getMessage(), e); } queryContextMap.put(context.getQueryHandle(), jdbcCtx); log.info("{} ExecuteAsync: {}", getFullyQualifiedName(), context.getQueryHandle()); }
JDBCDriver extends AbstractLensDriver { @Override public void executeAsync(QueryContext context) throws LensException { checkConfigured(); String rewrittenQuery = rewriteQuery(context); JdbcQueryContext jdbcCtx = new JdbcQueryContext(context); jdbcCtx.setRewrittenQuery(rewrittenQuery); try { Future<QueryResult> future = asyncQueryPool.submit(new QueryCallable(jdbcCtx, logSegregationContext)); jdbcCtx.setResultFuture(future); } catch (RejectedExecutionException e) { log.error("Query execution rejected: {} reason:{}", context.getQueryHandle(), e.getMessage(), e); throw new LensException("Query execution rejected: " + context.getQueryHandle() + " reason:" + e.getMessage(), e); } queryContextMap.put(context.getQueryHandle(), jdbcCtx); log.info("{} ExecuteAsync: {}", getFullyQualifiedName(), context.getQueryHandle()); } }
JDBCDriver extends AbstractLensDriver { @Override public void executeAsync(QueryContext context) throws LensException { checkConfigured(); String rewrittenQuery = rewriteQuery(context); JdbcQueryContext jdbcCtx = new JdbcQueryContext(context); jdbcCtx.setRewrittenQuery(rewrittenQuery); try { Future<QueryResult> future = asyncQueryPool.submit(new QueryCallable(jdbcCtx, logSegregationContext)); jdbcCtx.setResultFuture(future); } catch (RejectedExecutionException e) { log.error("Query execution rejected: {} reason:{}", context.getQueryHandle(), e.getMessage(), e); throw new LensException("Query execution rejected: " + context.getQueryHandle() + " reason:" + e.getMessage(), e); } queryContextMap.put(context.getQueryHandle(), jdbcCtx); log.info("{} ExecuteAsync: {}", getFullyQualifiedName(), context.getQueryHandle()); }  }
JDBCDriver extends AbstractLensDriver { @Override public void executeAsync(QueryContext context) throws LensException { checkConfigured(); String rewrittenQuery = rewriteQuery(context); JdbcQueryContext jdbcCtx = new JdbcQueryContext(context); jdbcCtx.setRewrittenQuery(rewrittenQuery); try { Future<QueryResult> future = asyncQueryPool.submit(new QueryCallable(jdbcCtx, logSegregationContext)); jdbcCtx.setResultFuture(future); } catch (RejectedExecutionException e) { log.error("Query execution rejected: {} reason:{}", context.getQueryHandle(), e.getMessage(), e); throw new LensException("Query execution rejected: " + context.getQueryHandle() + " reason:" + e.getMessage(), e); } queryContextMap.put(context.getQueryHandle(), jdbcCtx); log.info("{} ExecuteAsync: {}", getFullyQualifiedName(), context.getQueryHandle()); }  @Override void configure(Configuration conf, String driverType, String driverName); void init(); QueryCost calculateQueryCost(AbstractQueryContext qctx); @Override QueryCost estimate(AbstractQueryContext qctx); @Override DriverQueryPlan explain(AbstractQueryContext explainCtx); void validate(AbstractQueryContext pContext); @Override void prepare(PreparedQueryContext pContext); @Override DriverQueryPlan explainAndPrepare(PreparedQueryContext pContext); @Override void closePreparedQuery(QueryPrepareHandle handle); @Override LensResultSet execute(QueryContext context); @Override void executeAsync(QueryContext context); @Override void updateStatus(QueryContext context); @Override void closeResultSet(QueryHandle handle); @Override boolean cancelQuery(QueryHandle handle); @Override void closeQuery(QueryHandle handle); @Override void close(); @Override void registerDriverEventListener(LensEventListener<DriverEvent> driverEventListener); @Override void readExternal(ObjectInput arg0); @Override void writeExternal(ObjectOutput arg0); @Override StatusUpdateMethod getStatusUpdateMethod(); }
JDBCDriver extends AbstractLensDriver { @Override public void executeAsync(QueryContext context) throws LensException { checkConfigured(); String rewrittenQuery = rewriteQuery(context); JdbcQueryContext jdbcCtx = new JdbcQueryContext(context); jdbcCtx.setRewrittenQuery(rewrittenQuery); try { Future<QueryResult> future = asyncQueryPool.submit(new QueryCallable(jdbcCtx, logSegregationContext)); jdbcCtx.setResultFuture(future); } catch (RejectedExecutionException e) { log.error("Query execution rejected: {} reason:{}", context.getQueryHandle(), e.getMessage(), e); throw new LensException("Query execution rejected: " + context.getQueryHandle() + " reason:" + e.getMessage(), e); } queryContextMap.put(context.getQueryHandle(), jdbcCtx); log.info("{} ExecuteAsync: {}", getFullyQualifiedName(), context.getQueryHandle()); }  @Override void configure(Configuration conf, String driverType, String driverName); void init(); QueryCost calculateQueryCost(AbstractQueryContext qctx); @Override QueryCost estimate(AbstractQueryContext qctx); @Override DriverQueryPlan explain(AbstractQueryContext explainCtx); void validate(AbstractQueryContext pContext); @Override void prepare(PreparedQueryContext pContext); @Override DriverQueryPlan explainAndPrepare(PreparedQueryContext pContext); @Override void closePreparedQuery(QueryPrepareHandle handle); @Override LensResultSet execute(QueryContext context); @Override void executeAsync(QueryContext context); @Override void updateStatus(QueryContext context); @Override void closeResultSet(QueryHandle handle); @Override boolean cancelQuery(QueryHandle handle); @Override void closeQuery(QueryHandle handle); @Override void close(); @Override void registerDriverEventListener(LensEventListener<DriverEvent> driverEventListener); @Override void readExternal(ObjectInput arg0); @Override void writeExternal(ObjectOutput arg0); @Override StatusUpdateMethod getStatusUpdateMethod(); static final AtomicInteger THID; }