function [C, ia, ib] = intersect(A, B, varargin)
%INTERSECT Set intersect for codistributed array data.
%   C = intersect(A,B)    
%   [C,IA,IB] = intersect(A,B)    
%   [C,IA,IB] = intersect(A,B,'rows')    
%   [C,IA,IB] = intersect(A,B,'stable') 
%   [C,IA,IB] = intersect(A,B,'stable','rows') 
%   [C,IA,IB] = intersect(A,B,'sorted') 
%   [C,IA,IB] = intersect(A,B,'sorted','rows') 
%   
%   Limitations:
%   The 'legacy' flag is not supported.
%   1. Table, timetable, categorical, datetime and duration inputs are not supported.
%   2. Char and string inputs are not supported when A or B is a cell array of char vectors. Convert cell array of char vectors input arguments to string instead.
%   
%   end 
%   See also intersect, codistributed.


%   Copyright 2015-2023 The MathWorks, Inc.

% Allow 5 inputs so the right error can be thrown for the "legacy" flag"
narginchk(2, 5)
nargoutchk(0, 3)

% Default outputs
ia = [];
ib = [];

try
    [flags, A, B] = setfunCheckArgs(@intersect, A, B, varargin{:});
catch ME
    throwAsCaller(ME)
end

isRows = false;
if ~isempty(flags)
    isRows = any(contains(flags, "rows"));
end

% Get membership info from ismember:
% localLia is a logical column vector with same number of elements (rows) as
% localA with true where an element (or row) of B matches that element (or row) of A.
% locb is a double array same size as localLia that gives the linear index in
% B of the first match with the corresponding element or row in A.
% llocb is a logical array of size(numel(localA), numel(B)) or
% (numrows(localA), numrows(B)) with true for matches.

% if one input is a string matrix and the other is a char array, convert
% the char to a string.
if isstring(A) && ischar(B) && ~isRows
    B = string(distributedutil.CodistParser.gatherIfCodistributed(B));
elseif ischar(A) && isstring(B) && ~isRows
    A = string(distributedutil.CodistParser.gatherIfCodistributed(A));
end

% We only support 1D distributions right now. Convert rather than error.
A = pForce1DDistribution(A);
B = pForce1DDistribution(B);

if nargout < 2
    localLia = ismemberImpl(A, B, isRows, "none");
else
    [localLia, locb, llocb] = ismemberImpl(A, B, isRows, "first");
end
[fullC, lia] = setfunExtractMatch(A, localLia, isRows);
if isempty(fullC)
    if isRows
        fullC = codistributed.zeros(0, size(A, 2));
    else
        fullC = codistributed.zeros(1, 0);
    end
    if isstring(A) || isstring(B)
        fullC = codistributed(string.empty(size(fullC)));
    elseif iscell(A) || iscell(B)
        fullC = cell.empty(1, 0);
    else
        fullC = cast(fullC, underlyingType(A));
    end
end

if isrow(fullC) && ~isRows  % C must be a column vector unless isRows.
    fullC = fullC.';
end

if nargout < 2
    C = unique(fullC, flags{:});
else
    [C, cia] = unique(fullC, flags{:});
    ia = setfunGetIndices(cia, lia);
    if iscodistributed(A)
        codistA = getCodistributor(A);
        distDimA = codistA.Dimension;
        partA = codistA.Partition;
        if ~isRows
            locb = codistributed.pDoBuildFromLocalPart(locb, codistA);
        end
    end
    if isRows && ~isempty(llocb)
        locb = zeros(size(lia));
        [allLoca, allLocb] = find(llocb);
        allLoca = flipud(allLoca);   % Apply the first matches last -
        allLocb = flipud(allLocb);   % otherwise legacy results
        locb(allLoca) = allLocb;
    end
    if isempty(cia)  % 'stable' returns an empty double, but distributed empty C.
        ib = cia;
    elseif iscodistributed(cia)
        if isRows && iscodistributed(A) && distDimA == 1
            codistLia =  codistributor1d(1, partA, [size(A, 1) 1]);
            distLocb = codistributed.pDoBuildFromLocalPart(locb, codistLia);
        else
            distLocb = locb;
        end
        ib = subsref(distLocb, substruct('()', {ia}));
    else
        ib = locb(ia);
    end
    if ~iscodistributed(ib)
        ib = codistributed.pConstructFromReplicated(ib);
    end
    if isrow(ib)
        ib = ib';
    end
end

% Change type of output to inferior type of both inputs.
if isnumeric(C) && ~strcmp(underlyingType(A), underlyingType(B))
    if ischar(B)
        C = [C; codistributed(char.empty)];
    elseif isnumeric(B)
        C = [C; zeros([0 size(C, 2)], 'like', B)];
    end
end

if islogical(C) && isnumeric(B)
    C = double(C);
end

% When not in "rows" context, output should be row if both inputs are rows.
if isrow(A) && isrow(B) && iscolumn(C) && ~isRows
    C = C.';
end

% Finally restore correct sparsity if needed.
if issparse(A) || issparse(B)
    C = sparse(C);
end
if issparse(ia)
    ia = full(ia);
end
if issparse(ib)
    ib = full(ib);
end

if ~iscodistributed(C)
    C = codistributed.pConstructFromReplicated(C); 
end
end
