%evaluateReduceTask
% Entry point for a reduce task on Hadoop.

%   Copyright 2014-2018 The MathWorks, Inc.

function out = evaluateReduceTask(keyValuesSourceEndpointString, keyValueSinkEndpointFolder, serializedTaskAttemptID, outputFolder, finalOutputFolder, ~)
out = [];

import matlab.io.datastore.internal.ContextClassLoaderGuard;
import matlab.io.datastore.internal.HadoopConfiguration;
import parallel.internal.mapreduce.ReduceTask;
import parallel.internal.hadoop.KeyValueSinkAdapter;
import parallel.internal.hadoop.SequenceFileKeyValueSink;
import parallel.internal.hadoop.TextKeyValueSink;
import parallel.internal.hadoop.HadoopShuffleSortStrategy;
import parallel.internal.hadoop.HadoopWorker;
import com.mathworks.toolbox.parallel.mapreduce.HadoopTaskAttemptID;

guard = ContextClassLoaderGuard(); %#ok<NASGU>

% Deserialize all information from Hadoop
dctSchedulerMessage(6, 'Beginning Reduce Task %s', serializedTaskAttemptID);
taskAttemptID = HadoopTaskAttemptID.forName(serializedTaskAttemptID);
configuration = HadoopConfiguration.getGlobalConfiguration();
try
    taskID = taskAttemptID.getTaskID();
    jobInfo = HadoopWorker.Instance.Job.Info;
    
    % Currently there is no abstraction for the output writer. As such, we
    % switch on the string and create the output writer.
    switch (jobInfo.OutputType)
        case 'Binary'
            outputSink = SequenceFileKeyValueSink([outputFolder, '/', sprintf('part%i.seq', taskID)], configuration);
            outputKeyValueWriterFactory = @(id)outputSink;
        case 'TabularText'
            outputSink = TextKeyValueSink([outputFolder, '/', sprintf('part%i.txt', taskID)], configuration);
            outputKeyValueWriterFactory = @(id)outputSink;
        otherwise
            assert(false, 'Invalid output type passed to evaluateReduceTask');
    end
    
    % Delegate to ReduceTask to do the actual work.
    shuffleSortStrategy = HadoopShuffleSortStrategy(keyValuesSourceEndpointString, keyValueSinkEndpointFolder);
    task = ReduceTask.createRunOnSingle(taskID, jobInfo.ReduceFunction, outputKeyValueWriterFactory, shuffleSortStrategy);
    task.run();
catch err
    try
        iSaveErrorToOutputFolder(err, taskAttemptID, finalOutputFolder, configuration);
    catch saveErr
        dctSchedulerMessage(2, 'Unable to save error to output folder. The error was:\n%s\n', saveErr.getReport());
    end
    error(err.getReport());
end

% Helper function to convert a Hadoop path into an IRI.
function iri = iConvertToIri(hadoopPath)
hdfsLoader = com.mathworks.storage.hdfsloader.GlobalHdfsLoader.get();
hadoopPath = hdfsLoader.newPath(hadoopPath);
iri = char(hadoopPath.toUri());

% Helper function that saves error to the output folder in case of failure.
function iSaveErrorToOutputFolder(err, taskAttemptID, outputFolder, configuration)
import parallel.internal.hadoop.FileSystemTools;
fileSystemTools = FileSystemTools(configuration);
% We have to convert the output folder, which is in Hadoop path format,
% into an IRI. This is because FileSystemTools expects IRIs, where special
% characters are percent encoded.
fileSystemTools.saveErrorToOutputFolder(err, iConvertToIri(outputFolder), taskAttemptID);
