@Test public void testIdMapping() throws Throwable { assertEquals( "version 0 expected", GraphDataVersions.INITIAL.getVersion(), dataMigrationManager.getCurrentVersion( GraphMigrationPlugin.PLUGIN_NAME ) ); final ApplicationScope applicationScope = new ApplicationScopeImpl( IdGenerator.createId( "application" ) ); GraphManager gm = graphManagerFactory.createEdgeManager( applicationScope ); final Id sourceId1 = IdGenerator.createId( "source1" ); final Id sourceId2 = IdGenerator.createId( "source2" ); final Id target1 = IdGenerator.createId( "target1" ); final Id target2 = IdGenerator.createId( "target2" ); Edge s1t1 = createEdge( sourceId1, "test", target1 ); Edge s1t2 = createEdge( sourceId1, "baz", target2 ); Edge s2t1 = createEdge( sourceId2, "foo", target1 ); Edge s2t2 = createEdge( sourceId2, "bar", target2 ); gm.writeEdge( s1t1 ).toBlocking().last(); gm.writeEdge( s1t2 ).toBlocking().last(); gm.writeEdge( s2t1 ).toBlocking().last(); gm.writeEdge( s2t2 ).toBlocking().last(); final Observable<GraphNode> graphNodes = Observable .just( new GraphNode( applicationScope, sourceId1 ), new GraphNode( applicationScope, sourceId2 ) ); final MigrationDataProvider<GraphNode> testMigrationProvider = () -> graphNodes; final TestProgressObserver progressObserver = new TestProgressObserver(); EdgeDataMigrationImpl edgeDataMigrationImpl = new EdgeDataMigrationImpl( keyspace, graphManagerFactory, edgesFromSourceObservable, allVersions, edgeMetadataSerializationV2, testMigrationProvider ); final int returned = edgeDataMigrationImpl.migrate( GraphDataVersions.INITIAL.getVersion(), progressObserver ); assertEquals( "Correct version returned", returned, GraphDataVersions.META_SHARDING.getVersion() ); assertFalse( "Progress observer should not have failed", progressObserver.isFailed() ); assertTrue( "Progress observer should have update messages", progressObserver.getUpdates().size() > 0 ); List<String> source1Edges = gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( sourceId1, null, null ) ).toList().toBlocking().last(); assertTrue( "Edge type present", source1Edges.contains( "test" ) ); assertTrue( "Edge type present", source1Edges.contains( "baz" ) ); List<String> source2Edges = gm.getEdgeTypesFromSource( new SimpleSearchEdgeType( sourceId2, null, null ) ).toList().toBlocking().last(); assertTrue( "Edge type present", source2Edges.contains( "foo" ) ); assertTrue( "Edge type present", source2Edges.contains( "bar" ) ); }
@Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final AtomicLong counter = new AtomicLong(); final MigrationRelationship<EdgeMetadataSerialization> migration = allVersions.getMigrationRelationship( currentVersion ); final Observable<List<Edge>> observable = migrationDataProvider.getData().flatMap( graphNode -> { final GraphManager gm = graphManagerFactory.createEdgeManager( graphNode.applicationScope ); return edgesFromSourceObservable.edgesFromSourceDescending( gm, graphNode.entryNode, true).buffer( 1000 ) .doOnNext( edges -> { final MutationBatch batch = keyspace.prepareMutationBatch(); for ( Edge edge : edges ) { logger.info("Migrating meta for edge {}", edge); final MutationBatch edgeBatch = migration.to.writeEdge( graphNode.applicationScope, edge ); batch.mergeShallow( edgeBatch ); } try { batch.execute(); } catch ( ConnectionException e ) { throw new RuntimeException( "Unable to perform migration", e ); } final long newCount = counter.addAndGet( edges.size() ); observer.update( migration.to.getImplementationVersion(), String .format( "Currently running. Rewritten %d edge types", newCount ) ); } ).subscribeOn( Schedulers.io() ); }, 10 ); observable.countLong().toBlocking().last(); return migration.to.getImplementationVersion(); }
EdgeDataMigrationImpl implements DataMigration { @Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final AtomicLong counter = new AtomicLong(); final MigrationRelationship<EdgeMetadataSerialization> migration = allVersions.getMigrationRelationship( currentVersion ); final Observable<List<Edge>> observable = migrationDataProvider.getData().flatMap( graphNode -> { final GraphManager gm = graphManagerFactory.createEdgeManager( graphNode.applicationScope ); return edgesFromSourceObservable.edgesFromSourceDescending( gm, graphNode.entryNode, true).buffer( 1000 ) .doOnNext( edges -> { final MutationBatch batch = keyspace.prepareMutationBatch(); for ( Edge edge : edges ) { logger.info("Migrating meta for edge {}", edge); final MutationBatch edgeBatch = migration.to.writeEdge( graphNode.applicationScope, edge ); batch.mergeShallow( edgeBatch ); } try { batch.execute(); } catch ( ConnectionException e ) { throw new RuntimeException( "Unable to perform migration", e ); } final long newCount = counter.addAndGet( edges.size() ); observer.update( migration.to.getImplementationVersion(), String .format( "Currently running. Rewritten %d edge types", newCount ) ); } ).subscribeOn( Schedulers.io() ); }, 10 ); observable.countLong().toBlocking().last(); return migration.to.getImplementationVersion(); } }
EdgeDataMigrationImpl implements DataMigration { @Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final AtomicLong counter = new AtomicLong(); final MigrationRelationship<EdgeMetadataSerialization> migration = allVersions.getMigrationRelationship( currentVersion ); final Observable<List<Edge>> observable = migrationDataProvider.getData().flatMap( graphNode -> { final GraphManager gm = graphManagerFactory.createEdgeManager( graphNode.applicationScope ); return edgesFromSourceObservable.edgesFromSourceDescending( gm, graphNode.entryNode, true).buffer( 1000 ) .doOnNext( edges -> { final MutationBatch batch = keyspace.prepareMutationBatch(); for ( Edge edge : edges ) { logger.info("Migrating meta for edge {}", edge); final MutationBatch edgeBatch = migration.to.writeEdge( graphNode.applicationScope, edge ); batch.mergeShallow( edgeBatch ); } try { batch.execute(); } catch ( ConnectionException e ) { throw new RuntimeException( "Unable to perform migration", e ); } final long newCount = counter.addAndGet( edges.size() ); observer.update( migration.to.getImplementationVersion(), String .format( "Currently running. Rewritten %d edge types", newCount ) ); } ).subscribeOn( Schedulers.io() ); }, 10 ); observable.countLong().toBlocking().last(); return migration.to.getImplementationVersion(); } @Inject  EdgeDataMigrationImpl( final Keyspace keyspace, final GraphManagerFactory graphManagerFactory,
                                  final EdgesObservable edgesFromSourceObservable,
                                  final VersionedMigrationSet<EdgeMetadataSerialization> allVersions,
                                  final EdgeMetadataSerializationV2Impl edgeMetadataSerializationV2,
                                  final MigrationDataProvider<GraphNode> migrationDataProvider ); }
EdgeDataMigrationImpl implements DataMigration { @Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final AtomicLong counter = new AtomicLong(); final MigrationRelationship<EdgeMetadataSerialization> migration = allVersions.getMigrationRelationship( currentVersion ); final Observable<List<Edge>> observable = migrationDataProvider.getData().flatMap( graphNode -> { final GraphManager gm = graphManagerFactory.createEdgeManager( graphNode.applicationScope ); return edgesFromSourceObservable.edgesFromSourceDescending( gm, graphNode.entryNode, true).buffer( 1000 ) .doOnNext( edges -> { final MutationBatch batch = keyspace.prepareMutationBatch(); for ( Edge edge : edges ) { logger.info("Migrating meta for edge {}", edge); final MutationBatch edgeBatch = migration.to.writeEdge( graphNode.applicationScope, edge ); batch.mergeShallow( edgeBatch ); } try { batch.execute(); } catch ( ConnectionException e ) { throw new RuntimeException( "Unable to perform migration", e ); } final long newCount = counter.addAndGet( edges.size() ); observer.update( migration.to.getImplementationVersion(), String .format( "Currently running. Rewritten %d edge types", newCount ) ); } ).subscribeOn( Schedulers.io() ); }, 10 ); observable.countLong().toBlocking().last(); return migration.to.getImplementationVersion(); } @Inject  EdgeDataMigrationImpl( final Keyspace keyspace, final GraphManagerFactory graphManagerFactory,
                                  final EdgesObservable edgesFromSourceObservable,
                                  final VersionedMigrationSet<EdgeMetadataSerialization> allVersions,
                                  final EdgeMetadataSerializationV2Impl edgeMetadataSerializationV2,
                                  final MigrationDataProvider<GraphNode> migrationDataProvider ); @Override int migrate( final int currentVersion,
                        final ProgressObserver observer ); @Override boolean supports( final int currentVersion ); @Override int getMaxVersion(); }
EdgeDataMigrationImpl implements DataMigration { @Override public int migrate( final int currentVersion, final ProgressObserver observer ) { final AtomicLong counter = new AtomicLong(); final MigrationRelationship<EdgeMetadataSerialization> migration = allVersions.getMigrationRelationship( currentVersion ); final Observable<List<Edge>> observable = migrationDataProvider.getData().flatMap( graphNode -> { final GraphManager gm = graphManagerFactory.createEdgeManager( graphNode.applicationScope ); return edgesFromSourceObservable.edgesFromSourceDescending( gm, graphNode.entryNode, true).buffer( 1000 ) .doOnNext( edges -> { final MutationBatch batch = keyspace.prepareMutationBatch(); for ( Edge edge : edges ) { logger.info("Migrating meta for edge {}", edge); final MutationBatch edgeBatch = migration.to.writeEdge( graphNode.applicationScope, edge ); batch.mergeShallow( edgeBatch ); } try { batch.execute(); } catch ( ConnectionException e ) { throw new RuntimeException( "Unable to perform migration", e ); } final long newCount = counter.addAndGet( edges.size() ); observer.update( migration.to.getImplementationVersion(), String .format( "Currently running. Rewritten %d edge types", newCount ) ); } ).subscribeOn( Schedulers.io() ); }, 10 ); observable.countLong().toBlocking().last(); return migration.to.getImplementationVersion(); } @Inject  EdgeDataMigrationImpl( final Keyspace keyspace, final GraphManagerFactory graphManagerFactory,
                                  final EdgesObservable edgesFromSourceObservable,
                                  final VersionedMigrationSet<EdgeMetadataSerialization> allVersions,
                                  final EdgeMetadataSerializationV2Impl edgeMetadataSerializationV2,
                                  final MigrationDataProvider<GraphNode> migrationDataProvider ); @Override int migrate( final int currentVersion,
                        final ProgressObserver observer ); @Override boolean supports( final int currentVersion ); @Override int getMaxVersion();  }