<commit_before>package cz.cuni.mff.xrg.odcs.dpu.fusiontool;

import java.io.IOException;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.openrdf.model.Graph;
import org.openrdf.model.Model;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.impl.TreeModel;
import org.openrdf.model.vocabulary.OWL;
import org.openrdf.query.BindingSet;
import org.openrdf.query.MalformedQueryException;
import org.openrdf.query.QueryEvaluationException;
import org.openrdf.query.QueryLanguage;
import org.openrdf.query.TupleQueryResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cz.cuni.mff.odcleanstore.conflictresolution.ConflictResolver;
import cz.cuni.mff.odcleanstore.conflictresolution.ConflictResolverFactory;
import cz.cuni.mff.odcleanstore.conflictresolution.ResolutionFunctionRegistry;
import cz.cuni.mff.odcleanstore.conflictresolution.ResolvedStatement;
import cz.cuni.mff.odcleanstore.conflictresolution.URIMapping;
import cz.cuni.mff.odcleanstore.conflictresolution.exceptions.ConflictResolutionException;
import cz.cuni.mff.odcleanstore.conflictresolution.impl.DistanceMeasureImpl;
import cz.cuni.mff.odcleanstore.conflictresolution.quality.DummySourceQualityCalculator;
import cz.cuni.mff.odcleanstore.core.ODCSUtils;
import cz.cuni.mff.xrg.odcs.commons.dpu.DPUContext;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.config.ConfigContainer;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.config.FileOutput;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.exceptions.FusionToolDpuErrorCodes;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.exceptions.FusionToolDpuException;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.CanonicalUriFileReader;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.LargeCollectionFactory;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.MapdbCollectionFactory;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.MemoryCollectionFactory;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.file.FileOutputWriter;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.file.FileOutputWriterFactory;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.urimapping.AlternativeURINavigator;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.urimapping.URIMappingIterable;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.urimapping.URIMappingIterableImpl;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.EnumProfilingCounters;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.MemoryProfiler;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.PrefixDeclBuilder;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.PrefixDeclBuilderImpl;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.ProfilingTimeCounter;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.QuadLoader;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.UriQueue;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.UriQueueImpl;
import cz.cuni.mff.xrg.odcs.rdf.exceptions.InvalidQueryException;
import cz.cuni.mff.xrg.odcs.rdf.help.LazyTriples;
import cz.cuni.mff.xrg.odcs.rdf.interfaces.RDFDataUnit;

/**
 * Fuses RDF data from input using ODCS Conflict Resolution and writes the output to RDF outputs.
 * Conflict resolution includes resolution of owl:sameAs link, resolution of instance-level conflicts.
 * See sample configuration files (sample-config-full.xml) for overview of all processing options.
 * @author <NAME>
 */
public class FusionToolDpuExecutor {
    private static final Logger LOG = LoggerFactory.getLogger(FusionToolDpuExecutor.class);
    
    /** An instance of {@link FileOutputWriterFactory}. */
    protected static final FileOutputWriterFactory RDF_WRITER_FACTORY = new FileOutputWriterFactory();
    
    private ConfigContainer config;
    private DPUContext executionContext;
    private List<RDFDataUnit> rdfInputs;
    private RDFDataUnit sameAsInput;
    private RDFDataUnit metadataInput;
    private RDFDataUnit rdfOutput;
    private boolean hasFusionToolRan;
    private int fileOutputWrittenQuads;

    /**
     * Creates a new instance.
     * @param config configuration
     * @param executionContext execution context
     * @param rdfInputs RDF input data
     * @param sameAsInput input owl:sameAs links
     * @param metadataInput input metadata
     * @param rdfOutput RDF output data
     */
    public FusionToolDpuExecutor(ConfigContainer config, DPUContext executionContext, List<RDFDataUnit> rdfInputs,
            RDFDataUnit sameAsInput, RDFDataUnit metadataInput, RDFDataUnit rdfOutput) {
        this.config = config;
        this.executionContext = executionContext;
        this.rdfInputs = rdfInputs;
        this.sameAsInput = sameAsInput;
        this.metadataInput = metadataInput;
        this.rdfOutput = rdfOutput;
        this.hasFusionToolRan = false;
        this.fileOutputWrittenQuads = 0;
    }

    /**
     * Performs the actual ODCS-FusionTool task according to the given configuration.
     * @throws FusionToolDpuException error
     */
    public void runFusionTool() throws FusionToolDpuException {
        if (this.hasFusionToolRan) {
            throw new IllegalStateException("runFusionTool() can be called only once");
        }
        this.hasFusionToolRan = true;
        
        LargeCollectionFactory collectionFactory = 
                createLargeCollectionFactory(config.getEnableFileCache());
        ProfilingTimeCounter<EnumProfilingCounters> timeProfiler = ProfilingTimeCounter.createInstance(
                EnumProfilingCounters.class, config.isProfilingOn()); 
        MemoryProfiler memoryProfiler = MemoryProfiler.createInstance(config.isProfilingOn()); 
        timeProfiler.startCounter(EnumProfilingCounters.INITIALIZATION);
        List<FileOutputWriter> fileOutputWriters = null;
        try {
            // Load & resolve owl:sameAs links
            URIMappingIterable uriMapping = getURIMapping(); // TODO: canonical URI file
            AlternativeURINavigator alternativeURINavigator = new AlternativeURINavigator(uriMapping);
            Set<String> resolvedCanonicalURIs = collectionFactory.createSet();

            // Get iterator over subjects of relevant triples
            UriQueue queuedSubjects = getSeedSubjects(uriMapping, collectionFactory);
            final boolean isTransitive = config.getSeedResourceSparqlQuery() != null;

            // Initialize CR
            ConflictResolver conflictResolver = createConflictResolver(uriMapping);

            // File outputs
            fileOutputWriters = createRDFWriters(config.getFileOutputs(), config.getPrefixes());
            
            // Initialize triple counters
            long outputTriples = 0;
            long inputTriples = 0;
            
            // Load & process relevant triples (quads) subject by subject so that we can apply CR to them
            PrefixDeclBuilder nsPrefixes = new PrefixDeclBuilderImpl(config.getPrefixes());
            QuadLoader quadLoader = new QuadLoader(rdfInputs, alternativeURINavigator, nsPrefixes);
            timeProfiler.stopAddCounter(EnumProfilingCounters.INITIALIZATION);
            while (queuedSubjects.hasNext() && !executionContext.canceled()) {
                timeProfiler.startCounter(EnumProfilingCounters.BUFFERING);
                String uri = queuedSubjects.next();
                String canonicalURI = uriMapping.getCanonicalURI(uri);

                if (resolvedCanonicalURIs.contains(canonicalURI)) {
                    // avoid processing a URI multiple times
                    continue;
                }
                resolvedCanonicalURIs.add(canonicalURI);
                timeProfiler.stopAddCounter(EnumProfilingCounters.BUFFERING);

                // Load quads for the given subject
                timeProfiler.startCounter(EnumProfilingCounters.QUAD_LOADING);
                Collection<Statement> quads = quadLoader.loadQuadsForURI(canonicalURI);
                inputTriples += quads.size();
                timeProfiler.stopAddCounter(EnumProfilingCounters.QUAD_LOADING);

                // Resolve conflicts
                timeProfiler.startCounter(EnumProfilingCounters.CONFLICT_RESOLUTION);
                Collection<ResolvedStatement> resolvedQuads = conflictResolver.resolveConflicts(quads);
                timeProfiler.stopAddCounter(EnumProfilingCounters.CONFLICT_RESOLUTION);
                LOG.trace("Resolved {} quads for URI <{}> resulting in {} quads (processed totally {} quads)",
                        new Object[] { quads.size(), canonicalURI, resolvedQuads.size(), inputTriples});
                outputTriples += resolvedQuads.size();

                // Add objects filtered by CR for traversal
                if (isTransitive) {
                    timeProfiler.startCounter(EnumProfilingCounters.BUFFERING);
                    addDiscoveredObjects(queuedSubjects, resolvedQuads, uriMapping, resolvedCanonicalURIs);
                    timeProfiler.stopAddCounter(EnumProfilingCounters.BUFFERING);
                }

                // Write result to output
                timeProfiler.startCounter(EnumProfilingCounters.OUTPUT_WRITING);
                writeResults(resolvedQuads, fileOutputWriters);
                timeProfiler.stopAddCounter(EnumProfilingCounters.OUTPUT_WRITING);

                memoryProfiler.capture();<commit_msg>Removed trace log for each resolved subject<commit_after>package cz.cuni.mff.xrg.odcs.dpu.fusiontool;

import java.io.IOException;
import java.util.Collection;
import java.util.HashSet;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.openrdf.model.Graph;
import org.openrdf.model.Model;
import org.openrdf.model.Statement;
import org.openrdf.model.URI;
import org.openrdf.model.Value;
import org.openrdf.model.impl.TreeModel;
import org.openrdf.model.vocabulary.OWL;
import org.openrdf.query.BindingSet;
import org.openrdf.query.MalformedQueryException;
import org.openrdf.query.QueryEvaluationException;
import org.openrdf.query.QueryLanguage;
import org.openrdf.query.TupleQueryResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cz.cuni.mff.odcleanstore.conflictresolution.ConflictResolver;
import cz.cuni.mff.odcleanstore.conflictresolution.ConflictResolverFactory;
import cz.cuni.mff.odcleanstore.conflictresolution.ResolutionFunctionRegistry;
import cz.cuni.mff.odcleanstore.conflictresolution.ResolvedStatement;
import cz.cuni.mff.odcleanstore.conflictresolution.URIMapping;
import cz.cuni.mff.odcleanstore.conflictresolution.exceptions.ConflictResolutionException;
import cz.cuni.mff.odcleanstore.conflictresolution.impl.DistanceMeasureImpl;
import cz.cuni.mff.odcleanstore.conflictresolution.quality.DummySourceQualityCalculator;
import cz.cuni.mff.odcleanstore.core.ODCSUtils;
import cz.cuni.mff.xrg.odcs.commons.dpu.DPUContext;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.config.ConfigContainer;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.config.FileOutput;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.exceptions.FusionToolDpuErrorCodes;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.exceptions.FusionToolDpuException;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.CanonicalUriFileReader;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.LargeCollectionFactory;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.MapdbCollectionFactory;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.MemoryCollectionFactory;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.file.FileOutputWriter;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.io.file.FileOutputWriterFactory;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.urimapping.AlternativeURINavigator;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.urimapping.URIMappingIterable;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.urimapping.URIMappingIterableImpl;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.EnumProfilingCounters;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.MemoryProfiler;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.PrefixDeclBuilder;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.PrefixDeclBuilderImpl;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.ProfilingTimeCounter;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.QuadLoader;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.UriQueue;
import cz.cuni.mff.xrg.odcs.dpu.fusiontool.util.UriQueueImpl;
import cz.cuni.mff.xrg.odcs.rdf.exceptions.InvalidQueryException;
import cz.cuni.mff.xrg.odcs.rdf.help.LazyTriples;
import cz.cuni.mff.xrg.odcs.rdf.interfaces.RDFDataUnit;

/**
 * Fuses RDF data from input using ODCS Conflict Resolution and writes the output to RDF outputs.
 * Conflict resolution includes resolution of owl:sameAs link, resolution of instance-level conflicts.
 * See sample configuration files (sample-config-full.xml) for overview of all processing options.
 * @author <NAME>
 */
public class FusionToolDpuExecutor {
    private static final Logger LOG = LoggerFactory.getLogger(FusionToolDpuExecutor.class);
    
    /** An instance of {@link FileOutputWriterFactory}. */
    protected static final FileOutputWriterFactory RDF_WRITER_FACTORY = new FileOutputWriterFactory();
    
    private ConfigContainer config;
    private DPUContext executionContext;
    private List<RDFDataUnit> rdfInputs;
    private RDFDataUnit sameAsInput;
    private RDFDataUnit metadataInput;
    private RDFDataUnit rdfOutput;
    private boolean hasFusionToolRan;
    private int fileOutputWrittenQuads;

    /**
     * Creates a new instance.
     * @param config configuration
     * @param executionContext execution context
     * @param rdfInputs RDF input data
     * @param sameAsInput input owl:sameAs links
     * @param metadataInput input metadata
     * @param rdfOutput RDF output data
     */
    public FusionToolDpuExecutor(ConfigContainer config, DPUContext executionContext, List<RDFDataUnit> rdfInputs,
            RDFDataUnit sameAsInput, RDFDataUnit metadataInput, RDFDataUnit rdfOutput) {
        this.config = config;
        this.executionContext = executionContext;
        this.rdfInputs = rdfInputs;
        this.sameAsInput = sameAsInput;
        this.metadataInput = metadataInput;
        this.rdfOutput = rdfOutput;
        this.hasFusionToolRan = false;
        this.fileOutputWrittenQuads = 0;
    }

    /**
     * Performs the actual ODCS-FusionTool task according to the given configuration.
     * @throws FusionToolDpuException error
     */
    public void runFusionTool() throws FusionToolDpuException {
        if (this.hasFusionToolRan) {
            throw new IllegalStateException("runFusionTool() can be called only once");
        }
        this.hasFusionToolRan = true;
        
        LargeCollectionFactory collectionFactory = 
                createLargeCollectionFactory(config.getEnableFileCache());
        ProfilingTimeCounter<EnumProfilingCounters> timeProfiler = ProfilingTimeCounter.createInstance(
                EnumProfilingCounters.class, config.isProfilingOn()); 
        MemoryProfiler memoryProfiler = MemoryProfiler.createInstance(config.isProfilingOn()); 
        timeProfiler.startCounter(EnumProfilingCounters.INITIALIZATION);
        List<FileOutputWriter> fileOutputWriters = null;
        try {
            // Load & resolve owl:sameAs links
            URIMappingIterable uriMapping = getURIMapping(); // TODO: canonical URI file
            AlternativeURINavigator alternativeURINavigator = new AlternativeURINavigator(uriMapping);
            Set<String> resolvedCanonicalURIs = collectionFactory.createSet();

            // Get iterator over subjects of relevant triples
            UriQueue queuedSubjects = getSeedSubjects(uriMapping, collectionFactory);
            final boolean isTransitive = config.getSeedResourceSparqlQuery() != null;

            // Initialize CR
            ConflictResolver conflictResolver = createConflictResolver(uriMapping);

            // File outputs
            fileOutputWriters = createRDFWriters(config.getFileOutputs(), config.getPrefixes());
            
            // Initialize triple counters
            long outputTriples = 0;
            long inputTriples = 0;
            
            // Load & process relevant triples (quads) subject by subject so that we can apply CR to them
            PrefixDeclBuilder nsPrefixes = new PrefixDeclBuilderImpl(config.getPrefixes());
            QuadLoader quadLoader = new QuadLoader(rdfInputs, alternativeURINavigator, nsPrefixes);
            timeProfiler.stopAddCounter(EnumProfilingCounters.INITIALIZATION);
            while (queuedSubjects.hasNext() && !executionContext.canceled()) {
                timeProfiler.startCounter(EnumProfilingCounters.BUFFERING);
                String uri = queuedSubjects.next();
                String canonicalURI = uriMapping.getCanonicalURI(uri);

                if (resolvedCanonicalURIs.contains(canonicalURI)) {
                    // avoid processing a URI multiple times
                    continue;
                }
                resolvedCanonicalURIs.add(canonicalURI);
                timeProfiler.stopAddCounter(EnumProfilingCounters.BUFFERING);

                // Load quads for the given subject
                timeProfiler.startCounter(EnumProfilingCounters.QUAD_LOADING);
                Collection<Statement> quads = quadLoader.loadQuadsForURI(canonicalURI);
                inputTriples += quads.size();
                timeProfiler.stopAddCounter(EnumProfilingCounters.QUAD_LOADING);

                // Resolve conflicts
                timeProfiler.startCounter(EnumProfilingCounters.CONFLICT_RESOLUTION);
                Collection<ResolvedStatement> resolvedQuads = conflictResolver.resolveConflicts(quads);
                timeProfiler.stopAddCounter(EnumProfilingCounters.CONFLICT_RESOLUTION);
                
                // LOG.trace("Resolved {} quads for URI <{}> resulting in {} quads (processed totally {} quads)",
                // new Object[] { quads.size(), canonicalURI, resolvedQuads.size(), inputTriples});
                
                outputTriples += resolvedQuads.size();

                // Add objects filtered by CR for traversal
                if (isTransitive) {
                    timeProfiler.startCounter(EnumProfilingCounters.BUFFERING);
                    addDiscoveredObjects(queuedSubjects, resolvedQuads, uriMapping, resolvedCanonicalURIs);
                    timeProfiler.stopAddCounter(EnumProfilingCounters.BUFFERING);
                }

                // Write result to output
                timeProfiler.startCounter(EnumProfilingCounters.OUTPUT_WRITING);
                writeResults(resolvedQuads, fileOutputWriters);
                timeProfiler.stopAddCounter(EnumProfilingCounters.OUTPUT_WRITING);

                memoryProfiler.capture();
{'fraction_non_alphanumeric': 0.08477103301384452, 'fraction_numerical': 0.0005324813631522897, 'mean_word_length': 3.4117923420249, 'pattern_counts': {'":': 0, '<': 21, '<?xml version=': 0, '>': 21, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 2}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3232427', 'n_tokens_mistral': 5704, 'n_tokens_neox': 5483, 'n_words': 935}