<commit_before>﻿// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Diagnostics;
using System.Runtime.Serialization;
using System.Runtime.Serialization.Formatters.Binary;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Spark.CSharp.Interop;
using Microsoft.Spark.CSharp.Proxy;

namespace Microsoft.Spark.CSharp.Core
{

    /// <summary>
    /// Wraps C#-based transformations that can be executed within a stage. It helps avoid unnecessary Ser/De of data between
    /// JVM & CLR to execute C# transformations and pipelines them
    /// </summary>
    /// <typeparam name="U"></typeparam>
    [Serializable]
    public class PipelinedRDD<U> : RDD<U>
    {
        internal CSharpWorkerFunc workerFunc;
        internal bool preservesPartitioning;

        //TODO - give generic types a better id
        public override RDD<U1> MapPartitionsWithIndex<U1>(Func<int, IEnumerable<U>, IEnumerable<U1>> newFunc, bool preservesPartitioningParam = false)
        {
            if (IsPipelinable())
            {
                CSharpWorkerFunc newWorkerFunc = new CSharpWorkerFunc(
                    new MapPartitionsWithIndexHelper<U, U1>(newFunc, workerFunc.Func).Execute, workerFunc.StackTrace);

                var pipelinedRDD = new PipelinedRDD<U1>
                {
                    workerFunc = newWorkerFunc,
                    preservesPartitioning = preservesPartitioning && preservesPartitioningParam,
                    previousRddProxy = previousRddProxy,
                    prevSerializedMode = prevSerializedMode,
                    sparkContext = sparkContext,
                    rddProxy = null,
                    serializedMode = SerializedMode.Byte,
                    partitioner = preservesPartitioning ? partitioner : null
                };
                return pipelinedRDD;
            }

            return base.MapPartitionsWithIndex(newFunc, preservesPartitioningParam);
        }

        /// <summary>
        /// This class is defined explicitly instead of using anonymous method as delegate to prevent C# compiler from generating
        /// private anonymous type that is not serializable. Since the delegate has to be serialized and sent to the Spark workers
        /// for execution, it is necessary to have the type marked [Serializable]. This class is to work around the limitation
        /// on the serializability of compiler generated types
        /// </summary>
        [Serializable]
        private class MapPartitionsWithIndexHelper<I, O>
        {
            private readonly Func<int, IEnumerable<I>, IEnumerable<O>> newFunc;
            private readonly Func<int, IEnumerable<dynamic>, IEnumerable<dynamic>> prevFunc;
            internal MapPartitionsWithIndexHelper(Func<int, IEnumerable<I>, IEnumerable<O>> nFunc, Func<int, IEnumerable<dynamic>, IEnumerable<dynamic>> pFunc)
            {
                prevFunc = pFunc;
                newFunc = nFunc;
            }<commit_msg>adding qualifier back for improved readability<commit_after>﻿// Copyright (c) Microsoft. All rights reserved.
// Licensed under the MIT license. See LICENSE file in the project root for full license information.

using System;
using System.Collections.Generic;
using System.IO;
using System.Linq;
using System.Diagnostics;
using System.Runtime.Serialization;
using System.Runtime.Serialization.Formatters.Binary;
using System.Text;
using System.Threading.Tasks;
using Microsoft.Spark.CSharp.Interop;
using Microsoft.Spark.CSharp.Proxy;

namespace Microsoft.Spark.CSharp.Core
{

    /// <summary>
    /// Wraps C#-based transformations that can be executed within a stage. It helps avoid unnecessary Ser/De of data between
    /// JVM & CLR to execute C# transformations and pipelines them
    /// </summary>
    /// <typeparam name="U"></typeparam>
    [Serializable]
    public class PipelinedRDD<U> : RDD<U>
    {
        internal CSharpWorkerFunc workerFunc;
        internal bool preservesPartitioning;

        //TODO - give generic types a better id
        public override RDD<U1> MapPartitionsWithIndex<U1>(Func<int, IEnumerable<U>, IEnumerable<U1>> newFunc, bool preservesPartitioningParam = false)
        {
            if (IsPipelinable())
            {
                CSharpWorkerFunc newWorkerFunc = new CSharpWorkerFunc(
                    new MapPartitionsWithIndexHelper<U, U1>(newFunc, workerFunc.Func).Execute, workerFunc.StackTrace);

                var pipelinedRDD = new PipelinedRDD<U1>
                {
                    workerFunc = newWorkerFunc,
                    preservesPartitioning = preservesPartitioning && preservesPartitioningParam,
                    previousRddProxy = this.previousRddProxy,
                    prevSerializedMode = this.prevSerializedMode,
                    sparkContext = this.sparkContext,
                    rddProxy = null,
                    serializedMode = SerializedMode.Byte,
                    partitioner = preservesPartitioning ? partitioner : null
                };
                return pipelinedRDD;
            }

            return base.MapPartitionsWithIndex(newFunc, preservesPartitioningParam);
        }

        /// <summary>
        /// This class is defined explicitly instead of using anonymous method as delegate to prevent C# compiler from generating
        /// private anonymous type that is not serializable. Since the delegate has to be serialized and sent to the Spark workers
        /// for execution, it is necessary to have the type marked [Serializable]. This class is to work around the limitation
        /// on the serializability of compiler generated types
        /// </summary>
        [Serializable]
        private class MapPartitionsWithIndexHelper<I, O>
        {
            private readonly Func<int, IEnumerable<I>, IEnumerable<O>> newFunc;
            private readonly Func<int, IEnumerable<dynamic>, IEnumerable<dynamic>> prevFunc;
            internal MapPartitionsWithIndexHelper(Func<int, IEnumerable<I>, IEnumerable<O>> nFunc, Func<int, IEnumerable<dynamic>, IEnumerable<dynamic>> pFunc)
            {
                prevFunc = pFunc;
                newFunc = nFunc;
            }
{'fraction_non_alphanumeric': 0.07708006279434851, 'fraction_numerical': 0.0015698587127158557, 'mean_word_length': 2.441923284710967, 'pattern_counts': {'":': 0, '<': 59, '<?xml version=': 0, '>': 59, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1084896', 'n_tokens_mistral': 1611, 'n_tokens_neox': 1540, 'n_words': 487}