function dt = timetable(varargin)
%TIMETABLE Build a codistributed timetable from codistributed arrays
%   DT = TIMETABLE(DROWTIMES,D1,D2,...) creates a codistributed timetable DT from
%   codistributed arrays D1, D2, ... using the codistributed datetime or
%   duration vector DROWTIMES as the time vector. All arrays must be codistributed
%   and have the same number of rows.
%   
%   DT = TIMETABLE(D1, D2, ..., 'RowTimes',DROWTIMES) creates a timetable
%   using the specified codistributed datetime or duration vector, DROWTIMES, as 
%   the time vector. Other datetime or duration inputs become variables in DT.       
%   
%   DT = TIMETABLE(..., 'VariableNames', {'name1', ..., 'name_M'}) creates a
%   timetable containing variables that have the specified variable names.
%   The names must be valid MATLAB identifiers, and unique.
%   
%   DT = TIMETABLE(..., 'DimensionNames', {'dim1', 'dim2'}) creates a
%   timetable containing variables that have the specified dimension names.
%   The names must be valid MATLAB identifiers, and unique.
%   
%   Limitations:
%   1. The parameters 'SampleRate', 'TimeStep', and 'StartTime' are not supported.
%   2. The 'Events' property is not supported for codistributed timetables.
%   
%   See also TIMETABLE, CODISTRIBUTED.
%   


%   Copyright 2018-2022 The MathWorks, Inc.

% Attempt to deal with trailing p-v pairs.
numVars = matlab.bigdata.internal.util.countTableVarInputs(varargin);
vars = varargin(1:numVars);

pnames   = {'VariableNames' 'RowTimes'  'SampleRate'  'SamplingRate'  'TimeStep'  'StartTime'  'DimensionNames'}; % SamplingRate is legacy
dflts    = {            []         []            []              []          []    seconds(0)               [] };
priority = [             0          0             1               0           0             1                0 ]; % 'Sa' -> 'SampleRate', 'S' -> ambiguous
[varnames, rowtimes, ~, ~, ~, ~, dimnames, supplied] ...
    = matlab.internal.datatypes.parseArgsTabularConstructors(pnames, dflts, priority, ...
                                                             "MATLAB:timetable:StringParamNameNotSupported", ...
                                                             varargin{numVars+1:end});
if supplied.SampleRate || supplied.SamplingRate || supplied.TimeStep || supplied.StartTime
    error(message("parallel:distributed:TimetableUnsupportedParam"));
end

if ~supplied.VariableNames
    % Get the workspace names of the input arguments from inputname if
    % variable names were not provided. Need these names before looking
    % through vars for the time vector.
    varnames = repmat({''}, 1, numVars);
    for i = 1:numVars
        varnames{i} = inputname(i);
    end
end

% Setup rowtimes 
rowtimesName = getString(message("MATLAB:timetable:uistrings:DfltRowDimName"));
if ~supplied.RowTimes
    rowtimes = vars{1};
    % Without rowtimes, first argument must be datetime or duration
    if ~isdatetime(rowtimes) && ~isduration(rowtimes)
        error(message("MATLAB:timetable:NoTimeVector"));
    end
    
    vars(1) = [];
    if ~supplied.VariableNames
        if ~isempty(varnames{1})
            rowtimesName = varnames{1};
        end
        varnames(1) = [];
    end
    numVars = numVars - 1;
end

if supplied.DimensionNames
    rowtimesName = dimnames{1};
end

if ~supplied.VariableNames
    % Fill in default names for data args where inputname couldn't. Do
    % this after removing the time vector from the other vars, to get the
    % default names numbered correctly.
    empties = cellfun("isempty", varnames);
    if any(empties)
        varnames(empties) = matlab.internal.tabular.defaultVariableNames(find(empties));
    end
    % Make sure default names or names from inputname don't conflict
    varnames = matlab.lang.makeUniqueStrings(varnames, {}, namelengthmax);
end

if ~supplied.DimensionNames
    varDimNames = getString(message("MATLAB:timetable:uistrings:DfltVarDimName"));
    dimnames = {rowtimesName, varDimNames};
    matlab.bigdata.internal.util.checkTableVariableNames(varnames, dimnames, numVars);
end

% Data inputs must all be codistributed
if ~all(cellfun(@iscodistributed, vars))
    error(message("parallel:distributed:AllTableArgsDistributed"));
end

% Data inputs must all have same number of rows
nVars = numel(varnames);
nRows = cellfun(@(x) size(x,1), vars);
if isempty(nRows)
    nRows = 1;
elseif ~all(nRows == nRows(1))
    error(message("MATLAB:table:UnequalVarLengths"));
else
    nRows = nRows(1);
end

% We require the inputs to be distributed in dimension 1. We want even
% distribution of data, so use the number of rows to design the ideal
% distribution. We concatenate the times so that those get redistributed
% too.
allVars = cellfun(@(x) iRedistribute(x, nRows), [{rowtimes}, vars], "UniformOutput", false);
localParts = cellfun(@(x) getLocalPart(x), allVars, "UniformOutput", false);

% Now form the timetable on each worker
localTable = matlab.bigdata.internal.util.makeTabularChunk(@timetable, localParts(2:end), ...
    {'RowTimes', localParts{1}, 'VariableNames', varnames, 'DimensionNames', dimnames});

% Build a new codistributed from the local parts
codistr = codistributor1d(1, codistributor1d.unsetPartition, [nRows, nVars]);
dt = codistributed.pDoBuildFromLocalPart(localTable, codistr);

end

function out = iRedistribute(in, nRows)
% Redistribute one input
sz = size(in);
sz(1) = nRows;
codistr = codistributor1d(1, codistributor1d.unsetPartition, sz);
out = redistribute(in, codistr);
end
