Streaming aggregate cursor now requires 4.1.9.0 or later for continuation deserialization (FoundationDB#3246)

This cleans up the `StreamingAggregateCursor` and its helper classes to
remove the `createDefaultIfEmpty` option. We'd been carrying that option
around as we introduced that field in FoundationDB#3092 and wanted to preserve
behavior when upgrading from older versions. The intention had been that
all new plans would set the option starting in 4.1, but a bug (fixed in
FoundationDB#3211; see FoundationDB#3096) means that we didn't actually enable it until 4.1.9.0.
That means that after this change, we'd require 4.1.9.0 or newer in
order to safely continue these queries. In theory, we could wait with
this change, but 4.1.9.0 has enough fixes that I actually think our
recommendation should be that anyone upgrading from 4.0 or below go
straight to 4.1.9.0, and then they can proceed safely to a newer version
with this change.

I was able to validate that this change is compatible with 4.1.9.0 via
the cross-version tests run during PRB. When I ran the full mixed mode
tests with the `aggregate-index-tests.yamsql`, I found that only 4.1.9.0
worked, which is expected.

This resolves FoundationDB#3107.

Author: alecgrieser

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/cursors/aggregate/AggregateCursor.java

@@ -25,7 +25,6 @@
 import com.apple.foundationdb.record.RecordCursor;
 import com.apple.foundationdb.record.RecordCursorContinuation;
 import com.apple.foundationdb.record.RecordCursorResult;
-import com.apple.foundationdb.record.RecordCursorStartContinuation;
 import com.apple.foundationdb.record.RecordCursorVisitor;
 import com.apple.foundationdb.record.query.plan.plans.QueryResult;
 import com.google.common.base.Verify;
@@ -50,7 +49,6 @@ public class AggregateCursor<M extends Message> implements RecordCursor<QueryRes
     // group aggregator to break incoming records into groups
     @Nonnull
     private final StreamGrouping<M> streamGrouping;
-    private final boolean isCreateDefaultOnEmpty;
     // Previous record processed by this cursor
     @Nullable
     private RecordCursorResult<QueryResult> previousResult;
@@ -59,11 +57,9 @@ public class AggregateCursor<M extends Message> implements RecordCursor<QueryRes
     private RecordCursorResult<QueryResult> previousValidResult;
 
     public AggregateCursor(@Nonnull RecordCursor<QueryResult> inner,
-                           @Nonnull final StreamGrouping<M> streamGrouping,
-                           boolean isCreateDefaultOnEmpty) {
+                           @Nonnull final StreamGrouping<M> streamGrouping) {
         this.inner = inner;
         this.streamGrouping = streamGrouping;
-        this.isCreateDefaultOnEmpty = isCreateDefaultOnEmpty;
     }
 
     @Nonnull
@@ -77,7 +73,7 @@ public CompletableFuture<RecordCursorResult<QueryResult>> onNext() {
         return AsyncUtil.whileTrue(() -> inner.onNext().thenApply(innerResult -> {
             previousResult = innerResult;
             if (!innerResult.hasNext()) {
-                if (!isNoRecords() || (isCreateDefaultOnEmpty && streamGrouping.isResultOnEmpty())) {
+                if (!isNoRecords()) {
                     streamGrouping.finalizeGroup();
                 }
                 return false;
@@ -93,11 +89,7 @@ public CompletableFuture<RecordCursorResult<QueryResult>> onNext() {
         }), getExecutor()).thenApply(vignore -> {
             if (isNoRecords()) {
                 // Edge case where there are no records at all
-                if (isCreateDefaultOnEmpty && streamGrouping.isResultOnEmpty()) {
-                    return RecordCursorResult.withNextValue(QueryResult.ofComputed(streamGrouping.getCompletedGroupResult()), RecordCursorStartContinuation.START);
-                } else {
-                    return RecordCursorResult.exhausted();
-                }
+                return RecordCursorResult.exhausted();
             }
             // Use the last valid result for the continuation as we need non-terminal one here.
             RecordCursorContinuation continuation = Verify.verifyNotNull(previousValidResult).getContinuation();

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/cursors/aggregate/StreamGrouping.java

@@ -193,8 +193,4 @@ private Object evalGroupingKey(@Nullable final Object currentObject) {
         final EvaluationContext nestedContext = context.withBinding(Bindings.Internal.CORRELATION, alias, currentObject);
         return Objects.requireNonNull(groupingKeyValue).eval(store, nestedContext);
     }
-
-    public boolean isResultOnEmpty() {
-        return groupingKeyValue == null;
-    }
 }

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryStreamingAggregationPlan.java

@@ -27,6 +27,7 @@
 import com.apple.foundationdb.record.PlanDeserializer;
 import com.apple.foundationdb.record.PlanHashable;
 import com.apple.foundationdb.record.PlanSerializationContext;
+import com.apple.foundationdb.record.RecordCoreArgumentException;
 import com.apple.foundationdb.record.RecordCursor;
 import com.apple.foundationdb.record.cursors.aggregate.AggregateCursor;
 import com.apple.foundationdb.record.cursors.aggregate.StreamGrouping;
@@ -96,13 +97,6 @@ public class RecordQueryStreamingAggregationPlan implements RecordQueryPlanWithC
     private final CorrelationIdentifier aggregateAlias;
     @Nonnull
     private final Value completeResultValue;
-    //
-    // This flag is needed to distinguish if we need to create a default value on-empty or not (i.e.
-    // RecordQueryDefaultOnEmptyPlan will do that going forward). We will always plan with that flag set to false going
-    // forward, but we accept and honor this field coming from proto if we are continuing OR if it not there we imply
-    // true.
-    // https://github.com/FoundationDB/fdb-record-layer/issues/3107
-    private final boolean isCreateDefaultOnEmpty;
 
     /**
      * Construct a new plan.
@@ -119,15 +113,13 @@ private RecordQueryStreamingAggregationPlan(@Nonnull final Quantifier.Physical i
                                                 @Nonnull final AggregateValue aggregateValue,
                                                 @Nonnull final CorrelationIdentifier groupingKeyAlias,
                                                 @Nonnull final CorrelationIdentifier aggregateAlias,
-                                                @Nonnull final Value completeResultValue,
-                                                final boolean isCreateDefaultOnEmpty) {
+                                                @Nonnull final Value completeResultValue) {
         this.inner = inner;
         this.groupingKeyValue = groupingKeyValue;
         this.aggregateValue = aggregateValue;
         this.groupingKeyAlias = groupingKeyAlias;
         this.aggregateAlias = aggregateAlias;
         this.completeResultValue = completeResultValue;
-        this.isCreateDefaultOnEmpty = isCreateDefaultOnEmpty;
     }
 
     @Nonnull
@@ -148,7 +140,7 @@ public <M extends Message> RecordCursor<QueryResult> executePlan(@Nonnull FDBRec
                         (FDBRecordStoreBase<Message>)store,
                         context,
                         inner.getAlias());
-        return new AggregateCursor<>(innerCursor, streamGrouping, isCreateDefaultOnEmpty)
+        return new AggregateCursor<>(innerCursor, streamGrouping)
                 .skipThenLimit(executeProperties.getSkip(),
                         executeProperties.getReturnedRowLimit());
     }
@@ -206,8 +198,7 @@ public RecordQueryStreamingAggregationPlan translateCorrelations(@Nonnull final
                 translatedAggregateValue,
                 groupingKeyAlias,
                 aggregateAlias,
-                completeResultValue,
-                isCreateDefaultOnEmpty);
+                completeResultValue);
     }
 
     @Nonnull
@@ -218,8 +209,7 @@ public RecordQueryStreamingAggregationPlan withChild(@Nonnull final Reference ch
                 aggregateValue,
                 groupingKeyAlias,
                 aggregateAlias,
-                completeResultValue,
-                isCreateDefaultOnEmpty);
+                completeResultValue);
     }
 
     @Nonnull
@@ -371,7 +361,7 @@ public PRecordQueryStreamingAggregationPlan toProto(@Nonnull final PlanSerializa
         builder.setGroupingKeyAlias(groupingKeyAlias.getId())
                 .setAggregateAlias(aggregateAlias.getId())
                 .setCompleteResultValue(completeResultValue.toValueProto(serializationContext))
-                .setIsCreateDefaultOnEmpty(isCreateDefaultOnEmpty);
+                .setIsCreateDefaultOnEmpty(false);
         return builder.build();
     }
 
@@ -396,7 +386,10 @@ public static RecordQueryStreamingAggregationPlan fromProto(@Nonnull final PlanS
         final CorrelationIdentifier aggregateAlias = CorrelationIdentifier.of(Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getAggregateAlias()));
         final Value completeResultValue = Value.fromValueProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getCompleteResultValue()));
         final boolean isCreateDefaultOnEmpty = recordQueryStreamingAggregationPlanProto.hasIsCreateDefaultOnEmpty() ? recordQueryStreamingAggregationPlanProto.getIsCreateDefaultOnEmpty() : true;
-        return new RecordQueryStreamingAggregationPlan(inner, groupingKeyValue, aggregateValue, groupingKeyAlias, aggregateAlias, completeResultValue, isCreateDefaultOnEmpty);
+        if (isCreateDefaultOnEmpty) {
+            throw new RecordCoreArgumentException("cannot create streaming aggregate plan with default value on empty");
+        }
+        return new RecordQueryStreamingAggregationPlan(inner, groupingKeyValue, aggregateValue, groupingKeyAlias, aggregateAlias, completeResultValue);
     }
 
     @Nonnull
@@ -428,7 +421,7 @@ public static RecordQueryStreamingAggregationPlan of(@Nonnull final Quantifier.P
         final var referencedAggregateValue = ObjectValue.of(aggregateAlias, aggregateValue.getResultType());
 
         return new RecordQueryStreamingAggregationPlan(inner, groupingKeyValue, aggregateValue, groupingKeyAlias, aggregateAlias,
-                resultValueFunction.apply(referencedGroupingKeyValue, referencedAggregateValue), false);
+                resultValueFunction.apply(referencedGroupingKeyValue, referencedAggregateValue));
     }
 
     /**