Serialize missing field createDefaultIfEmpty into streaming aggregate plan continuations (#3211)

There is missing state that was introduced to the `AggregateCursor` and
its accompanying plan in #3092. However, this state was not added to the
continuation, which results in unexpected behavior when we try to
deserialize the plan's continuation (effectively, it reverts to the old
behavior of the plan operator). We were already correctly
_de_serializing this information, so the upshot is that any plan that is
started after this current version should exhibit the correct behavior,
but we get incorrect behavior when a plan is started on an older
version.

This fixes #3096. I was able to get test coverage of this part of the
code by enabling `FORCE_CONTINUATIONS` mode on `aggregateIndexTests`.

Author: alecgrieser

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

@@ -370,7 +370,8 @@ public PRecordQueryStreamingAggregationPlan toProto(@Nonnull final PlanSerializa
         }
         builder.setGroupingKeyAlias(groupingKeyAlias.getId())
                 .setAggregateAlias(aggregateAlias.getId())
-                .setCompleteResultValue(completeResultValue.toValueProto(serializationContext));
+                .setCompleteResultValue(completeResultValue.toValueProto(serializationContext))
+                .setIsCreateDefaultOnEmpty(isCreateDefaultOnEmpty);
         return builder.build();
     }
 

yaml-tests/src/test/java/YamlIntegrationTests.java

@@ -129,8 +129,6 @@ public void createDropCreateTemplate(YamlTest.Runner runner) throws Exception {
     }
 
     @TestTemplate
-    @ExcludeYamlTestConfig(value = YamlTestConfigFilters.DO_NOT_FORCE_CONTINUATIONS,
-            reason = "Continuation verification (https://github.com/FoundationDB/fdb-record-layer/issues/3096)")
     public void aggregateIndexTests(YamlTest.Runner runner) throws Exception {
         runner.runYamsql("aggregate-index-tests.yamsql");
     }

yaml-tests/src/test/resources/aggregate-index-tests.yamsql

@@ -132,8 +132,20 @@ test_block:
        # At some point, should be able to roll up values from the aggregate index. However, even
        # controlling for that, it can still use the index
        - query: select max(col2) from T1 use index (mv8);
+       - supported_version: !current_version
        - explain: "ISCAN(MV8 <,>) | MAP (_ AS _0) | AGG (max_l(_._0.COL2) AS _0) | ON EMPTY NULL | MAP (_._0._0 AS _0)"
        - result: [{!l 13}]
+    -
+       - query: select max(col2) from T1 use index (mv8);
+       - maxRows: 1
+       # Cannot use FORCE_CONTINUATIONS with older versions due to: https://github.com/FoundationDB/fdb-record-layer/issues/3096
+       # (Extra values being produced after exhausting source of an aggregate cursor)
+       # Can remove once we do not care about backwards compatibility before !current_version
+       - initialVersionLessThan: !current_version
+       # Different (incorrect) behavior for different past versions
+       - initialVersionAtLeast: !current_version
+       - result: [{!l 13}]
+       - result: []
     -
        # Min/max indexes need keep what amounts to a standard value index on their keys (in order to properly look up
        # the min/max). That index should be usable for normal queries just like a value index. Note that the scan is
@@ -147,14 +159,43 @@ test_block:
        - result: [{!l 5}, {!l 4}, {!l 3}, {!l 2}, {!l 1}]
     -
       - query: select min(col3) from T2 group by col1, col2;
+      - supported_version: !current_version
       - explain: "ISCAN(MV2 <,>) | MAP (_ AS _0) | AGG (min_l(_._0.COL3) AS _0) GROUP BY (_._0.COL1 AS _0, _._0.COL2 AS _1) | MAP (_._1._0 AS _0)"
       - result: [{!l 1}, {!l 2}, {!l 3}]
+    -
+      - query: select min(col3) from T2 group by col1, col2;
+       # Cannot use FORCE_CONTINUATIONS with older versions due to: https://github.com/FoundationDB/fdb-record-layer/issues/3096
+       # (Extra values being produced after exhausting source of an aggregate cursor)
+       # Can remove once we do not care about backwards compatibility before !current_version
+      - maxRows: 1
+      - initialVersionLessThan: !current_version
+       # Different (incorrect) behavior for different past versions
+      - initialVersionAtLeast: !current_version
+      - result: [{!l 1}]
+      - result: [{!l 2}]
+      - result: [{!l 3}]
+      - result: []
     -
       # this should use the aggregate index in the future, for now, it is using streaming aggregate
       # over base table scan.
       - query: select max(col2) from t2;
+      - supported_version: !current_version
       - explain: "ISCAN(MV3 <,>) | MAP (_ AS _0) | AGG (max_l(_._0.COL2) AS _0) | ON EMPTY NULL | MAP (_._0._0 AS _0)"
       - result: [{!l 2}]
+    -
+      - query: select max(col2) from t2;
+       # Cannot use FORCE_CONTINUATIONS with older versions due to: https://github.com/FoundationDB/fdb-record-layer/issues/3096
+       # (Extra values being produced after exhausting source of an aggregate cursor)
+       # Can remove once we do not care about backwards compatibility before !current_version
+      - supported_version: !current_version
+      - maxRows: 1
+      - initialVersionLessThan: !current_version
+      - result: [{!l 2}]
+      - result: [{!null _}]
+      - result: [{!l 2}] # ad infinitum
+      - initialVersionAtLeast: !current_version
+      - result: [{!l 2}]
+      - result: []
     -
       - query: select col1, sum(col2) from T1 USE INDEX (vi1) group by col1;
       - explain: "ISCAN(VI1 <,>) | MAP (_ AS _0) | AGG (sum_l(_._0.COL2) AS _0) GROUP BY (_._0.COL1 AS _0) | MAP (_._0._0 AS COL1, _._1._0 AS _1)"
@@ -208,12 +249,28 @@ test_block:
     -
       # Permuted max index can also be used to evaluate other aggregate functions via aggregation and roll-up
       - query: select col3, sum(col2) as s from t2 use index (mv9) where col1 = 1 group by col1, col3 order by col3 asc;
+      - supported_version: !current_version
       - explain: "ISCAN(MV9 [EQUALS promote(@c20 AS LONG)]) | MAP (_ AS _0) | AGG (sum_l(_._0.COL2) AS _0) GROUP BY (_._0.COL1 AS _0, _._0.COL3 AS _1) | MAP (_._0._1 AS COL3, _._1._0 AS S)"
       - result: [{COL3: 1, S: 1}, {COL3: 2, S: 2}, {COL3: 100, S: 1}, {COL3: 200, S: 2}]
+    -
+      - query: select col3, sum(col2) as s from t2 use index (mv9) where col1 = 1 group by col1, col3 order by col3 asc;
+      # Cannot use FORCE_CONTINUATIONS with older versions due to: https://github.com/FoundationDB/fdb-record-layer/issues/3096
+      # (Extra values being produced after exhausting source of an aggregate cursor)
+      # Can remove once we do not care about backwards compatibility before !current_version
+      - maxRows: 0
+      - result: [{COL3: 1, S: 1}, {COL3: 2, S: 2}, {COL3: 100, S: 1}, {COL3: 200, S: 2}]
     -
       - query: select col3, sum(col2) as s from t2 use index (mv9) where col1 = 1 group by col1, col3 order by col3 desc;
+      - supported_version: !current_version
       - explain: "ISCAN(MV9 [EQUALS promote(@c20 AS LONG)] REVERSE) | MAP (_ AS _0) | AGG (sum_l(_._0.COL2) AS _0) GROUP BY (_._0.COL1 AS _0, _._0.COL3 AS _1) | MAP (_._0._1 AS COL3, _._1._0 AS S)"
       - result: [{COL3: 200, S: 2}, {COL3: 100, S: 1}, {COL3: 2, S: 2}, {COL3: 1, S: 1}]
+    -
+      - query: select col3, sum(col2) as s from t2 use index (mv9) where col1 = 1 group by col1, col3 order by col3 desc;
+      # Cannot use FORCE_CONTINUATIONS with older versions due to: https://github.com/FoundationDB/fdb-record-layer/issues/3096
+      # (Extra values being produced after exhausting source of an aggregate cursor)
+      # Can remove once we do not care about backwards compatibility before !current_version
+      - maxRows: 0
+      - result: [{COL3: 200, S: 2}, {COL3: 100, S: 1}, {COL3: 2, S: 2}, {COL3: 1, S: 1}]
 #    -
 #      # grouping by constant is not yet supported.
 #      - query: select sum(col2) from t1 group by 3,2,1;
@@ -230,6 +287,9 @@ test_block:
       - result: [{!l 100}, {!l 200}, {!l 400}]
     -
       - query: select min_ever(col3) from t2
+       # Cannot enable FORCE_CONTINUATIONS with ungrouped aggregate scan because of: https://github.com/FoundationDB/fdb-record-layer/issues/3206
+      - maxRows: 0
+      - explain: "AISCAN(MV7 <,> BY_GROUP -> [_0: VALUE:[0]]) | MAP (_ AS _0) | ON EMPTY NULL | MAP (_._0._0 AS _0)"
       - result: [{!l 1}]
     -
       - query: select min_ever(col3) from t2