Deprecate BloomFilter.invert and add difference (A NOT B)

src/main/java/org/apache/datasketches/filters/bloomfilter/BitArray.java

@@ -88,6 +88,9 @@ boolean isEmpty() {
 
   abstract void invert();
 
+  // applies logical AND-NOT (this &= ~other), matching BitSet.andNot
+  abstract void andNot(final BitArray other);
+
   // prints the raw BitArray as 0s and 1s, one long per row
   @Override
   public String toString() {

src/main/java/org/apache/datasketches/filters/bloomfilter/BloomFilter.java

@@ -703,13 +703,50 @@ public void intersect(final BloomFilter other) {
 
   /**
    * Inverts all the bits of the BloomFilter. Approximately inverts the notion of set-membership.
+   *
+   * @deprecated Bit inversion has no sound set-membership interpretation. An inverted filter is a
+   * strictly worse absence oracle than the original, and updates after inversion have no checkable
+   * meaning. Use {@link #difference(BloomFilter)} for the approximate set-difference (A NOT B) use
+   * case {@code invert} was meant to enable. See
+   * <a href="https://github.com/apache/datasketches-java/issues/766">#766</a>.
    */
+  @Deprecated
   public void invert() {
     bitArray_.invert();
   }
 
   /**
-   * Helps identify if two BloomFilters may be unioned or intersected.
+   * Computes the approximate set difference with another filter via bitwise AND-NOT
+   * ({@code this &= ~other}). After this operation, the filter approximates the set of items
+   * inserted into this filter but not into {@code other}:
+   * <ul>
+   *   <li>Items inserted into {@code other} always query {@code false}: they are excluded exactly.</li>
+   *   <li>Items inserted only into this filter keep querying {@code true} as long as none of their
+   *       hash positions is occupied in {@code other}. Unlike {@link #union(BloomFilter)} and
+   *       {@link #intersect(BloomFilter)}, this operation can drop items, with a probability that
+   *       grows with {@code other}'s load factor.</li>
+   *   <li>Items never inserted into this filter may still query {@code true} (false positives), at a
+   *       rate no higher than this filter's false positive rate before the operation.</li>
+   * </ul>
+   * This is the Bloom-filter form of the A NOT B operation exposed elsewhere in DataSketches
+   * (for example Theta {@code AnotB}). Compatible with the Rust {@code BloomFilter::difference}
+   * API.
+   *
+   * @param other A BloomFilter to subtract from this one. A {@code null} argument is a no-op.
+   * @throws SketchesArgumentException if the filters are not compatible (different seeds, hash
+   *         counts, or sizes)
+   */
+  public void difference(final BloomFilter other) {
+    if (other == null) { return; }
+    if (!isCompatible(other)) {
+      throw new SketchesArgumentException("Cannot difference sketches with different seeds, hash functions, or sizes");
+    }
+
+    bitArray_.andNot(other.bitArray_);
+  }
+
+  /**
+   * Helps identify if two BloomFilters may be unioned, intersected, or differenced.
    * @param other A BloomFilter to check for compatibility with this one
    * @return True if the filters are compatible, otherwise false
    */

src/main/java/org/apache/datasketches/filters/bloomfilter/DirectBitArray.java

@@ -198,6 +198,21 @@ void invert() {
     wseg_.set(JAVA_LONG_UNALIGNED, NUM_BITS_OFFSET, numBitsSet_);
   }
 
+  @Override
+  void andNot(final BitArray other) {
+    if (getCapacity() != other.getCapacity()) {
+      throw new SketchesArgumentException("Cannot andNot bit arrays with unequal lengths");
+    }
+
+    numBitsSet_ = 0;
+    for (int i = 0; i < dataLength_; ++i) {
+      final long val = getLong(i) & ~other.getLong(i);
+      numBitsSet_ += Long.bitCount(val);
+      setLong(i, val);
+    }
+    wseg_.set(JAVA_LONG_UNALIGNED, NUM_BITS_OFFSET, numBitsSet_);
+  }
+
   @Override
   protected void setLong(final int arrayIndex, final long value) {
     wseg_.set(JAVA_LONG_UNALIGNED, DATA_OFFSET + (arrayIndex << 3), value);

src/main/java/org/apache/datasketches/filters/bloomfilter/DirectBitArrayR.java

@@ -155,6 +155,11 @@ void invert() {
     throw new SketchesReadOnlyException("Attempt to call invert() on read-only MemorySegment");
   }
 
+  @Override
+  void andNot(final BitArray other) {
+    throw new SketchesReadOnlyException("Attempt to call andNot() on read-only MemorySegment");
+  }
+
   @Override
   protected void setLong(final int arrayIndex, final long value) {
     throw new SketchesReadOnlyException("Attempt to call setLong() on read-only MemorySegment");

src/main/java/org/apache/datasketches/filters/bloomfilter/HeapBitArray.java

@@ -202,6 +202,22 @@ void invert() {
     }
   }
 
+  // applies logical AND-NOT (this &= ~other)
+  @Override
+  void andNot(final BitArray other) {
+    if (getCapacity() != other.getCapacity()) {
+      throw new SketchesArgumentException("Cannot andNot bit arrays with unequal lengths");
+    }
+
+    numBitsSet_ = 0;
+    for (int i = 0; i < data_.length; ++i) {
+      final long val = data_[i] & ~other.getLong(i);
+      numBitsSet_ += Long.bitCount(val);
+      data_[i] = val;
+    }
+    isDirty_ = false;
+  }
+
   void writeToSegmentAsStream(final PositionalSegment posSeg) { //position = 16
     posSeg.setInt(data_.length);
     posSeg.setInt(0); // unused

src/test/java/org/apache/datasketches/filters/bloomfilter/BloomFilterTest.java

@@ -180,7 +180,9 @@ public void basicFilterOperationsTest() {
   }
 
   @Test
+  @SuppressWarnings("deprecation")
   public void inversionTest() {
+    // Deprecated path retained until invert() is removed; prefer difference().
     final long numBits = 8192;
     final int numHashes = 3;
 
@@ -195,12 +197,12 @@ public void inversionTest() {
     bf.invert();
     assertEquals(bf.getBitsUsed(), numBits - numBitsSet);
 
-    // original items should be mostly not-present
+    // inserted items are always absent after inversion (all their positions flipped to 0)
     int count = 0;
     for (int i = 0; i < n; ++i) {
       count += bf.query(Integer.toString(i)) ? 1 : 0;
     }
-    assertTrue(count < (numBits / 10));
+    assertEquals(count, 0);
 
     // many other items should be present
     count = 0;
@@ -227,6 +229,7 @@ public void incompatibleSetOperationsTest() {
     // mismatched seed
     final BloomFilter bf4 = BloomFilterBuilder.createBySize(numBits, numHashes, bf1.getSeed() - 1);
     assertThrows(SketchesArgumentException.class, () -> bf1.union(bf4));
+    assertThrows(SketchesArgumentException.class, () -> bf1.difference(bf4));
   }
 
   @Test
@@ -290,6 +293,68 @@ public void basicIntersectionTest() {
     assertTrue(count < (numBits / 10)); // not being super strict
   }
 
+  @Test
+  public void basicDifferenceTest() {
+    final long numBits = 8192;
+    final int numHashes = 5;
+
+    final BloomFilter left = BloomFilterBuilder.createBySize(numBits, numHashes);
+    final BloomFilter right = BloomFilterBuilder.createBySize(numBits, numHashes, left.getSeed());
+
+    final int n = 1024;
+    for (int i = 0; i < n; ++i) {
+      left.queryAndUpdate(i);
+      right.queryAndUpdate((n / 2) + i); // overlap [n/2, n)
+    }
+
+    final long bitsBefore = left.getBitsUsed();
+    left.difference(null); // no-op
+    left.difference(right);
+
+    // items only in the right filter / overlap are excluded exactly
+    for (int i = n / 2; i < (n + n / 2); ++i) {
+      assertFalse(left.query(i), "item " + i + " should be excluded by difference");
+    }
+    assertTrue(left.getBitsUsed() <= bitsBefore);
+
+    // disjoint left-only items should mostly remain; allow for hash collisions with right
+    int retained = 0;
+    for (int i = 0; i < (n / 2); ++i) {
+      retained += left.query(i) ? 1 : 0;
+    }
+    assertTrue(retained > (n / 4), "expected most left-only items retained, got " + retained);
+  }
+
+  @Test
+  public void differenceWithSelfClearsFilter() {
+    final BloomFilter bf = BloomFilterBuilder.createBySize(4096, 4);
+    for (int i = 0; i < 200; ++i) {
+      bf.queryAndUpdate(i);
+    }
+    assertFalse(bf.isEmpty());
+    final BloomFilter same = BloomFilter.heapify(MemorySegment.ofArray(bf.toByteArray()));
+    bf.difference(same);
+    assertTrue(bf.isEmpty());
+    assertEquals(bf.getBitsUsed(), 0);
+    assertFalse(bf.query(0));
+  }
+
+  @Test
+  public void differenceWithEmptyIsIdentity() {
+    final BloomFilter left = BloomFilterBuilder.createBySize(4096, 4, 42L);
+    left.queryAndUpdate("apple");
+    left.queryAndUpdate("banana");
+    final long bits = left.getBitsUsed();
+    final byte[] before = left.toByteArray();
+
+    final BloomFilter empty = BloomFilterBuilder.createBySize(4096, 4, 42L);
+    left.difference(empty);
+    assertEquals(left.getBitsUsed(), bits);
+    assertTrue(left.query("apple"));
+    assertTrue(left.query("banana"));
+    assertEquals(left.toByteArray(), before);
+  }
+
   @Test
   public void emptySerializationTest() {
     final long numBits = 32768;

src/test/java/org/apache/datasketches/filters/bloomfilter/DirectBitArrayRTest.java

@@ -167,5 +167,6 @@ public void checkInvalidMethods() {
     assertThrows(SketchesReadOnlyException.class, () -> dba.invert());
     assertThrows(SketchesReadOnlyException.class, () -> dba.intersect(hba));
     assertThrows(SketchesReadOnlyException.class, () -> dba.union(hba));
+    assertThrows(SketchesReadOnlyException.class, () -> dba.andNot(hba));
   }
 }

src/test/java/org/apache/datasketches/filters/bloomfilter/DirectBitArrayTest.java

@@ -217,6 +217,7 @@ public void invalidUnionIntersectionTest() {
     final DirectBitArray dba = DirectBitArray.writableWrap(wseg, false);
     assertThrows(SketchesArgumentException.class, () -> dba.union(new HeapBitArray(64)));
     assertThrows(SketchesArgumentException.class, () -> dba.intersect(new HeapBitArray(512)));
+    assertThrows(SketchesArgumentException.class, () -> dba.andNot(new HeapBitArray(512)));
   }
 
   @Test
@@ -243,4 +244,29 @@ public void validUnionAndIntersectionTest() {
     ba3.union(ba2);
     assertEquals(ba3.getNumBitsSet(), (3 * n) / 2);
   }
+
+  @Test
+  public void validAndNotTest() {
+    final long numBits = 64;
+    final int sizeBytes = (int) BitArray.getSerializedSizeBytes(64);
+    final DirectBitArray ba1 = DirectBitArray.initialize(numBits, MemorySegment.ofArray(new byte[sizeBytes]));
+    final DirectBitArray ba2 = DirectBitArray.initialize(numBits, MemorySegment.ofArray(new byte[sizeBytes]));
+
+    final int n = 10;
+    for (int i = 0; i < n; ++i) {
+      ba1.getAndSetBit(i);
+      ba2.getAndSetBit(i + (n / 2));
+    }
+    assertEquals(ba1.getNumBitsSet(), n);
+    assertEquals(ba2.getNumBitsSet(), n);
+
+    ba1.andNot(ba2);
+    assertEquals(ba1.getNumBitsSet(), n / 2);
+    for (int i = 0; i < (n / 2); ++i) {
+      assertTrue(ba1.getBit(i));
+    }
+    for (int i = n / 2; i < n; ++i) {
+      assertFalse(ba1.getBit(i));
+    }
+  }
 }

src/test/java/org/apache/datasketches/filters/bloomfilter/HeapBitArrayTest.java

@@ -147,6 +147,36 @@ public void validUnionAndIntersectionTest() {
     assertEquals(ba3.getNumBitsSet(), (3 * n) / 2);
   }
 
+  @Test(expectedExceptions = SketchesArgumentException.class)
+  public void invalidAndNotTest() {
+    final HeapBitArray ba = new HeapBitArray(128);
+    ba.andNot(new HeapBitArray(64));
+  }
+
+  @Test
+  public void validAndNotTest() {
+    final HeapBitArray ba1 = new HeapBitArray(64);
+    final HeapBitArray ba2 = new HeapBitArray(64);
+
+    final int n = 10;
+    for (int i = 0; i < n; ++i) {
+      ba1.getAndSetBit(i);
+      ba2.getAndSetBit(i + (n / 2));
+    }
+    assertEquals(ba1.getNumBitsSet(), n);
+    assertEquals(ba2.getNumBitsSet(), n);
+
+    ba1.andNot(ba2);
+    // bits [0, n/2) remain; bits [n/2, n) cleared
+    assertEquals(ba1.getNumBitsSet(), n / 2);
+    for (int i = 0; i < (n / 2); ++i) {
+      assertTrue(ba1.getBit(i));
+    }
+    for (int i = n / 2; i < n; ++i) {
+      assertFalse(ba1.getBit(i));
+    }
+  }
+
   @Test
   public void serializeEmptyTest() {
     final HeapBitArray ba = new HeapBitArray(64);