Fix continuation deserialization errors by deserializing streaming aggregate plans in the correct order (#3215)

The error we were seeing was that for certain plans, we were getting
errors trying to deserialize the elements of a
`FDBStreamingAggregatePlan` continuation. In particular, we were seeing
the type's reference before the type's definition. This was because the
serialization and deserialization of the plan happened in different
orders, which meant that at deserialization time, we were no longer
guaranteed to come across the definition first. I've kept the order of
serialization the same, and then updated the deserialization code path
to align it with existing continuations that are out in the wild.

This fixes #3214. I've enabled FORCE_CONTINUATIONS mode on that test,
though it mostly just works on the embedded vs. current configuration
(which is enough to surface the bug). More test coverage is added by
having the `StreamingAggregate` plans in one of the Record Layer tests
all go through `verifySerialization`, which previously wasn't being done
on those plans.

Author: alecgrieser

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

@@ -35,7 +35,6 @@
 import com.apple.foundationdb.record.provider.common.StoreTimer;
 import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase;
 import com.apple.foundationdb.record.provider.foundationdb.FDBStoreTimer;
-import com.apple.foundationdb.record.query.plan.explain.ExplainPlanVisitor;
 import com.apple.foundationdb.record.query.plan.cascades.AliasMap;
 import com.apple.foundationdb.record.query.plan.cascades.CorrelationIdentifier;
 import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
@@ -50,6 +49,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.values.ObjectValue;
 import com.apple.foundationdb.record.query.plan.cascades.values.Value;
 import com.apple.foundationdb.record.query.plan.cascades.values.translation.TranslationMap;
+import com.apple.foundationdb.record.query.plan.explain.ExplainPlanVisitor;
 import com.apple.foundationdb.record.query.plan.serialization.PlanSerialization;
 import com.google.auto.service.AutoService;
 import com.google.common.collect.ImmutableList;
@@ -385,15 +385,18 @@ public PRecordQueryPlan toRecordQueryPlanProto(@Nonnull final PlanSerializationC
     @Nonnull
     public static RecordQueryStreamingAggregationPlan fromProto(@Nonnull final PlanSerializationContext serializationContext,
                                                                 @Nonnull final PRecordQueryStreamingAggregationPlan recordQueryStreamingAggregationPlanProto) {
-        return new RecordQueryStreamingAggregationPlan(Quantifier.Physical.fromProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getInner())),
-                PlanSerialization.getFieldOrNull(recordQueryStreamingAggregationPlanProto,
-                        PRecordQueryStreamingAggregationPlan::hasGroupingKeyValue,
-                        m -> Value.fromValueProto(serializationContext, m.getGroupingKeyValue())),
-                (AggregateValue)Value.fromValueProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getAggregateValue())),
-                CorrelationIdentifier.of(Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getGroupingKeyAlias())),
-                CorrelationIdentifier.of(Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getAggregateAlias())),
-                Value.fromValueProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getCompleteResultValue())),
-                recordQueryStreamingAggregationPlanProto.hasIsCreateDefaultOnEmpty() ? recordQueryStreamingAggregationPlanProto.getIsCreateDefaultOnEmpty() : true);
+        // Note: it is important for proper deserialization (at least of things that interact with the serializationContext's cache of
+        // referenced values and plans) that we deserialize the values in the same order as they are serialized, or we may
+        // not
+        final Quantifier.Physical inner = Quantifier.Physical.fromProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getInner()));
+        final AggregateValue aggregateValue =  (AggregateValue) Value.fromValueProto(serializationContext, Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getAggregateValue()));
+        @Nullable final Value groupingKeyValue = PlanSerialization.getFieldOrNull(recordQueryStreamingAggregationPlanProto, PRecordQueryStreamingAggregationPlan::hasGroupingKeyValue,
+                m -> Value.fromValueProto(serializationContext, m.getGroupingKeyValue()));
+        final CorrelationIdentifier groupingKeyAlias = CorrelationIdentifier.of(Objects.requireNonNull(recordQueryStreamingAggregationPlanProto.getGroupingKeyAlias()));
+        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);
     }
 
     @Nonnull

Diff for: fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/FDBStreamAggregationTest.java

@@ -334,7 +334,8 @@ private void populateDB(final int numRecords) throws Exception {
     }
 
     @Nonnull
-    private RecordCursor<QueryResult> executePlan(final RecordQueryPlan plan, final int rowLimit, final byte[] continuation) {
+    private RecordCursor<QueryResult> executePlan(final RecordQueryPlan originalPlan, final int rowLimit, final byte[] continuation) {
+        final RecordQueryPlan plan = verifySerialization(originalPlan);
         final var types = plan.getDynamicTypes();
         final var typeRepository = TypeRepository.newBuilder().addAllTypes(types).build();
         ExecuteProperties executeProperties = ExecuteProperties.SERIAL_EXECUTE;

Diff for: yaml-tests/src/test/java/YamlIntegrationTests.java

@@ -141,8 +141,6 @@ public void aggregateEmptyTable(YamlTest.Runner runner) throws Exception {
     }
 
     @TestTemplate
-    @ExcludeYamlTestConfig(value = YamlTestConfigFilters.DO_NOT_FORCE_CONTINUATIONS,
-            reason = "Infinite continuation loop (https://github.com/FoundationDB/fdb-record-layer/issues/3095)")
     public void aggregateIndexTestsCount(YamlTest.Runner runner) throws Exception {
         runner.runYamsql("aggregate-index-tests-count.yamsql");
     }

Diff for: yaml-tests/src/test/resources/aggregate-index-tests-count.yamsql

@@ -47,6 +47,8 @@ test_block:
     -
       - query: select count(*) from t1
       - explain: "AISCAN(MV1 <,> BY_GROUP -> [_0: VALUE:[0]]) | MAP (_ AS _0) | ON EMPTY NULL | MAP (coalesce_long(_._0._0, promote(0l AS LONG)) AS _0)"
+      # Cannot run with FORCE_CONTINUATIONS due to: https://github.com/FoundationDB/fdb-record-layer/issues/3206
+      - maxRows: 0
       - result: [{4}]
     -
       - query: select count(*) from t1 group by col2
@@ -55,6 +57,8 @@ test_block:
     -
       - query: select count(col1) from t1
       - explain: "AISCAN(MV3 <,> BY_GROUP -> [_0: VALUE:[0]]) | MAP (_ AS _0) | ON EMPTY NULL | MAP (coalesce_long(_._0._0, promote(0l AS LONG)) AS _0)"
+      # Cannot run with FORCE_CONTINUATIONS due to: https://github.com/FoundationDB/fdb-record-layer/issues/3206
+      - maxRows: 0
       - result: [{2}]
     -
       - query: select count(col1) from t1 group by col2
@@ -81,17 +85,47 @@ test_block:
     -
       - query: select count(*) from t2
       - explain: "ISCAN(MV5 <,>) | MAP (_ AS _0) | AGG (count_star(*) AS _0) | ON EMPTY NULL | MAP (coalesce_long(_._0._0, promote(0l AS LONG)) AS _0)"
+      # Cannot run with FORCE_CONTINUATIONS due to: https://github.com/FoundationDB/fdb-record-layer/issues/3206
+      - maxRows: 0
       - result: [{4}]
     -
       - query: select count(*) from t2 group by col2
+      # Plan deserialization previously failed : https://github.com/FoundationDB/fdb-record-layer/issues/3214
+      - supported_version: !current_version
       - explain: "ISCAN(MV5 <,>) | MAP (_ AS _0) | AGG (count_star(*) AS _0) GROUP BY (_._0.COL2 AS _0) | MAP (_._1._0 AS _0)"
       - result: [{1}, {3}]
+    -
+      # Same as above test, but tests serialization upgrades from before !current_version. Can be removed once we no longer
+      # care about upgrading to that version from older versions
+      - query: select count(*) from t2 group by col2
+      - maxRows: 1
+      - initialVersionLessThan: !current_version
+      - result: [{1}]
+      - result: [{2}] # Off by one due to: https://github.com/FoundationDB/fdb-record-layer/issues/3097 (also fixed in !current_version)
+      - error: 'XX000' # plan fails to deserialize on older server
+      - initialVersionAtLeast: !current_version
+      # Covered in above test case
     -
       - query: select count(col1) from t2
       - explain: "ISCAN(MV5 <,>) | MAP (_ AS _0) | AGG (count(_._0.COL1) AS _0) | ON EMPTY NULL | MAP (coalesce_long(_._0._0, promote(0l AS LONG)) AS _0)"
+      # Cannot run with FORCE_CONTINUATIONS due to: https://github.com/FoundationDB/fdb-record-layer/issues/3206
+      - maxRows: 0
       - result: [{2}]
     -
       - query: select count(col1) from t2 group by col2
+      # Plan deserialization previously failed : https://github.com/FoundationDB/fdb-record-layer/issues/3214
+      - supported_version: !current_version
       - explain: "ISCAN(MV5 <,>) | MAP (_ AS _0) | AGG (count(_._0.COL1) AS _0) GROUP BY (_._0.COL2 AS _0) | MAP (_._1._0 AS _0)"
       - result: [{1}, {1}]
+    -
+      # Same as above test, but tests serialization upgrades from before !current_version. Can be removed once we no longer
+      # care about upgrading to that version from older versions
+      - query: select count(col1) from t2 group by col2
+      - maxRows: 1
+      - initialVersionLessThan: !current_version
+      - result: [{1}]
+      - result: [{1}]
+      - error: 'XX000' # plan fails to deserialize on older server
+      - initialVersionAtLeast: !current_version
+      # Covered in above test case
 ...