ExecutorQueryService implements Query.Service { @Override public Query prepareQuery( String sqlString, int maxRows, Map<String, String> parameters ) throws KettleException { SQL sql = new SQL( sqlString ); Query query; try { IMetaStore metaStore = metastoreLocator != null ? metastoreLocator.getMetastore() : null; DataServiceExecutor executor = resolver.createBuilder( sql ) .rowLimit( maxRows ) .parameters( parameters ) .metastore( metaStore ) .build(); query = new ExecutorQuery( executor ); } catch ( Exception e ) { Throwables.propagateIfInstanceOf( e, KettleException.class ); throw new KettleException( e ); } return query; } ExecutorQueryService( DataServiceResolver resolver, MetastoreLocator metastoreLocator ); @Override Query prepareQuery( String sqlString, int maxRows, Map<String, String> parameters ); @Override Query prepareQuery( String sqlString, IDataServiceClientService.StreamingMode windowMode,
                                       long windowSize, long windowEvery, long windowLimit,
                                       final Map<String, String> parameters ); static DataOutputStream asDataOutputStream( OutputStream outputStream );  }
@Test public void testQueryBuildsWithMetastore() throws Exception { result = executorQueryService.prepareQuery( sql.getSqlString(), rowLimit, parameters ); assertEquals( ImmutableList.of( serviceTrans, genTrans ), result.getTransList() ); verify( builder ).rowLimit( rowLimit ); verify( builder ).parameters( parameters ); verify( builder ).metastore( metastore ); }
@Test public void testQueryBuildsWithWindowRowBased() throws Exception { result = executorQueryService.prepareQuery( sql.getSqlString(), IDataServiceClientService.StreamingMode.ROW_BASED, 10, 1, 15000, parameters ); assertEquals( ImmutableList.of( serviceTrans, genTrans ), result.getTransList() ); verify( builder ).rowLimit( 0 ); verify( builder ).parameters( parameters ); verify( builder ).metastore( metastore ); verify( builder ).windowLimit( 15000 ); verify( builder ).windowSize( 10 ); verify( builder ).windowEvery( 1 ); verify( builder ).windowMode( IDataServiceClientService.StreamingMode.ROW_BASED ); }
@Test public void testQueryBuildsWithWindowTimeBased() throws Exception { result = executorQueryService.prepareQuery( sql.getSqlString(), IDataServiceClientService.StreamingMode.TIME_BASED, 100, 10, 22000, parameters ); assertEquals( ImmutableList.of( serviceTrans, genTrans ), result.getTransList() ); verify( builder ).rowLimit( 0 ); verify( builder ).parameters( parameters ); verify( builder ).metastore( metastore ); verify( builder ).windowLimit( 22000 ); verify( builder ).windowSize( 100 ); verify( builder ).windowEvery( 10 ); verify( builder ).windowMode( IDataServiceClientService.StreamingMode.TIME_BASED ); }
@Test public void testExecutorQueryInnerClass() throws Exception { SQL sql = new SQL( "select field from table" ); ExecutorQueryService executorQueryService = new ExecutorQueryService( dataServiceResolver, metastoreLocator ); Query executorQuery = executorQueryService.prepareQuery( sql.getSqlString(), rowLimit, parameters ); DataOutputStream dataOutputStreamMock = mock( DataOutputStream.class ); doReturn( dataServiceExecutor ).when( dataServiceExecutor ).executeQuery( dataOutputStreamMock ); executorQuery.writeTo( dataOutputStreamMock ); verify( dataServiceExecutor, times( 1 ) ).waitUntilFinished(); }
@Test public void testExecutorQueryInnerClassNullExecutor() throws Exception { SQL sql = new SQL( "select field from table" ); ExecutorQueryService executorQueryService = new ExecutorQueryService( dataServiceResolver, metastoreLocator ); Query executorQuery = executorQueryService.prepareQuery( sql.getSqlString(), rowLimit, parameters ); DataOutputStream dataOutputStreamMock = mock( DataOutputStream.class ); doReturn( null ).when( dataServiceExecutor ).executeQuery( dataOutputStreamMock ); executorQuery.writeTo( dataOutputStreamMock ); verify( dataServiceExecutor, times( 0 ) ).waitUntilFinished(); }