mono/mcs/class/System.Core/System.Linq/ParallelEnumerable.cs

//
// ParallelEnumerable.cs
//
// Author:
//       Jérémie "Garuma" Laval <[email protected]>
//
// Copyright (c) 2010 Jérémie "Garuma" Laval
//
// Permission is hereby granted, free of charge, to any person obtaining a copy
// of this software and associated documentation files (the "Software"), to deal
// in the Software without restriction, including without limitation the rights
// to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
// copies of the Software, and to permit persons to whom the Software is
// furnished to do so, subject to the following conditions:
//
// The above copyright notice and this permission notice shall be included in
// all copies or substantial portions of the Software.
//
// THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
// IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
// FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
// AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
// LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
// OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
// THE SOFTWARE.

using System;
using System.Threading;
using System.Collections;
using System.Collections.Generic;
using System.Collections.Concurrent;
using System.Linq.Parallel;
using System.Linq.Parallel.QueryNodes;

namespace System.Linq
{
	public static class ParallelEnumerable
	{
		#region Range & Repeat
		public static ParallelQuery<int> Range (int start, int count)
		{
			if (int.MaxValue - start < count - 1)
				throw new ArgumentOutOfRangeException ("count", "start + count - 1 is larger than Int32.MaxValue");
			if (count < 0)
				throw new ArgumentOutOfRangeException ("count", "count is less than 0");

			return (new RangeList (start, count)).AsParallel ();
		}

		public static ParallelQuery<TResult> Repeat<TResult> (TResult element, int count)
		{
			if (count < 0)
				throw new ArgumentOutOfRangeException ("count", "count is less than 0");

			return (new RepeatList<TResult> (element, count)).AsParallel ();
		}
		#endregion

		#region Empty
		public static ParallelQuery<TResult> Empty<TResult> ()
		{
			return Repeat<TResult> (default (TResult), 0);
		}
		#endregion

		#region AsParallel
		public static ParallelQuery<TSource> AsParallel<TSource> (this IEnumerable<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			/* Someone might be trying to use AsParallel a bit too much, if the query was in fact
			 * already a ParallelQuery, just cast it
			 */
			ParallelQuery<TSource> query = source as ParallelQuery<TSource>;

			return query ?? new ParallelQuery<TSource> (new QueryStartNode<TSource> (source));
		}

		public static ParallelQuery<TSource> AsParallel<TSource> (this Partitioner<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			/* Someone might be trying to use AsParallel a bit too much, if the query was in fact
			 * already a ParallelQuery, just cast it
			 */
			ParallelQuery<TSource> query = source as ParallelQuery<TSource>;

			return query ?? new ParallelQuery<TSource> (new QueryStartNode<TSource> (source));
		}

		public static ParallelQuery AsParallel (this IEnumerable source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			/* Someone might be trying to use AsParallel a bit too much, if the query was in fact
			 * already a ParallelQuery, just cast it
			 */
			ParallelQuery query = source as ParallelQuery;

			return query ?? new ParallelQuery<object> (new QueryStartNode<object> (source.Cast<object> ()));
		}

		public static IEnumerable<TSource> AsEnumerable<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.AsSequential ();
		}

		public static IEnumerable<TSource> AsSequential<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Node.GetSequential ();
		}
		#endregion

		#region AsOrdered / AsUnordered
		public static ParallelQuery<TSource> AsOrdered<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return new ParallelQuery<TSource> (new QueryAsOrderedNode<TSource> (source.Node));
		}

		public static ParallelQuery<TSource> AsUnordered<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return new ParallelQuery<TSource> (new QueryAsUnorderedNode<TSource> (source.Node));
		}

		public static ParallelQuery AsOrdered (this ParallelQuery source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.TypedQuery.AsOrdered ();
		}
		#endregion

		#region With*
		public static ParallelQuery<TSource> WithExecutionMode<TSource> (this ParallelQuery<TSource> source,
		                                                                 ParallelExecutionMode executionMode)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return new ParallelQuery<TSource> (new ParallelExecutionModeNode<TSource> (executionMode, source.Node));
		}

		public static ParallelQuery<TSource> WithCancellation<TSource> (this ParallelQuery<TSource> source,
		                                                                CancellationToken cancellationToken)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return new ParallelQuery<TSource> (new CancellationTokenNode<TSource> (cancellationToken, source.Node));
		}

		public static ParallelQuery<TSource> WithMergeOptions<TSource> (this ParallelQuery<TSource> source,
		                                                                ParallelMergeOptions mergeOptions)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return new ParallelQuery<TSource> (new ParallelMergeOptionsNode<TSource> (mergeOptions, source.Node));
		}

		public static ParallelQuery<TSource> WithDegreeOfParallelism<TSource> (this ParallelQuery<TSource> source,
		                                                                       int degreeOfParallelism)
		{
			if (degreeOfParallelism < 1 || degreeOfParallelism > 63)
				throw new ArgumentException ("degreeOfParallelism is less than 1 or greater than 63", "degreeOfParallelism");
			if (source == null)
				throw new ArgumentNullException ("source");

			return new ParallelQuery<TSource> (new DegreeOfParallelismNode<TSource> (degreeOfParallelism, source.Node));
		}

		internal static ParallelQuery<TSource> WithImplementerToken<TSource> (this ParallelQuery<TSource> source,
		                                                                      CancellationTokenSource token)
		{
			return new ParallelQuery<TSource> (new ImplementerTokenNode<TSource> (token, source.Node));
		}
		#endregion

		#region Select
		public static ParallelQuery<TResult> Select<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return new ParallelQuery<TResult> (new QuerySelectNode<TResult, TSource> (source.Node, selector));
		}

		public static ParallelQuery<TResult> Select<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, int, TResult> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return new ParallelQuery<TResult> (new QuerySelectNode<TResult, TSource> (source.Node, selector));
		}
		#endregion
		
		#region SelectMany
		public static ParallelQuery<TResult> SelectMany<TSource, TResult> (this ParallelQuery<TSource> source,
		                                                                   Func<TSource, IEnumerable<TResult>> selector)
		{
			return source.SelectMany (selector, (s, e) => e);
		}

		public static ParallelQuery<TResult> SelectMany<TSource, TResult> (this ParallelQuery<TSource> source,
		                                                                   Func<TSource, int, IEnumerable<TResult>> selector)
		{
			return source.SelectMany (selector, (s, e) => e);
		}
		
		public static ParallelQuery<TResult> SelectMany<TSource, TCollection, TResult> (this ParallelQuery<TSource> source,
		                                                                                Func<TSource, IEnumerable<TCollection>> collectionSelector,
		                                                                                Func<TSource, TCollection, TResult> resultSelector)
		{
			return new ParallelQuery<TResult> (new QuerySelectManyNode<TSource, TCollection, TResult> (source.Node,
			                                                                                           collectionSelector,
			                                                                                           resultSelector));
		}
		
		public static ParallelQuery<TResult> SelectMany<TSource, TCollection, TResult> (this ParallelQuery<TSource> source,
		                                                                                Func<TSource, int, IEnumerable<TCollection>> collectionSelector,
		                                                                                Func<TSource, TCollection, TResult> resultSelector)
		{
			return new ParallelQuery<TResult> (new QuerySelectManyNode<TSource, TCollection, TResult> (source.Node,
			                                                                                           collectionSelector,
			                                                                                           resultSelector));
		}
		#endregion

		#region Where
		public static ParallelQuery<TSource> Where<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return new ParallelQuery<TSource> (new QueryWhereNode<TSource> (source.Node, predicate));
		}

		public static ParallelQuery<TSource> Where<TSource> (this ParallelQuery<TSource> source, Func<TSource, int, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return new ParallelQuery<TSource> (new QueryWhereNode<TSource> (source.Node, predicate));
		}
		#endregion

		#region Aggregate
		public static TSource Aggregate<TSource> (this ParallelQuery<TSource> source, Func<TSource, TSource, TSource> func)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (func == null)
				throw new ArgumentNullException ("func");

			return source.Aggregate<TSource, TSource, TSource> ((Func<TSource>)null,
			                                                    func,
			                                                    func,
			                                                    (e) => e);
		}

		public static TAccumulate Aggregate<TSource, TAccumulate> (this ParallelQuery<TSource> source,
		                                                             TAccumulate seed,
		                                                             Func<TAccumulate, TSource, TAccumulate> func)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (func == null)
				throw new ArgumentNullException ("func");

			return source.Aggregate (seed, func, (e) => e);
		}

		public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
		                                                                  TAccumulate seed,
		                                                                  Func<TAccumulate, TSource, TAccumulate> func,
		                                                                  Func<TAccumulate, TResult> resultSelector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (func == null)
				throw new ArgumentNullException ("func");
			if (resultSelector == null)
				throw new ArgumentNullException ("resultSelector");

			TAccumulate accumulator = seed;

			foreach (TSource value in source)
				accumulator = func (accumulator, value);

			return resultSelector (accumulator);
		}

		public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
		                                                                  TAccumulate seed,
		                                                                  Func<TAccumulate, TSource, TAccumulate> updateAccumulatorFunc,
		                                                                  Func<TAccumulate, TAccumulate, TAccumulate> combineAccumulatorsFunc,
		                                                                  Func<TAccumulate, TResult> resultSelector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (updateAccumulatorFunc == null)
				throw new ArgumentNullException ("updateAccumulatorFunc");
			if (combineAccumulatorsFunc == null)
				throw new ArgumentNullException ("combineAccumulatorsFunc");
			if (resultSelector == null)
				throw new ArgumentNullException ("resultSelector");

			return source.Aggregate (() => seed, updateAccumulatorFunc, combineAccumulatorsFunc, resultSelector);
		}

		public static TResult Aggregate<TSource, TAccumulate, TResult> (this ParallelQuery<TSource> source,
		                                                                  Func<TAccumulate> seedFactory,
		                                                                  Func<TAccumulate, TSource, TAccumulate> updateAccumulatorFunc,
		                                                                  Func<TAccumulate, TAccumulate, TAccumulate> combineAccumulatorsFunc,
		                                                                  Func<TAccumulate, TResult> resultSelector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (seedFactory == null)
				throw new ArgumentNullException ("seedFactory");
			if (updateAccumulatorFunc == null)
				throw new ArgumentNullException ("updateAccumulatorFunc");
			if (combineAccumulatorsFunc == null)
				throw new ArgumentNullException ("combineAccumulatorsFunc");
			if (resultSelector == null)
				throw new ArgumentNullException ("resultSelector");

			TAccumulate accumulator = default (TAccumulate);

			ParallelExecuter.ProcessAndAggregate<TSource, TAccumulate> (source.Node, seedFactory, updateAccumulatorFunc, (list) => {
				accumulator = list [0];
				for (int i = 1; i < list.Count; i++)
					accumulator = combineAccumulatorsFunc (accumulator, list[i]);
			});

			return resultSelector (accumulator);;
		}
		#endregion

		#region ForAll
		public static void ForAll<TSource> (this ParallelQuery<TSource> source, Action<TSource> action)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (action == null)
				throw new ArgumentNullException ("action");

			ParallelExecuter.ProcessAndBlock (source.Node, (e, c) => action (e));
		}
		#endregion

		#region OrderBy
		public static OrderedParallelQuery<TSource> OrderByDescending<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                                              Func<TSource, TKey> keySelector,
		                                                                              IComparer<TKey> comparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (keySelector == null)
				throw new ArgumentNullException ("keySelector");
			if (comparer == null)
				comparer = Comparer<TKey>.Default;

			Comparison<TSource> comparison = (e1, e2) => -comparer.Compare (keySelector (e1), keySelector (e2));

			return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
		}

		public static OrderedParallelQuery<TSource> OrderByDescending<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                                              Func<TSource, TKey> keySelector)
		{
			return OrderByDescending (source, keySelector, Comparer<TKey>.Default);
		}

		public static OrderedParallelQuery<TSource> OrderBy<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                                    Func<TSource, TKey> keySelector)
		{
			return OrderBy (source, keySelector, Comparer<TKey>.Default);
		}

		public static OrderedParallelQuery<TSource> OrderBy<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                                    Func<TSource, TKey> keySelector,
		                                                                    IComparer<TKey> comparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (keySelector == null)
				throw new ArgumentNullException ("keySelector");
			if (comparer == null)
				comparer = Comparer<TKey>.Default;

			Comparison<TSource> comparison = (e1, e2) => comparer.Compare (keySelector (e1), keySelector (e2));

			return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
		}
		#endregion

		#region ThenBy
		public static OrderedParallelQuery<TSource> ThenBy<TSource, TKey> (this OrderedParallelQuery<TSource> source,
		                                                                   Func<TSource, TKey> keySelector)
		{
			return ThenBy (source, keySelector, Comparer<TKey>.Default);
		}

		public static OrderedParallelQuery<TSource> ThenBy<TSource, TKey> (this OrderedParallelQuery<TSource> source,
		                                                                   Func<TSource, TKey> keySelector,
		                                                                   IComparer<TKey> comparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (keySelector == null)
				throw new ArgumentNullException ("keySelector");
			if (comparer == null)
				comparer = Comparer<TKey>.Default;

			Comparison<TSource> comparison = (e1, e2) => comparer.Compare (keySelector (e1), keySelector (e2));

			return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
		}

		public static OrderedParallelQuery<TSource> ThenByDescending<TSource, TKey> (this OrderedParallelQuery<TSource> source,
		                                                                             Func<TSource, TKey> keySelector)
		{
			return ThenByDescending (source, keySelector, Comparer<TKey>.Default);
		}

		public static OrderedParallelQuery<TSource> ThenByDescending<TSource, TKey> (this OrderedParallelQuery<TSource> source,
		                                                                             Func<TSource, TKey> keySelector,
		                                                                             IComparer<TKey> comparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (keySelector == null)
				throw new ArgumentNullException ("keySelector");
			if (comparer == null)
				comparer = Comparer<TKey>.Default;

			Comparison<TSource> comparison = (e1, e2) => -comparer.Compare (keySelector (e1), keySelector (e2));

			return new OrderedParallelQuery<TSource> (new QueryOrderByNode<TSource> (source.Node, comparison));
		}
		#endregion

		#region All
		public static bool All<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			CancellationTokenSource src = new CancellationTokenSource ();
			ParallelQuery<TSource> innerQuery = source.WithImplementerToken (src);

			bool result = true;
			try {
				innerQuery.ForAll ((e) => {
						if (!predicate (e)) {
							result = false;
							src.Cancel ();
						}
					});
			} catch (OperationCanceledException e) {
				if (e.CancellationToken != src.Token)
					throw e;
			}

			return result;
		}
		#endregion

		#region Any
		public static bool Any<TSource> (this ParallelQuery<TSource> source)
		{
			return Any<TSource> (source, (_) => true);
		}

		public static bool Any<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return !source.All ((e) => !predicate (e));
		}
		#endregion

		#region Contains
		public static bool Contains<TSource> (this ParallelQuery<TSource> source, TSource value)
		{
			return Contains<TSource> (source, value, EqualityComparer<TSource>.Default);
		}

		public static bool Contains<TSource> (this ParallelQuery<TSource> source, TSource value, IEqualityComparer<TSource> comparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (comparer == null)
				comparer = EqualityComparer<TSource>.Default;

			return Any<TSource> (source, (e) => comparer.Equals (value, e));
		}
		#endregion

		#region SequenceEqual
		public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
		                                           ParallelQuery<TSource> second)
		{
			if (first == null)
				throw new ArgumentNullException ("first");
			if (second == null)
				throw new ArgumentNullException ("second");

			return first.SequenceEqual (second, EqualityComparer<TSource>.Default);
		}

		public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
		                                           ParallelQuery<TSource> second,
		                                           IEqualityComparer<TSource> comparer)
		{
			if (first == null)
				throw new ArgumentNullException ("first");
			if (second == null)
				throw new ArgumentNullException ("second");
			if (comparer == null)
				comparer = EqualityComparer<TSource>.Default;

			var source = new CancellationTokenSource ();
			var zip = new QueryZipNode<TSource, TSource, bool> (comparer.Equals, first.Node, second.Node) {	Strict = true };
			var innerQuery = new ParallelQuery<bool> (zip).WithImplementerToken (source);

			bool result = true;

			try {
				innerQuery.ForAll (value => {
					if (!value) {
						result = false;
						source.Cancel ();
					}
				});
			} catch (AggregateException ex) {
				if (ex.InnerException is QueryZipException)
					return false;
				else
					throw ex;
			} catch (OperationCanceledException e) {
				if (e.CancellationToken != source.Token)
					throw e;
			}

			return result;
		}

		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
		{
			throw new NotSupportedException ();
		}

		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static bool SequenceEqual<TSource> (this ParallelQuery<TSource> first,
		                                              IEnumerable<TSource> second,
		                                              IEqualityComparer<TSource> comparer)
		{
			throw new NotSupportedException ();
		}

		#endregion
		
		#region GroupBy
		public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                                              Func<TSource, TKey> keySelector)
		{
			return source.GroupBy (keySelector, EqualityComparer<TKey>.Default);
		}
		
		public static ParallelQuery<IGrouping<TKey, TSource>> GroupBy<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                                              Func<TSource, TKey> keySelector,
		                                                                              IEqualityComparer<TKey> comparer)
		{
			return source.GroupBy (keySelector, new Identity<TSource> ().Apply, comparer);
		}
		
		public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
		                                                                                         Func<TSource, TKey> keySelector,
		                                                                                         Func<TSource, TElement> elementSelector)
		{
			return source.GroupBy (keySelector, elementSelector, EqualityComparer<TKey>.Default);
		}
		
		public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> (this ParallelQuery<TSource> source,
		                                                                      Func<TSource, TKey> keySelector,
		                                                                      Func<TKey, IEnumerable<TSource>, TResult> resultSelector)
		{
			return source.GroupBy (keySelector)
				.Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
		}
		
		public static ParallelQuery<IGrouping<TKey, TElement>> GroupBy<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
		                                                                                         Func<TSource, TKey> keySelector,
		                                                                                         Func<TSource, TElement> elementSelector,
		                                                                                         IEqualityComparer<TKey> comparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (keySelector == null)
				throw new ArgumentNullException ("keySelector");
			if (elementSelector == null)
				throw new ArgumentNullException ("elementSelector");
			if (comparer == null)
				comparer = EqualityComparer<TKey>.Default;

			return new ParallelQuery<IGrouping<TKey, TElement>> (new QueryGroupByNode<TSource, TKey, TElement> (source.Node, keySelector, elementSelector, comparer));
		}
		
		public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
		                                                                                Func<TSource, TKey> keySelector,
		                                                                                Func<TSource, TElement> elementSelector,
		                                                                                Func<TKey, IEnumerable<TElement>, TResult> resultSelector)
		{
			return source.GroupBy (keySelector, elementSelector)
				.Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
		}
		
		public static ParallelQuery<TResult> GroupBy<TSource, TKey, TResult> (this ParallelQuery<TSource> source,
		                                                                      Func<TSource, TKey> keySelector,
		                                                                      Func<TKey, IEnumerable<TSource>, TResult> resultSelector,
		                                                                      IEqualityComparer<TKey> comparer)
		{
			return source.GroupBy (keySelector, comparer)
				.Select ((g) => resultSelector (g.Key, (IEnumerable<TSource>)g));
		}
		
		public static ParallelQuery<TResult> GroupBy<TSource, TKey, TElement, TResult> (this ParallelQuery<TSource> source,
		                                                                                Func<TSource, TKey> keySelector,
		                                                                                Func<TSource, TElement> elementSelector,
		                                                                                Func<TKey, IEnumerable<TElement>, TResult> resultSelector,
		                                                                                IEqualityComparer<TKey> comparer)
		{
			return source.GroupBy (keySelector, elementSelector, comparer)
				.Select ((g) => resultSelector (g.Key, (IEnumerable<TElement>)g));
		}
		#endregion

		#region GroupJoin
		public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
		                                                                               ParallelQuery<TInner> inner,
		                                                                               Func<TOuter, TKey> outerKeySelector,
		                                                                               Func<TInner, TKey> innerKeySelector,
		                                                                               Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
		{
			return outer.GroupJoin (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
		}
		
		public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
		                                                                               ParallelQuery<TInner> inner,
		                                                                               Func<TOuter, TKey> outerKeySelector,
		                                                                               Func<TInner, TKey> innerKeySelector,
		                                                                               Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
		                                                                               IEqualityComparer<TKey> comparer)
		{
			return outer.Join (inner.GroupBy (innerKeySelector, (e) => e), outerKeySelector, (e) => e.Key, (e1, e2) => resultSelector (e1, e2), comparer);
		}
		
		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
		                                                                               IEnumerable<TInner> inner,
		                                                                               Func<TOuter, TKey> outerKeySelector,
		                                                                               Func<TInner, TKey> innerKeySelector,
		                                                                               Func<TOuter, IEnumerable<TInner>, TResult> resultSelector)
		{
			throw new NotSupportedException ();
		}
		
		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TResult> GroupJoin<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
		                                                                               IEnumerable<TInner> inner,
		                                                                               Func<TOuter, TKey> outerKeySelector,
		                                                                               Func<TInner, TKey> innerKeySelector,
		                                                                               Func<TOuter, IEnumerable<TInner>, TResult> resultSelector,
		                                                                               IEqualityComparer<TKey> comparer)
		{
			throw new NotSupportedException ();
		}
		#endregion

		#region ElementAt
		public static TSource ElementAt<TSource> (this ParallelQuery<TSource> source, int index)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (index < 0)
				throw new ArgumentOutOfRangeException ("index");
			if (index == 0) {
				try {
					return source.First ();
				} catch (InvalidOperationException) {
					throw new ArgumentOutOfRangeException ("index");
				}
			}

			TSource result = default (TSource);

			ParallelQuery<TSource> innerQuery = source.Where ((e, i) => i == index);

			try {
				result = innerQuery.First ();
			} catch (InvalidOperationException) {
				throw new ArgumentOutOfRangeException ("index");
			}

			return result;
		}

		public static TSource ElementAtOrDefault<TSource> (this ParallelQuery<TSource> source, int index)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			try {
				return source.ElementAt (index);
			} catch (ArgumentOutOfRangeException) {
				return default (TSource);
			}
		}
		#endregion

		#region Intersect
		public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
		                                                         ParallelQuery<TSource> second)
		{
			return Intersect<TSource> (first, second, EqualityComparer<TSource>.Default);
		}

		public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
		                                                         ParallelQuery<TSource> second,
		                                                         IEqualityComparer<TSource> comparer)
		{
			if (first == null)
				throw new ArgumentNullException ("first");
			if (second == null)
				throw new ArgumentNullException ("second");
			if (comparer == null)
				comparer = EqualityComparer<TSource>.Default;

			return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Intersect, comparer, first.Node, second.Node));
		}

		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first, IEnumerable<TSource> second)
		{
			throw new NotSupportedException ();
		}

		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TSource> Intersect<TSource> (this ParallelQuery<TSource> first,
		                                                         IEnumerable<TSource> second,
		                                                         IEqualityComparer<TSource> comparer)
		{
			throw new NotSupportedException ();
		}
		#endregion

		#region Join
		public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
		                                                                          ParallelQuery<TInner> inner,
		                                                                          Func<TOuter, TKey> outerKeySelector,
		                                                                          Func<TInner, TKey> innerKeySelector,
		                                                                          Func<TOuter, TInner, TResult> resultSelector)
		{
			return outer.Join (inner, outerKeySelector, innerKeySelector, resultSelector, EqualityComparer<TKey>.Default);
		}
		
		public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
		                                                                          ParallelQuery<TInner> inner,
		                                                                          Func<TOuter, TKey> outerKeySelector,
		                                                                          Func<TInner, TKey> innerKeySelector,
		                                                                          Func<TOuter, TInner, TResult> resultSelector,
		                                                                          IEqualityComparer<TKey> comparer)
		{
			return new ParallelQuery<TResult> (new QueryJoinNode<TOuter, TInner, TKey, TResult> (outer.Node, inner.Node, outerKeySelector, innerKeySelector, resultSelector, comparer));
		}
		
		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
		                                                                          IEnumerable<TInner> inner,
		                                                                          Func<TOuter, TKey> outerKeySelector,
		                                                                          Func<TInner, TKey> innerKeySelector,
		                                                                          Func<TOuter, TInner, TResult> resultSelector)
		{
			throw new NotSupportedException ();
		}

		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TResult> Join<TOuter, TInner, TKey, TResult> (this ParallelQuery<TOuter> outer,
		                                                                          IEnumerable<TInner> inner,
		                                                                          Func<TOuter, TKey> outerKeySelector,
		                                                                          Func<TInner, TKey> innerKeySelector,
		                                                                          Func<TOuter, TInner, TResult> resultSelector,
		                                                                          IEqualityComparer<TKey> comparer)
		{
			throw new NotSupportedException ();
		}
		#endregion

		#region Except
		public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
		                                                      ParallelQuery<TSource> second)
		{
			return Except<TSource> (first, second, EqualityComparer<TSource>.Default);
		}

		public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
		                                                      ParallelQuery<TSource> second,
		                                                      IEqualityComparer<TSource> comparer)
		{
			if (first == null)
				throw new ArgumentNullException ("first");
			if (second == null)
				throw new ArgumentNullException ("second");
			if (comparer == null)
				comparer = EqualityComparer<TSource>.Default;

			return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Except,
			                                                              comparer, first.Node, second.Node));
		}

		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
		                                                      IEnumerable<TSource> second)
		{
			throw new NotSupportedException ();
		}

		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TSource> Except<TSource> (this ParallelQuery<TSource> first,
		                                                      IEnumerable<TSource> second,
		                                                      IEqualityComparer<TSource> comparer)
		{
			throw new NotSupportedException ();
		}
		#endregion

		#region Distinct
		public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source)
		{
			return Distinct<TSource> (source, EqualityComparer<TSource>.Default);
		}

		public static ParallelQuery<TSource> Distinct<TSource> (this ParallelQuery<TSource> source, IEqualityComparer<TSource> comparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (comparer == null)
				comparer = EqualityComparer<TSource>.Default;

			return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Distinct, comparer,
			                                                              source.Node, null));
		}
		#endregion

		#region Union
		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
		                                                     IEnumerable<TSource> second)
		{
			throw new NotSupportedException ();
		}

		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TSource> Union<TSource>(this ParallelQuery<TSource> first,
		                                                    IEnumerable<TSource> second,
		                                                    IEqualityComparer<TSource> comparer)
		{
			throw new NotSupportedException ();
		}

		public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
		                                                     ParallelQuery<TSource> second)
		{
			return first.Union (second, EqualityComparer<TSource>.Default);
		}

		public static ParallelQuery<TSource> Union<TSource> (this ParallelQuery<TSource> first,
		                                                     ParallelQuery<TSource> second,
		                                                     IEqualityComparer<TSource> comparer)
		{
			if (first == null)
				throw new ArgumentNullException ("first");
			if (second == null)
				throw new ArgumentNullException ("second");
			if (comparer == null)
				comparer = EqualityComparer<TSource>.Default;

			return new ParallelQuery<TSource> (new QuerySetNode<TSource> (SetInclusionDefaults.Union, comparer, first.Node, second.Node));
		}
		#endregion

		#region Take
		public static ParallelQuery<TSource> Take<TSource> (this ParallelQuery<TSource> source, int count)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, count));
		}

		public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
		                                                         Func<TSource, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, (e, _) => predicate (e), false));
		}

		public static ParallelQuery<TSource> TakeWhile<TSource> (this ParallelQuery<TSource> source,
		                                                         Func<TSource, int, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return new ParallelQuery<TSource> (new QueryHeadWorkerNode<TSource> (source.Node, predicate, true));
		}
		#endregion

		#region Skip
		public static ParallelQuery<TSource> Skip<TSource> (this ParallelQuery<TSource> source, int count)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Node.IsOrdered () ?
				source.Where ((e, i) => i >= count) :
				source.Where ((e) => count < 0 || Interlocked.Decrement (ref count) < 0);

		}

		public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
		                                                         Func<TSource, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return source.Node.IsOrdered () ?
				source.SkipWhile ((e, i) => predicate (e)) :
				source.Where ((e) => !predicate (e));
		}

		public static ParallelQuery<TSource> SkipWhile<TSource> (this ParallelQuery<TSource> source,
		                                                         Func<TSource, int, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			int indexCache = int.MaxValue;

			return source.Where ((e, i) => i >= indexCache || (!predicate (e, i) && (indexCache = i) == i));
		}
		#endregion

		#region Single
		static TSource SingleInternal<TSource> (this ParallelQuery<TSource> source, params TSource[] init)
		{
			TSource result = default(TSource);
			bool hasValue = false;

			foreach (TSource element in source) {
				if (hasValue)
					throw new InvalidOperationException ("The input sequence contains more than one element.");

				result = element;
				hasValue = true;
			}

			if (!hasValue && init.Length != 0) {
				result = init[0];
				hasValue = true;
			}

			if (!hasValue)
				throw new InvalidOperationException ("The input sequence is empty.");

			return result;
		}

		public static TSource Single<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return SingleInternal<TSource> (source);
		}

		public static TSource Single<TSource> (this ParallelQuery<TSource> source,
		                                       Func<TSource, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return source.Where (predicate).Single ();
		}

		public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return SingleInternal<TSource> (source, default (TSource));
		}

		public static TSource SingleOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return source.Where (predicate).SingleOrDefault ();
		}
		#endregion

		#region Count
		public static int Count<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			var helper = new CountAggregateHelper<TSource> ();
			return source.Aggregate<TSource, int, int> (helper.Seed,
			                                            helper.Intermediate,
			                                            helper.Reducer,
			                                            helper.Final);
		}

		public static int Count<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return source.Where (predicate).Count ();
		}

		class CountAggregateHelper<TSource>
		{
			public int Seed ()
			{
				return 0;
			}

			public int Intermediate (int acc, TSource e)
			{
				return acc + 1;
			}

			public int Reducer (int acc1, int acc2)
			{
				return acc1 + acc2;
			}

			public int Final (int acc)
			{
				return acc;
			}
		}

		public static long LongCount<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			var helper = new LongCountAggregateHelper<TSource> ();
			return source.Aggregate<TSource, long, long> (helper.Seed,
			                                              helper.Intermediate,
			                                              helper.Reducer,
			                                              helper.Final);
		}

		public static long LongCount<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (predicate == null)
				throw new ArgumentNullException ("predicate");

			return source.Where (predicate).LongCount ();
		}

		class LongCountAggregateHelper<TSource>
		{
			public long Seed ()
			{
				return 0;
			}

			public long Intermediate (long acc, TSource e)
			{
				return acc + 1;
			}

			public long Reducer (long acc1, long acc2)
			{
				return acc1 + acc2;
			}

			public long Final (long acc)
			{
				return acc;
			}
		}
		#endregion

		#region Average
		public static double Average (this ParallelQuery<int> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate (() => new int[2],
			                        (acc, e) => { acc[0] += e; acc[1]++; return acc; },
			                        (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
			                        (acc) => acc[0] / ((double)acc[1]));
		}

		public static double Average (this ParallelQuery<long> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate (() => new long[2],
			                        (acc, e) => { acc[0] += e; acc[1]++; return acc; },
			                        (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
			                        (acc) => acc[0] / ((double)acc[1]));
		}

		public static decimal Average (this ParallelQuery<decimal> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate (() => new decimal[2],
			                        (acc, e) => { acc[0] += e; acc[1]++; return acc; },
			                        (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
			                        (acc) => acc[0] / acc[1]);
		}

		public static double Average (this ParallelQuery<double> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate (() => new double[2],
			                        (acc, e) => { acc[0] += e; acc[1]++; return acc; },
			                        (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
			                        (acc) => acc[0] / ((double)acc[1]));
		}

		public static float Average (this ParallelQuery<float> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate (() => new float[2],
			                        (acc, e) => { acc[0] += e; acc[1]++; return acc; },
			                        (acc1, acc2) => { acc1[0] += acc2[0]; acc1[1] += acc2[1]; return acc1; },
			                        (acc) => acc[0] / acc[1]);
		}
		#endregion

		#region More Average
		public static double? Average (this ParallelQuery<int?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();;
		}

		public static double? Average (this ParallelQuery<long?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
		}

		public static decimal? Average (this ParallelQuery<decimal?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
		}

		public static double? Average (this ParallelQuery<double?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
		}

		public static float? Average (this ParallelQuery<float?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : 0).Average ();
		}

		public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}

		public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}

		public static float Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}

		public static double Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}

		public static decimal Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}

		public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}

		public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}

		public static float? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}

		public static double? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}

		public static decimal? Average<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Average ();
		}
		#endregion

		#region Sum
		public static int Sum (this ParallelQuery<int> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate (0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
		}

		public static long Sum (this ParallelQuery<long> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate ((long)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
		}

		public static float Sum (this ParallelQuery<float> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate (0.0f, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
		}

		public static double Sum (this ParallelQuery<double> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate (0.0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
		}

		public static decimal Sum (this ParallelQuery<decimal> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Aggregate ((decimal)0, (e1, e2) => e1 + e2, (sum1, sum2) => sum1 + sum2, (sum) => sum);
		}

		public static int? Sum (this ParallelQuery<int?> source)
		{
			return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
		}

		public static long? Sum (this ParallelQuery<long?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
		}

		public static float? Sum (this ParallelQuery<float?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
		}

		public static double? Sum (this ParallelQuery<double?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
		}

		public static decimal? Sum (this ParallelQuery<decimal?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : 0).Sum ();
		}

		public static int Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}

		public static long Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}

		public static decimal Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}

		public static float Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}

		public static double Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}

		public static int? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}

		public static long? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}

		public static decimal? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}

		public static float? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}

		public static double? Sum<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Sum ();
		}
		#endregion

		#region Min-Max
		static T BestOrder<T> (ParallelQuery<T> source, BestOrderComparer<T> bestOrderComparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			T best = source.Aggregate (bestOrderComparer.Seed,
			                           bestOrderComparer.Intermediate,
			                           bestOrderComparer.Intermediate,
			                           new Identity<T> ().Apply);
			return best;
		}

		class BestOrderComparer<T>
		{
			IComparer<T> comparer;
			int inverter;
			T seed;

			public BestOrderComparer (IComparer<T> comparer, int inverter, T seed)
			{
				this.comparer = comparer;
				this.inverter = inverter;
				this.seed = seed;
			}

			public T Seed ()
			{
				return seed;
			}

			public T Intermediate (T first, T second)
			{
				return Better (first, second) ? first : second;
			}

			bool Better (T first, T second)
			{
				return (inverter * comparer.Compare (first, second)) > 0;
			}
		}

		public static int Min (this ParallelQuery<int> source)
		{
			return BestOrder (source, new BestOrderComparer<int> (Comparer<int>.Default, -1, int.MaxValue));
		}

		public static long Min (this ParallelQuery<long> source)
		{
			return BestOrder (source, new BestOrderComparer<long> (Comparer<long>.Default, -1, long.MaxValue));
		}

		public static float Min (this ParallelQuery<float> source)
		{
			return BestOrder (source, new BestOrderComparer<float> (Comparer<float>.Default, -1, float.MaxValue));
		}

		public static double Min (this ParallelQuery<double> source)
		{
			return BestOrder (source, new BestOrderComparer<double> (Comparer<double>.Default, -1, double.MaxValue));
		}

		public static decimal Min (this ParallelQuery<decimal> source)
		{
			return BestOrder (source, new BestOrderComparer<decimal> (Comparer<decimal>.Default, -1, decimal.MaxValue));
		}

		public static TSource Min<TSource> (this ParallelQuery<TSource> source)
		{
			IComparer<TSource> comparer = Comparer<TSource>.Default;
			return BestOrder (source, new BestOrderComparer<TSource> (comparer, -1, default (TSource)));
		}

		public static TResult Min<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static int? Min (this ParallelQuery<int?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : int.MaxValue).Min ();
		}

		public static long? Min (this ParallelQuery<long?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : long.MaxValue).Min ();
		}

		public static float? Min (this ParallelQuery<float?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : float.MaxValue).Min ();
		}

		public static double? Min (this ParallelQuery<double?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : double.MaxValue).Min ();
		}

		public static decimal? Min (this ParallelQuery<decimal?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : decimal.MaxValue).Min ();
		}

		public static int Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static long Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static float Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static double Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static decimal Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static int? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static long? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static float? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static double? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static decimal? Min<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Min ();
		}

		public static int Max (this ParallelQuery<int> source)
		{
			return BestOrder (source, new BestOrderComparer<int> (Comparer<int>.Default, 1, int.MinValue));
		}

		public static long Max (this ParallelQuery<long> source)
		{
			return BestOrder (source, new BestOrderComparer<long> (Comparer<long>.Default, 1, long.MinValue));
		}

		public static float Max (this ParallelQuery<float> source)
		{
			return BestOrder (source, new BestOrderComparer<float> (Comparer<float>.Default, 1, float.MinValue));
		}

		public static double Max (this ParallelQuery<double> source)
		{
			return BestOrder (source, new BestOrderComparer<double> (Comparer<double>.Default, 1, double.MinValue));
		}

		public static decimal Max (this ParallelQuery<decimal> source)
		{
			return BestOrder (source, new BestOrderComparer<decimal> (Comparer<decimal>.Default, 1, decimal.MinValue));
		}

		public static TSource Max<TSource> (this ParallelQuery<TSource> source)
		{
			IComparer<TSource> comparer = Comparer<TSource>.Default;
			return BestOrder (source, new BestOrderComparer<TSource> (comparer, 1, default (TSource)));
		}

		public static TResult Max<TSource, TResult> (this ParallelQuery<TSource> source, Func<TSource, TResult> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static int? Max (this ParallelQuery<int?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : int.MinValue).Max ();
		}

		public static long? Max (this ParallelQuery<long?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : long.MinValue).Max ();
		}

		public static float? Max (this ParallelQuery<float?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : float.MinValue).Max ();
		}

		public static double? Max (this ParallelQuery<double?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : double.MinValue).Max ();
		}

		public static decimal? Max (this ParallelQuery<decimal?> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.Select ((e) => e.HasValue ? e.Value : decimal.MinValue).Max ();
		}

		public static int Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static long Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static float Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static double Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static decimal Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static int? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, int?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static long? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, long?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static float? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, float?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static double? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, double?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}

		public static decimal? Max<TSource> (this ParallelQuery<TSource> source, Func<TSource, decimal?> selector)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (selector == null)
				throw new ArgumentNullException ("selector");

			return source.Select (selector).Max ();
		}
		#endregion

		#region Cast / OfType
		public static ParallelQuery<TResult> Cast<TResult> (this ParallelQuery source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.TypedQuery.Select ((e) => (TResult)e);
		}

		public static ParallelQuery<TResult> OfType<TResult> (this ParallelQuery source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return source.TypedQuery.Where ((e) => e is TResult).Cast<TResult> ();
		}
		#endregion

		#region Reverse
		public static ParallelQuery<TSource> Reverse<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			return new ParallelQuery<TSource> (new QueryReverseNode<TSource> (source));
		}
		#endregion

		#region ToArray - ToList - ToDictionary - ToLookup
		public static List<TSource> ToList<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			if (source.Node.IsOrdered ())
				return ToListOrdered (source);

			var helper = new ListAggregateHelper<TSource> ();
			List<TSource> temp = source.Aggregate (helper.Seed,
			                                       helper.Intermediate,
			                                       helper.Reducer,
			                                       helper.Final);
			return temp;
		}

		class ListAggregateHelper<TSource>
		{
			public List<TSource> Seed ()
			{
				return new List<TSource> (50);
			}

			public List<TSource> Intermediate (List<TSource> list, TSource e)
			{
				list.Add (e);
				return list;
			}

			public List<TSource> Reducer (List<TSource> list, List<TSource> list2)
			{
				list.AddRange (list2);
				return list;
			}

			public List<TSource> Final (List<TSource> list)
			{
				return list;
			}
		}

		internal static List<TSource> ToListOrdered<TSource> (this ParallelQuery<TSource> source)
		{
			List<TSource> result = new List<TSource> ();

			foreach (TSource element in source)
				result.Add (element);

			return result;
		}

		public static TSource[] ToArray<TSource> (this ParallelQuery<TSource> source)
		{
			if (source == null)
				throw new ArgumentNullException ("source");

			if (source.Node.IsOrdered ())
				return ToListOrdered (source).ToArray ();

			var helper = new ArrayAggregateHelper<TSource> ();
			ParallelExecuter.ProcessAndAggregate<TSource, List<TSource>> (source.Node,
			                                                              helper.Seed,
			                                                              helper.Intermediate,
			                                                              helper.Final);

			return helper.Result;
		}

		class ArrayAggregateHelper<TSource>
		{
			TSource[] result;

			public TSource[] Result {
				get {
					return result;
				}
			}

			internal List<TSource> Seed ()
			{
				return new List<TSource> ();
			}

			internal List<TSource> Intermediate (List<TSource> list, TSource e)
			{
				list.Add (e);
				return list;
			}

			internal void Final (IList<List<TSource>> list)
			{
				int count = 0;

				for (int i = 0; i < list.Count; i++)
					count += list[i].Count;

				result = new TSource[count];
				int insertIndex = -1;

				for (int i = 0; i < list.Count; i++)
					for (int j = 0; j < list[i].Count; j++)
						result [++insertIndex] = list[i][j];
			}
		}

		public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                                     Func<TSource, TKey> keySelector,
		                                                                     IEqualityComparer<TKey> comparer)
		{
			return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
		}

		public static Dictionary<TKey, TSource> ToDictionary<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                                     Func<TSource, TKey> keySelector)
		{
			return ToDictionary<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
		}

		public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
		                                                                                  Func<TSource, TKey> keySelector,
		                                                                                  Func<TSource, TElement> elementSelector)
		{
			return ToDictionary<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
		}

		public static Dictionary<TKey, TElement> ToDictionary<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
		                                                                                  Func<TSource, TKey> keySelector,
		                                                                                  Func<TSource, TElement> elementSelector,
		                                                                                  IEqualityComparer<TKey> comparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (keySelector == null)
				throw new ArgumentNullException ("keySelector");
			if (comparer == null)
				comparer = EqualityComparer<TKey>.Default;
			if (elementSelector == null)
				throw new ArgumentNullException ("elementSelector");

			var helper = new DictionaryAggregateHelper<TSource, TKey, TElement> (comparer, keySelector, elementSelector);
			return source.Aggregate (helper.Seed,
			                         helper.Intermediate,
			                         helper.Reducer,
			                         helper.Final);
		}

		class DictionaryAggregateHelper<TSource, TKey, TElement>
		{
			IEqualityComparer<TKey> comparer;
			Func<TSource, TKey> keySelector;
			Func<TSource, TElement> elementSelector;

			public DictionaryAggregateHelper (IEqualityComparer<TKey> comparer,
			                                  Func<TSource, TKey> keySelector,
			                                  Func<TSource, TElement> elementSelector)
			{
				this.comparer = comparer;
				this.keySelector = keySelector;
				this.elementSelector = elementSelector;
			}

			public Dictionary<TKey, TElement> Seed ()
			{
				return new Dictionary<TKey, TElement> (comparer);
			}

			public Dictionary<TKey, TElement> Intermediate (Dictionary<TKey, TElement> d, TSource e)
			{
				d.Add (keySelector (e), elementSelector (e));
				return d;
			}

			public Dictionary<TKey, TElement> Reducer (Dictionary<TKey, TElement> d1, Dictionary<TKey, TElement> d2)
			{
				foreach (var couple in d2) d1.Add (couple.Key, couple.Value);
				return d1;
			}

			public Dictionary<TKey, TElement> Final (Dictionary<TKey, TElement> d)
			{
				return d;
			}
		}

		public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                              Func<TSource, TKey> keySelector)
		{
			return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, EqualityComparer<TKey>.Default);
		}

		public static ILookup<TKey, TSource> ToLookup<TSource, TKey> (this ParallelQuery<TSource> source,
		                                                              Func<TSource, TKey> keySelector,
		                                                              IEqualityComparer<TKey> comparer)
		{
			return ToLookup<TSource, TKey, TSource> (source, keySelector, (e) => e, comparer);
		}

		public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
		                                                                         Func<TSource, TKey> keySelector,
		                                                                         Func<TSource, TElement> elementSelector)
		{
			return ToLookup<TSource, TKey, TElement> (source, keySelector, elementSelector, EqualityComparer<TKey>.Default);
		}

		public static ILookup<TKey, TElement> ToLookup<TSource, TKey, TElement> (this ParallelQuery<TSource> source,
		                                                                         Func<TSource, TKey> keySelector,
		                                                                         Func<TSource, TElement> elementSelector,
		                                                                         IEqualityComparer<TKey> comparer)
		{
			if (source == null)
				throw new ArgumentNullException ("source");
			if (keySelector == null)
				throw new ArgumentNullException ("keySelector");
			if (comparer == null)
				comparer = EqualityComparer<TKey>.Default;
			if (elementSelector == null)
				throw new ArgumentNullException ("elementSelector");

			ConcurrentLookup<TKey, TElement> lookup = new ConcurrentLookup<TKey, TElement> (comparer);
			source.ForAll ((e) => lookup.Add (keySelector (e), elementSelector (e)));

			return lookup;
		}
		#endregion

		#region Concat
		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather than "
		                   + "System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() extension method "
		                   + "to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TSource> Concat<TSource>(this ParallelQuery<TSource> first,
		                                                     IEnumerable<TSource> second)
		{
			throw new NotSupportedException ();
		}

		public static ParallelQuery<TSource> Concat<TSource> (this ParallelQuery<TSource> first, ParallelQuery<TSource> second)
		{
			return new ParallelQuery<TSource> (new QueryConcatNode<TSource> (first.Node, second.Node));
		}
		#endregion
		
		#region DefaultIfEmpty
		public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source)
		{
			return source.DefaultIfEmpty (default (TSource));
		}
		
		public static ParallelQuery<TSource> DefaultIfEmpty<TSource> (this ParallelQuery<TSource> source, TSource defaultValue)
		{
			return new ParallelQuery<TSource> (new QueryDefaultEmptyNode<TSource> (source.Node, defaultValue));
		}
		#endregion
		
		#region First
		public static TSource First<TSource> (this ParallelQuery<TSource> source)
		{
			CancellationTokenSource src = new CancellationTokenSource ();
			IEnumerator<TSource> enumerator = source.WithImplementerToken (src).GetEnumerator ();
			
			if (enumerator == null || !enumerator.MoveNext ())
				throw new InvalidOperationException ("source contains no element");
			
			TSource result = enumerator.Current;
			src.Cancel ();
			enumerator.Dispose ();
			
			return result;
		}
		
		public static TSource First<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			return source.Where (predicate).First ();
		}
		
		public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source)
		{
			return source.DefaultIfEmpty ().First ();
		}
		
		public static TSource FirstOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			return source.Where (predicate).FirstOrDefault ();
		}
		#endregion
		
		#region Last
		public static TSource Last<TSource> (this ParallelQuery<TSource> source)
		{
			return source.Reverse ().First ();
		}
		
		public static TSource Last<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			return source.Reverse ().First (predicate);
		}
		
		public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source)
		{
			return source.Reverse ().FirstOrDefault ();
		}
		
		public static TSource LastOrDefault<TSource> (this ParallelQuery<TSource> source, Func<TSource, bool> predicate)
		{
			return source.Reverse ().FirstOrDefault (predicate);
		}
		#endregion

		#region Zip
		public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
		                                                                      ParallelQuery<TSecond> second,
		                                                                      Func<TFirst, TSecond, TResult> resultSelector)
		{
			if (first == null)
				throw new ArgumentNullException ("first");
			if (second == null)
				throw new ArgumentNullException ("second");
			if (resultSelector == null)
				throw new ArgumentNullException ("resultSelector");

			return new ParallelQuery<TResult> (new QueryZipNode<TFirst, TSecond, TResult> (resultSelector, first.Node, second.Node));
		}

		[ObsoleteAttribute("The second data source of a binary operator must be of type System.Linq.ParallelQuery<T> rather "
		                   + "than System.Collections.Generic.IEnumerable<T>. To fix this problem, use the AsParallel() "
		                   + "extension method to convert the right data source to System.Linq.ParallelQuery<T>.")]
		public static ParallelQuery<TResult> Zip<TFirst, TSecond, TResult> (this ParallelQuery<TFirst> first,
		                                                                      IEnumerable<TSecond> second,
		                                                                      Func<TFirst, TSecond, TResult> resultSelector)
		{
			throw new NotSupportedException ();
		}
		#endregion

		#region Helpers

		class Identity<T>
		{
			public T Apply (T input)
			{
				return input;
			}
		}
		#endregion
	}
}