DataServiceExecutor { public Trans getServiceTrans() { return serviceTrans; } private  DataServiceExecutor( Builder builder ); static void writeMetadata( DataOutputStream dos, String... metadatas ); DataServiceExecutor executeQuery( final DataOutputStream dos ); DataServiceExecutor executeQuery( Observer<RowMetaAndData> consumer ); DataServiceExecutor executeStreamingQuery( final Observer<List<RowMetaAndData>> streamingConsumer, boolean pollingMode ); String getStreamingGenTransCacheKey(); DataServiceExecutor executeDefaultQuery( final Observer<RowMetaAndData> consumer ); DataServiceExecutor executeQuery(); void executeListeners( ExecutionPoint... stages ); RowProducer addRowProducer(); void waitUntilFinished(); TransMeta getServiceTransMeta(); TransMeta getGenTransMeta(); DataServiceMeta getService(); Trans getServiceTrans(); Trans getGenTrans(); String getServiceName(); Map<String, String> getParameters(); String getId(); Boolean hasErrors(); static String calculateTransname( SQL sql, boolean isService ); void stop( boolean stopTrans ); boolean isStopped(); SQL getSql(); String getResultStepName(); int getRowLimit(); int getServiceRowLimit(); ListMultimap<ExecutionPoint, Runnable> getListenerMap();  }
@Test public void testBuilderBuildStreamingServiceContext() throws Exception { SQL sql = new SQL( "SELECT * FROM " + DATA_SERVICE_NAME ); when( serviceTransExecutor.getServiceTrans() ).thenReturn( serviceTrans ); when( serviceTransExecutor.getKey() ).thenReturn( key ); when( serviceTrans.getTransMeta() ).thenReturn( dataService.getServiceTrans() ); context.addServiceTransExecutor( serviceTransExecutor ); dataService.setStreaming( true ); IMetaStore metastore = mock( IMetaStore.class ); DataServiceExecutor executor = new DataServiceExecutor.Builder( sql, dataService, context ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). metastore( metastore ). enableMetrics( false ). normalizeConditions( false ). rowLimit( 50 ). windowMode( IDataServiceClientService.StreamingMode.ROW_BASED ). build(); verify( serviceTransExecutor, times( 2 ) ).getServiceTrans(); assertSame( executor.getServiceTrans(), serviceTrans ); when( serviceTrans.getTransMeta() ).thenReturn( serviceTransMetaChanged ); when( serviceTransExecutor.getServiceTrans() ).thenReturn( serviceTransChanged ); executor = new DataServiceExecutor.Builder( sql, dataService, context ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). metastore( metastore ). enableMetrics( false ). normalizeConditions( false ). rowLimit( 50 ). windowMode( IDataServiceClientService.StreamingMode.ROW_BASED ). build(); assertNotSame( executor.getServiceTrans(), serviceTrans ); }
@Test public void testBuilderBuildServiceServiceTransNotNull() throws Exception { SQL sql = new SQL( "SELECT * FROM " + DATA_SERVICE_NAME ); RowMeta rowMeta = new RowMeta(); ValueMetaInterface vm = new ValueMetaString( "aBinaryStoredString" ); vm.setStorageType( ValueMetaInterface.STORAGE_TYPE_BINARY_STRING ); vm.setStorageMetadata( new ValueMetaString() ); rowMeta.addValueMeta( vm ); when( serviceTrans.getTransMeta() ).thenReturn( serviceTransMeta ); when( serviceTransMeta.realClone( false ) ).thenReturn( serviceTransMeta ); when( serviceTransMeta.listVariables() ).thenReturn( new String[]{} ); when( serviceTransMeta.listParameters() ).thenReturn( new String[]{} ); when( serviceTransMeta.getStepFields( DATA_SERVICE_STEP ) ).thenReturn( rowMeta ); dataService.setServiceTrans( serviceTrans.getTransMeta() ); dataService.setStreaming( true ); IMetaStore metastore = mock( IMetaStore.class ); DataServiceExecutor executor = new DataServiceExecutor.Builder( sql, dataService, context ) .sqlTransGenerator( sqlTransGenerator ) .genTrans( genTrans ) .metastore( metastore ) .enableMetrics( false ) .normalizeConditions( false ) .rowLimit( 50 ) .windowMode( IDataServiceClientService.StreamingMode.ROW_BASED ) .build(); assertSame( dataService.getServiceTrans(), serviceTrans.getTransMeta() ); }
@Test public void testStreamServiceCachePushdownParameter() throws Exception { when( serviceTrans.getTransMeta() ).thenReturn( dataService.getServiceTrans() ); IMetaStore metastore = mock( IMetaStore.class ); dataService.setStreaming( true ); dataService.setStreaming( true ); context = Mockito.spy( context ); SQL[] queries = new SQL[] { new SQL( "SELECT * FROM " + DATA_SERVICE_NAME + " where PARAMETER('param_param')='tutuu'" ), new SQL( "SELECT * FROM " + DATA_SERVICE_NAME + " where PARAMETER('param_param')='rururuu'" ), new SQL( "SELECT * FROM " + DATA_SERVICE_NAME + " where p=2 and PARAMETER('param_param') = 'tutuu'" ) }; for ( SQL sql : queries ) { new DataServiceExecutor.Builder( sql, dataService, context ). sqlTransGenerator( sqlTransGenerator ). genTrans( genTrans ). metastore( metastore ). enableMetrics( false ). normalizeConditions( true ). rowLimit( 50 ). windowEvery( 1 ). windowSize( 10 ). windowLimit( 10000 ). windowMode( IDataServiceClientService.StreamingMode.ROW_BASED ). build(); } ArgumentCaptor<StreamServiceKey> keyCaptor = ArgumentCaptor.forClass( StreamServiceKey.class ); verify( context, times( 3 ) ).getServiceTransExecutor( keyCaptor.capture() ); assertNotEquals( keyCaptor.getAllValues().get( 0 ), keyCaptor.getAllValues().get( 1 ) ); assertEquals( keyCaptor.getAllValues().get( 0 ), keyCaptor.getAllValues().get( 2 ) ); }