diff --git a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java index f4e7605a7f..c206cb88ed 100644 --- a/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java +++ b/hugegraph-store/hg-store-rocksdb/src/main/java/org/apache/hugegraph/rocksdb/access/RocksDBSession.java @@ -1080,6 +1080,16 @@ public boolean equals(Object obj) { return Bytes.equals(this.name, other.name) && Bytes.equals(this.value, other.value); } + + /** + * Hashes the same name and value bytes that equals() compares. A scan allocates a new + * column per row, so the hash-based union, intersection and limit-dedup iterators rely on + * byte-equal columns landing on one key. + */ + @Override + public int hashCode() { + return 31 * Arrays.hashCode(this.name) + Arrays.hashCode(this.value); + } } class RefCounter { diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java index 52afd5d8ab..1a42eb233e 100644 --- a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/core/CoreSuiteTest.java @@ -23,6 +23,7 @@ import org.apache.hugegraph.store.core.store.meta.PartitionMetaCodecTest; import org.apache.hugegraph.store.util.FileObjectIteratorTest; import org.apache.hugegraph.store.util.FilterIteratorCleanupTest; +import org.apache.hugegraph.store.util.FilterIteratorColumnEqualityTest; import org.apache.hugegraph.store.util.SortShuffleSerializerTest; import org.apache.hugegraph.store.util.SortShuffleTest; import org.apache.hugegraph.store.util.TypeTransIteratorTest; @@ -54,6 +55,7 @@ @Suite.SuiteClasses({ FileObjectIteratorTest.class, FilterIteratorCleanupTest.class, + FilterIteratorColumnEqualityTest.class, SortShuffleSerializerTest.class, SortShuffleTest.class, TypeTransIteratorTest.class, diff --git a/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/util/FilterIteratorColumnEqualityTest.java b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/util/FilterIteratorColumnEqualityTest.java new file mode 100644 index 0000000000..bc485a90aa --- /dev/null +++ b/hugegraph-store/hg-store-test/src/main/java/org/apache/hugegraph/store/util/FilterIteratorColumnEqualityTest.java @@ -0,0 +1,186 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.hugegraph.store.util; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotEquals; + +import java.nio.ByteBuffer; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Set; +import java.util.stream.Collectors; +import java.util.stream.IntStream; + +import org.apache.hugegraph.rocksdb.access.RocksDBSession.BackendColumn; +import org.apache.hugegraph.rocksdb.access.ScanIterator; +import org.apache.hugegraph.store.business.itrv2.IntersectionFilterIterator; +import org.apache.hugegraph.store.business.itrv2.IntersectionWrapper; +import org.apache.hugegraph.store.business.itrv2.MapLimitIterator; +import org.apache.hugegraph.store.business.itrv2.UnionFilterIterator; +import org.apache.hugegraph.store.business.itrv2.io.SortShuffleSerializer; +import org.junit.Test; + +/** + * A RocksDB scan allocates a new BackendColumn for every row, so the hash-based union, + * intersection and limit-dedup iterators must treat byte-equal columns from separate scans as + * one key. Every row below is built from fresh arrays; reusing one object per key would hide a + * missing hashCode(). + */ +public class FilterIteratorColumnEqualityTest { + + private static final int KEYS = 100; + private static final Set ALL_KEYS = + IntStream.range(0, KEYS).boxed().collect(Collectors.toSet()); + + @Test + public void testEqualColumnsFromSeparateArraysShareHashCode() { + BackendColumn left = column(7); + BackendColumn right = column(7); + + assertEquals(left, right); + assertEquals(left.hashCode(), right.hashCode()); + + Set set = new HashSet<>(); + set.add(left); + set.add(right); + assertEquals(1, set.size()); + } + + @Test + public void testColumnsWithEqualNamesButDifferentValuesStayDistinct() { + BackendColumn left = BackendColumn.of(name(7), new byte[]{1}); + BackendColumn right = BackendColumn.of(name(7), new byte[]{2}); + + assertNotEquals(left, right); + + Set set = new HashSet<>(); + set.add(left); + set.add(right); + assertEquals(2, set.size()); + } + + @Test + public void testIntersectionCombinesEqualRowsFromFreshObjects() { + IntersectionWrapper wrapper = wrapper(); + ScanIterator filter = new IntersectionFilterIterator(scanTwice(), wrapper, 2); + + assertEquals(ALL_KEYS, drain(filter)); + } + + @Test + public void testUnionBelowSortThresholdCombinesEqualRowsFromFreshObjects() { + IntersectionWrapper wrapper = wrapper(); + ScanIterator filter = new UnionFilterIterator<>( + scanTwice(), wrapper, (left, right) -> Arrays.compare(left.name, right.name), + SortShuffleSerializer.ofBackendColumnSerializer()); + + assertEquals(ALL_KEYS, drain(filter)); + } + + @Test + public void testLimitDedupDropsEqualRowsFromFreshObjects() { + ScanIterator filter = new MapLimitIterator(scanTwice()); + + assertEquals(ALL_KEYS, drain(filter)); + } + + /** + * Marks every key as a repeated candidate, the way a multi-list query does once each key + * has been seen in more than one source. + */ + private static IntersectionWrapper wrapper() { + IntersectionWrapper wrapper = new IntersectionWrapper<>( + scanTwice(), column -> ByteBuffer.wrap(column.name).getInt()); + wrapper.proc(); + return wrapper; + } + + /** + * Each key appears twice, each time as a newly allocated column with its own arrays. + */ + private static ScanIterator scanTwice() { + List rows = new ArrayList<>(); + for (int round = 0; round < 2; round++) { + for (int key = 0; key < KEYS; key++) { + rows.add(column(key)); + } + } + return new ListScanIterator(rows); + } + + private static BackendColumn column(int key) { + return BackendColumn.of(name(key), name(key)); + } + + private static byte[] name(int key) { + return ByteBuffer.allocate(Integer.BYTES).putInt(key).array(); + } + + /** + * Returns the keys the iterator emits, failing on a key emitted twice so that a wrong row + * count can't hide behind a set. + */ + private static Set drain(ScanIterator iterator) { + Set keys = new HashSet<>(); + try { + while (iterator.hasNext()) { + BackendColumn column = iterator.next(); + int key = ByteBuffer.wrap(column.name).getInt(); + if (!keys.add(key)) { + throw new AssertionError("key " + key + " emitted twice"); + } + } + } finally { + iterator.close(); + } + return keys; + } + + private static class ListScanIterator implements ScanIterator { + + private final Iterator rows; + + private ListScanIterator(List rows) { + this.rows = rows.iterator(); + } + + @Override + public boolean hasNext() { + return this.rows.hasNext(); + } + + @Override + public boolean isValid() { + return this.rows.hasNext(); + } + + @Override + @SuppressWarnings("unchecked") + public T next() { + return (T) this.rows.next(); + } + + @Override + public void close() { + } + } +}