Implementation of grouping size aggregator (#3577)

This implementation is creating a flavor of the
`SizeStatisticsCollectorCursor` that can return sizes aggregated by
sub-subspaces.
This relies on the original implementation but had to go into a separate
class as the generic type of the cursor has changed.
Eventually, the original cursor can be replaced by this one with the
aggregation size being "0".
For backwards compatibility, the original class was marked as
"deprecated". The original `SizeStatisticsResults` was marked as
deprecated and a new (non-embedded version) of it was added.

Initially, I thought that we can make a backwards-compatible change to
the `SizeStatisticsCollectorCursor` class, but that proved more
complicated with the changes to the result type, so I decided to go with
a new implementation instead.
 
Resolves #3576

Author: ohadzeliger

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/cursors/SizeStatisticsCollectorCursor.java

@@ -65,8 +65,9 @@
  * for all keys and values in the specified subspace. Any cursor results emitted prior to that are a manifestation of
  * hitting an execution limit while scanning the subspace. These results provide a continuation that can be used
  * to resume aggregating the remaining keys and values of the subspace.
+ * Deprecated, use the {@link SizeStatisticsGroupingCursor} instead.
  */
-@API(API.Status.EXPERIMENTAL)
+@API(API.Status.DEPRECATED)
 public class SizeStatisticsCollectorCursor implements RecordCursor<SizeStatisticsCollectorCursor.SizeStatisticsResults> {
     @Nonnull
     private final SubspaceProvider subspaceProvider;
@@ -318,7 +319,9 @@ public static SizeStatisticsCollectorCursor ofSubspace(@Nonnull Subspace subspac
 
     /**
      * Encapsulates the distribution statistics returned by a SizeStatisticsCollectorCursor.
+     * This is deprecated. Use the {@link com.apple.foundationdb.record.provider.foundationdb.cursors.SizeStatisticsResults} instead
      */
+    @API(API.Status.DEPRECATED)
     public static class SizeStatisticsResults {
         private long keyCount;
         private long keySize;

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/cursors/SizeStatisticsGroupedResults.java

@@ -0,0 +1,59 @@
+/*
+ * SizeStatisticsGroupedResults.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2025 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed 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 com.apple.foundationdb.record.provider.foundationdb.cursors;
+
+import com.apple.foundationdb.annotation.API;
+import com.apple.foundationdb.tuple.Tuple;
+
+import javax.annotation.Nonnull;
+
+/**
+ * The result of a grouped size calculation.
+ * The result is aggregated for a subspace whose key suffix is specified below, within the containing subspace.
+ */
+@API(API.Status.EXPERIMENTAL)
+public class SizeStatisticsGroupedResults {
+    private final Tuple aggregationKey;
+    private final SizeStatisticsResults stats;
+
+    public SizeStatisticsGroupedResults(@Nonnull final Tuple aggregationKey, @Nonnull SizeStatisticsResults stats) {
+        this.aggregationKey = aggregationKey;
+        this.stats = stats;
+    }
+
+    /**
+     * The Tuple that represents the key whose stats are aggregated.
+     * @return the Tuple representing the subspace key aggregated
+     */
+    @Nonnull
+    public Tuple getAggregationKey() {
+        return aggregationKey;
+    }
+
+    /**
+     * The aggregated stats.
+     * @return the statistics collected for the subspace
+     */
+    @Nonnull
+    public SizeStatisticsResults getStats() {
+        return stats;
+    }
+}

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/cursors/SizeStatisticsGroupingContinuation.java

@@ -0,0 +1,127 @@
+/*
+ * SizeStatisticsGroupingContinuation.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2025 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed 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 com.apple.foundationdb.record.provider.foundationdb.cursors;
+
+import com.apple.foundationdb.KeyValue;
+import com.apple.foundationdb.record.RecordCursorContinuation;
+import com.apple.foundationdb.record.RecordCursorEndContinuation;
+import com.apple.foundationdb.record.RecordCursorProto;
+import com.apple.foundationdb.record.RecordCursorResult;
+import com.apple.foundationdb.tuple.Tuple;
+import com.google.protobuf.ByteString;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+/**
+ * A continuation for the {@link SizeStatisticsGroupingCursor}.
+ */
+class SizeStatisticsGroupingContinuation implements RecordCursorContinuation {
+    /**
+     * A continuation for the case where we have one last result to send, after which the cursor would be done.
+      */
+    public static final SizeStatisticsGroupingContinuation LAST_RESULT_CONTINUATION = new SizeStatisticsGroupingContinuation();
+
+    /** Whether the continuation is a marker for the last result being sent before the cursor is done. */
+    private final boolean lastResultContinuation;
+    @Nullable
+    private RecordCursorContinuation innerContinuation;
+    @Nullable
+    private SizeStatisticsResults partialResults;
+    private Tuple currentGroupingKey;
+    @Nullable
+    private byte[] cachedBytes;
+    @Nullable
+    private ByteString cachedByteString;
+
+    /**
+     * The inner cursor is done, and we sent the last result, cannot continue afterward.
+     */
+    private SizeStatisticsGroupingContinuation() {
+        lastResultContinuation = true;
+    }
+
+    /**
+     * The inner cursor still has results, we can continue (e.g. group break).
+     */
+    SizeStatisticsGroupingContinuation(@Nonnull RecordCursorResult<KeyValue> currentKvResult,
+                                       @Nonnull SizeStatisticsResults partialResults,
+                                       @Nonnull Tuple currentGroupingKey) {
+        lastResultContinuation = false;
+        this.innerContinuation = currentKvResult.getContinuation();
+        this.partialResults = partialResults.copy(); //cache an immutable snapshot of the partial aggregate state
+        this.currentGroupingKey = currentGroupingKey;
+    }
+
+    /**
+     * Whether the given protobuf is the same as the "LAST_RESULT_CONTINUATION".
+     * @param statsContinuation the incoming protobuf
+     * @return TRUE if the protobuf marks a continuation with last result
+     */
+    public static boolean isLastResultContinuation(final RecordCursorProto.SizeStatisticsGroupingContinuation statsContinuation) {
+        // Default to false if unset
+        return (statsContinuation.getLastResultContinuation());
+    }
+
+    public boolean isLastResultContinuation() {
+        return lastResultContinuation;
+    }
+
+    @Nullable
+    @Override
+    public byte[] toBytes() {
+        // form bytes exactly once
+        if (this.cachedBytes == null) {
+            this.cachedBytes = toByteString().toByteArray();
+        }
+        return cachedBytes;
+    }
+
+    @Override
+    @Nonnull
+    public ByteString toByteString() {
+        if (cachedByteString == null) {
+            final RecordCursorProto.SizeStatisticsGroupingContinuation.Builder builder = RecordCursorProto.SizeStatisticsGroupingContinuation.newBuilder();
+            builder.setLastResultContinuation(lastResultContinuation);
+            if (innerContinuation != null) {
+                builder.setUnderlyingContinuation(innerContinuation.toByteString());
+            }
+            if (partialResults != null) {
+                builder.setPartialResults(partialResults.toProto());
+            }
+            if (currentGroupingKey != null) {
+                builder.setCurrentGroupingKey(ByteString.copyFrom(currentGroupingKey.pack()));
+            }
+            cachedByteString = builder.build().toByteString();
+        }
+        return cachedByteString;
+    }
+
+    /**
+     * Whether this continuation is the end of the stream.
+     * This is always false, since the {@link RecordCursorEndContinuation#END} would be used once the cursor is exhausted.
+     * @return TRUE if the cursor is exhausted
+     */
+    @Override
+    public boolean isEnd() {
+        return false;
+    }
+}