This is an automated email from the ASF dual-hosted git repository.
dsmiley pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/solr.git
The following commit(s) were added to refs/heads/main by this push:
new ba6f29cbf8b SOLR-18418: Fix complement()/intersect() with asymmetric
on= (#4873)
ba6f29cbf8b is described below
commit ba6f29cbf8bdfc6f0087c2f92adecc44ce350552
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]>
---
...8418-fix-complement-intersect-asymmetric-on.yml | 9 ++
.../pages/stream-decorator-reference.adoc | 16 ++-
.../solr/client/solrj/io/eq/FieldEqualitor.java | 46 ++++++++
.../client/solrj/io/eq/MultipleFieldEqualitor.java | 49 +++++++++
.../solr/client/solrj/io/eq/StreamEqualitor.java | 78 ++++++++++++++
.../solr/client/solrj/io/stream/BiJoinStream.java | 55 +---------
.../client/solrj/io/stream/ComplementStream.java | 22 +++-
.../client/solrj/io/stream/IntersectStream.java | 28 +++--
.../solrj/io/stream/StreamDecoratorTest.java | 116 +++++++++++++++++++++
9 files changed, 349 insertions(+), 70 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 ca52627a712..120f871f5f6 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 f17b62afc9a..78ca78a9fdc 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 class FieldEqualitor implements StreamEqualitor {
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 1889ce2fd8a..7384402aafa 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 class MultipleFieldEqualitor implements
StreamEqualitor {
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 9eaeb4973e1..122f367ebf5 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,79 @@ 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 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 566ed39f720..86d17614b49 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,58 +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 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 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 4b1acf26959..6337f626189 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;
@@ -4895,6 +4897,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()
+ .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<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()
+ .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<Tuple> dedupedTuples = getTuples(deduped);
+ assertEquals(2, dedupedTuples.size());
+ assertOrder(dedupedTuples, 20, 22);
+ } finally {
+ solrClientCache.close();
+ }
+ }
+
@Test
public void testCartesianProductStream() throws Exception {
