function [LPY, LPI, codistr] = hMinMaxImpl(codistr, fcnMinMax, LP, dims, wantI, linearI)
; %#ok<NOSEM> % Undocumented
%hMinMaxImpl Implementation for codistributor1d.

%   Copyright 2009-2022 The MathWorks, Inc.

% First the workers independently calculate the min/max on their local part.
if wantI
    [LPY, LPI] = fcnMinMax(LP, [], dims);
    % Adjust LPI to be global indices or subscripts
    if linearI
        LPI = codistr.hLocalToGlobalLinearIndices(LPI, spmdIndex);
    else
        assert(numel(dims)==1, "Two output multi-dimensional MIN/MAX without 'linear' option")
        LPI = codistr.hLocalToGlobalIndices(LPI, dims, spmdIndex);
    end
else
    LPY = fcnMinMax(LP, [], dims);
    LPI = [];
end

if any(codistr.Dimension == dims)
    % In general, we need to communicate for min/max across the distribution
    % dimension, but we try to find special cases where that is not necessary.
    if nnz(codistr.Partition) == 1
        [LPY, LPI, codistr] = iHandleTrivialPartition(codistr, LPY, LPI, wantI);
    else
        [LPY, LPI, codistr] = iMinMaxAcrossWorkers(codistr, fcnMinMax, LPY, LPI, wantI, linearI);
    end
else
    % Operation was not along the distribution dimension, so output has
    % the same partition as the input.  However, we have reduced
    % the size in dim(s) down to 1 (unless already 0).
    outSize = iGetPaddedGlobalSize(codistr, dims);
    outSize(dims) = min(outSize(dims), 1);
    codistr = codistributor1d(codistr.Dimension, codistr.Partition, outSize);
end
    
end % End of hMinMaxImpl.

function [LPY, LPI, codistr] = iMinMaxAcrossWorkers(codistr, fcnMinMax, LPY, LPI, wantI, linearI)
% The input LPY and LPI are the data and the indices after calculating min/max
% on the local parts.
% We perform the second stage of min/max where we operate across the workers.
if wantI
    % Gather all the local min/max and the indices from the first min/max
    % operation.  Pack the data into a cell array and then unpack so that
    % we only communicate once.
    % Opportunity for performance improvements by spmdCat-ing only to worker 1.
    data = spmdCat({LPY, LPI}, 2);
    repFirstData = cat(codistr.Dimension, data{1:2:end-1});
    repFirstInd = cat(codistr.Dimension, data{2:2:end});

    % Perform the second stage min/max across all the local parts.
    [repY, indToFirst] = fcnMinMax(repFirstData, [], codistr.Dimension);
    
    % The 'linear' flag may have been baked into fcnMinMax. If not then we
    % need to convert subscripts in the reduction dimension into linear
    % indices.
    if ~linearI
        indToFirst = distributedutil.IndexManip.minmaxIndToLinear( ...
            size(repFirstData), codistr.Dimension, indToFirst);
    end
    
    % Extract the original indices that were selected by the second
    % reduction.
    repI = repFirstInd(indToFirst);
    
    % Construct local parts from the replicated data.
    % Size in codistr.Dimension is 1.  We should change this to use the last
    % non-singleton dimension instead.
    codistr = codistributor1d(codistr.Dimension);
    [LPY, codistr] = codistr.hBuildFromReplicatedImpl(0, repY);
    LPI = codistr.hBuildFromReplicatedImpl(0, repI);
else
    % Opportunity for performance improvements by spmdCat-ing only to worker 1.
    repFirstData = spmdCat(LPY, codistr.Dimension); % Gather local min/max
    repY = fcnMinMax(repFirstData, [], codistr.Dimension);    % Compute global min/max
    % Size in codistr.Dimension is 1.  We should change this to use the last
    % non-singleton dimension instead.
    codistr = codistributor1d(codistr.Dimension);
    [LPY, codistr] = codistr.hBuildFromReplicatedImpl(0, repY);
end
end % End of iMinMaxInDistDim.

function [LPY, LPI, codistr] = iHandleTrivialPartition(codistr, LPY, LPI, wantI)
% The input LPY and LPI are the data and the indices after calculating min/max
% on the local parts.
% We handle the special case where the min/max calculations can be completed
% without performing any communication because the entire array resides on a
% single worker.
assert(nnz(codistr.Partition) == 1, ...
    "MIN/MAX trivial partitioning called with non-trivial partitions");

part = codistr.Partition;
part(part ~= 0) = 1;
gsize = size(LPY);
gsize(end+1:codistr.Dimension-1) = 1; % Pad any trailing dims with 1
% codistr.Dimension is guaranteed to be <= number of dimensions, so the
% following is valid:
gsize(codistr.Dimension) = 1;
codistr = codistributor1d(codistr.Dimension, part, gsize);
if codistr.Partition(spmdIndex) == 0
    LPY = distributedutil.Allocator.create(codistr.hLocalSize(), LPY);
    if wantI
        LPI = distributedutil.Allocator.create(codistr.hLocalSize(), LPI);
    end
end

end % iHandleTrivialPartition.

function gsize = iGetPaddedGlobalSize(codistr, dims)
% Get the global array size padded with ones as required if any dims>ndims
gsize = codistr.Cached.GlobalSize;
gsize(end+1:max(dims)) = 1;
end % iGetPaddedGlobalSize
