%SqliteKeyValuesSource
% An implementation of AbstractKeyValuesSource, which reads from a given
% table in the given database file.

%   Copyright 2014-2016 The MathWorks, Inc.

classdef (Sealed, Hidden) SqliteKeyValuesSource < parallel.internal.mapreduce.AbstractKeyValuesSource
    properties (GetAccess = private, SetAccess = immutable)
        SqliteStore; % An SQLite database instance.
    end
    
    properties (Access = private)
        HasData = false;
        HasDataForCurrentKey = false;
    end
    
    methods
        % Construct a source that reads the table of given reduceTaskIndex
        % from the database file sourceFile.
        function this = SqliteKeyValuesSource(sourceFile, reduceTaskIndex)
            this.SqliteStore = matlab.mapreduce.internal.KeyValueStore(sourceFile, 'noCreate', 'doNotDelete');
            
            % This selects a store to read ordered by key.
            storeName = sprintf('KeyValueStore%i', reduceTaskIndex);
            if this.SqliteStore.storeExists(storeName) 
                this.SqliteStore.selectStoreToRead(storeName, true);
                this.HasData = this.SqliteStore.hasNextKey();
            end
        end
        
        % Returns true if and only if there is more data overall to read.
        function result = hasData(this)
            result = this.HasData;
        end
        
        % Returns true if and only if there is more data associated with the
        % current key.
        function result = hasDataForCurrentKey(this)
            result = this.HasDataForCurrentKey;
        end
        
        % Indicate that this source should start reading values from the
        % next key. This cannot be called when there is still data for the
        % current key.
        function key = nextKey(this)
            assert(~this.HasDataForCurrentKey);
            key = this.SqliteStore.getNextKey();
            this.HasDataForCurrentKey = this.SqliteStore.hasNextValue();
            this.HasData = this.HasDataForCurrentKey;
        end
        
        % Discard any remaining data associated with the current key.
        function discardRemainingDataForCurrentKey(this)
            this.HasDataForCurrentKey = false;
            this.HasData = this.SqliteStore.hasNextKey();
        end
        
        % Return a chunk of values up-to maxLength number of entries
        % associated with the current key.
        function values = read(this, numEntries)
            % The SqliteStore MCOS object currently only supports returning
            % 1 value at a time. As soon as that changes, this can be
            % removed.
            values = cell(numEntries, 1);
            
            sqliteStore = this.SqliteStore;
            for ii = 1:numEntries
                if ~this.HasDataForCurrentKey
                    values(ii:end) = [];
                    break;
                end
                values{ii} = sqliteStore.getNextValue();
                this.HasDataForCurrentKey = sqliteStore.hasNextValue();
                this.HasData = this.HasDataForCurrentKey;
            end
            
            if ~this.HasDataForCurrentKey
                this.HasData = sqliteStore.hasNextKey();
            end
        end
    end
end
