%SequenceFileKeyValueSink
% Helper class that writes output to a Sequence File.
%

%   Copyright 2014-2018 The MathWorks, Inc.

classdef (Sealed, Hidden) SequenceFileKeyValueSink < parallel.internal.mapreduce.AbstractKeyValueSink

    properties (Access = private)
        InternalSequenceWriter = []; % The internal java object.
    end
    
    methods
        % Construct a SequenceFileKeyValueSink that outputs to the sequence
        % file of given name.
        function obj = SequenceFileKeyValueSink(filename, ~)
            import matlab.io.datastore.internal.ContextClassLoaderGuard;
            import matlab.io.datastore.internal.HadoopConfiguration;
            hdfsLoader = com.mathworks.storage.hdfsloader.GlobalHdfsLoader.get();
            tmp = hdfsLoader.newMxArrayWritable2();
            outputClass = tmp.getUnderlyingClass();
            
            guard = ContextClassLoaderGuard(); %#ok<NASGU>
            
            path = hdfsLoader.newPath(filename);
            fileSystem = HadoopConfiguration.getGlobalFileSystem(path.toUri());
            obj.InternalSequenceWriter = hdfsLoader.newSequenceFileWriter(...
                fileSystem, fileSystem.getConf(), path, outputClass, outputClass);
        end
        
        % Add the given <key, value> pairs to the underlying data storage.
        % The keys and values input must be vectors of the same length.
        function add(obj, keys, values)
            writer = obj.InternalSequenceWriter;
            writer.sync();
            hdfsLoader = com.mathworks.storage.hdfsloader.GlobalHdfsLoader.get();
            for ii = 1:numel(keys)
                key = mxarraywritableproxyserialize(hdfsLoader, keys(ii));
                value = mxarraywritableproxyserialize(hdfsLoader, values{ii});
                writer.append(key, value);
            end
        end
        
        % In a Hadoop context, commit is dealt with by Hadoop.
        function commit(~)
        end
        
        % Reduce output does not have results.
        function results = getTaskOutputMetadata(~)
            results = [];
        end
        
        % Cleanup of internal resources
        function delete(obj)
            obj.InternalSequenceWriter.close();
        end
    end

end
