planNestedFieldChild stop condition isn't quite right
The old planner optimization in #1143 can get confused about whether it can continue with more keys / conditions after the nested child, generating plans that use only part of the nested key's fields followed by non-nested, even though these follow some unbound nested keys.
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/RecordQueryPlanner.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/RecordQueryPlanner.java
index e571b004..54509dd7 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/RecordQueryPlanner.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/RecordQueryPlanner.java
@@ -1827,6 +1827,10 @@ public class RecordQueryPlanner implements QueryPlanner {
          * True if the current child of the index {@link ThenKeyExpression Then} clause has a corresponding equality comparison in the filter.
          */
         private boolean foundComparison;
+        /**
+         * True if {@code foundComparison} completely accounted for the child.
+         */
+        private boolean foundCompleteComparison;
         /**
          * Accumulate matching comparisons here.
          */
@@ -1881,8 +1885,8 @@ public class RecordQueryPlanner implements QueryPlanner {
             for (KeyExpression child : indexChildren) {
                 if (!doneComparing) {
                     planChild(child);
-                    if (!comparisons.isEquality() || !foundComparison) {
-                        // Didn't add another equality; done matching filters to index.
+                    if (!comparisons.isEquality() || !foundCompleteComparison) {
+                        // Didn't add another equality or only did part of child; done matching filters to index.
                         doneComparing = true;
                     }
                 }
@@ -1941,12 +1945,13 @@ public class RecordQueryPlanner implements QueryPlanner {
         }
 
         private void planChild(@Nonnull KeyExpression child) {
-            foundComparison = false;
+            foundCompleteComparison = foundComparison = false;
             if (child instanceof RecordTypeKeyExpression) {
                 if (candidateScan.planContext.query.getRecordTypes().size() == 1) {
                     // Can scan just the one requested record type.
                     final RecordTypeKeyComparison recordTypeKeyComparison = new RecordTypeKeyComparison(candidateScan.planContext.query.getRecordTypes().iterator().next());
                     addToComparisons(recordTypeKeyComparison.getComparison());
+                    foundCompleteComparison = true;
                 }
                 return;
             }
@@ -1971,10 +1976,14 @@ public class RecordQueryPlanner implements QueryPlanner {
                     }
                     if (nestedFilters.size() > 1) {
                         final NestedField nestedAnd = new NestedField(parent.getFieldName(), Query.and(nestedChildren));
-                        if (planNestedFieldChild(child, nestedAnd, null)) {
-                            unsatisfiedFilters.removeAll(nestedFilters);
+                        final List<QueryComponent> saveUnsatisfiedFilters = new ArrayList<>(unsatisfiedFilters);
+                        unsatisfiedFilters.removeAll(nestedFilters);
+                        unsatisfiedFilters.add(nestedAnd);
+                        if (planNestedFieldChild(child, nestedAnd, nestedAnd)) {
                             return;
                         }
+                        unsatisfiedFilters.clear();
+                        unsatisfiedFilters.addAll(saveUnsatisfiedFilters);
                     }
                 }
             }
@@ -1998,7 +2007,7 @@ public class RecordQueryPlanner implements QueryPlanner {
             }
         }
 
-        private boolean planNestedFieldChild(@Nonnull KeyExpression child, @Nonnull NestedField filterField, @Nullable QueryComponent filterChild) {
+        private boolean planNestedFieldChild(@Nonnull KeyExpression child, @Nonnull NestedField filterField, @Nonnull QueryComponent filterChild) {
             ScoredPlan scoredPlan = planNestedField(candidateScan, child, filterField, null);
             ScanComparisons nextComparisons = getPlanComparisons(scoredPlan);
             if (nextComparisons != null) {
@@ -2014,15 +2023,17 @@ public class RecordQueryPlanner implements QueryPlanner {
                     } else if (currentSort != null) {
                         // Didn't plan to equality, need to try with sorting.
                         scoredPlan = planNestedField(candidateScan, child, filterField, currentSort);
+                        if (scoredPlan != null) {
+                            advanceCurrentSort();
+                        }
                     }
                     if (scoredPlan != null) {
-                        if (filterChild != null) {
-                            unsatisfiedFilters.remove(filterChild);
-                        }
+                        unsatisfiedFilters.remove(filterChild);
                         unsatisfiedFilters.addAll(scoredPlan.unsatisfiedFilters);
                         comparisons.addAll(nextComparisons);
                         if (nextComparisons.isEquality()) {
                             foundComparison = true;
+                            foundCompleteComparison = nextComparisons.getEqualitySize() == child.getColumnSize();
                         }
                         return true;
                     }
@@ -2053,10 +2064,7 @@ public class RecordQueryPlanner implements QueryPlanner {
                 FieldKeyExpression indexField = (FieldKeyExpression) child;
                 if (Objects.equals(field.getFieldName(), indexField.getFieldName())) {
                     if (addToComparisons(field.getComparison())) {
-                        unsatisfiedFilters.remove(filterChild);
-                        if (foundComparison && currentSortMatches(child)) {
-                            advanceCurrentSort();
-                        }
+                        addedComparison(child, filterChild);
                     }
                 }
             }
@@ -2065,10 +2073,7 @@ public class RecordQueryPlanner implements QueryPlanner {
         private void planWithComparisonChild(@Nonnull KeyExpression child, @Nonnull QueryKeyExpressionWithComparison queryKeyExpression, @Nonnull QueryComponent filterChild) {
             if (child.equals(queryKeyExpression.getKeyExpression())) {
                 if (addToComparisons(queryKeyExpression.getComparison())) {
-                    unsatisfiedFilters.remove(filterChild);
-                    if (foundComparison && currentSortMatches(child)) {
-                        advanceCurrentSort();
-                    }
+                    addedComparison(child, filterChild);
                 }
             }
         }
@@ -2078,10 +2083,7 @@ public class RecordQueryPlanner implements QueryPlanner {
                 FieldKeyExpression indexField = (FieldKeyExpression) child;
                 if (Objects.equals(oneOfThem.getFieldName(), indexField.getFieldName()) && indexField.getFanType() == FanType.FanOut) {
                     if (addToComparisons(oneOfThem.getComparison())) {
-                        unsatisfiedFilters.remove(filterChild);
-                        if (foundComparison && currentSortMatches(child)) {
-                            advanceCurrentSort();
-                        }
+                        addedComparison(child, filterChild);
                     }
                 }
             }
@@ -2090,10 +2092,7 @@ public class RecordQueryPlanner implements QueryPlanner {
         private void planWithVersionComparisonChild(@Nonnull KeyExpression child, @Nonnull QueryRecordFunctionWithComparison filter, @Nonnull QueryComponent filterChild) {
             if (child instanceof VersionKeyExpression) {
                 if (addToComparisons(filter.getComparison())) {
-                    unsatisfiedFilters.remove(filterChild);
-                    if (foundComparison && currentSortMatches(child)) {
-                        advanceCurrentSort();
-                    }
+                    addedComparison(child, filterChild);
                 }
             }
         }
@@ -2119,6 +2118,16 @@ public class RecordQueryPlanner implements QueryPlanner {
             return false;
         }
 
+        private void addedComparison(@Nonnull KeyExpression child, @Nonnull QueryComponent filterChild) {
+            unsatisfiedFilters.remove(filterChild);
+            if (foundComparison) {
+                foundCompleteComparison = true;
+                if (currentSortMatches(child)) {
+                    advanceCurrentSort();
+                }
+            }
+        }
+
     }
 
 }
diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/FDBNestedFieldQueryTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/FDBNestedFieldQueryTest.java
index 12eca779..4e30d392 100644
--- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/FDBNestedFieldQueryTest.java
+++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/FDBNestedFieldQueryTest.java
@@ -69,8 +69,10 @@ import static com.apple.foundationdb.record.query.plan.match.PlanMatchers.indexS
 import static com.apple.foundationdb.record.query.plan.match.PlanMatchers.primaryKeyDistinct;
 import static com.apple.foundationdb.record.query.plan.match.PlanMatchers.queryPredicateDescendant;
 import static com.apple.foundationdb.record.query.plan.match.PlanMatchers.scan;
+import static com.apple.foundationdb.record.query.plan.match.PlanMatchers.typeFilter;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.allOf;
+import static org.hamcrest.Matchers.contains;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -246,6 +248,33 @@ public class FDBNestedFieldQueryTest extends FDBRecordStoreQueryTestBase {
         assertEquals(Collections.singletonList(101L), fetchResultValues(plan, TestRecords4Proto.RestaurantRecord.REST_NO_FIELD_NUMBER,
                 this::openNestedRecordStore,
                 TestHelpers::assertDiscardedNone));
+
+        QueryComponent reviewFilter = Query.field("reviews").oneOfThem().matches(Query.and(
+                Query.field("rating").equalsValue(5),
+                Query.field("reviewer").equalsValue(1L)));
+        query = RecordQuery.newBuilder()
+                .setRecordType("RestaurantRecord")
+                .setFilter(reviewFilter)
+                .build();
+        plan = planner.plan(query);
+        if (planner instanceof RecordQueryPlanner) {
+            assertThat(plan, filter(reviewFilter, primaryKeyDistinct(indexScan(allOf(indexName("review_rating"), bounds(hasTupleString("[[5],[5]]")))))));
+            assertEquals(1252155441, plan.planHash(PlanHashable.PlanHashKind.LEGACY));
+            assertEquals(-1754925686, plan.planHash(PlanHashable.PlanHashKind.FOR_CONTINUATION));
+            assertEquals(1387591835, plan.planHash(PlanHashable.PlanHashKind.STRUCTURAL_WITHOUT_LITERALS));
+            assertEquals(Collections.singletonList(102L), fetchResultValues(plan, TestRecords4Proto.RestaurantRecord.REST_NO_FIELD_NUMBER,
+                    this::openNestedRecordStore,
+                    TestHelpers::assertDiscardedNone));
+        } else {
+            // TODO: Costing issue with full scan versus index scan.
+            assertThat(plan, filter(reviewFilter, typeFilter(contains("RestaurantRecord"), scan())));
+            assertEquals(277825167, plan.planHash(PlanHashable.PlanHashKind.LEGACY));
+            assertEquals(49070805, plan.planHash(PlanHashable.PlanHashKind.FOR_CONTINUATION));
+            assertEquals(-1447620295, plan.planHash(PlanHashable.PlanHashKind.STRUCTURAL_WITHOUT_LITERALS));
+            assertEquals(Collections.singletonList(102L), fetchResultValues(plan, TestRecords4Proto.RestaurantRecord.REST_NO_FIELD_NUMBER,
+                    this::openNestedRecordStore,
+                    context -> TestHelpers.assertDiscardedAtMost(3, context)));
+        }
     }
 
     /**
@@ -436,6 +465,41 @@ public class FDBNestedFieldQueryTest extends FDBRecordStoreQueryTestBase {
         assertEquals(plan, planner.plan(query));
     }
 
+    /**
+     * Verify that matching part of a nested field only uses part of the index.
+     */
+    @DualPlannerTest
+    public void nestedThenWithAndPartial() throws Exception {
+        final RecordMetaDataHook hook = metaData -> {
+            metaData.addIndex("RestaurantReviewer", "hometownEmail", concat(field("stats").nest(concatenateFields("hometown", "school_name", "start_date")), field("email")));
+        };
+        nestedWithAndSetup(hook);
+
+        RecordQuery query = RecordQuery.newBuilder()
+                .setRecordType("RestaurantReviewer")
+                .setFilter(Query.and(
+                        Query.field("stats").matches(Query.and(
+                                Query.field("hometown").equalsValue("Home Town"),
+                                Query.field("school_name").equalsValue("University of Learning"))),
+                        Query.field("email").equalsValue("pmp@example.com")))
+                .build();
+        RecordQueryPlan plan = planner.plan(query);
+        assertThat(plan, filter(Query.field("email").equalsValue("pmp@example.com"),
+                indexScan(allOf(indexName("hometownEmail"), bounds(hasTupleString("[[Home Town, University of Learning],[Home Town, University of Learning]]"))))));
+        if (planner instanceof RecordQueryPlanner) {
+            assertEquals(895882018, plan.planHash(PlanHashable.PlanHashKind.LEGACY));
+            assertEquals(1929345776, plan.planHash(PlanHashable.PlanHashKind.FOR_CONTINUATION));
+            assertEquals(391991162, plan.planHash(PlanHashable.PlanHashKind.STRUCTURAL_WITHOUT_LITERALS));
+        } else {
+            assertEquals(-1385621911, plan.planHash(PlanHashable.PlanHashKind.LEGACY));
+            assertEquals(-1566008386, plan.planHash(PlanHashable.PlanHashKind.FOR_CONTINUATION));
+            assertEquals(1191604296, plan.planHash(PlanHashable.PlanHashKind.STRUCTURAL_WITHOUT_LITERALS));
+        }
+        assertEquals(Collections.singletonList(1L), fetchResultValues(plan, TestRecords4Proto.RestaurantReviewer.ID_FIELD_NUMBER,
+                context -> openNestedRecordStore(context, hook),
+                TestHelpers::assertDiscardedNone));
+    }
+
     /**
      * Verify that an AND query on a nested record store that can be mostly implemented by a scan of a concatenated index
      * still filters on predicates that are not satisfied by scanning that index.
diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/match/FilterMatcherWithComponent.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/match/FilterMatcherWithComponent.java
index 6cb6be61..b2ddf367 100644
--- a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/match/FilterMatcherWithComponent.java
+++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/match/FilterMatcherWithComponent.java
@@ -86,7 +86,7 @@ public class FilterMatcherWithComponent extends PlanMatcherWithChild {
                 return predicate.semanticEquals(componentAsPredicateSupplier.get(), AliasMap.of(planBaseAlias, baseAlias))
                        && super.matchesSafely(plan);
             } else if (predicate instanceof QueryComponentPredicate) {
-                return component.equals(((QueryComponentPredicate)predicate).getQueryComponent());
+                return component.equals(((QueryComponentPredicate)predicate).getQueryComponent()) && super.matchesSafely(plan);
             } else {
                 return false;
             }
['fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/query/plan/match/FilterMatcherWithComponent.java', 'fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/FDBNestedFieldQueryTest.java', 'fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/RecordQueryPlanner.java']