From 15152fb0ec22a86c0a01a4288301dfb9a30c87ae Mon Sep 17 00:00:00 2001 From: David Smiley Date: Thu, 3 Sep 2026 21:32:46 -0400 Subject: [PATCH] SOLR-18418: Fix complement()/intersect() returning wrong results with asymmetric on= complement() and intersect() streaming expressions silently returned wrong results whenever their on= clause mapped two differently-named fields (e.g. on="_parent_document_id=document_id"), with no exception raised. Root cause: both decorators compared tuples across streamA/streamB using streamA.getStreamSort() - a FieldComparator whose left and right field names are both streamA's own field. Applied to a streamB tuple, this always read a missing field as null, and FieldComparator's null-handling branch returns a constant, non-negative result. The on= mapping was honored by eq.test() but never by the comparison that drives the merge, so a non-matching pair was never recognized as "streamA's value is less", and instead of advancing streamB it discarded it, one tuple at a time. The first streamA value absent from streamB therefore drained streamB to EOF, after which every remaining streamA tuple hit the "streamB is EOF" branch: complement() emitted its entire input and intersect() emitted nothing. Fix: reuse the equalitor-derived comparator that innerJoin/leftOuterJoin/ fullOuterJoin already build correctly (BiJoinStream.createIterationComparator, now promoted to StreamEqualitor.deriveComparator so BiJoinStream, Complement- Stream, and IntersectStream share one implementation). Also: - Added StreamEqualitor.isDerivedFromLeft/isDerivedFromRight so Complement/ IntersectStream's precondition check validates each stream against the correct side of an asymmetric on=, instead of the old isDerivedFrom(), which OR's the two field checks together and can't detect an asymmetric mismatch. - ComplementStream/IntersectStream deduped streamB with the full (asymmetric) equalitor, which compared streamB tuples using streamA's field name and so never matched; fixed via StreamEqualitor.deriveRightEqualitor(eq). - Added StreamEqualitor.assertFieldsPresent(), called at the cross-stream comparison site, to fail loudly if an on= field is entirely absent from a tuple (a wiring bug) rather than silently treating it as null. - Added regression tests using an asymmetric on= where streamA's first value is absent from streamB (the condition that drains streamB), asserting exact membership plus the |complement| + |intersect| == |streamA| invariant, and a focused test for the streamB dedup fix. - Updated the complement/intersect ref guide sections with the sort/on= precondition and the sanity-check invariant. Co-Authored-By: Claude Opus 5 (1M context) --- ...fix-complement-intersect-asymmetric-on.yml | 9 ++ .../pages/stream-decorator-reference.adoc | 16 ++- .../client/solrj/io/eq/FieldEqualitor.java | 46 +++++++ .../solrj/io/eq/MultipleFieldEqualitor.java | 49 ++++++++ .../client/solrj/io/eq/StreamEqualitor.java | 74 +++++++++++ .../client/solrj/io/stream/BiJoinStream.java | 55 +-------- .../solrj/io/stream/ComplementStream.java | 22 +++- .../solrj/io/stream/IntersectStream.java | 28 +++-- .../solrj/io/stream/StreamDecoratorTest.java | 116 ++++++++++++++++++ 9 files changed, 345 insertions(+), 70 deletions(-) create mode 100644 changelog/unreleased/SOLR-18418-fix-complement-intersect-asymmetric-on.yml diff --git a/changelog/unreleased/SOLR-18418-fix-complement-intersect-asymmetric-on.yml b/changelog/unreleased/SOLR-18418-fix-complement-intersect-asymmetric-on.yml new file mode 100644 index 000000000000..951b51a39546 --- /dev/null +++ b/changelog/unreleased/SOLR-18418-fix-complement-intersect-asymmetric-on.yml @@ -0,0 +1,9 @@ +title: Fix complement() and intersect() streaming expressions silently returning wrong + results when on= maps two differently-named fields +type: fixed +authors: + - name: David Smiley + url: https://home.apache.org/phonebook.html?uid=dsmiley +links: + - name: SOLR-18418 + url: https://issues.apache.org/jira/browse/SOLR-18418 diff --git a/solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc b/solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc index ca52627a7123..120f871f5f60 100644 --- a/solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc +++ b/solr/solr-ref-guide/modules/query-guide/pages/stream-decorator-reference.adoc @@ -467,7 +467,13 @@ commit( The `complement` function wraps two streams (A and B) and emits tuples from A which do not exist in B. The tuples are emitted in the order in which they appear in stream A. -Both streams must be sorted by the fields being used to determine equality (using the `on` parameter). +Both streams must be sorted ascending by the field(s) named on their own side of the `on` parameter +(e.g. for `on="fieldNameInLeft=fieldNameInRight"`, A must be sorted by `fieldNameInLeft` and B by +`fieldNameInRight`) - this holds whether or not the two sides name the same field. + +As a sanity check, `complement` and `intersect` (below) partition A: for the same two input streams +and the same `on`, the number of tuples emitted by `complement` plus the number emitted by +`intersect` always equals the number of tuples in A. === complement Parameters @@ -1025,9 +1031,15 @@ innerJoin( The `intersect` function wraps two streams, A and B, and emits tuples from A which *DO* exist in B. The tuples are emitted in the order in which they appear in stream A. -Both streams must be sorted by the fields being used to determine equality (the `on` parameter). +Both streams must be sorted ascending by the field(s) named on their own side of the `on` parameter +(e.g. for `on="fieldNameInLeft=fieldNameInRight"`, A must be sorted by `fieldNameInLeft` and B by +`fieldNameInRight`) - this holds whether or not the two sides name the same field. Only tuples from A are emitted. +As a sanity check, `intersect` and `complement` (above) partition A: for the same two input streams +and the same `on`, the number of tuples emitted by `complement` plus the number emitted by +`intersect` always equals the number of tuples in A. + === intersect Parameters * `StreamExpression for StreamA` diff --git a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/FieldEqualitor.java b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/FieldEqualitor.java index f17b62afc9a9..78ca78a9fdcf 100644 --- a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/FieldEqualitor.java +++ b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/FieldEqualitor.java @@ -137,4 +137,50 @@ public boolean isDerivedFrom(StreamComparator base) { return false; } + + @Override + public boolean isDerivedFromLeft(StreamComparator base) { + if (null == base) { + return false; + } + if (base instanceof FieldComparator baseComp) { + return leftFieldName.equals(baseComp.getLeftFieldName()); + } else if (base instanceof MultipleFieldComparator baseComps) { + // must equal the first one + if (baseComps.getComps().length > 0) { + return isDerivedFromLeft(baseComps.getComps()[0]); + } + } + + return false; + } + + @Override + public boolean isDerivedFromRight(StreamComparator base) { + if (null == base) { + return false; + } + if (base instanceof FieldComparator baseComp) { + return rightFieldName.equals(baseComp.getRightFieldName()); + } else if (base instanceof MultipleFieldComparator baseComps) { + // must equal the first one + if (baseComps.getComps().length > 0) { + return isDerivedFromRight(baseComps.getComps()[0]); + } + } + + return false; + } + + @Override + public void assertFieldsPresent(Tuple left, Tuple right) throws IOException { + if (!left.getFields().containsKey(leftFieldName)) { + throw new IOException( + "Field '" + leftFieldName + "' is missing (not just null) from tuple: " + left); + } + if (!right.getFields().containsKey(rightFieldName)) { + throw new IOException( + "Field '" + rightFieldName + "' is missing (not just null) from tuple: " + right); + } + } } diff --git a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/MultipleFieldEqualitor.java b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/MultipleFieldEqualitor.java index 1889ce2fd8ae..7384402aafa3 100644 --- a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/MultipleFieldEqualitor.java +++ b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/MultipleFieldEqualitor.java @@ -119,4 +119,53 @@ public boolean test(Tuple t1, Tuple t2) { return true; } + + @Override + public boolean isDerivedFromLeft(StreamComparator base) { + if (null == base) { + return false; + } + if (!(base instanceof MultipleFieldComparator baseComps)) { + return false; + } + + if (baseComps.getComps().length >= eqs.length) { + for (int idx = 0; idx < eqs.length; ++idx) { + if (!eqs[idx].isDerivedFromLeft(baseComps.getComps()[idx])) { + return false; + } + } + return true; + } + + return false; + } + + @Override + public boolean isDerivedFromRight(StreamComparator base) { + if (null == base) { + return false; + } + if (!(base instanceof MultipleFieldComparator baseComps)) { + return false; + } + + if (baseComps.getComps().length >= eqs.length) { + for (int idx = 0; idx < eqs.length; ++idx) { + if (!eqs[idx].isDerivedFromRight(baseComps.getComps()[idx])) { + return false; + } + } + return true; + } + + return false; + } + + @Override + public void assertFieldsPresent(Tuple left, Tuple right) throws IOException { + for (StreamEqualitor eq : eqs) { + eq.assertFieldsPresent(left, right); + } + } } diff --git a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/StreamEqualitor.java b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/StreamEqualitor.java index 9eaeb4973e17..b53c8dab6002 100644 --- a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/StreamEqualitor.java +++ b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/eq/StreamEqualitor.java @@ -16,8 +16,11 @@ */ package org.apache.solr.client.solrj.io.eq; +import java.io.IOException; import java.io.Serializable; import org.apache.solr.client.solrj.io.Tuple; +import org.apache.solr.client.solrj.io.comp.FieldComparator; +import org.apache.solr.client.solrj.io.comp.MultipleFieldComparator; import org.apache.solr.client.solrj.io.comp.StreamComparator; import org.apache.solr.client.solrj.io.stream.expr.Expressible; @@ -26,4 +29,75 @@ public interface StreamEqualitor extends Equalitor, Expressible, Serializ public boolean isDerivedFrom(StreamEqualitor base); public boolean isDerivedFrom(StreamComparator base); + + /** + * Whether this equalitor's left-hand field(s) are exactly the field(s) that {@code base} - a + * single stream's own sort comparator, whose left/right field names are necessarily identical - + * sorts on. Used to validate the stream feeding the left side of a two-stream equality (e.g. + * streamA in complement/intersect), as opposed to {@link #isDerivedFrom(StreamComparator)} which + * matches either side and so cannot validate an asymmetric {@code on=} clause correctly. + */ + boolean isDerivedFromLeft(StreamComparator base); + + /** Right-hand counterpart of {@link #isDerivedFromLeft(StreamComparator)}. */ + boolean isDerivedFromRight(StreamComparator base); + + /** + * Verifies that this equalitor's field(s) are actually present in the given tuples, as opposed to + * merely holding a null value, throwing if not. A missing field compares as null just like a + * present-but-null field, which would otherwise silently mask a stream wiring bug (e.g. an {@code + * on=} clause naming a field that one side's search/select never returns). + */ + default void assertFieldsPresent(Tuple left, Tuple right) throws IOException { + // no-op by default; overridden by equalitors that know their own field names + } + + /** + * Builds a {@link StreamComparator} that orders tuples the way this equalitor pairs fields across + * two streams, taking the sort order(s) from {@code comp} (typically one of the two streams' own + * sort). Unlike {@code comp} itself - whose left and right field names are identical, since it's + * a single stream's own sort - the returned comparator carries this equalitor's (possibly + * different) left/right field names, so it can correctly order a tuple from one stream against a + * tuple from the other even when {@code on=} maps differently-named fields. + */ + static StreamComparator deriveComparator(StreamEqualitor eq, StreamComparator comp) + throws IOException { + if (eq instanceof MultipleFieldEqualitor multiEq + && comp instanceof MultipleFieldComparator multiComp) { + // comp is at least as long as eq because tuple order has already been validated + StreamComparator[] compoundComps = new StreamComparator[multiEq.getEqs().length]; + for (int idx = 0; idx < compoundComps.length; ++idx) { + compoundComps[idx] = deriveComparator(multiEq.getEqs()[idx], multiComp.getComps()[idx]); + } + return new MultipleFieldComparator(compoundComps); + } else if (comp instanceof MultipleFieldComparator multiComp) { + return deriveComparator(eq, multiComp.getComps()[0]); + } else if (eq instanceof FieldEqualitor fieldEq && comp instanceof FieldComparator fieldComp) { + return new FieldComparator( + fieldEq.getLeftFieldName(), fieldEq.getRightFieldName(), fieldComp.getOrder()); + } else { + throw new IOException( + "Failed to derive a comparator from equalitor " + eq + " and comparator " + comp); + } + } + + /** + * Derives an equalitor referencing only this equalitor's right-hand field(s), for comparing two + * tuples that both come from the "right" stream (e.g. de-duplicating streamB against itself in + * complement/intersect, where using the original, possibly asymmetric equalitor would compare the + * wrong field on one side). + */ + static StreamEqualitor deriveRightEqualitor(StreamEqualitor eq) throws IOException { + if (eq instanceof MultipleFieldEqualitor multiEq) { + StreamEqualitor[] rightEqs = new StreamEqualitor[multiEq.getEqs().length]; + for (int idx = 0; idx < rightEqs.length; ++idx) { + rightEqs[idx] = deriveRightEqualitor(multiEq.getEqs()[idx]); + } + return new MultipleFieldEqualitor(rightEqs); + } else if (eq instanceof FieldEqualitor fieldEq) { + return new FieldEqualitor(fieldEq.getRightFieldName()); + } else { + throw new IOException("Failed to derive a right-side equalitor from " + eq); + } + } } diff --git a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/BiJoinStream.java b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/BiJoinStream.java index 566ed39f7202..86d17614b49f 100644 --- a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/BiJoinStream.java +++ b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/BiJoinStream.java @@ -20,7 +20,6 @@ import org.apache.solr.client.solrj.io.comp.FieldComparator; import org.apache.solr.client.solrj.io.comp.MultipleFieldComparator; import org.apache.solr.client.solrj.io.comp.StreamComparator; -import org.apache.solr.client.solrj.io.eq.FieldEqualitor; import org.apache.solr.client.solrj.io.eq.MultipleFieldEqualitor; import org.apache.solr.client.solrj.io.eq.StreamEqualitor; import org.apache.solr.client.solrj.io.stream.expr.Expressible; @@ -67,7 +66,7 @@ private void init() throws IOException { // easily be done by grabbing the first N parts of each comp where N is the number of parts in // the equalitor. Because we've already validated tuple order (the comps) then we know this can // be done. - iterationComparator = createIterationComparator(eq, leftStream.getStreamSort()); + iterationComparator = StreamEqualitor.deriveComparator(eq, leftStream.getStreamSort()); leftStreamComparator = createSideComparator(eq, leftStream.getStreamSort()); rightStreamComparator = createSideComparator(eq, rightStream.getStreamSort()); } @@ -80,58 +79,6 @@ protected void validateTupleOrder() throws IOException { } } - private StreamComparator createIterationComparator(StreamEqualitor eq, StreamComparator comp) - throws IOException { - if (eq instanceof MultipleFieldEqualitor && comp instanceof MultipleFieldComparator) { - // we know the comp is at least as long as the eq because we've already validated the tuple - // order - StreamComparator[] compoundComps = - new StreamComparator[((MultipleFieldEqualitor) eq).getEqs().length]; - for (int idx = 0; idx < compoundComps.length; ++idx) { - StreamEqualitor sourceEqualitor = ((MultipleFieldEqualitor) eq).getEqs()[idx]; - StreamComparator sourceComparator = ((MultipleFieldComparator) comp).getComps()[idx]; - - if (sourceEqualitor instanceof FieldEqualitor fieldEqualitor - && sourceComparator instanceof FieldComparator fieldComparator) { - compoundComps[idx] = - new FieldComparator( - fieldEqualitor.getLeftFieldName(), - fieldEqualitor.getRightFieldName(), - fieldComparator.getOrder()); - } else { - throw new IOException("Failed to create an iteration comparator"); - } - } - return new MultipleFieldComparator(compoundComps); - } else if (comp instanceof MultipleFieldComparator) { - StreamEqualitor sourceEqualitor = eq; - StreamComparator sourceComparator = ((MultipleFieldComparator) comp).getComps()[0]; - - if (sourceEqualitor instanceof FieldEqualitor fieldEqualitor - && sourceComparator instanceof FieldComparator fieldComparator) { - return new FieldComparator( - fieldEqualitor.getLeftFieldName(), - fieldEqualitor.getRightFieldName(), - fieldComparator.getOrder()); - } else { - throw new IOException("Failed to create an iteration comparator"); - } - } else { - StreamEqualitor sourceEqualitor = eq; - StreamComparator sourceComparator = comp; - - if (sourceEqualitor instanceof FieldEqualitor fieldEqualitor - && sourceComparator instanceof FieldComparator fieldComparator) { - return new FieldComparator( - fieldEqualitor.getLeftFieldName(), - fieldEqualitor.getRightFieldName(), - fieldComparator.getOrder()); - } else { - throw new IOException("Failed to create an iteration comparator"); - } - } - } - private StreamComparator createSideComparator(StreamEqualitor eq, StreamComparator comp) throws IOException { if (eq instanceof MultipleFieldEqualitor && comp instanceof MultipleFieldComparator) { diff --git a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/ComplementStream.java b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/ComplementStream.java index 8698099d71b5..5d503d1f9976 100644 --- a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/ComplementStream.java +++ b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/ComplementStream.java @@ -47,6 +47,7 @@ public class ComplementStream extends TupleStream implements Expressible { private PushBackStream streamB; private TupleStream originalStreamB; private StreamEqualitor eq; + private StreamComparator crossStreamComparator; public ComplementStream(TupleStream streamA, TupleStream streamB, StreamEqualitor eq) throws IOException { @@ -94,15 +95,23 @@ public ComplementStream(StreamExpression expression, StreamFactory factory) thro private void init(TupleStream streamA, TupleStream streamB, StreamEqualitor eq) throws IOException { this.streamA = new PushBackStream(streamA); - this.streamB = new PushBackStream(new UniqueStream(streamB, eq)); + // dedup streamB using only its own (right-side) field(s); using the full, possibly + // asymmetric eq here would compare streamB's field against a field it doesn't have. + this.streamB = + new PushBackStream(new UniqueStream(streamB, StreamEqualitor.deriveRightEqualitor(eq))); this.originalStreamB = streamB; // hold onto this for toExpression this.eq = eq; // streamA and streamB must both be sorted so that comp can be derived from - if (!eq.isDerivedFrom(streamA.getStreamSort()) || !eq.isDerivedFrom(streamB.getStreamSort())) { + if (!eq.isDerivedFromLeft(this.streamA.getStreamSort()) + || !eq.isDerivedFromRight(this.streamB.getStreamSort())) { throw new IOException( "Invalid ComplementStream - both substream comparators (sort) must be a superset of this stream's equalitor."); } + + // comparator used to order a streamA tuple against a streamB tuple; unlike either stream's own + // sort, it carries eq's (possibly different) left/right field names. + crossStreamComparator = StreamEqualitor.deriveComparator(eq, this.streamA.getStreamSort()); } @Override @@ -201,9 +210,12 @@ public Tuple read() throws IOException { } // if a != b && a < b then we know there is no b which a might equal so return a - if (!eq.test(a, b) && streamA.getStreamSort().compare(a, b) < 0) { - streamB.pushBack(b); - return a; + if (!eq.test(a, b)) { + eq.assertFieldsPresent(a, b); + if (crossStreamComparator.compare(a, b) < 0) { + streamB.pushBack(b); + return a; + } } // if a == b then ignore a cause it exists in b diff --git a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/IntersectStream.java b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/IntersectStream.java index e0fed5a49f93..f9e6680ef7fb 100644 --- a/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/IntersectStream.java +++ b/solr/solrj-streaming/src/java/org/apache/solr/client/solrj/io/stream/IntersectStream.java @@ -47,6 +47,7 @@ public class IntersectStream extends TupleStream implements Expressible { private PushBackStream streamB; private TupleStream originalStreamB; private StreamEqualitor eq; + private StreamComparator crossStreamComparator; public IntersectStream(TupleStream streamA, TupleStream streamB, StreamEqualitor eq) throws IOException { @@ -94,15 +95,23 @@ public IntersectStream(StreamExpression expression, StreamFactory factory) throw private void init(TupleStream streamA, TupleStream streamB, StreamEqualitor eq) throws IOException { this.streamA = new PushBackStream(streamA); - this.streamB = new PushBackStream(new UniqueStream(streamB, eq)); + // dedup streamB using only its own (right-side) field(s); using the full, possibly + // asymmetric eq here would compare streamB's field against a field it doesn't have. + this.streamB = + new PushBackStream(new UniqueStream(streamB, StreamEqualitor.deriveRightEqualitor(eq))); this.originalStreamB = streamB; // hold onto this for toExpression this.eq = eq; // streamA and streamB must both be sorted so that comp can be derived from - if (!eq.isDerivedFrom(streamA.getStreamSort()) || !eq.isDerivedFrom(streamB.getStreamSort())) { + if (!eq.isDerivedFromLeft(this.streamA.getStreamSort()) + || !eq.isDerivedFromRight(this.streamB.getStreamSort())) { throw new IOException( "Invalid IntersectStream - both substream comparators (sort) must be a superset of this stream's equalitor."); } + + // comparator used to order a streamA tuple against a streamB tuple; unlike either stream's own + // sort, it carries eq's (possibly different) left/right field names. + crossStreamComparator = StreamEqualitor.deriveComparator(eq, this.streamA.getStreamSort()); } @Override @@ -203,13 +212,14 @@ public Tuple read() throws IOException { } // We're not at the end, and they're not equal. We now need to decide which we can - // throw away. This is accomplished by checking which is less than the other. The - // one that is less (determined by the sort) can be tossed. The other should - // be pushed back and the loop continued. We don't have to worry about an == 0 - // result because we already know tuples a and b are not equal. And because eq - // is derived from the sorts of both streamA and streamB we can rest assured that - // equality is not a possibility. - int aComp = streamA.getStreamSort().compare(a, b); + // throw away. This is accomplished by checking which is less than the other, using + // crossStreamComparator - a comparator built from eq's (possibly different) left/right + // field names, since streamA's own sort comparator only knows streamA's field and would + // read null off of b. The one that is less can be tossed. The other should be pushed back + // and the loop continued. We don't have to worry about an == 0 result because we already + // know tuples a and b are not equal. + eq.assertFieldsPresent(a, b); + int aComp = crossStreamComparator.compare(a, b); if (aComp < 0) { streamB.pushBack(b); } else { diff --git a/solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java b/solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java index 4b1acf269593..6337f6261894 100644 --- a/solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java +++ b/solr/solrj-streaming/src/test/org/apache/solr/client/solrj/io/stream/StreamDecoratorTest.java @@ -33,6 +33,8 @@ import org.apache.solr.client.solrj.io.Tuple; import org.apache.solr.client.solrj.io.comp.ComparatorOrder; import org.apache.solr.client.solrj.io.comp.FieldComparator; +import org.apache.solr.client.solrj.io.eq.FieldEqualitor; +import org.apache.solr.client.solrj.io.eq.StreamEqualitor; import org.apache.solr.client.solrj.io.eval.AddEvaluator; import org.apache.solr.client.solrj.io.eval.AndEvaluator; import org.apache.solr.client.solrj.io.eval.EqualToEvaluator; @@ -4895,6 +4897,120 @@ public void testComplementStream() throws Exception { } } + @Test + public void testIntersectComplementAsymmetricOn() throws Exception { + // Regression test: complement()/intersect() must order tuples across streamA/streamB using + // a comparator derived from the (possibly asymmetric) on= equalitor, not streamA's own sort + // comparator (whose left/right field names are both streamA's field, so comparing it against + // a streamB tuple always reads a missing field as null and returns a constant, non-negative + // result). That bug caused streamB to be fully drained the first time a streamA value that + // isn't present in streamB was compared, silently making complement() return all of streamA + // and intersect() return nothing. + // + // streamA's first value (x_i=1) is deliberately absent from streamB's y_i values, which is + // exactly what drained streamB under the bug. fl restricts each side's tuples so that the + // other side's field is genuinely absent (not merely null). + new UpdateRequest() + .add(id, "10", "a_s", "setA", "x_i", "1") // no match in streamB + .add(id, "11", "a_s", "setA", "x_i", "2") // matches streamB y_i=2 + .add(id, "12", "a_s", "setA", "x_i", "4") // matches streamB y_i=4 + .add(id, "13", "a_s", "setB", "y_i", "2") + .add(id, "14", "a_s", "setB", "y_i", "3") + .add(id, "15", "a_s", "setB", "y_i", "4") + .commit(cluster.getSolrClient(), COLLECTIONORALIAS); + + StreamContext streamContext = new StreamContext(); + SolrClientCache solrClientCache = new SolrClientCache(); + streamContext.setSolrClientCache(solrClientCache); + + StreamFactory factory = + new StreamFactory() + .withCollectionUseThisConnection("collection1", getSolrConnection()) + .withFunctionName("search", CloudSolrStream.class) + .withFunctionName("intersect", IntersectStream.class) + .withFunctionName("complement", ComplementStream.class); + + try { + StreamExpression intersectExpr = + StreamExpressionParser.parse( + "intersect(" + + "search(collection1, q=a_s:setA, fl=\"id,x_i\", sort=\"x_i asc\")," + + "search(collection1, q=a_s:setB, fl=\"id,y_i\", sort=\"y_i asc\")," + + "on=\"x_i=y_i\")"); + TupleStream intersectStream = new IntersectStream(intersectExpr, factory); + intersectStream.setStreamContext(streamContext); + List intersectTuples = getTuples(intersectStream); + + assertEquals(2, intersectTuples.size()); + assertOrder(intersectTuples, 11, 12); + + StreamExpression complementExpr = + StreamExpressionParser.parse( + "complement(" + + "search(collection1, q=a_s:setA, fl=\"id,x_i\", sort=\"x_i asc\")," + + "search(collection1, q=a_s:setB, fl=\"id,y_i\", sort=\"y_i asc\")," + + "on=\"x_i=y_i\")"); + TupleStream complementStream = new ComplementStream(complementExpr, factory); + complementStream.setStreamContext(streamContext); + List complementTuples = getTuples(complementStream); + + assertEquals(1, complementTuples.size()); + assertOrder(complementTuples, 10); + + // sanity check invariant: complement and intersect partition streamA + assertEquals(3, complementTuples.size() + intersectTuples.size()); + } finally { + solrClientCache.close(); + } + } + + @Test + public void testUniqueStreamRightSideEqualitorDedup() throws Exception { + // Regression test: complement()/intersect() dedup streamB using an equalitor derived from + // only the right-hand side of the (possibly asymmetric) on= equalitor. Using the full, + // asymmetric equalitor directly (the pre-fix behavior) compares tuple.get(leftFieldName) - a + // field streamB doesn't have - against tuple.get(rightFieldName), so two equal streamB tuples + // never test as equal and dedup silently never fires. + new UpdateRequest() + .add(id, "20", "a_s", "setB", "y_i", "7") + .add(id, "21", "a_s", "setB", "y_i", "7") + .add(id, "22", "a_s", "setB", "y_i", "9") + .commit(cluster.getSolrClient(), COLLECTIONORALIAS); + + StreamContext streamContext = new StreamContext(); + SolrClientCache solrClientCache = new SolrClientCache(); + streamContext.setSolrClientCache(solrClientCache); + + StreamFactory factory = + new StreamFactory() + .withCollectionUseThisConnection("collection1", getSolrConnection()) + .withFunctionName("search", CloudSolrStream.class); + + try { + StreamEqualitor asymmetricEq = new FieldEqualitor("x_i", "y_i"); + // sort includes "id asc" as a tiebreaker so which of the two y_i=7 tuples survives dedup is + // deterministic + String searchExpr = + "search(collection1, q=a_s:setB, fl=\"id,y_i\", sort=\"y_i asc, id asc\")"; + + TupleStream undeduped = new UniqueStream(factory.constructStream(searchExpr), asymmetricEq); + undeduped.setStreamContext(streamContext); + // dedup never fires against the asymmetric equalitor: the duplicate y_i=7 isn't collapsed + assertEquals(3, getTuples(undeduped).size()); + + TupleStream deduped = + new UniqueStream( + factory.constructStream(searchExpr), + StreamEqualitor.deriveRightEqualitor(asymmetricEq)); + deduped.setStreamContext(streamContext); + List dedupedTuples = getTuples(deduped); + assertEquals(2, dedupedTuples.size()); + assertOrder(dedupedTuples, 20, 22); + } finally { + solrClientCache.close(); + } + } + @Test public void testCartesianProductStream() throws Exception {