Merge experiments

Keys.java

package test;

import java.util.ArrayList;
import java.util.Random;

public class Keys implements Comparable<Keys> {
    private final int[] keys;

    public Keys(final int[] keys) {
        this.keys = keys;
    }

    public static Keys generateRandom(final int K) {
        final Random random = new Random();
        final int[] buf = new int[K];
        for (int ix = 0; ix < K; ++ix) buf[ix] = Math.abs(random.nextInt());
        return new Keys(buf);
    }

    /**
     * Implements a null-aware ordering. Null keys are treated as being bigger than any non-null key.
     */
    public static int compare(final Keys a, final Keys b) {
        if (a == null) {
            if (b == null) return 0;
            return 1;
        }
        if (b == null) return -1;
        return a.compareTo(b);
    }

    @Override
    public int compareTo(final Keys o) {
        final int len = Math.min(keys.length, o.keys.length);
        for (int ix = 0; ix < len; ++ix) {
            int diff = Integer.compare(keys[ix], o.keys[ix]);
            if (diff != 0) return diff;
        }
        return Integer.compare(keys.length, o.keys.length);
    }

    @Override
    public String toString() {
        final StringBuilder sb = new StringBuilder();
        for (int ix = 0; ix < keys.length; ++ix) {
            if (ix > 2) {
                sb.append(", ...");
                break;
            }
            if (ix != 0) sb.append(", ");
            sb.append(keys[ix]);
        }
        return sb.toString();
    }
}

Main.java

package test;

import java.io.FileOutputStream;
import java.io.OutputStreamWriter;
import java.util.ArrayList;
import java.util.TreeMap;

class BenchmarkResults {
    public int count;
    public long ms;

    public BenchmarkResults(int count, long ms) {
        this.count = count;
        this.ms = ms;
    }

    @Override
    public String toString() {
        return "produced " + count + " entries in " + ms + " ms";
    }
}

public class Main {
    public static BenchmarkResults benchmark(final int M, final int N, final int K) {
        final Producer[] iterators = new Producer[M];
        for (int ix = 0; ix < M; ++ix) iterators[ix] = Producer.generateRandom(N, K);
        long start = System.currentTimeMillis();
        final Merge merge = new Merge(iterators);
        Keys last = null;
        int count = 0;
        while (merge.hasNext()) {
            final Keys keys = merge.next();
            if (last != null)
                assert last.compareTo(keys) > 0;
            last = keys;
            ++count;
        }
        return new BenchmarkResults(count, System.currentTimeMillis() - start);
    }

    public static void main(final String[] args) {
        final int K = 10;
        benchmark(3, 1000, K);  // Warm-up.

        final TreeMap<Integer, TreeMap<Integer, Long>> table = new TreeMap();

        for (int M = 5; M <= 15; M += 5)
            for (int N = 3000; N <= 50000; N += 3000) {
                int runs = 20;
                long ms = 0;
                for (int num = 0; num < runs; ++num) {
                    final BenchmarkResults bench = benchmark(M, N, K);
                    System.out.println("Benchmarking M " + M + ", N " + N + ", K " + K + "; run " + (num + 1) + ": " + bench + ".");
                    ms += bench.ms;
                }
                if (!table.containsKey(N)) table.put(N, new TreeMap());
                table.get(N).put(M, ms / runs);
            }

        // Merge: https://live.amcharts.com/VlZTc/
        // Merge2: https://live.amcharts.com/ZmQzY/
        try {
            final OutputStreamWriter csv = new OutputStreamWriter(new FileOutputStream("graph.csv"));
            csv.write("N");
            for (Integer M : table.firstEntry().getValue().keySet()) csv.write(",M" + M);
            csv.write("\n");
            for (Integer N : table.keySet()) {
                csv.write(N.toString());
                for (Integer M : table.get(N).keySet())
                    csv.write("," + table.get(N).get(M));
                csv.write("\n");
            }
            csv.close();
        } catch (final Exception ex) {
            ex.printStackTrace();
        }
    }
}

Merge.java

package test;

import java.util.Iterator;

/**
 * Given a list of ordered iterators produce an ordered merged stream, leaving out the duplicates.
 */
public class Merge implements Iterator<Keys> {
    private final Producer[] producers;
    private final Keys[] keys;
    Keys minimal;

    public Merge(final Producer[] producers) {
        this.producers = producers;
        this.keys = new Keys[producers.length];
        for (int ix = 0; ix < producers.length; ++ix) if (producers[ix].hasNext()) keys[ix] = producers[ix].next();
        for (int ix = 0; ix < keys.length; ++ix) if (keys[ix] != null && (minimal == null || keys[ix].compareTo(minimal) < 0)) minimal = keys[ix];
    }

    @Override
    public boolean hasNext() {
        return minimal != null;
    }

    @Override
    public Keys next() {
        final Keys minimal = this.minimal;
        Keys nextMinimal = null;
        skipProducer: for (int ix = 0; ix < keys.length; ++ix) {
            if (keys[ix] == null) continue skipProducer;  // This stream already ended.
            while (keys[ix].compareTo(minimal) <= 0) {
                keys[ix] = producers[ix].hasNext() ? producers[ix].next() : null;
                if (keys[ix] == null) continue skipProducer;  // Reached the end of this stream.
            }
            if (nextMinimal == null || keys[ix].compareTo(nextMinimal) < 0) nextMinimal = keys[ix];
        }
        this.minimal = nextMinimal;
        return minimal;
    }
}

Merge2.java

package test;

import java.util.Arrays;
import java.util.Iterator;

class ProducerPosition implements Comparable<ProducerPosition> {
    Keys key;
    Producer producer;

    public ProducerPosition(final Producer producer) {
        this.key = producer.hasNext() ? producer.next() : null;
        this.producer = producer;
    }

    @Override
    public int compareTo(final ProducerPosition o) {
        return Keys.compare(key, o.key);
    }

    public Keys advance() {
        return key = producer.hasNext() ? producer.next() : null;
    }
}

/**
 * Given a list of ordered iterators produce an ordered merged stream.
 */
public class Merge2 implements Iterator<Keys> {
    private final ProducerPosition[] positions;

    public Merge2(final Producer[] producers) {
        positions = new ProducerPosition[producers.length];
        for (int ix = 0; ix < producers.length; ++ix) positions[ix] = new ProducerPosition(producers[ix]);
        Arrays.sort(positions);
    }

    @Override
    public boolean hasNext() {
        return positions[0].key != null;
    }

    @Override
    public Keys next() {
        final Keys minimal = positions[0].key;
        positions[0].advance();

        // Reposition the new item, maintaining the `positions` ordering.
        for (int ix = 0; ix < positions.length - 1; ++ix) {
            if (Keys.compare(positions[ix].key, positions[ix+1].key) > 0) {
                ProducerPosition tmp = positions[ix];
                positions[ix] = positions[ix+1];
                positions[ix+1] = tmp;
            } else break;  // O(log n)
        }

        return minimal;
    }
}

Producer.java

package test;

import java.util.Arrays;
import java.util.Iterator;
import java.util.Random;

public class Producer implements Iterator<Keys> {
    private int idx = 0;
    final private Keys[] items;

    public Producer(Keys[] items) {
        this.items = items;
    }

    public static Producer generateRandom(int N, int K) {
        final Random random = new Random();
        final Keys[] items = new Keys[N];
        for (int ix = 0; ix < N; ++ix) items[ix] = Keys.generateRandom(K);
        Arrays.sort(items);
        return new Producer(items);
    }

    @Override
    public boolean hasNext() {
        return idx < items.length;
    }

    @Override
    public Keys next() {
        return items[idx++];
    }
}