function tt = table2timetable(t,varargin)
%TABLE2TIMETABLE Convert codistributed table to codistributed timetable.
%   TT = TABLE2TIMETABLE(T)
%   TT = TABLE2TIMETABLE(T,'RowTimes',TIMEVARNAME)
%   TT = TABLE2TIMETABLE(T,'RowTimes',ROWTIMES)
%   
%   Limitations:
%   The parameters 'SampleRate', 'TimeStep', and 'StartTime' are not
%   supported.
%   
%   See also TABLE2TIMETABLE, codistributed.
%   


%   Copyright 2018-2022 The MathWorks, Inc.

if ~istable(t)
    error(message("MATLAB:table2timetable:NonTable"));
end

% Extract the table header, irrespective of distribution scheme
tmpl = getTableTemplate(t);
varNames = tmpl.Properties.VariableNames;

if nargin == 1
    % Take the time vector as the first datetime/duration variable in the table.
    % If the table is n-by-p, the timetable will be n-by-(p-1).
    varClasses = cellfun(@(name) class(tmpl.(name)), varNames, 'UniformOutput', false);
    rowtimesCandidates = ismember(varClasses, {'datetime','duration'});
    rowtimesIndex = find(rowtimesCandidates, 1);
    if isempty(rowtimesIndex)
        error(message("MATLAB:table2timetable:NoTimeVarFound"));
    end
    tt = iConvertUsingExistingTimeVariable(t, rowtimesIndex);
else
    % Take the time vector as the RowTimes input, or as the specified variable in the table.
    % If the table is n-by-p, the timetable will be n-by-p, or n-by-(p-1), respectively.
    pnames = {'VariableNames', 'RowTimes'};
    dflts =  {             [],         []};
    [~, rowtimes, supplied] = matlab.internal.datatypes.parseArgs(pnames, dflts, varargin{:});
    if supplied.VariableNames
        error(message("MATLAB:table2timetable:VariableNamesNotAccepted"));
    end
    if supplied.RowTimes
        if matlab.internal.datatypes.isScalarText(rowtimes)
            rowtimesIndex = find(strcmp(rowtimes, varNames), 1, 'first');
            if isempty(rowtimesIndex)
                error(message("MATLAB:table:UnrecognizedVarName", rowtimes));
            end
            tt = iConvertUsingExistingTimeVariable(t, rowtimesIndex);
            
        elseif matlab.internal.datatypes.isScalarInt(rowtimes, 1)
            if rowtimes > width(t)
                error(message("MATLAB:table:VarIndexOutOfRange"));
            end
            tt = iConvertUsingExistingTimeVariable(t, rowtimes);
            
        elseif any(strcmp(underlyingType(rowtimes), {'datetime', 'duration'}))
            tt = iConvertUsingNewTimeVariable(t, rowtimes);
            
        else
            % Not a name or a valid datetime/duration.
            error(message("MATLAB:table2timetable:InvalidRowTimes"));
            
        end
    end
end

end


% Convert a table to a timetable using one of the existing table variables
% as the time vector. For an input table of size MxN the output will be
% Mx(N-1).
function tt = iConvertUsingExistingTimeVariable(t, rowtimesIndex)

% Tables and timetables are always 1D distribution in first or second
% dimension.
codistr = getCodistributor(t);
part = codistr.Partition;
dim = codistr.Dimension;
inSz = size(t);
outSz = inSz - [0 1];
lp = getLocalPart(t);
assert(isa(codistr, 'codistributor1d') && codistr.Dimension<=2, ...
       "Tabular distributed arrays should always be 1D distribution.");

if codistr.Dimension==1
    % Each worker has all columns, and the converion is "slicewise" (i.e.
    % independent on each worker).
    outLP = table2timetable(lp, 'RowTimes', rowtimesIndex);
    
else
    % Each worker has a few columns. The worker that has the specified time
    % column needs to transmit it to the others and loses that column.
    idxInWorker = rowtimesIndex - cumsum([0, part(1:end-1)]);
    workerWithTimes = find(idxInWorker>0 & idxInWorker<=part, 1, 'first');
    if workerWithTimes == spmdIndex
        % We have this column. Grab it to send to the others and remove
        % from the local table.
        rowtimes = lp{:, idxInWorker(spmdIndex)};
        rowtimesName = lp.Properties.VariableNames{idxInWorker(spmdIndex)};
        lp(:, idxInWorker(spmdIndex)) = [];
    else
        % Someone else has this column
        rowtimes = [];
        rowtimesName = '';
    end
    part(workerWithTimes) = part(workerWithTimes)-1;
    rowtimes = spmdBroadcast(workerWithTimes, rowtimes);
    rowtimesName = spmdBroadcast(workerWithTimes, rowtimesName);

    outLP = table2timetable(lp, 'RowTimes', rowtimes);
    % Fix the time dimension name
    outLP.Properties.DimensionNames{1} = rowtimesName;
    
end
outCodistr = codistributor1d(dim, part, outSz);
tt = codistributed.pDoBuildFromLocalPart(outLP, outCodistr);

end


% Convert a table to a timetable using a new time vector. The result will
% be the same size and distribution scheme as the input.
function tt = iConvertUsingNewTimeVariable(t, rowtimes)

% We cheat here by appending a new column for the time and then calling
% the existing variable case. Note that if the table already has a variable
% "time" then throw an error (as per the in-memory version).
timeT = table(rowtimes, 'VariableNames', {'Time'});

% CAT will redistribute the time vector if necessary and will throw a
% collective error if the time vector is the wrong size.
try
    t = cat(2, t, timeT); % NB: must put at end to preserve properties of t
catch
    error(message("MATLAB:table2timetable:IncorrectNumberOfRowTimes"));
end
tt = iConvertUsingExistingTimeVariable(t, width(t));
end
