@@ -0,0 +1,1821 @@
+package com.jasongj.spark.utils;%0A%0Aimport com.google.common.base.Preconditions;%0Aimport com.jasongj.spark.model.Tuple;%0Aimport com.jasongj.spark.reader.BucketReaderIterator;%0A%0Aimport javax.validation.constraints.NotNull;%0Aimport java.util.Comparator;%0Aimport java.util.Iterator;%0A%0A/**%0A * Created by Jason Guo (jason.guo.vip@gmail.com)%0A */%0Apublic class SortMergeJoinIterator implements Iterator%3CTuple%3E%7B%0A%0A    @NotNull private static final Comparator%3CTuple%3E comparator = new TupleComparator();%0A    @NotNull private BucketReaderIterator baseIterator;%0A    @NotNull private BucketReaderIterator deltaIterator;%0A    private Tuple baseHeader;%0A    private Tuple deltaHeader;%0A%0A%0A    public SortMergeJoinIterator(BucketReaderIterator baseIterator, BucketReaderIterator deltaIterator) %7B%0A        Preconditions.checkNotNull(baseIterator);%0A        Preconditions.checkNotNull(deltaIterator);%0A        this.baseIterator = baseIterator;%0A        this.deltaIterator = deltaIterator;%0A    %7D%0A%0A    @Override%0A    public boolean hasNext() %7B%0A        if(baseIterator.hasNext() %7C%7C deltaIterator.hasNext()) %7B%0A            return true;%0A        %7D%0A        return false;%0A    %7D%0A%0A    @Override%0A    public Tuple next() %7B%0A        baseHeader = baseIterator.preFetch();%0A        deltaHeader = deltaIterator.preFetch();%0A        Tuple tuple = null;%0A        if(baseHeader == null) %7B%0A            tuple = deltaIterator.next();%0A        %7D else if(deltaHeader == null)%7B%0A            tuple = baseIterator.next();%0A        %7D else %7B%0A            int flag = comparator.compare(baseHeader, deltaHeader);%0A            if(flag %3E= 0) %7B%0A                tuple = deltaIterator.next();%0A                if(flag == 0) %7B%0A                    baseIterator.next();%0A                %7D%0A            %7D else %7B%0A                tuple = baseIterator.next();%0A            %7D%0A        %7D%0A        return tuple;%0A    %7D%0A%7D%0A

Implement SortMergeJoinIterator