function varargout = minkMaxkOp(fcnMinkMaxk, A, k, varargin)
%MINKMAXKOP - template for maxk and mink

%   Copyright 2019-2022 The MathWorks, Inc.

% Make sure all arguments other than A are gathered.
distributedutil.CodistParser.verifyNonCodistributedInputs([{A}, {k}, varargin]);
fcnStr = func2str(fcnMinkMaxk);
argList = distributedutil.CodistParser.gatherElements([{k}, varargin]);
k = argList{1};

try
    % if A isn't codistributed, call MATLAB function.
    if ~iscodistributed(A)
        % Only the nondata arguments were codistributed.
        [varargout{(1:nargout)}] = fcnMinkMaxk(A, argList{:});
        return
    end

    % Use the local function to check syntax.
    distributedutil.validateSyntax(fcnMinkMaxk, [{A}, argList]);

    if nargin > 3
        sortArgs = iParseArgs(A, fcnStr, argList{2:end});
        sortDim = sortArgs{1};
    else
        sortDim = distributedutil.Sizes.firstNonSingletonDimension(size(A));
        if strcmpi(fcnStr, 'maxk')
            sortArgs = [{sortDim} {'descend'}];
        else
            sortArgs = [{sortDim} {'ascend'}];
        end
    end
    minkmaxkArgs = sortArgs;
    minkmaxkArgs(2) = [];  % sort direction not used when calling mink or maxk

    % If not 1D, redistribute to 1D.
    codistrA = getCodistributor(A);
    if ~isa(codistrA, 'codistributor1d')
        % Try to distribute on a dimension we're not sorting on.
        % Distribute on the highest dimension unless it is the sorting one.
        if sortDim == ndims(A)
            A = redistribute(A, codistributor1d(sortDim - 1));
        else 
            A = redistribute(A, codistributor1d(ndims(A)));
        end
        codistrA = getCodistributor(A);
    end
    distDim = codistrA.Dimension;

    % k shouldn't be greater than the size of A in the sorting dimension.
    k = min(k, size(A, sortDim));
    if size(A, sortDim) == k  % Everything is a max or min!
        [varargout{(1:nargout)}] = sort(A, sortArgs{:});
        return
    elseif k == 1
        fcnStr = func2str(fcnMinkMaxk);
        fcnMinMax = str2func(fcnStr(1:end-1));
        [varargout{(1:nargout)}] = fcnMinMax(A, [], minkmaxkArgs{:});
        return
     elseif k >= size(A, sortDim)/spmdSize && distDim == sortDim
        % Requested k is too large to do reduction efficiently, so sort
        % and do subsref on result.
        [varargout{(1:nargout)}] = sort(A, sortArgs{:});
        % Truncate all outputs using subs
        subs = repmat({':'}, 1, ndims(varargout{1}));
        subs{sortDim} = matlab.internal.ColonDescriptor(1, 1, k);
        varargout = cellfun(@(x)subsref(x, substruct('()', subs)), varargout, 'UniformOutput', false);
        return
    else
        % Not distributed in sort dimension, or distributed in sort 
        % dimension and k is small enough so we can do reduction.
        [varargout{(1:nargout)}] = minkMaxkReduction(fcnMinkMaxk, A, k, minkmaxkArgs);
    end
catch E
    if strfind(E.identifier, 'complexIntegersNotSupported')
        % Error stack should only show mink or maxk, not minkmaxkOp.
        E = MException(['MATLAB:', fcnStr, ':complexIntegers'], E.message);
    end
    throwAsCaller(E);
end

end

function argout = iParseArgs(A, fcnStr, varargin)
% Looks at the optional arguments to see if the sort dimension was specified.
firstArg = varargin{1};
if isnumeric(firstArg) && isscalar(firstArg)
    sortDim = firstArg;
    % Take it out (we put it back later).
    varargin(1) = [];
else
    sortDim = distributedutil.Sizes.firstNonSingletonDimension(size(A));
end
if strcmpi(fcnStr, 'maxk')
    argout = [{sortDim} {'descend'} varargin];
else
    argout = [{sortDim} {'ascend'} varargin];
end
end
