ServiceCache extends StepOptimization { @Override public boolean activate( final DataServiceExecutor executor, StepInterface stepInterface ) { if ( !executor.getService().isStreaming() ) { final LogChannelInterface logChannel = executor.getGenTrans().getLogChannel(); for ( CachedService availableCache : getAvailableCache( executor ).values() ) { try { ListenableFuture<Integer> replay = factory.createCachedServiceLoader( availableCache ).replay( executor ); addReplayCallback( logChannel, replay ); return true; } catch ( Throwable e ) { logChannel.logError( "Unable to replay from cache", e ); } } CachedService.CacheKey rootKey = createRootKey( executor ); final Map<CachedService.CacheKey, ServiceObserver> runningServices = factory.getRunningServices(); if ( runningServices.containsKey( rootKey ) ) { try { ListenableFuture<Integer> replay = factory.createCachedServiceLoader( () -> runningServices.get( rootKey ).rows() ).replay( executor ); addReplayCallback( logChannel, replay ); return true; } catch ( KettleException e ) { logChannel.logError( "Unable to replay from running service" ); } } ServiceObserver serviceObserver = factory.createObserver( executor ); if ( CachedService.calculateRank( executor ) == Integer.MAX_VALUE ) { runningServices.put( rootKey, serviceObserver ); } Futures.addCallback( serviceObserver.install(), new FutureCallback<CachedService>() { @Override public void onSuccess( CachedService result ) { if ( executor.isStopped() || executor.hasErrors() ) { runningServices.remove( rootKey ); return; } Cache<CachedService.CacheKey, CachedService> cache = factory.getCache( ServiceCache.this, executor.getServiceName() ); CachedService.CacheKey key = createRootKey( executor ); if ( result.isComplete() ) { key = key.withoutOrder(); } if ( cache.putIfAbsent( key, result ) ) { logChannel.logBasic( "Service Transformation results cached", key ); } else { try { CachedService existing = checkNotNull( cache.get( key ) ); if ( !existing.answersQuery( executor ) && cache.replace( key, existing, result ) ) { logChannel.logBasic( "Service Transformation cache updated", key ); } else { logChannel.logDetailed( "Service Transformation cache was not updated", key ); } } catch ( Throwable t ) { onFailure( t ); } } runningServices.remove( rootKey ); } @Override public void onFailure( Throwable t ) { runningServices.remove( rootKey ); logChannel.logError( "Cache failed to observe service transformation", t ); } }, factory.getExecutorService() ); } return false; } ServiceCache( ServiceCacheFactory factory ); @Override void init( TransMeta transMeta, DataServiceMeta dataService, PushDownOptimizationMeta optMeta ); @Override boolean activate( final DataServiceExecutor executor, StepInterface stepInterface ); @Override OptimizationImpactInfo preview( DataServiceExecutor executor, StepInterface stepInterface ); CachedService.CacheKey createRootKey( DataServiceExecutor executor ); String getTemplateName(); void setTemplateName( String templateName ); String getTimeToLive(); void setTimeToLive( String timeToLive ); String getConfiguredTimeToLive(); Map<String, String> getTemplateOverrides(); static final String NAME; static final String SERVICE_CACHE_TEMPLATE_NAME; static final String SERVICE_CACHE_TTL; }
@Test public void testActivateObserve() throws Exception { DataServiceExecutor executor = dataServiceExecutor( "SELECT * FROM MOCK_SERVICE ORDER BY ID" ); CachedService.CacheKey key = CachedService.CacheKey.create( executor ); CachedService cachedService = CachedService.complete( ImmutableList.<RowMetaAndData>of() ); ServiceObserver observer = mock( ServiceObserver.class ); when( factory.createObserver( executor ) ).thenReturn( observer ); when( observer.install() ).thenReturn( Futures.immediateFuture( cachedService ) ); when( cache.get( any( CachedService.CacheKey.class ) ) ).thenReturn( null ); when( cache.putIfAbsent( key.withoutOrder(), cachedService ) ).thenReturn( true ); assertThat( serviceCache.activate( executor, serviceStep ), is( false ) ); verify( cache ).putIfAbsent( key.withoutOrder(), cachedService ); verifyNoMoreInteractions( ignoreStubs( cache ) ); }
@Test public void testActivateObserveStreaming() throws Exception { DataServiceExecutor executor = dataServiceStreamingExecutor( "SELECT * FROM MOCK_SERVICE ORDER BY ID" ); CachedService.CacheKey key = CachedService.CacheKey.create( executor ); CachedService cachedService = CachedService.complete( ImmutableList.<RowMetaAndData>of() ); ServiceObserver observer = mock( ServiceObserver.class ); when( factory.createObserver( executor ) ).thenReturn( observer ); when( observer.install() ).thenReturn( Futures.immediateFuture( cachedService ) ); when( cache.get( any( CachedService.CacheKey.class ) ) ).thenReturn( null ); when( cache.putIfAbsent( key.withoutOrder(), cachedService ) ).thenReturn( true ); assertThat( serviceCache.activate( executor, serviceStep ), is( false ) ); verify( cache, times( 0 ) ).putIfAbsent( key.withoutOrder(), cachedService ); verifyNoMoreInteractions( ignoreStubs( cache ) ); }
@Test public void testRunningServiceIsRemovedOnError() throws Exception { DataServiceExecutor executor = spy( dataServiceExecutor( "SELECT * FROM MOCK_SERVICE ORDER BY ID" ) ); CachedService cachedService = CachedService.complete( ImmutableList.<RowMetaAndData>of() ); doReturn( true ).when( executor ).hasErrors(); ServiceObserver observer = mock( ServiceObserver.class ); when( factory.createObserver( executor ) ).thenReturn( observer ); Map<CachedService.CacheKey, ServiceObserver> runningServices = new HashMap<>(); when( factory.getRunningServices() ).thenReturn( runningServices ); when( observer.install() ).thenReturn( Futures.immediateFuture( cachedService ) ); assertThat( serviceCache.activate( executor, serviceStep ), is( false ) ); assertTrue( runningServices.isEmpty() ); }
@Test public void testObserveWithoutCondition() throws Exception { String select = "SELECT * FROM MOCK_SERVICE"; String selectWithCondition = select + " WHERE A = 2"; for ( Map.Entry<String, List<PushDownOptimizationMeta>> testEntry : ImmutableMultimap .<String, List<PushDownOptimizationMeta>>builder() .put( select, ImmutableList.of( serviceCacheOpt ) ) .put( select, ImmutableList.of( serviceCacheOpt, otherOpt ) ) .put( selectWithCondition, ImmutableList.of( serviceCacheOpt ) ) .put( selectWithCondition, ImmutableList.<PushDownOptimizationMeta>of() ) .build().entries() ) { dataServiceMeta.setPushDownOptimizationMeta( testEntry.getValue() ); DataServiceExecutor executor = dataServiceExecutor( testEntry.getKey() ); CachedService cachedService = CachedService.complete( ImmutableList.<RowMetaAndData>of() ); CachedService.CacheKey key = CachedService.CacheKey.create( executor ).withoutCondition(); ServiceObserver observer = mock( ServiceObserver.class ); when( factory.createObserver( executor ) ).thenReturn( observer ); when( observer.install() ).thenReturn( Futures.immediateFuture( cachedService ) ); when( cache.get( any( CachedService.CacheKey.class ) ) ).thenReturn( null ); when( cache.putIfAbsent( key, cachedService ) ).thenReturn( true ); assertThat( serviceCache.activate( executor, serviceStep ), is( false ) ); try { verify( cache ).putIfAbsent( key, cachedService ); verify( cache ).getConfiguration( CompleteConfiguration.class ); verifyNoMoreInteractions( ignoreStubs( cache ) ); } catch ( AssertionError e ) { throw new AssertionError( testEntry.toString(), e ); } reset( (Cache<CachedService.CacheKey, CachedService>) cache ); } }
@Test public void testObserveUpdateExisting() throws Exception { DataServiceExecutor executor = dataServiceExecutor( "SELECT * FROM MOCK_SERVICE WHERE A = 2" ); CachedService.CacheKey key = CachedService.CacheKey.create( executor ); CachedService existingCache = mock( CachedService.class ); when( cache.get( key.withoutCondition() ) ).thenReturn( null ); when( cache.get( key ) ).thenReturn( existingCache ); when( existingCache.answersQuery( executor ) ).thenReturn( false ); ServiceObserver observer = mock( ServiceObserver.class ); when( factory.createObserver( executor ) ).thenReturn( observer ); CachedService cachedService = CachedService.complete( ImmutableList.<RowMetaAndData>of() ); when( observer.install() ).thenReturn( Futures.immediateFuture( cachedService ) ); when( cache.putIfAbsent( key, cachedService ) ).thenReturn( false ); when( cache.replace( key, existingCache, cachedService ) ).thenReturn( true ); assertThat( serviceCache.activate( executor, serviceStep ), is( false ) ); verify( cache ).replace( key, existingCache, cachedService ); verifyNoMoreInteractions( ignoreStubs( cache ) ); }
@Test public void testReplay() throws Exception { DataServiceExecutor executor = dataServiceExecutor( "SELECT * FROM MOCK_SERVICE WHERE A = 2" ); CachedService.CacheKey key = CachedService.CacheKey.create( executor ); CachedService cachedService = mock( CachedService.class ); CachedServiceLoader cachedServiceLoader = mock( CachedServiceLoader.class ); when( cache.get( key ) ).thenReturn( cachedService ); when( cachedService.answersQuery( executor ) ).thenReturn( true ); when( factory.createCachedServiceLoader( cachedService ) ).thenReturn( cachedServiceLoader ); when( cachedServiceLoader.replay( executor ) ).thenReturn( Futures.immediateFuture( 2000 ) ); assertThat( serviceCache.activate( executor, serviceStep ), is( true ) ); verify( cachedServiceLoader ).replay( executor ); }
@Test public void testReplayComplete() throws Exception { DataServiceExecutor executor = dataServiceExecutor( "SELECT * FROM MOCK_SERVICE WHERE A = 2" ); CachedService.CacheKey key = CachedService.CacheKey.create( executor ); CachedService cachedService = mock( CachedService.class ); final CachedServiceLoader cachedServiceLoader = mock( CachedServiceLoader.class ); when( cache.get( key ) ).thenReturn( null ); when( cache.get( key.withoutCondition() ) ).thenReturn( cachedService ); when( cachedService.isComplete() ).thenReturn( true ); when( factory.createCachedServiceLoader( cachedService ) ).thenReturn( cachedServiceLoader ); when( cachedServiceLoader.replay( executor ) ).thenReturn( Futures.immediateFuture( 2000 ) ); assertThat( serviceCache.activate( executor, serviceStep ), is( true ) ); verify( cachedServiceLoader ).replay( executor ); }
@Test public void testReplayFromRunning() throws Exception { DataServiceExecutor executor = dataServiceExecutor( "SELECT * FROM MOCK_SERVICE WHERE A = 2" ); CachedService.CacheKey key = CachedService.CacheKey.create( executor ); final CachedServiceLoader cachedServiceLoader = mock( CachedServiceLoader.class ); when( cache.get( key ) ).thenReturn( null ); when( cache.get( key.withoutCondition() ) ).thenReturn( null ); HashMap<CachedService.CacheKey, ServiceObserver> runningServices = new HashMap<>(); ServiceObserver serviceObserver = new ServiceObserver( executor ); runningServices.put( key, serviceObserver ); when( factory.getRunningServices() ).thenReturn( runningServices ); when( factory.createCachedServiceLoader( any( java.util.function.Supplier.class ) ) ).thenReturn( cachedServiceLoader ); when( cachedServiceLoader.replay( executor ) ).thenReturn( Futures.immediateFuture( 2000 ) ); assertThat( serviceCache.activate( executor, serviceStep ), is( true ) ); verify( cachedServiceLoader ).replay( executor ); }