SOLR-18418: Fix complement()/intersect() with asymmetric on=

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

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`

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);
+    }
+  }
 }

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);
+    }
+  }
 }

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<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.
+   */
+  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);
+    }
+  }
 }

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) {

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