Fix ResolvedKeySpacePath.equals and hashCode

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/keyspace/KeySpaceDirectory.java

@@ -710,7 +710,8 @@ public Object getValue() {
         return value;
     }
 
-    protected static boolean areEqual(Object o1, Object o2) {
+    @SuppressWarnings("PMD.CompareObjectsWithEquals") // we use ref
+    protected static boolean areEqual(@Nullable Object o1, @Nullable Object o2) {
         if (o1 == null) {
             return o2 == null;
         } else {
@@ -719,6 +720,12 @@ protected static boolean areEqual(Object o1, Object o2) {
             }
         }
 
+        // Handle ANY_VALUE specially - typeOf does not support ANY_VALUE
+        boolean isAnyValue = (o1 == ANY_VALUE || o2 == ANY_VALUE);
+        if (isAnyValue) {
+            return Objects.equals(o1, o2);
+        }
+
         KeyType o1Type = KeyType.typeOf(o1);
         if (o1Type != KeyType.typeOf(o2)) {
             return false;
@@ -740,6 +747,31 @@ protected static boolean areEqual(Object o1, Object o2) {
         }
     }
 
+    protected static int valueHashCode(@Nullable Object value) {
+        if (value == null) {
+            return 0;
+        }
+
+        // Handle ANY_VALUE specially
+        if (value == ANY_VALUE) {
+            return System.identityHashCode(value);
+        }
+
+        switch (KeyType.typeOf(value)) {
+            case BYTES:
+                return Arrays.hashCode((byte[]) value);
+            case LONG:
+            case STRING:
+            case FLOAT:
+            case DOUBLE:
+            case BOOLEAN:
+            case UUID:
+                return Objects.hashCode(value);
+            default:
+                throw new RecordCoreException("Unexpected key type " + KeyType.typeOf(value));
+        }
+    }
+
     /**
      * Returns the path that leads up to this directory (including this directory), and returns it as a string
      * that looks something like a filesystem path.
@@ -898,6 +930,27 @@ public static KeyType typeOf(@Nullable Object value) {
         }
     }
 
+    /**
+     * Compares two KeySpaceDirectory objects for equality based on their intrinsic properties,
+     * ignoring their position in the directory hierarchy (parent, subdirectories) and wrapper functions.
+     *
+     * @param other the other KeySpaceDirectory to compare with
+     * @return true if the directories have the same name, keyType, and value; false otherwise
+     */
+    @SuppressWarnings("PMD.CompareObjectsWithEquals")
+    public boolean equalsIgnoringHierarchy(@Nullable Object other) {
+        if (this == other) {
+            return true;
+        }
+        if (other == null || getClass() != other.getClass()) {
+            return false;
+        }
+        KeySpaceDirectory that = (KeySpaceDirectory) other;
+        return Objects.equals(name, that.name) &&
+                Objects.equals(keyType, that.keyType) &&
+                areEqual(value, that.value);
+    }
+
     private static class AnyValue {
         private AnyValue() {
         }

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/keyspace/KeySpacePathImpl.java

@@ -276,27 +276,22 @@ public boolean equals(Object obj) {
         }
         KeySpacePath that = (KeySpacePath) obj;
 
-        // Check that the KeySpaceDirectories of the two paths are "equal enough".
-        // Even this is probably overkill since the isCompatible check in KeySpaceDirectory
-        // will keep us from doing anything too bad. We could move this check into KeySpaceDirectory
-        // but comparing two directories by value would necessitate traversing the entire directory
-        // tree, so instead we will use a narrower definition of equality here.
-        boolean directoriesEqual = Objects.equals(this.getDirectory().getKeyType(), that.getDirectory().getKeyType()) &&
-                                   Objects.equals(this.getDirectory().getName(), that.getDirectory().getName()) &&
-                                   Objects.equals(this.getDirectory().getValue(), that.getDirectory().getValue());
+        // We don't care whether the two objects exist in the same hierarchy, we validate the relevant bits by comparing
+        // parents.
+        boolean directoriesEqual = this.getDirectory().equalsIgnoringHierarchy(that.getDirectory());
 
+        // the values might be byte[]
         return directoriesEqual &&
-               Objects.equals(this.getValue(), that.getValue()) &&
-               Objects.equals(this.getParent(), that.getParent());
+                KeySpaceDirectory.areEqual(this.getValue(), that.getValue()) &&
+                Objects.equals(this.getParent(), that.getParent());
     }
 
     @Override
     public int hashCode() {
         return Objects.hash(
                 getDirectory().getKeyType(),
                 getDirectory().getName(),
-                getDirectory().getValue(),
-                getValue(),
+                KeySpaceDirectory.valueHashCode(getValue()),
                 parent);
     }
 

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/keyspace/PathValue.java

@@ -24,13 +24,14 @@
 
 import javax.annotation.Nullable;
 import java.util.Arrays;
+import java.util.Objects;
 
 /**
  * A class to represent the value stored at a particular element of a {@link KeySpacePath}. The <code>resolvedValue</code>
  * is the object that will appear in the {@link com.apple.foundationdb.tuple.Tuple} when
  * {@link KeySpacePath#toTuple(com.apple.foundationdb.record.provider.foundationdb.FDBRecordContext)} is invoked.
  * The <code>metadata</code> is left null by {@link KeySpaceDirectory} but other implementations may make use of
- * it (e.g. {@link DirectoryLayerDirectory}.
+ * it (e.g. {@link DirectoryLayerDirectory}).
  */
 @API(API.Status.UNSTABLE)
 public class PathValue {
@@ -69,4 +70,22 @@ public Object getResolvedValue() {
     public byte[] getMetadata() {
         return metadata == null ? null : Arrays.copyOf(metadata, metadata.length);
     }
+
+    @Override
+    public boolean equals(Object other) {
+        if (this == other) {
+            return true;
+        }
+        if (!(other instanceof PathValue)) {
+            return false;
+        }
+        PathValue that = (PathValue) other;
+        return KeySpaceDirectory.areEqual(this.resolvedValue, that.resolvedValue) &&
+                Arrays.equals(this.metadata, that.metadata);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(KeySpaceDirectory.valueHashCode(resolvedValue), Arrays.hashCode(metadata));
+    }
 }

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/keyspace/ResolvedKeySpacePath.java

@@ -229,13 +229,15 @@ public boolean equals(Object other) {
         }
 
         ResolvedKeySpacePath otherPath = (ResolvedKeySpacePath) other;
-        return this.inner.equals(otherPath.inner)
-               && Objects.equals(this.getResolvedValue(), otherPath.getResolvedValue());
+        return Objects.equals(this.getResolvedPathValue(), otherPath.getResolvedPathValue()) &&
+                Objects.equals(this.getParent(), otherPath.getParent()) &&
+                this.inner.equals(otherPath.inner) &&
+                Objects.equals(this.remainder, otherPath.remainder);
     }
 
     @Override
     public int hashCode() {
-        return Objects.hash(inner, getResolvedPathValue());
+        return Objects.hash(inner, getResolvedPathValue(), remainder, getParent());
     }
 
     @Override

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/keyspace/KeySpaceDirectoryTest.java

@@ -48,6 +48,9 @@
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.RegisterExtension;
 import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.FieldSource;
+import org.junit.jupiter.params.provider.MethodSource;
 
 import javax.annotation.Nonnull;
 import javax.annotation.Nullable;
@@ -68,6 +71,7 @@
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
+import java.util.stream.Stream;
 
 import static com.apple.foundationdb.record.TestHelpers.assertThrows;
 import static com.apple.foundationdb.record.TestHelpers.eventually;
@@ -110,11 +114,16 @@ public KeyTypeValue(KeyType keyType, @Nullable Object value, @Nullable Object va
             assertTrue(keyType.isMatch(value));
             assertTrue(keyType.isMatch(generator.get()));
         }
+
+        @Override
+        public String toString() {
+            return "KeyTypeValue{" + keyType + '}';
+        }
     }
 
-    private final Random random = new Random();
+    private static final Random random = new Random();
 
-    private final List<KeyTypeValue> valueOfEveryType = new ImmutableList.Builder<KeyTypeValue>()
+    private static final List<KeyTypeValue> valueOfEveryType = new ImmutableList.Builder<KeyTypeValue>()
             .add(new KeyTypeValue(KeyType.NULL, null, null, () -> null))
             .add(new KeyTypeValue(KeyType.BYTES, new byte[] { 0x01, 0x02 }, new byte[] { 0x03, 0x04 }, () -> {
                 int size = random.nextInt(10) + 1;
@@ -1218,6 +1227,123 @@ public void testListAcrossTransactions() {
         assertEquals(directoryEntries.size(), idx);
     }
 
+    static Stream<Arguments> testEqualsIgnoringHierarchyWithVariousTypes() {
+        // Basic type equality tests
+        KeySpaceDirectory stringDir1 = new KeySpaceDirectory("dir", KeyType.STRING, "hello");
+        KeySpaceDirectory stringDir2 = new KeySpaceDirectory("dir", KeyType.STRING, "hello");
+        KeySpaceDirectory stringDir3 = new KeySpaceDirectory("dir", KeyType.STRING, "goodbye");
+        KeySpaceDirectory stringDir4 = new KeySpaceDirectory("different_name", KeyType.STRING, "hello");
+
+        KeySpaceDirectory longDir1 = new KeySpaceDirectory("dir", KeyType.LONG, 42L);
+        KeySpaceDirectory longDir2 = new KeySpaceDirectory("dir", KeyType.LONG, 42L);
+        KeySpaceDirectory longDir3 = new KeySpaceDirectory("dir", KeyType.LONG, 100L);
+        KeySpaceDirectory longDirAsString = new KeySpaceDirectory("dir", KeyType.STRING, "42");
+
+        // bytes is, important here because testEqualsIgnoringHierarchyWithAllKeyTypes could pass if calling
+        // .equals on the two values, but this one would not.
+        KeySpaceDirectory bytesDir1 = new KeySpaceDirectory("dir", KeyType.BYTES, new byte[] { 0x01, 0x02 });
+        KeySpaceDirectory bytesDir2 = new KeySpaceDirectory("dir", KeyType.BYTES, new byte[] { 0x01, 0x02 });
+        KeySpaceDirectory bytesDir3 = new KeySpaceDirectory("dir", KeyType.BYTES, new byte[] { 0x05, 0x06 });
+
+        // Wrapper tests
+        Function<KeySpacePath, KeySpacePath> wrapper1 = TestWrapper1::new;
+        Function<KeySpacePath, KeySpacePath> wrapper2 = TestWrapper2::new;
+        KeySpaceDirectory dirWithWrapper1 = new KeySpaceDirectory("test", KeyType.STRING, "value", wrapper1);
+        KeySpaceDirectory dirWithWrapper2 = new KeySpaceDirectory("test", KeyType.STRING, "value", wrapper2);
+        KeySpaceDirectory dirWithoutWrapper = new KeySpaceDirectory("test", KeyType.STRING, "value", null);
+
+        // ANY_VALUE tests
+        KeySpaceDirectory anyValueDir1 = new KeySpaceDirectory("test", KeyType.STRING);
+        KeySpaceDirectory anyValueDir2 = new KeySpaceDirectory("test", KeyType.STRING);
+        KeySpaceDirectory specificValueDir = new KeySpaceDirectory("test", KeyType.STRING, "specificValue");
+
+        return Stream.of(
+                // Basic equality/inequality
+                Arguments.of(stringDir1, stringDir2, true),
+                Arguments.of(stringDir1, stringDir3, false),
+                Arguments.of(stringDir4, stringDir1, false),
+                Arguments.of(longDir1, longDir2, true),
+                Arguments.of(longDir1, longDir3, false),
+                Arguments.of(longDir1, longDirAsString, false),
+                Arguments.of(bytesDir1, bytesDir2, true),
+                Arguments.of(bytesDir1, bytesDir3, false),
+                Arguments.of(stringDir1, longDir1, false),
+
+                // Wrapper tests - different wrappers should be equal
+                Arguments.of(dirWithWrapper1, dirWithWrapper2, true),
+                Arguments.of(dirWithWrapper1, dirWithoutWrapper, true),
+                Arguments.of(dirWithWrapper2, dirWithoutWrapper, true),
+
+                // ANY_VALUE tests
+                Arguments.of(anyValueDir1, anyValueDir2, true),
+                Arguments.of(anyValueDir1, specificValueDir, false)
+        );
+    }
+
+    @ParameterizedTest
+    @MethodSource("testEqualsIgnoringHierarchyWithVariousTypes")
+    void testEqualsIgnoringHierarchyWithVariousTypes(KeySpaceDirectory dir1, KeySpaceDirectory dir2, boolean shouldBeEqual) {
+        if (shouldBeEqual) {
+            assertTrue(dir1.equalsIgnoringHierarchy(dir2), "Directories should be equal for " + dir1.getKeyType());
+            assertTrue(dir2.equalsIgnoringHierarchy(dir1), "Directories should be equal for " + dir2.getKeyType() + " (reversed)");
+        } else {
+            assertFalse(dir1.equalsIgnoringHierarchy(dir2), "Directories should not be equal");
+            assertFalse(dir2.equalsIgnoringHierarchy(dir1), "Directories should not be equal (reversed)");
+        }
+        assertTrue(dir1.equalsIgnoringHierarchy(dir1), "Directories should always equal themselves");
+        assertTrue(dir2.equalsIgnoringHierarchy(dir2), "Directories should always equal themselves");
+        assertFalse(dir1.equalsIgnoringHierarchy(null), "Directory should not equal null");
+        assertFalse(dir1.equalsIgnoringHierarchy(dir1.value), "Directory should not equal other classes");
+    }
+
+    @Test
+    void testEqualsIgnoringHierarchyWithSubdirectories() {
+        KeySpaceDirectory parent1 = new KeySpaceDirectory("parent", KeyType.STRING, "test");
+        KeySpaceDirectory parent2 = new KeySpaceDirectory("parent", KeyType.STRING, "test");
+
+        KeySpaceDirectory child1 = new KeySpaceDirectory("child", KeyType.LONG);
+        KeySpaceDirectory child2 = new KeySpaceDirectory("child", KeyType.LONG);
+
+        parent1.addSubdirectory(child1);
+        parent2.addSubdirectory(child2);
+
+        assertTrue(parent1.equalsIgnoringHierarchy(parent2), "Parents with equivalent subdirectories should be equal when ignoring hierarchy");
+        assertTrue(child1.equalsIgnoringHierarchy(child2), "Children should be equal when ignoring hierarchy");
+
+        KeySpaceDirectory parent3 = new KeySpaceDirectory("parent", KeyType.STRING, "test");
+        KeySpaceDirectory child3 = new KeySpaceDirectory("child", KeyType.STRING);
+        parent3.addSubdirectory(child3);
+
+        assertTrue(parent1.equalsIgnoringHierarchy(parent3), "Parents should be equal when ignoring subdirectories");
+    }
+
+    @Test
+    void testEqualsIgnoringHierarchyWithParentRelationship() {
+        KeySpaceDirectory root1 = new KeySpaceDirectory("root", KeyType.STRING, "test");
+        KeySpaceDirectory root2 = new KeySpaceDirectory("root", KeyType.STRING, "test");
+
+        KeySpaceDirectory child1 = new KeySpaceDirectory("child", KeyType.LONG);
+        KeySpaceDirectory child2 = new KeySpaceDirectory("child", KeyType.LONG);
+
+        root1.addSubdirectory(child1);
+        root2.addSubdirectory(child2);
+
+        assertTrue(child1.equalsIgnoringHierarchy(child2), "Children with different parents should be equal when ignoring hierarchy");
+    }
+
+    @ParameterizedTest
+    @FieldSource("valueOfEveryType")
+    void testEqualsIgnoringHierarchyWithAllKeyTypes(KeyTypeValue keyTypeValue) {
+        KeySpaceDirectory dir1 = new KeySpaceDirectory("dir", keyTypeValue.keyType, keyTypeValue.value);
+        KeySpaceDirectory dir2 = new KeySpaceDirectory("dir", keyTypeValue.keyType, keyTypeValue.value);
+
+        assertTrue(dir1.equalsIgnoringHierarchy(dir2), "Directories should be equal");
+        if (keyTypeValue.keyType != KeyType.NULL) {
+            assertFalse(dir1.equalsIgnoringHierarchy(new KeySpaceDirectory("dir", keyTypeValue.keyType, keyTypeValue.value2)),
+                    "Directories should not be equal");
+        }
+    }
+
     private static class TestWrapper1 extends KeySpacePathWrapper {
         public TestWrapper1(KeySpacePath inner) {
             super(inner);