%Spark Interact with a Spark cluster
%   The Spark cluster type allows you to perform mapreduce using workers
%   running under the given Spark environment.
%
%   parallel.cluster.Spark properties:
%      ClusterMatlabRoot       - Path to MATLAB on the cluster
%      LicenseNumber           - License number to use with online licensing
%      Modified                - True if any properties in this cluster have been modified
%      NumThreads              - Number of computational threads for workers
%      OperatingSystem         - Operating System of the cluster
%      Profile                 - Profile used to build this cluster
%      RequiresOnlineLicensing - Cluster uses online licensing
%      Type                    - Type of this cluster
%      SparkInstallFolder      - The installation of Spark on the local machine
%      SparkProperties         - A map of name-value property pairs to be given to Spark
%      UserData                - Data associated with a cluster object in this client session

%   Copyright 2022-2023 The MathWorks, Inc.

classdef (ConstructOnLoad = true, Sealed) Spark < parallel.Cluster

    properties (PCTGetSet, Transient, PCTConstraint = "string")
        %SparkInstallFolder The installation of Spark on the local machine
        %   Specifies the path to Spark on the local client machine. This will
        %   be the installation of Spark used to perform tall expressions
        %   on a Spark cluster.
        SparkInstallFolder
    end

    properties (PCTGetSet, Transient, SetAccess = private, PCTConstraint = "sparkproperties")
        %SparkProperties A map of name-value property pairs to be given to Spark
        %   A map of name-value pairs that specify Spark properties. These
        %   will be passed directly to Spark when evaluating tall expressions.
        SparkProperties;
    end

    properties (PCTGetSet, Transient, PCTConstraint = "string")
        %ClusterMatlabRoot Specifies the path to MATLAB for workers to use
        %   ClusterMatlabRoot specifies the path name to MATLAB for the
        %   cluster to use for starting MATLAB worker processes.
        ClusterMatlabRoot
    end

    properties (PCTGetSet, Transient, PCTConstraint = "logical")
        %RequiresOnlineLicensing Cluster uses online licensing
        %   RequiresOnlineLicensing indicates whether the cluster uses
        %   online licensing.  You must provide your MathWorks account
        %   login information if this value is true.  If you have more than one
        %   MATLAB Parallel Server license associated with your
        %   MathWorks account you will also need to supply the LicenseNumber to
        %   use.
        %
        %   See also parallel.cluster.Spark.LicenseNumber
        RequiresOnlineLicensing = false;
    end

    properties (PCTGetSet, Transient, PCTConstraint = "string")
        %LicenseNumber License number to use with online licensing
        %   License number specifies the license number to use on the workers
        %   when the cluster is using online licensing.  This string
        %   must be the same as one of the licenses listed in your MathWorks
        %   account.
        %
        %   LicenseNumber cannot be defined if RequiresOnlineLicensing is
        %   false.
        %
        %   See also parallel.cluster.Spark.RequiresOnlineLicensing
        LicenseNumber = '';
    end

    properties (PCTGetSet, Transient, PCTConstraint = "logical", Hidden)
        %AutoAttachFiles Determines whether dependent files will be automatically sent to the workers
        %   If true the files required to run the job's task functions will be calculated
        %   and sent to the workers at job submission time.
        %   See also parallel.cluster.Spark/AttachedFiles, parallel.cluster.Spark/AdditionalPaths.
        AutoAttachFiles = true;
    end

    properties (PCTGetSet, Transient, PCTConstraint = "nonemptycellstr", Hidden)
        %AttachedFiles A list of files to be sent to the workers during a mapreduce call
        %   AttachedFiles can be modified only before the job is submitted.
        AttachedFiles  = {};

        %AdditionalPaths A list of paths to be added to the worker MATLAB path
        %   AdditionalPaths can be modified only before the job is submitted.
        AdditionalPaths = {};
    end

    properties (SetAccess = immutable, Hidden)
        %Host Identifies the headnode of this cluster
        Host = ''
    end

    properties (PCTGetSet, Transient, ...
            PCTConstraint = 'enum:parallel.internal.types.OperatingSystem')
        %OperatingSystem Operating System of the cluster
        %   OperatingSystem specifies the operating system of the nodes in
        %   the cluster. Valid values are 'windows', 'unix', and 'mixed'.
        %
        %   See also parallel.Cluster.Host
        OperatingSystem
    end

    properties (GetAccess = {?matlab.mapreduce.ParallelSparkMapReducer}, SetAccess = immutable, Transient)
        Impl;
    end

    properties (Access = private, Transient)
        MatlabLogoutListener;
    end

    properties (Hidden, Dependent)
        RequiresMathWorksHostedLicensing
    end

    methods
        function val = get.RequiresMathWorksHostedLicensing(obj)
            val = obj.RequiresOnlineLicensing;
        end
        function set.RequiresMathWorksHostedLicensing(obj, val)
            obj.RequiresOnlineLicensing = val;
        end
    end

    % Unsupported properties
    properties (Dependent, Hidden)
        JobStorageLocation
        NumWorkers
    end

    methods
        function obj = Spark(varargin)
            %Spark Create a Spark cluster instance
            %   parallel.cluster.Spark(p1, v1, p2, v2, ...) builds a generic
            %   cluster with the specified property values. The properties can
            %   include any of the properties of the Spark cluster class.
            %
            %   Example:
            %   myCluster = parallel.cluster.Spark("SparkInstallFolder", "/usr/local/hadoop");
            import parallel.internal.cluster.ConstructorArgsHelper
            try
                [~, cProf] = ...
                    ConstructorArgsHelper.interpretClusterCtorArgs(...
                    ?parallel.cluster.Spark, varargin{:});
            catch err
                throw(err)
            end
            obj = obj@parallel.Cluster('Spark', cProf);
            try
                obj.Impl = parallel.internal.bigdata.SparkCluster(varargin{:});
                obj.Impl.hEnsureSparkInstallFolderIsSet();
                obj.setPropStorage(obj.Impl.PropStorage);
            catch err
                throw(err);
            end
        end

        function jobStorageLocation = get.JobStorageLocation(~)
            jobStorageLocation = getString(message('parallel:bigdata:SparkJobStorageLocation'));
        end
        function set.JobStorageLocation(~, ~)
            error(message("parallel:bigdata:UnsupportedOnSpark"));
        end
        function numWorkers = get.NumWorkers(~)
            numWorkers = Inf;
        end
        function set.NumWorkers(~, ~)
            error(message("parallel:bigdata:UnsupportedOnSpark"));
        end
    end

    methods (Hidden)
        % Hidden method to forward all property implementations to the impl
        % object. Overridden from parallel.internal.customattr.CustomGetSet.
        function hSetProperty(obj, names, values)
            try
                hSetProperty(obj.Impl, names, values);

                % If we have set RequiresOnlineLicensing to true,
                % we need to register a logout listener with the current
                % desktop client (if one has not already been registered)
                if obj.RequiresOnlineLicensing && isempty(obj.MatlabLogoutListener)
                    confirmLogout = true;
                    obj.MatlabLogoutListener = parallel.internal.mwa.addLogoutListener(...
                        "InvalidatingClusters", @obj.onInvalidatingClusters, confirmLogout);
                end
            catch err
                throw(err);
            end
        end
        function values = hGetProperty(obj, names)
            values = hGetProperty(obj.Impl, names);
        end

        % This will be true unless the cluster object has been loaded from a mat file.
        function out = hIsSparkClusterValid( obj )
            out = ~isempty(obj.Impl) && isvalid(obj.Impl);
        end

        % Actions to do as part of connect to cluster profile validation
        function [isNetworkCluster, latency, dataSend, dataReceive] = hTestClusterConnection(~)
            isNetworkCluster = false;
            latency = 0;
            dataSend = Inf;
            dataReceive = Inf;
        end

        % Display items
        function [clusterPropertyMap, propNames] = hGetDisplayItems(obj, diFactory)
            % Add default displayable items to map.
            names = {...
                'Type', ...
                'Profile', ...
                'Modified', ...
                'NumThreads', ...
                'RequiresOnlineLicensing', ...
                'LicenseNumber', ...
                'ClusterMatlabRoot'};
            values = diFactory.makeMultipleItems(@createDefaultItem, obj.hGetDisplayPropertiesNoError(names));
            clusterPropertyMap = containers.Map(names, values);
            clusterPropertyMap('Separator') = diFactory.createSeparator();

            % Add Spark-specific properties
            clusterPropertyMap('SparkInstallFolder') = diFactory.createDefaultItem(obj.hGetDisplayPropertiesNoError('SparkInstallFolder'));
            clusterPropertyMap('SparkProperties') = diFactory.createDefaultItem(obj.SparkProperties);

            propNames = {...
                'Type', ...
                'Profile', ...
                'Modified', ...
                'NumThreads', ...
                'RequiresOnlineLicensing', ...
                'ClusterMatlabRoot', ...
                'Separator', ...
                'SparkInstallFolder', ...
                'SparkProperties'};
        end
    end

    methods (Access = protected)
        function jobs = getJobs(obj) %#ok<MANU>
            % Simply return empty as this cluster class doesn't support
            % creating or querying jobs.
            jobs = parallel.job.CJSIndependentJob.empty(0, 1);
        end

        function props = getAllMementoProperties(obj, clusterMementoType)
            props = getAllMementoProperties(obj.Impl, clusterMementoType);
        end

        function onInvalidatingClusters(obj)
            % This type If the cluster is using online licensing at the
            % time of logout, invalidate the cluster impl object and this
            % wrapper class of the impl.
            if obj.RequiresOnlineLicensing
                delete(obj.Impl);
                delete(obj);
            end
        end
    end

    % Override methods not supported on Spark
    methods (Access = protected)
        function job = buildIndependentJob(obj) %#ok<STOUT,MANU>
            error(message("parallel:bigdata:UnsupportedCreateJobOnSpark"));
        end
        function job = buildCommunicatingJob(obj, variant, executionMode) %#ok<INUSD,STOUT>
            error(message("parallel:bigdata:UnsupportedCreateJobOnSpark"));
        end
        function job = buildConcurrentJob(obj) %#ok<STOUT,MANU>
            error(message("parallel:bigdata:UnsupportedCreateJobOnSpark"));
        end
        function connMgr = buildConnectionManager(obj)  %#ok<STOUT,MANU>
            error(message("parallel:bigdata:UnsupportedPoolOnSpark"));
        end
        function num = getNumConcurrentTasks(obj, job) %#ok<INUSD,STOUT>
            error(message("parallel:bigdata:UnsupportedOnSpark"));
        end
    end

    % Hidden methods not supported on Spark
    methods (Hidden)
        function varargout = hCreateJobStatusProvider(varargin) %#ok<STOUT>
            error(message("parallel:bigdata:UnsupportedOnSpark"));
        end
        function varargout = hCreatePoolEndpointFactory(varargin) %#ok<STOUT>
            error(message("parallel:bigdata:UnsupportedPoolOnSpark"));
        end
    end
end
