Redo bounds checking for decoding and encoding of double-records

* We are not allowed to throw exceptions when decoding records until after we have verified that our read was consistent, with `PageCursor.shouldRetry()`
* Double-records needs two concurrently open PageCursors cursors do decode.
* … but the outer retry loop in CommonAbstractStore is only looking at the cursor used to decode the first record.
* To solve this, the concept of linked page cursors is introduced:
* A page cursor can open a linked page cursor, and calls to `shouldRetry` and `checkAndClearOutOfBoundsFlag` on the parent cursor will also automatically delegate to the linked cursor.
* This way, the retry loop on CommonAbstractStore ends up covering both cursors with its calls to `shouldRetry` and its bounds check.
* To simplify decoding of records that span multiple record-units, the CompositePageCursor is introduced:
* A CompositePageCursor presents a seamless view of parts of two distinct page cursors.
* Using a composite page cursor, double-records can be decoded as if they were a single record.
* This simplifies record decoding logic.
* This simplification has made the DataAdapter concept, and its SecondaryRead and WriteCursorAdapters redundant.
* As part of this work the `PageCursor.getUnsignedInt()` method has been removed and inlined into all of its call sites. This simplifies testing of the cursor interface, which already has a large number of IO methods.
* The `RecordFormatTest` was making assertions about the number of `PagedFile.io()` and `PageCursor.next()` calls the decoding logic makes during its work. These assertions have been removed because those measurements cannot be trusted, when we are also using an adversarial page cache implementation that causes random retries of reads, etc.

community/io/src/main/java/org/neo4j/io/pagecache/PageCursor.java

@@ -71,129 +71,73 @@ public interface PageCursor extends AutoCloseable
 
     /**
      * Get the signed byte at the current page offset, and then increment the offset by one.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the current offset is not within the page bounds.
      */
     byte getByte();
 
     /**
      * Get the signed byte at the given offset into the page.
      * Leaves the current page offset unchanged.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the given offset is not within the page bounds.
      */
     byte getByte( int offset );
 
     /**
      * Set the signed byte at the current offset into the page, and then increment the offset by one.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the current offset is not within the page bounds.
      */
     void putByte( byte value );
 
     /**
      * Set the signed byte at the given offset into the page.
      * Leaves the current page offset unchanged.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the given offset is not within the page bounds.
      */
     void putByte( int offset, byte value );
 
     /**
      * Get the signed long at the current page offset, and then increment the offset by one.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the current offset is not within the page bounds.
      */
     long getLong();
 
     /**
      * Get the signed long at the given offset into the page.
      * Leaves the current page offset unchanged.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the given offset is not within the page bounds.
      */
     long getLong( int offset );
 
     /**
      * Set the signed long at the current offset into the page, and then increment the offset by one.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the current offset is not within the page bounds.
      */
     void putLong( long value );
 
     /**
      * Set the signed long at the given offset into the page.
      * Leaves the current page offset unchanged.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the given offset is not within the page bounds.
      */
     void putLong( int offset, long value );
 
     /**
      * Get the signed int at the current page offset, and then increment the offset by one.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the current offset is not within the page bounds.
      */
     int getInt();
 
     /**
      * Get the signed int at the given offset into the page.
      * Leaves the current page offset unchanged.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the given offset is not within the page bounds.
      */
     int getInt( int offset );
 
     /**
      * Set the signed int at the current offset into the page, and then increment the offset by one.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the current offset is not within the page bounds.
      */
     void putInt( int value );
 
     /**
      * Set the signed int at the given offset into the page.
      * Leaves the current page offset unchanged.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the given offset is not within the page bounds.
      */
     void putInt( int offset, int value );
 
-    /**
-     * Get the unsigned int at the current page offset, and then increment the offset by one.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the current offset is not within the page bounds.
-     */
-    long getUnsignedInt();
-
-    /**
-     * Get the unsigned int at the given offset into the page.
-     * Leaves the current page offset unchanged.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the given offset is not within the page bounds.
-     */
-    long getUnsignedInt( int offset );
-
     /**
      * Fill the given array with bytes from the page, beginning at the current offset into the page,
      * and then increment the current offset by the length of the array.
-     *
-     * @throws IndexOutOfBoundsException
-     * if the current offset plus the length of the array reaches beyond the end of the page.
      */
     void getBytes( byte[] data );
 
@@ -318,7 +262,7 @@ public interface PageCursor extends AutoCloseable
 
     /**
      * Relinquishes all resources associated with this cursor, including the
-     * cursor itself. The cursor cannot be used after this call.
+     * cursor itself, and any linked cursors opened through it. The cursor cannot be used after this call.
      * @see AutoCloseable#close()
      */
     void close();
@@ -353,7 +297,25 @@ public interface PageCursor extends AutoCloseable
      * Discern whether an out-of-bounds access has occurred since the last call to {@link #next()} or
      * {@link #next(long)}, or since the last call to {@link #shouldRetry()} that returned {@code true}, or since the
      * last call to this method.
-     * @return {@code true} if an access was out of bounds.
+     * @return {@code true} if an access was out of bounds, or the {@link #raiseOutOfBounds()} method has been called.
      */
     boolean checkAndClearBoundsFlag();
+
+    /**
+     * Explicitly raise the out-of-bounds flag.
+     * @see #checkAndClearBoundsFlag()
+     */
+    void raiseOutOfBounds();
+
+    /**
+     * Open a new page cursor with the same pf_flags as this cursor, as if calling the {@link PagedFile#io(long, int)}
+     * on the relevant paged file. This cursor will then also delegate to the linked cursor when checking
+     * {@link #shouldRetry()} and {@link #checkAndClearBoundsFlag()}.
+     *
+     * Opening a linked cursor on a cursor that already has a linked cursor, will close the older linked cursor.
+     * Closing a cursor also closes any linked cursor.
+     * @param pageId The page id that the linked cursor will be placed at after its first call to {@link #next()}.
+     * @return A cursor that is linked with this cursor.
+     */
+    PageCursor openLinkedCursor( long pageId );
 }