Самый эффективный алгоритм объединения отсортированных IEnumerable <T>

Question

У меня есть несколько огромных отсортированных перечислимых последовательностей, которые я хочу объединить . Списки тезисов обрабатываются как IEnumerable, но уже отсортированы . Поскольку входные списки отсортированы, должна быть возможность объединить их за одну поездку, ничего не сортируя заново.

Я бы хотел сохранить поведение отложенного выполнения.

Я пытался написать наивный алгоритм, который делает это (см. Ниже). Тем не менее, это выглядит довольно некрасиво, и я уверен, что это можно оптимизировать. Может существовать более академичный алгоритм ...

IEnumerable<T> MergeOrderedLists<T, TOrder>(IEnumerable<IEnumerable<T>> orderedlists, 
                                            Func<T, TOrder> orderBy)
{
    var enumerators = orderedlists.ToDictionary(l => l.GetEnumerator(), l => default(T));
    IEnumerator<T> tag = null;

    var firstRun = true;
    while (true)
    {
        var toRemove = new List<IEnumerator<T>>();
        var toAdd = new List<KeyValuePair<IEnumerator<T>, T>>();
        foreach (var pair in enumerators.Where(pair => firstRun || tag == pair.Key))
        {
            if (pair.Key.MoveNext())
                toAdd.Add(pair);
            else
                toRemove.Add(pair.Key);
        }

        foreach (var enumerator in toRemove)
            enumerators.Remove(enumerator);

        foreach (var pair in toAdd)
            enumerators[pair.Key] = pair.Key.Current;

        if (enumerators.Count == 0)
            yield break;

        var min = enumerators.OrderBy(t => orderBy(t.Value)).FirstOrDefault();
        tag = min.Key;
        yield return min.Value;

        firstRun = false;
    }
}

Метод можно использовать так:

// Person lists are already sorted by age
MergeOrderedLists(orderedList, p => p.Age);

при условии, что где-то существует следующий класс Person:

    public class Person
    {
        public int Age { get; set; }
    }

Дубликаты должны быть сохранены, нас не волнует их порядок в новой последовательности. Вы видите какую-то очевидную оптимизацию, которую я мог бы использовать?

Answer 1

Вот мой четвертый (спасибо @tanascius за то, что он продвинул это к чему-то гораздо большему LINQ):

public static IEnumerable<T> MergePreserveOrder3<T, TOrder>(
    this IEnumerable<IEnumerable<T>> aa,
    Func<T, TOrder> orderFunc)
where TOrder : IComparable<TOrder>
{
    var items = aa.Select(xx => xx.GetEnumerator()).Where(ee => ee.MoveNext())
        .OrderBy(ee => orderFunc(ee.Current)).ToList();

    while (items.Count > 0)
    {
        yield return items[0].Current;

        var next = items[0];
        items.RemoveAt(0);
        if (next.MoveNext())
        {
            // simple sorted linear insert
            var value = orderFunc(next.Current);
            var ii = 0;
            for ( ; ii < items.Count; ++ii)
            {
                if (value.CompareTo(orderFunc(items[ii].Current)) <= 0)
                {
                    items.Insert(ii, next);
                    break;
                }
            }

            if (ii == items.Count) items.Add(next);
        }
        else next.Dispose(); // woops! can't forget IDisposable
    }
}

Результаты:

for (int p = 0; p < people.Count; ++p)
{
    Console.WriteLine("List {0}:", p + 1);
    Console.WriteLine("\t{0}", String.Join(", ", people[p].Select(x => x.Name)));
}

Console.WriteLine("Merged:");
foreach (var person in people.MergePreserveOrder(pp => pp.Age))
{
    Console.WriteLine("\t{0}", person.Name);
}

List 1:
        8yo, 22yo, 47yo, 49yo
List 2:
        35yo, 47yo, 60yo
List 3:
        28yo, 55yo, 64yo
Merged:
        8yo
        22yo
        28yo
        35yo
        47yo
        47yo
        49yo
        55yo
        60yo
        64yo

Улучшено благодаря поддержке Tuple .Net 4.0:

public static IEnumerable<T> MergePreserveOrder4<T, TOrder>(
    this IEnumerable<IEnumerable<T>> aa,
    Func<T, TOrder> orderFunc) where TOrder : IComparable<TOrder>
{
    var items = aa.Select(xx => xx.GetEnumerator())
                  .Where(ee => ee.MoveNext())
                  .Select(ee => Tuple.Create(orderFunc(ee.Current), ee))
                  .OrderBy(ee => ee.Item1).ToList();

    while (items.Count > 0)
    {
        yield return items[0].Item2.Current;

        var next = items[0];
        items.RemoveAt(0);
        if (next.Item2.MoveNext())
        {
            var value = orderFunc(next.Item2.Current);
            var ii = 0;
            for (; ii < items.Count; ++ii)
            {
                if (value.CompareTo(items[ii].Item1) <= 0)
                {   // NB: using a tuple to minimize calls to orderFunc
                    items.Insert(ii, Tuple.Create(value, next.Item2));
                    break;
                }
            }

            if (ii == items.Count) items.Add(Tuple.Create(value, next.Item2));
        }
        else next.Item2.Dispose(); // woops! can't forget IDisposable
    }
}

Answer 2

Можно предположить, что это может улучшить четкость и производительность:

• Создайте приоритетную очередь для пар T, IEnumerable<T>, упорядоченных в соответствии с вашей функцией сравнения на T
• Для каждого объединяемого IEnumerable<T> добавьте элемент в очередь приоритетов с аннотацией со ссылкой на IEnumerable<T>, из которой он возник
• Пока приоритетная очередь не пуста
  • Извлечение минимального элемента из очереди приоритетов
  • Перейдите IEnumerable<T> в его аннотации к следующему элементу
  • Если MoveNext() вернул true, добавьте следующий элемент в очередь приоритетов с аннотацией со ссылкой на IEnumerable<T>, который вы только что добавили
  • Если MoveNext() вернул false, ничего не добавляйте в приоритетную очередь
  • Выход из вытесненного элемента

Answer 3

Вот решение с очень хорошим анализом сложности, которое значительно короче, чем другие предложенные решения.

public static IEnumerable<T> Merge<T>(this IEnumerable<IEnumerable<T>> self) 
    where T : IComparable<T>
{
    var es = self.Select(x => x.GetEnumerator()).Where(e => e.MoveNext());
    var tmp = es.ToDictionary(e => e.Current);
    var dict = new SortedDictionary<T, IEnumerator<T>>(tmp);
    while (dict.Count > 0)
    {
        var key = dict.Keys.First();
        var cur = dict[key];
        dict.Remove(key);
        yield return cur.Current;
        if (cur.MoveNext())
            dict.Add(cur.Current, cur);                    
    }
}

Answer 4

Вот решение без сортировки ... только минимальное количество сравнений. (Я упустил фактический порядок передачи функций для простоты). Обновлено для построения сбалансированного дерева: -

    /// <summary>
    /// Merge a pair of ordered lists
    /// </summary>
    public static IEnumerable<T> Merge<T>(IEnumerable<T> aList, IEnumerable<T> bList)
        where T:IComparable<T>
    {
        var a = aList.GetEnumerator();
        bool aOK = a.MoveNext();

        foreach (var b in bList)
        {
            while (aOK && a.Current.CompareTo(b) <= 0) {yield return a.Current; aOK = a.MoveNext();}
            yield return b;
        }
        // And anything left in a
        while (aOK) { yield return a.Current; aOK = a.MoveNext(); }
    }

    /// <summary>
    /// Merge lots of sorted lists
    /// </summary>
    public static IEnumerable<T> Merge<T>(IEnumerable<IEnumerable<T>> listOfLists)
        where T : IComparable<T>
    {
        int n = listOfLists.Count();
        if (n < 2) 
            return listOfLists.FirstOrDefault();
        else
            return Merge (Merge(listOfLists.Take(n/2)), Merge(listOfLists.Skip(n/2)));
    }


public static void Main(string[] args)
{

    var sample = Enumerable.Range(1, 5).Select((i) => Enumerable.Range(i, i+5).Select(j => string.Format("Test {0:00}", j)));

    Console.WriteLine("Merged:");
    foreach (var result in Merge(sample))
    {
        Console.WriteLine("\t{0}", result);
    }

Answer 5

Сколько списков вы ожидаете объединить? Похоже, ваш алгоритм не будет эффективным, если у вас есть много разных списков для объединения. Эта строка является проблемой:

var min = enumerators.OrderBy(t => orderBy(t.Value)).FirstOrDefault();

Это будет выполнено один раз для каждого элемента во всех списках, поэтому время выполнения будет O (n * m), где n - ОБЩЕЕ количество элементов во всех списках, а n - количество списков. Выраженная в терминах средней длины списка в списке списков, время выполнения равно O (a * m ^ 2).

Если вам нужно объединить много списков, я бы предложил использовать heap . Затем на каждой итерации вы можете удалить наименьшее значение из кучи и добавить следующий элемент в кучу из списка, из которого получено наименьшее значение.

Answer 6

Вот мое решение:
Алгоритм берет первый элемент каждого списка и помещает их в небольшой вспомогательный класс (отсортированный список, который принимает элементы mutliple с одинаковым значением). В этом отсортированном списке используется двоичная вставка .
Таким образом, первый элемент в этом списке - это элемент, который мы хотим вернуть следующим. После этого мы удаляем его из отсортированного списка и вставляем следующий элемент из исходного списка источников (по крайней мере, пока этот список содержит больше элементов). Опять же, мы можем вернуть первый элемент нашего отсортированного списка. Когда отсортированный список пуст один раз, мы использовали все элементы из всех разных исходных списков и сделали.

Это решение использует меньше операторов foreach и не OrderBy на каждом шаге, что должно улучшить поведение во время выполнения. Только двоичная вставка должна быть сделана снова и снова.

IEnumerable<T> MergeOrderedLists<T, TOrder>( IEnumerable<IEnumerable<T>> orderedlists, Func<T, TOrder> orderBy )
{
    // Get an enumerator for each list, create a sortedList
    var enumerators = orderedlists.Select( enumerable => enumerable.GetEnumerator() );
    var sortedEnumerators = new SortedListAllowingDoublets<TOrder, IEnumerator<T>>();

    // Point each enumerator onto the first element
    foreach( var enumerator in enumerators )
    {
        // Missing: assert true as the return value
        enumerator.MoveNext();

        //  Initially add the first value
        sortedEnumerators.AddSorted( orderBy( enumerator.Current ), enumerator );
    }

    // Continue as long as we have elements to return
    while( sortedEnumerators.Count != 0 )
    {
        // The first element of the sortedEnumerator list always
        // holds the next element to return
        var enumerator = sortedEnumerators[0].Value;

        // Return this enumerators current value
        yield return enumerator.Current;

        // Remove the element we just returned
        sortedEnumerators.RemoveAt( 0 );

        // Check if there is another element in the list of the enumerator
        if( enumerator.MoveNext() )
        {
            // Ok, so add it to the sorted list
            sortedEnumerators.AddSorted( orderBy( enumerator.Current ), enumerator );
        }
    }

Мой вспомогательный класс (с использованием простой двоичной вставки):

private class SortedListAllowingDoublets<TOrder, T> : Collection<KeyValuePair<TOrder, T>> where T : IEnumerator
{
    public void AddSorted( TOrder value, T enumerator )
    {
        Insert( GetSortedIndex( value, 0, Count - 1 ), new KeyValuePair<TOrder, T>( value, enumerator ) );
    }

    private int GetSortedIndex( TOrder item, int startIndex, int endIndex )
    {
        if( startIndex > endIndex )
        {
            return startIndex;
        }
        var midIndex = startIndex + ( endIndex - startIndex ) / 2;
        return Comparer<TOrder>.Default.Compare( this[midIndex].Key, item ) < 0 ? GetSortedIndex( item, midIndex + 1, endIndex ) : GetSortedIndex( item, startIndex, midIndex - 1 );
    }
}

Что сейчас не реализовано: проверьте наличие пустого списка, что приведет к проблемам.
И класс SortedListAllowingDoublets может быть улучшен, чтобы использовать компаратор вместо использования Comparer<TOrder>.Default самостоятельно.

Answer 7

Вот удобное решение Linq на основе Wintellect's OrderedBag :

public static IEnumerable<T> MergeOrderedLists<T, TOrder>(this IEnumerable<IEnumerable<T>> orderedLists, Func<T, TOrder> orderBy)
    where TOrder : IComparable<TOrder>
{
    var enumerators = new OrderedBag<IEnumerator<T>>(orderedLists
        .Select(enumerable => enumerable.GetEnumerator())
        .Where(enumerator => enumerator.MoveNext()),
        (x, y) => orderBy(x.Current).CompareTo(orderBy(y.Current)));
    while (enumerators.Count > 0)
    {
        IEnumerator<T> minEnumerator = enumerators.RemoveFirst();
        T minValue = minEnumerator.Current;
        if (minEnumerator.MoveNext())
            enumerators.Add(minEnumerator);
        else
            minEnumerator.Dispose();
        yield return minValue;
    }
}

Если вы используете какое-либо решение на основе Enumerator, не забудьте позвонить Dispose ()

А вот простой тест:

[Test]
public void ShouldMergeInOrderMultipleOrderedListWithDuplicateValues()
{
    // given
    IEnumerable<IEnumerable<int>> orderedLists = new[]
    {
        new [] {1, 5, 7},
        new [] {1, 2, 4, 6, 7}
    };

    // test
    IEnumerable<int> merged = orderedLists.MergeOrderedLists(i => i);

    // expect
    merged.ShouldAllBeEquivalentTo(new [] { 1, 1, 2, 4, 5, 6, 7, 7 });
}

Answer 8

Моя версия ответа Sixlettervariables. Я уменьшил количество обращений к orderFunc (каждый элемент проходит через orderFunc только один раз), а в случае связей сортировка пропускается. Это оптимизировано для небольшого количества источников, большего количества элементов в каждом источнике и, возможно, дорогого порядка Func.

public static IEnumerable<T> MergePreserveOrder<T, TOrder>(
  this IEnumerable<IEnumerable<T>> sources, 
  Func<T, TOrder> orderFunc)  
  where TOrder : IComparable<TOrder> 
{
  Dictionary<TOrder, List<IEnumerable<T>>> keyedSources =
    sources.Select(source => source.GetEnumerator())
      .Where(e => e.MoveNext())
      .GroupBy(e => orderFunc(e.Current))
      .ToDictionary(g => g.Key, g => g.ToList()); 

  while (keyedSources.Any())
  {
     //this is the expensive line
    KeyValuePair<TOrder, List<IEnumerable<T>>> firstPair = keyedSources
      .OrderBy(kvp => kvp.Key).First();

    keyedSources.Remove(firstPair.Key);
    foreach(IEnumerable<T> e in firstPair.Value)
    {
      yield return e.Current;
      if (e.MoveNext())
      {
        TOrder newKey = orderFunc(e.Current);
        if (!keyedSources.ContainsKey(newKey))
        {
          keyedSources[newKey] = new List<IEnumerable<T>>() {e};
        }
        else
        {
          keyedSources[newKey].Add(e);
        }
      }
    }
  }
}

Могу поспорить, что это может быть улучшено с помощью SortedDictionary, но я не достаточно смел, чтобы попробовать решение, использующее его без редактора.

Answer 9

Это альтернативное решение:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.Reflection;
using System.Data;
using System.Text.RegularExpressions;

namespace ConsoleApplication1
{

    class Person
    {
        public string Name
        {
            get;
            set;
        }

        public int Age
        {
            get;
            set;
        }
    }

    public class Program
    {
        public static void Main()
        {
            Person[] persons1 = new Person[] { new Person() { Name = "Ahmed", Age = 20 }, new Person() { Name = "Ali", Age = 40 } };
            Person[] persons2 = new Person[] { new Person() { Name = "Zaid", Age = 21 }, new Person() { Name = "Hussain", Age = 22 } };
            Person[] persons3 = new Person[] { new Person() { Name = "Linda", Age = 19 }, new Person() { Name = "Souad", Age = 60 } };

            Person[][] personArrays = new Person[][] { persons1, persons2, persons3 };

            foreach(Person person in MergeOrderedLists<Person, int>(personArrays, person => person.Age))
            {
                Console.WriteLine("{0} {1}", person.Name, person.Age);
            }

            Console.ReadLine();
        }

        static IEnumerable<T> MergeOrderedLists<T, TOrder>(IEnumerable<IEnumerable<T>> orderedlists, Func<T, TOrder> orderBy)
        {
            List<IEnumerator<T>> enumeratorsWithData = orderedlists.Select(enumerable => enumerable.GetEnumerator())
                                                                   .Where(enumerator => enumerator.MoveNext()).ToList();

            while (enumeratorsWithData.Count > 0)
            {
                IEnumerator<T> minEnumerator = enumeratorsWithData[0];
                for (int i = 1; i < enumeratorsWithData.Count; i++)
                    if (((IComparable<TOrder>)orderBy(minEnumerator.Current)).CompareTo(orderBy(enumeratorsWithData[i].Current)) >= 0)
                        minEnumerator = enumeratorsWithData[i];

                yield return minEnumerator.Current;

                if (!minEnumerator.MoveNext())
                    enumeratorsWithData.Remove(minEnumerator);
            }             
        }
    }   
}

Answer 10

Я выбрал более функциональный подход, надеюсь, это хорошо читается.

Прежде всего, это сам метод слияния:

public static IEnumerable<T> MergeSorted<T>(IEnumerable<IEnumerable<T>> xss) where T :IComparable
{
    var stacks = xss.Select(xs => new EnumerableStack<T>(xs)).ToList();

    while (true)
    {
        if (stacks.All(x => x.IsEmpty)) yield break;

        yield return 
            stacks
                .Where(x => !x.IsEmpty)
                .Select(x => new { peek = x.Peek(), x })
                .MinBy(x => x.peek)
                .x.Pop();
    }
}

Идея состоит в том, что мы превращаем каждый IEnumerable в EnumerableStack, который имеет Peek(), Pop() и IsEmpty членов.

Он работает так же, как обычный стек. Обратите внимание, что при вызове IsEmpty может быть перечислено значение IEnumerable.

Вот код:

/// <summary>
/// Wraps IEnumerable in Stack like wrapper
/// </summary>
public class EnumerableStack<T>
{
    private enum StackState
    {
        Pending,
        HasItem,
        Empty
    }

    private readonly IEnumerator<T> _enumerator;

    private StackState _state = StackState.Pending;

    public EnumerableStack(IEnumerable<T> xs)
    {
        _enumerator = xs.GetEnumerator();
    }

    public T Pop()
    {
        var res = Peek(isEmptyMessage: "Cannot Pop from empty EnumerableStack");
        _state = StackState.Pending;
        return res;
    }

    public T Peek()
    {
        return Peek(isEmptyMessage: "Cannot Peek from empty EnumerableStack");
    }

    public bool IsEmpty
    {
        get
        {
            if (_state == StackState.Empty) return true;
            if (_state == StackState.HasItem) return false;
            ReadNext();
            return _state == StackState.Empty;
        }
    }

    private T Peek(string isEmptyMessage)
    {
        if (_state != StackState.HasItem)
        {
            if (_state == StackState.Empty) throw new InvalidOperationException(isEmptyMessage);
            ReadNext();
            if (_state == StackState.Empty) throw new InvalidOperationException(isEmptyMessage);
        }
        return _enumerator.Current;
    }

    private void ReadNext()
    {
        _state = _enumerator.MoveNext() ? StackState.HasItem : StackState.Empty;
    }
}

Наконец, вот метод расширения MinBy на тот случай, если вы еще не написали его самостоятельно:

public static T MinBy<T, TS>(this IEnumerable<T> xs, Func<T, TS> selector) where TS : IComparable
{
    var en = xs.GetEnumerator();
    if (!en.MoveNext()) throw new Exception();

    T max = en.Current;
    TS maxVal = selector(max);
    while(en.MoveNext())
    {
        var x = en.Current;
        var val = selector(x);
        if (val.CompareTo(maxVal) < 0)
        {
            max = x;
            maxVal = val;
        }
    }

    return max;
}