class DatasetJoiner {

	public static Dataset join(Dataset d1, Dataset d2, String joinColumn) {
		Validate.notNull(d1, "d1 must not be null");
		Validate.notNull(d2, "d2 must not be null");
		Validate.notEmpty(joinColumn, "joinColumn must not be null");

		ProgressReporter progress = new ProgressMonitor();
		progress.startTask("Joining datasets", -1);

		ProgressReporter p1 = progress.createSubProgress(.5);
		p1.startTask("Reading d1", d1.size());

		MultiMap<Value, FeatureVector> d1vectors = DefaultMultiMap.createWithList();
		for (Instance i1 : d1) {
			Value key = i1.getVector().get(joinColumn);
			d1vectors.add(key, i1.getVector());
			p1.increment();
		}

		ProgressReporter p2 = progress.createSubProgress(.5);
		p2.startTask("Reading d2", d2.size());
		
		return new AbstractDataset() {
			@Override
			public long size() {
				return -1;
			}
			@Override
			public CloseableIterator<Instance> iterator() {
				return new CloseableIterator<Instance>() {
					CloseableIterator<Instance> i2 = d2.iterator();
					Queue<FeatureVector> vectorsToJoin = new LinkedList<>();
					Instance instance2;

					@Override
					public boolean hasNext() {
						return vectorsToJoin.size() > 0 || i2.hasNext();
					}
					@Override
					public Instance next() {
						FeatureVector vector1 = vectorsToJoin.poll();
						if (vector1 != null) {
							return join(vector1, instance2);
						}
						for (;;) {
							instance2 = i2.next();
							vectorsToJoin.addAll(d1vectors.get(instance2.getVector().get(joinColumn)));
							vector1 = vectorsToJoin.poll();
							if (vector1 != null) {
								return join(vector1, instance2);
							}
						}
					}
					@Override
					public void close() throws IOException {
						i2.close();
					}
				};
			}
			@Override
			public FeatureInformation getFeatureInformation() {
				FeatureInformationBuilder builder = new FeatureInformationBuilder();
				builder.add(d1.getFeatureInformation());
				builder.add(d2.getFeatureInformation());
				return builder.create();
			}
		};
	}

	private  DatasetJoiner();

}