function tallArray = tall(distributedArray)
%TALL Convert distributed array into the equivalent tall array
%   
%   T = TALL(X)
%   
%   If X is distributed in the first dimension, the tall output has the same
%   distribution and this conversion requires no communication. If X is
%   distributed in any other dimension or by 2D block cyclic, T contains
%   a redistribution of the underlying data distributed in the first dimension.
%   
%   Example:
%      N = 1000;
%      dX = distributed.colon(1, N)';
%      tX = tall(dX);
%      gather(sum(tX))
%   
%   Converts the distributed (1:N)' into the equivalent tall array and
%   calculates its sum.
%   
%   See also DISTRIBUTED.

%   Copyright 2022-2023 The MathWorks, Inc.

import matlab.bigdata.internal.lazyeval.LazyPartitionedArray
import matlab.bigdata.internal.PartitionMetadata

% Ensure the distributed is valid first because this allows us to make the
% assumption that gcp represents this array.
if ~hCheckIsValid( distributedArray )
    error(message('parallel:distributed:InvalidDistributed'));
end

% Ensuring gcmr and gcp align is important as alternate gcmr environments
% will not support distributed array data.
parallel.internal.bigdata.distributed.throwIfGcpGcmrDifferent();

% Tall arrays only support 1d distributed in first dimension, and also
% require type/size information on the client.
spmd
    distributedArray = redistribute(distributedArray, codistributor1d(1));
    if spmdIndex == 1
        try
            adaptorOrErr = matlab.bigdata.internal.adaptors.getAdaptor(getLocalPart(distributedArray));
        catch err
            adaptorOrErr = err;
        end
    end
end

% We issue adaptor error on client because this gives a reasonable error
% for underlying types not supported for tall arrays (E.G. sparse).
adaptorOrErr = adaptorOrErr{1};
if isa(adaptorOrErr, "MException")
    throw(adaptorOrErr);
else
    adaptor = adaptorOrErr.resetTallSize().setTallSize(size(distributedArray, 1));
end


% Define a tall array with a partition strategy + operation to match
% distributed arrays.
operation = parallel.internal.bigdata.distributed.DistributedGetOperation(...
    distributedArray.KeyVector, distributedArray.ResSetHolder);
partitionMetadata = PartitionMetadata(operation.PartitionStrategy);
partitionedArray = LazyPartitionedArray.applyOperation(operation, partitionMetadata);
% Ensures we chunk in cases where local parts are larger than optimal for
% tall arrays.
partitionedArray = resizechunks(partitionedArray);
tallArray = tall(partitionedArray, adaptor);
