@Override
protected IScalarEvaluator createEvaluator(IHyracksTaskContext ctx, IScalarEvaluator[] args)
throws AlgebricksException {
final ArrayBackedValueStorage abvs = new ArrayBackedValueStorage();
final SequenceBuilder sb = new SequenceBuilder();
final SequencePointable seq = new SequencePointable();
final DynamicContext dCtx = (DynamicContext) ctx.getJobletContext().getGlobalJobData();
final AbstractValueComparisonOperation aOp = new ValueEqComparisonOperation();
final TaggedValuePointable tvpDistinct = (TaggedValuePointable) TaggedValuePointable.FACTORY.createPointable();
final TaggedValuePointable tvpCheck = (TaggedValuePointable) TaggedValuePointable.FACTORY.createPointable();
final VoidPointable p = (VoidPointable) VoidPointable.FACTORY.createPointable();
final UTF8StringPointable stringp = (UTF8StringPointable) UTF8StringPointable.FACTORY.createPointable();
final TypedPointables tp1 = new TypedPointables();
final TypedPointables tp2 = new TypedPointables();
return new AbstractTaggedValueArgumentScalarEvaluator(args) {
@Override
protected void evaluate(TaggedValuePointable[] args, IPointable result) throws SystemException {
try {
abvs.reset();
sb.reset(abvs);
TaggedValuePointable tvp1 = args[0];
// Second parameter is optional.
if (args.length > 1) {
TaggedValuePointable tvp2 = args[1];
if (tvp2.getTag() != ValueTag.XS_STRING_TAG) {
throw new SystemException(ErrorCode.FORG0006);
}
tvp2.getValue(stringp);
}
// TODO use the third value as collation
if (tvp1.getTag() == ValueTag.SEQUENCE_TAG) {
tvp1.getValue(seq);
int seqLen = seq.getEntryCount();
// Add remaining distinct item.
boolean found = false;
for (int j = 0; j < seqLen; ++j) {
seq.getEntry(j, p);
tvpDistinct.set(p.getByteArray(), p.getStartOffset(), p.getLength());
// Check if item is distinct.
if (j + 1 < seqLen) {
for (int k = j + 1; k < seqLen; ++k) {
seq.getEntry(k, p);
tvpCheck.set(p.getByteArray(), p.getStartOffset(), p.getLength());
if (FunctionHelper.compareTaggedValues(aOp, tvpDistinct, tvpCheck, dCtx, tp1, tp2)) {
found = true;
break;
}