@Test public void processesAllInputStatementsWhenHasVirtuosoSource() throws Exception { FusionExecutor executor = getLDFusionToolExecutor(Long.MAX_VALUE, true); TestInputLoader inputLoader = new TestInputLoader(ImmutableList.of( (Collection<Statement>) ImmutableList.<Statement>of(), ImmutableList.of( createHttpStatement("s1", "p1", "o1", "g1"), createHttpStatement("s1", "p1", "o1", "g2")), ImmutableList.of( createHttpStatement("s2", "p2", "o1", "g3"), createHttpStatement("s2", "p2", "o2", "g3")) )); TestRDFWriter rdfWriter = new TestRDFWriter(); executor.fuse(new TestConflictResolver(), inputLoader, rdfWriter); List<ResolvedStatement> resolvedStatements = rdfWriter.getCollectedResolvedStatements(); assertThat(resolvedStatements.get(0).getStatement(), contextAwareStatementIsEqual(createHttpStatement("s1", "p1", "o1", "g1"))); assertThat(resolvedStatements.get(1).getStatement(), contextAwareStatementIsEqual(createHttpStatement("s1", "p1", "o1", "g2"))); assertThat(resolvedStatements.get(2).getStatement(), contextAwareStatementIsEqual(createHttpStatement("s2", "p2", "o1", "g3"))); assertThat(resolvedStatements.get(3).getStatement(), contextAwareStatementIsEqual(createHttpStatement("s2", "p2", "o2", "g3"))); }
@Override public void fuse(ResourceDescriptionConflictResolver conflictResolver, InputLoader inputLoader, CloseableRDFWriter rdfWriter) throws LDFusionToolException, ConflictResolutionException, IOException { long outputTriples = 0; long inputTriples = 0; boolean checkMaxOutputTriples = maxOutputTriples != null && maxOutputTriples >= 0; LOG.info("Starting conflict resolution"); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); while (!isCanceled() && inputLoader.hasNext()) { timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.QUAD_LOADING); ResourceDescription resourceDescription = inputLoader.next(); inputTriples += resourceDescription.getDescribingStatements().size(); timeProfiler.stopAddCounter(EnumFusionCounters.QUAD_LOADING); timeProfiler.startCounter(EnumFusionCounters.INPUT_FILTERING); boolean accept = this.resourceDescriptionFilter.accept(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.INPUT_FILTERING); if (!accept) { LOG.debug("Resource {} doesn't match filter, skipping", resourceDescription.getResource()); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); continue; } timeProfiler.startCounter(EnumFusionCounters.CONFLICT_RESOLUTION); Collection<ResolvedStatement> resolvedQuads = conflictResolver.resolveConflicts(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.CONFLICT_RESOLUTION); LOG.debug("Resolved {} quads resulting in {} quads (processed totally {} quads)", new Object[] {resourceDescription.getDescribingStatements().size(), resolvedQuads.size(), inputTriples}); if (checkMaxOutputTriples && outputTriples + resolvedQuads.size() > maxOutputTriples) { break; } outputTriples += resolvedQuads.size(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); inputLoader.updateWithResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.OUTPUT_WRITING); rdfWriter.writeResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.OUTPUT_WRITING); memoryProfiler.capture(); fixVirtuosoOpenedStatements(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); } timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); if (isCanceled()) { LOG.warn("The execution was canceled!"); } LOG.info(String.format("Processed %,d quads which were resolved to %,d output quads.", inputTriples, outputTriples)); }
LDFusionToolExecutor implements FusionExecutor { @Override public void fuse(ResourceDescriptionConflictResolver conflictResolver, InputLoader inputLoader, CloseableRDFWriter rdfWriter) throws LDFusionToolException, ConflictResolutionException, IOException { long outputTriples = 0; long inputTriples = 0; boolean checkMaxOutputTriples = maxOutputTriples != null && maxOutputTriples >= 0; LOG.info("Starting conflict resolution"); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); while (!isCanceled() && inputLoader.hasNext()) { timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.QUAD_LOADING); ResourceDescription resourceDescription = inputLoader.next(); inputTriples += resourceDescription.getDescribingStatements().size(); timeProfiler.stopAddCounter(EnumFusionCounters.QUAD_LOADING); timeProfiler.startCounter(EnumFusionCounters.INPUT_FILTERING); boolean accept = this.resourceDescriptionFilter.accept(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.INPUT_FILTERING); if (!accept) { LOG.debug("Resource {} doesn't match filter, skipping", resourceDescription.getResource()); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); continue; } timeProfiler.startCounter(EnumFusionCounters.CONFLICT_RESOLUTION); Collection<ResolvedStatement> resolvedQuads = conflictResolver.resolveConflicts(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.CONFLICT_RESOLUTION); LOG.debug("Resolved {} quads resulting in {} quads (processed totally {} quads)", new Object[] {resourceDescription.getDescribingStatements().size(), resolvedQuads.size(), inputTriples}); if (checkMaxOutputTriples && outputTriples + resolvedQuads.size() > maxOutputTriples) { break; } outputTriples += resolvedQuads.size(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); inputLoader.updateWithResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.OUTPUT_WRITING); rdfWriter.writeResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.OUTPUT_WRITING); memoryProfiler.capture(); fixVirtuosoOpenedStatements(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); } timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); if (isCanceled()) { LOG.warn("The execution was canceled!"); } LOG.info(String.format("Processed %,d quads which were resolved to %,d output quads.", inputTriples, outputTriples)); } }
LDFusionToolExecutor implements FusionExecutor { @Override public void fuse(ResourceDescriptionConflictResolver conflictResolver, InputLoader inputLoader, CloseableRDFWriter rdfWriter) throws LDFusionToolException, ConflictResolutionException, IOException { long outputTriples = 0; long inputTriples = 0; boolean checkMaxOutputTriples = maxOutputTriples != null && maxOutputTriples >= 0; LOG.info("Starting conflict resolution"); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); while (!isCanceled() && inputLoader.hasNext()) { timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.QUAD_LOADING); ResourceDescription resourceDescription = inputLoader.next(); inputTriples += resourceDescription.getDescribingStatements().size(); timeProfiler.stopAddCounter(EnumFusionCounters.QUAD_LOADING); timeProfiler.startCounter(EnumFusionCounters.INPUT_FILTERING); boolean accept = this.resourceDescriptionFilter.accept(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.INPUT_FILTERING); if (!accept) { LOG.debug("Resource {} doesn't match filter, skipping", resourceDescription.getResource()); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); continue; } timeProfiler.startCounter(EnumFusionCounters.CONFLICT_RESOLUTION); Collection<ResolvedStatement> resolvedQuads = conflictResolver.resolveConflicts(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.CONFLICT_RESOLUTION); LOG.debug("Resolved {} quads resulting in {} quads (processed totally {} quads)", new Object[] {resourceDescription.getDescribingStatements().size(), resolvedQuads.size(), inputTriples}); if (checkMaxOutputTriples && outputTriples + resolvedQuads.size() > maxOutputTriples) { break; } outputTriples += resolvedQuads.size(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); inputLoader.updateWithResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.OUTPUT_WRITING); rdfWriter.writeResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.OUTPUT_WRITING); memoryProfiler.capture(); fixVirtuosoOpenedStatements(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); } timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); if (isCanceled()) { LOG.warn("The execution was canceled!"); } LOG.info(String.format("Processed %,d quads which were resolved to %,d output quads.", inputTriples, outputTriples)); } LDFusionToolExecutor(
            boolean hasVirtuosoSource,
            Long maxOutputTriples,
            ResourceDescriptionFilter resourceDescriptionFilter,
            ProfilingTimeCounter<EnumFusionCounters> timeProfiler,
            MemoryProfiler memoryProfiler); }
LDFusionToolExecutor implements FusionExecutor { @Override public void fuse(ResourceDescriptionConflictResolver conflictResolver, InputLoader inputLoader, CloseableRDFWriter rdfWriter) throws LDFusionToolException, ConflictResolutionException, IOException { long outputTriples = 0; long inputTriples = 0; boolean checkMaxOutputTriples = maxOutputTriples != null && maxOutputTriples >= 0; LOG.info("Starting conflict resolution"); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); while (!isCanceled() && inputLoader.hasNext()) { timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.QUAD_LOADING); ResourceDescription resourceDescription = inputLoader.next(); inputTriples += resourceDescription.getDescribingStatements().size(); timeProfiler.stopAddCounter(EnumFusionCounters.QUAD_LOADING); timeProfiler.startCounter(EnumFusionCounters.INPUT_FILTERING); boolean accept = this.resourceDescriptionFilter.accept(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.INPUT_FILTERING); if (!accept) { LOG.debug("Resource {} doesn't match filter, skipping", resourceDescription.getResource()); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); continue; } timeProfiler.startCounter(EnumFusionCounters.CONFLICT_RESOLUTION); Collection<ResolvedStatement> resolvedQuads = conflictResolver.resolveConflicts(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.CONFLICT_RESOLUTION); LOG.debug("Resolved {} quads resulting in {} quads (processed totally {} quads)", new Object[] {resourceDescription.getDescribingStatements().size(), resolvedQuads.size(), inputTriples}); if (checkMaxOutputTriples && outputTriples + resolvedQuads.size() > maxOutputTriples) { break; } outputTriples += resolvedQuads.size(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); inputLoader.updateWithResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.OUTPUT_WRITING); rdfWriter.writeResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.OUTPUT_WRITING); memoryProfiler.capture(); fixVirtuosoOpenedStatements(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); } timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); if (isCanceled()) { LOG.warn("The execution was canceled!"); } LOG.info(String.format("Processed %,d quads which were resolved to %,d output quads.", inputTriples, outputTriples)); } LDFusionToolExecutor(
            boolean hasVirtuosoSource,
            Long maxOutputTriples,
            ResourceDescriptionFilter resourceDescriptionFilter,
            ProfilingTimeCounter<EnumFusionCounters> timeProfiler,
            MemoryProfiler memoryProfiler); @Override void fuse(ResourceDescriptionConflictResolver conflictResolver, InputLoader inputLoader, CloseableRDFWriter rdfWriter); Long getMaxOutputTriples(); ResourceDescriptionFilter getResourceDescriptionFilter(); void setIsCanceledCallback(IsCanceledCallback isCanceledCallback); }
LDFusionToolExecutor implements FusionExecutor { @Override public void fuse(ResourceDescriptionConflictResolver conflictResolver, InputLoader inputLoader, CloseableRDFWriter rdfWriter) throws LDFusionToolException, ConflictResolutionException, IOException { long outputTriples = 0; long inputTriples = 0; boolean checkMaxOutputTriples = maxOutputTriples != null && maxOutputTriples >= 0; LOG.info("Starting conflict resolution"); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); while (!isCanceled() && inputLoader.hasNext()) { timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.QUAD_LOADING); ResourceDescription resourceDescription = inputLoader.next(); inputTriples += resourceDescription.getDescribingStatements().size(); timeProfiler.stopAddCounter(EnumFusionCounters.QUAD_LOADING); timeProfiler.startCounter(EnumFusionCounters.INPUT_FILTERING); boolean accept = this.resourceDescriptionFilter.accept(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.INPUT_FILTERING); if (!accept) { LOG.debug("Resource {} doesn't match filter, skipping", resourceDescription.getResource()); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); continue; } timeProfiler.startCounter(EnumFusionCounters.CONFLICT_RESOLUTION); Collection<ResolvedStatement> resolvedQuads = conflictResolver.resolveConflicts(resourceDescription); timeProfiler.stopAddCounter(EnumFusionCounters.CONFLICT_RESOLUTION); LOG.debug("Resolved {} quads resulting in {} quads (processed totally {} quads)", new Object[] {resourceDescription.getDescribingStatements().size(), resolvedQuads.size(), inputTriples}); if (checkMaxOutputTriples && outputTriples + resolvedQuads.size() > maxOutputTriples) { break; } outputTriples += resolvedQuads.size(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); inputLoader.updateWithResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); timeProfiler.startCounter(EnumFusionCounters.OUTPUT_WRITING); rdfWriter.writeResolvedStatements(resolvedQuads); timeProfiler.stopAddCounter(EnumFusionCounters.OUTPUT_WRITING); memoryProfiler.capture(); fixVirtuosoOpenedStatements(); timeProfiler.startCounter(EnumFusionCounters.BUFFERING); } timeProfiler.stopAddCounter(EnumFusionCounters.BUFFERING); if (isCanceled()) { LOG.warn("The execution was canceled!"); } LOG.info(String.format("Processed %,d quads which were resolved to %,d output quads.", inputTriples, outputTriples)); } LDFusionToolExecutor(
            boolean hasVirtuosoSource,
            Long maxOutputTriples,
            ResourceDescriptionFilter resourceDescriptionFilter,
            ProfilingTimeCounter<EnumFusionCounters> timeProfiler,
            MemoryProfiler memoryProfiler); @Override void fuse(ResourceDescriptionConflictResolver conflictResolver, InputLoader inputLoader, CloseableRDFWriter rdfWriter); Long getMaxOutputTriples(); ResourceDescriptionFilter getResourceDescriptionFilter(); void setIsCanceledCallback(IsCanceledCallback isCanceledCallback);  }