### Question:
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(); static Dataset join(Dataset d1, Dataset d2, String joinColumn); static void main(String[] args);  }### Answer:
@Test public void testJoiner() { List<Instance> instances1 = new ArrayList<>(); instances1.add(new InstanceBuilder().set("join", "a").set("value1", "1").create(true)); instances1.add(new InstanceBuilder().set("join", "b").set("value1", "2").create(true)); instances1.add(new InstanceBuilder().set("join", "b").set("value1", "3").create(true)); instances1.add(new InstanceBuilder().set("join", "c").set("value1", "4").create(true)); List<Instance> instances2 = new ArrayList<>(); instances2.add(new InstanceBuilder().set("join", "a").set("value2", "q").create(true)); instances2.add(new InstanceBuilder().set("join", "a").set("value2", "r").create(true)); instances2.add(new InstanceBuilder().set("join", "b").set("value2", "s").create(true)); instances2.add(new InstanceBuilder().set("join", "b").set("value2", "t").create(true)); DefaultDataset dataset1 = new DefaultDataset(instances1); DefaultDataset dataset2 = new DefaultDataset(instances2); Dataset joined = DatasetJoiner.join(dataset1, dataset2, "join"); int numResults = CollectionHelper.count(joined.iterator()); assertEquals(6, numResults); assertEquals(3, joined.getFeatureInformation().count()); }