Как выполнить сортировку + индексацию, одновременно удаляя дубликаты?

Question

Я использую следующий код для сортировки элементов и индекса: В массивах values и index имеется около 256 *1024* 1024 записей.
Несжатый массив values занимает около 10 ГБ, поэтому я хочу сжать его, сгруппировав дубликаты значений, которых много.

Я предложил следующее доказательство концептуального кода, но есть проблема в методе Compress. Каждый раз, когда он находит дубликат, он выполняет поиск в индексе, который стоит O (N ² ) времени.

Мне нужно сохранить индекс, потому что мне нужно иметь доступ к элементам в массиве, как если бы дедупликация не происходила.

Это делается с использованием простого свойства массива свойств default, таким образом имитируя исходный массив:

function TLookupTable.GetItems(Index: integer): TSlice;
begin   
  Result:= FData[FIndex[Index]];
end;

Доказательство кодекса понятия (которое является медленным) выглядит следующим образом.

TMyArray = class
  class procedure QuickSort<T,Idx>(var Values: array of T; var Index: array of Idx; const Comparer: IComparer<T>;
  L, R: Integer);
  class procedure Compress<T>(const Values: array of T; var Index: array of Integer; out CompressedValues: TArray<T>; const Comparer: IComparer<T>);
end;

class procedure TMyArray.QuickSort<T,Idx>(var Values: array of T; var Index: array of Idx; const Comparer: IComparer<T>;
  L, R: Integer);
var
  I, J: Integer;
  pivot, temp: T;
  TempIdx: Idx;
begin
  if (Length(Values) = 0) or ((R - L) <= 0) then
    Exit;
  repeat
    I := L;
    J := R;
    pivot := Values[L + (R - L) shr 1];
    repeat
      while Comparer.Compare(Values[I], pivot) < 0 do Inc(I);
      while Comparer.Compare(Values[J], pivot) > 0 do Dec(J);
      if I <= J then begin
        if I <> J then begin
          temp := Values[I];
          Values[I] := Values[J];
          Values[J] := temp;

          //Keep the index in sync
          tempIdx := Index[I];
          Index[I] := Index[J];
          Index[J] := tempIdx;
        end;
        Inc(I);
        Dec(J);
      end;
    until I > J;
    if L < J then QuickSort<T,Idx>(Values, Index, Comparer, L, J);
    L := I;
  until I >= R;
end;

class procedure TMyArray.Compress<T>(const Values: array of T; var Index: array of integer; out CompressedValues: TArray<T>; const Comparer: IComparer<T>);
var
  i,j: integer;
  Count: integer;
  Duplicate: integer;
begin
  Count:= 256*1024*1024;
  SetLength(CompressedValues, Count);
  CompressedValues[0]:= Values[0];
  Duplicate:= 0;
  for i := 1 to High(Values) do begin
    //Compress duplicate values
    if Comparer.Compare(Values[i], CompressedValues[Duplicate]) = 0 then begin
      //Search for the indexed item
      //Very time consuming: O(N*N)
      for j:= i to High(Index) do if Index[j] = i then begin
        Index[j]:= Duplicate;  //Fix up the index
        Break;
      end; {for j}
    end else begin
      Inc(Duplicate);
      if Duplicate >= Count  then begin
        Inc(Count, 1024*1024);
        SetLength(CompressedValues, Count);
      end;
      CompressedValues[Duplicate]:= Values[i];
    end; 
  end; {for i}
  SetLength(CompressedValues, Duplicate+1)
end;

Как я могу ускорить этап сжатия, чтобы он занимал время O (N)?

Если есть способ и сохранить индекс, и удалить дубликаты, и отсортировать их одновременно, это было бы здорово. Мой ответ ниже разделяет сортировку и дедупликацию на две отдельные стадии.

Answer 1

Хитрость заключается в том, чтобы оставить исходный массив данных в покое и просто отсортировать индекс. Затем мы можем использовать тот факт, что исходные данные находятся в правильном порядке, и использовать их для построения нового индекса.

Кроме того, это означает, что нам больше не нужна пользовательская функция Sort; это также перемещает намного меньше данных вокруг.

Создайте индекс примерно так:

  FIndex: TArray<integer>;
  ....
  SetLength(FIndex, Length(FAllData));
  for i:= 0 to count-1 do begin
    FIndex[i]:= i;
  end;
  TArray.Sort<Integer>(FIndex, TDelegatedComparer<integer>.Construct(
  function(const Left, Right: Integer): Integer
  begin
    if FAllData[Left] > FAllData[Right] then Exit(1);
    if FAllData[Left] < FAllData[Right] then Exit(-1);
    Result:= 0;
  end));

Измените класс TMyArray следующим образом:

TMyArray = class
  class procedure Compress<T>(const Values: array of T; var Index: TArray<integer>; out CompressedValues: TArray<T>; const Comparer: IComparer<T>);
end;

class procedure TMyArray.Compress<T>(const Values: array of T; var Index: TArray<integer>; out CompressedValues: TArray<T>; const Comparer: IComparer<T>);
const
  Equal = 0;
var
  i,j: integer;
  Count: integer;
  Duplicate: integer;
  IndexEntry: integer;
  OutIndex: TArray<integer>;
begin

  Count:= 16*1024*1024;  //Start with something reasonable
  SetLength(CompressedValues, Count);
  SetLength(OutIndex, Length(Index));
  Duplicate:= 0;
  CompressedValues[0]:= Values[Index[0]];
  OutIndex[Index[0]]:= 0;
  for i:= 1 to High(Index) do begin
    if Comparer.Compare(Values[Index[i]], CompressedValues[Duplicate]) = Equal then begin
      OutIndex[Index[i]]:= Duplicate;
    end else begin
      Inc(Duplicate);
      //Grow as needed
      if (Duplicate >= Length(CompressedValues)) then SetLength(CompressedValues, Length(CompressedValues) + 1024*1024);
      CompressedValues[Duplicate]:= Values[Index[i]];
      OutIndex[Index[i]]:= Duplicate;
    end;
  end;
  Index:= OutIndex;
  SetLength(CompressedValues, Duplicate+1);
end;