LogicalInput rhsInput = getInputs().get(RHS_INPUT_NAME);
Reader lhsReaderRaw = lhsInput.getReader();
Reader rhsReaderRaw = rhsInput.getReader();
Preconditions.checkState(lhsReaderRaw instanceof KeyValuesReader);
Preconditions.checkState(rhsReaderRaw instanceof KeyValuesReader);
KeyValuesReader lhsReader = (KeyValuesReader) lhsReaderRaw;
KeyValuesReader rhsReader = (KeyValuesReader) rhsReaderRaw;
TezCounter lhsMissingKeyCounter = getContext().getCounters().findCounter(COUNTER_GROUP_NAME,
MISSING_KEY_COUNTER_NAME);
while (lhsReader.next()) {
if (rhsReader.next()) {
if (!lhsReader.getCurrentKey().equals(rhsReader.getCurrentKey())) {
LOG.info("MismatchedKeys: " + "lhs=" + lhsReader.getCurrentKey() + ", rhs=" + rhsReader.getCurrentKey());
lhsMissingKeyCounter.increment(1);
}
} else {
lhsMissingKeyCounter.increment(1);
LOG.info("ExtraKey in lhs: " + lhsReader.getClass());
break;
}
}
if (rhsReader.next()) {
lhsMissingKeyCounter.increment(1);
LOG.info("ExtraKey in rhs: " + lhsReader.getClass());
}
}