This is an automated email from the ASF dual-hosted git repository. dsmiley pushed a commit to branch branch_9x in repository https://gitbox.apache.org/repos/asf/solr.git
commit 92558ad8c9a4fa4a914db72867ea6d4a0a85e04d Author: David Smiley <[email protected]> AuthorDate: Tue Sep 8 20:27:07 2026 -0400 SOLR-18418: Fix complement()/intersect() with asymmetric on= (#4873) Fix complement() and intersect() streaming expressions silently returning wrong results when on= maps two differently-named fields Co-authored-by: Claude Opus 5 (1M context) <[email protected]> Co-authored-by: Copilot Autofix powered by AI <[email protected]> (cherry picked from commit ba6f29cbf8bdfc6f0087c2f92adecc44ce350552) (cherry picked from commit 07c253a4d04a9c28f4b5128754d9d06947be69c9) --- ...8418-fix-complement-intersect-asymmetric-on.yml | 9 ++ .../pages/stream-decorator-reference.adoc | 16 ++- .../solr/client/solrj/io/eq/FieldEqualitor.java | 50 +++++++++ .../client/solrj/io/eq/MultipleFieldEqualitor.java | 51 +++++++++ .../solr/client/solrj/io/eq/StreamEqualitor.java | 82 +++++++++++++++ .../solr/client/solrj/io/stream/BiJoinStream.java | 61 +---------- .../client/solrj/io/stream/ComplementStream.java | 22 +++- .../client/solrj/io/stream/IntersectStream.java | 28 +++-- .../solrj/io/stream/StreamDecoratorTest.java | 116 +++++++++++++++++++++ 9 files changed, 359 insertions(+), 76 deletions(-) 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 00000000000..951b51a3954 --- /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 d9cadb16c49..a9a92a12b0f 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 95dcb5bebfd..590bb09b93b 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 @@ -141,4 +141,54 @@ public class FieldEqualitor implements StreamEqualitor { return false; } + + @Override + public boolean isDerivedFromLeft(StreamComparator base) { + if (null == base) { + return false; + } + if (base instanceof FieldComparator) { + FieldComparator baseComp = (FieldComparator) base; + return leftFieldName.equals(baseComp.getLeftFieldName()); + } else if (base instanceof MultipleFieldComparator) { + // must equal the first one + MultipleFieldComparator baseComps = (MultipleFieldComparator) base; + 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) { + FieldComparator baseComp = (FieldComparator) base; + return rightFieldName.equals(baseComp.getRightFieldName()); + } else if (base instanceof MultipleFieldComparator) { + // must equal the first one + MultipleFieldComparator baseComps = (MultipleFieldComparator) base; + 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 92765bae52b..031a7cb9be7 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 @@ -120,4 +120,55 @@ public class MultipleFieldEqualitor implements StreamEqualitor { return true; } + + @Override + public boolean isDerivedFromLeft(StreamComparator base) { + if (null == base) { + return false; + } + if (!(base instanceof MultipleFieldComparator)) { + return false; + } + MultipleFieldComparator baseComps = (MultipleFieldComparator) base; + + 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)) { + return false; + } + MultipleFieldComparator baseComps = (MultipleFieldComparator) base; + + 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 9eaeb4973e1..c145a9f62a6 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,83 @@ public interface StreamEqualitor extends Equalitor<Tuple>, 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. + */ + default boolean isDerivedFromLeft(StreamComparator base) { + return isDerivedFrom(base); + } + + /** Right-hand counterpart of {@link #isDerivedFromLeft(StreamComparator)}. */ + default boolean isDerivedFromRight(StreamComparator base) { + return isDerivedFrom(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 && comp instanceof MultipleFieldComparator) { + MultipleFieldEqualitor multiEq = (MultipleFieldEqualitor) eq; + MultipleFieldComparator multiComp = (MultipleFieldComparator) comp; + // 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) { + return deriveComparator(eq, ((MultipleFieldComparator) comp).getComps()[0]); + } else if (eq instanceof FieldEqualitor && comp instanceof FieldComparator) { + FieldEqualitor fieldEq = (FieldEqualitor) eq; + FieldComparator fieldComp = (FieldComparator) comp; + 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) { + MultipleFieldEqualitor multiEq = (MultipleFieldEqualitor) eq; + 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) { + return new FieldEqualitor(((FieldEqualitor) eq).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 ac8ec398233..e0debbaf31b 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 java.io.IOException; 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 @@ public abstract class BiJoinStream extends JoinStream implements Expressible { // 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,64 +79,6 @@ public abstract class BiJoinStream extends JoinStream implements Expressible { } } - 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 - && sourceComparator instanceof FieldComparator) { - FieldEqualitor fieldEqualitor = (FieldEqualitor) sourceEqualitor; - FieldComparator fieldComparator = (FieldComparator) sourceComparator; - 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 - && sourceComparator instanceof FieldComparator) { - FieldEqualitor fieldEqualitor = (FieldEqualitor) sourceEqualitor; - FieldComparator fieldComparator = (FieldComparator) sourceComparator; - 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 - && sourceComparator instanceof FieldComparator) { - FieldEqualitor fieldEqualitor = (FieldEqualitor) sourceEqualitor; - FieldComparator fieldComparator = (FieldComparator) sourceComparator; - 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 8698099d71b..5d503d1f997 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 class ComplementStream extends TupleStream implements Expressible { 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 class ComplementStream extends TupleStream implements Expressible { } // 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 e0fed5a49f9..f9e6680ef7f 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 class IntersectStream extends TupleStream implements Expressible { 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 class IntersectStream extends TupleStream implements Expressible { } // 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 c3b477eba5b..1447c3dc0e7 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.SolrClientCache; 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; @@ -4952,6 +4954,120 @@ public class StreamDecoratorTest extends SolrCloudTestCase { } } + @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() + .withCollectionZkHost(COLLECTIONORALIAS, cluster.getZkServer().getZkAddress()) + .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<Tuple> 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<Tuple> 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() + .withCollectionZkHost(COLLECTIONORALIAS, cluster.getZkServer().getZkAddress()) + .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<Tuple> dedupedTuples = getTuples(deduped); + assertEquals(2, dedupedTuples.size()); + assertOrder(dedupedTuples, 20, 22); + } finally { + solrClientCache.close(); + } + } + @Test public void testCartesianProductStream() throws Exception {
