diff --git a/docs/ReleaseNotes.md b/docs/ReleaseNotes.md
index 2ccd8fe8fb..33d7893ba0 100644
--- a/docs/ReleaseNotes.md
+++ b/docs/ReleaseNotes.md
@@ -31,7 +31,7 @@ Starting with version [3.4.455.0](#344550), the semantics of `UnnestedRecordType
 * **Feature** Feature 2 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
 * **Feature** Feature 3 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
 * **Feature** Feature 4 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
-* **Feature** Feature 5 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
+* **Feature** Add a RecordCursor that flattens recursive cursors from previously returned items [(Issue #2859)](https://github.com/FoundationDB/fdb-record-layer/issues/2859)
 * **Breaking change** Change 1 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
 * **Breaking change** Change 2 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
 * **Breaking change** Change 3 [(Issue #NNN)](https://github.com/FoundationDB/fdb-record-layer/issues/NNN)
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/cursors/RecursiveCursor.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/cursors/RecursiveCursor.java
new file mode 100644
index 0000000000..92289a5cd0
--- /dev/null
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/cursors/RecursiveCursor.java
@@ -0,0 +1,458 @@
+/*
+ * RecursiveCursor.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2024 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.cursors;
+
+import com.apple.foundationdb.annotation.API;
+import com.apple.foundationdb.async.AsyncUtil;
+import com.apple.foundationdb.record.ByteArrayContinuation;
+import com.apple.foundationdb.record.RecordCoreException;
+import com.apple.foundationdb.record.RecordCursor;
+import com.apple.foundationdb.record.RecordCursorContinuation;
+import com.apple.foundationdb.record.RecordCursorProto;
+import com.apple.foundationdb.record.RecordCursorResult;
+import com.apple.foundationdb.record.RecordCursorStartContinuation;
+import com.apple.foundationdb.record.RecordCursorVisitor;
+import com.apple.foundationdb.tuple.ByteArrayUtil2;
+import com.google.protobuf.ByteString;
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.ZeroCopyByteString;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
+import java.util.function.Function;
+
+/**
+ * A cursor that flattens a tree of cursors.
+ * A root cursor seeds the output and then each output element is additionally mapped into a child cursor.
+ * @param <T> the type of elements of the cursors
+ */
+@API(API.Status.EXPERIMENTAL)
+@SuppressWarnings("PMD.CloseResource")
+public class RecursiveCursor<T> implements RecordCursor<RecursiveCursor.RecursiveValue<T>> {
+
+    @Nonnull
+    private final ChildCursorFunction<T> childCursorFunction;
+    @Nullable
+    private final Function<T, byte[]> checkValueFunction;
+    @Nonnull
+    private final List<RecursiveNode<T>> nodes;
+
+    private int currentDepth;
+
+    @Nullable
+    private RecordCursorResult<RecursiveValue<T>> lastResult;
+
+    private RecursiveCursor(@Nonnull ChildCursorFunction<T> childCursorFunction,
+                            @Nullable Function<T, byte[]> checkValueFunction,
+                            @Nonnull List<RecursiveNode<T>> nodes) {
+        this.childCursorFunction = childCursorFunction;
+        this.checkValueFunction = checkValueFunction;
+        this.nodes = nodes;
+    }
+
+    /**
+     * Create a recursive cursor.
+     * @param rootCursorFunction a function that given a continuation or {@code null} returns the children of the root
+     * @param childCursorFunction a function that given a value and a continuation returns the children of that level
+     * @param checkValueFunction a function to recognize changes to the database since a continuation
+     * @param continuation a continuation from a previous cursor
+     * @param <T> the type of elements of the cursors
+     * @return a cursor over the recursive tree determined by the cursor functions
+     */
+    @Nonnull
+    public static <T> RecursiveCursor<T> create(@Nonnull Function<byte[], ? extends RecordCursor<T>> rootCursorFunction,
+                                                @Nonnull ChildCursorFunction<T> childCursorFunction,
+                                                @Nullable Function<T, byte[]> checkValueFunction,
+                                                @Nullable byte[] continuation) {
+        final List<RecursiveNode<T>> nodes = new ArrayList<>();
+        if (continuation == null) {
+            nodes.add(RecursiveNode.forRoot(RecordCursorStartContinuation.START, rootCursorFunction.apply(null)));
+        } else {
+            RecordCursorProto.RecursiveContinuation parsed;
+            try {
+                parsed = RecordCursorProto.RecursiveContinuation.parseFrom(continuation);
+            } catch (InvalidProtocolBufferException ex) {
+                throw new RecordCoreException("error parsing continuation", ex)
+                        .addLogInfo("raw_bytes", ByteArrayUtil2.loggable(continuation));
+            }
+            final int totalDepth = parsed.getLevelsCount();
+            for (int depth = 0; depth < totalDepth; depth++) {
+                final RecordCursorProto.RecursiveContinuation.LevelCursor parentLevel = parsed.getLevels(depth);
+                final RecordCursorContinuation priorContinuation;
+                if (parentLevel.hasContinuation()) {
+                    priorContinuation = ByteArrayContinuation.fromNullable(parentLevel.getContinuation().toByteArray());
+                } else {
+                    priorContinuation = RecordCursorStartContinuation.START;
+                }
+                if (depth == 0) {
+                    nodes.add(RecursiveNode.forRoot(priorContinuation, rootCursorFunction.apply(priorContinuation.toBytes())));
+                } else {
+                    byte[] checkValue = null;
+                    if (checkValueFunction != null && depth < totalDepth - 1) {
+                        final RecordCursorProto.RecursiveContinuation.LevelCursor childLevel = parsed.getLevels(depth + 1);
+                        if (childLevel.hasCheckValue()) {
+                            checkValue = childLevel.getCheckValue().toByteArray();
+                        }
+                    }
+                    nodes.add(RecursiveNode.forContinuation(priorContinuation, checkValue));
+                }
+            }
+        }
+        return new RecursiveCursor<>(childCursorFunction, checkValueFunction, nodes);
+    }
+
+    @Nonnull
+    @Override
+    public CompletableFuture<RecordCursorResult<RecursiveValue<T>>> onNext() {
+        if (lastResult != null && !lastResult.hasNext()) {
+            return CompletableFuture.completedFuture(lastResult);
+        }
+        return AsyncUtil.whileTrue(this::recursionLoop, getExecutor()).thenApply(vignore -> lastResult);
+    }
+
+    @Override
+    public void close() {
+        for (RecursiveNode<T> node : nodes) {
+            CompletableFuture<RecordCursorResult<T>> childFuture = node.childFuture;
+            if (childFuture != null) {
+                if (childFuture.cancel(false)) {
+                    node.childFuture = null;
+                } else {
+                    continue;
+                }
+            }
+            RecordCursor<T> childCursor = node.childCursor;
+            if (childCursor != null) {
+                childCursor.close();
+            }
+        }
+    }
+
+    @Override
+    public boolean isClosed() {
+        for (RecursiveNode<T> node : nodes) {
+            if (node.childFuture != null) {
+                return false;
+            }
+            RecordCursor<T> childCursor = node.childCursor;
+            if (childCursor != null && !childCursor.isClosed()) {
+                return false;
+            }
+        }
+        return true;
+    }
+
+    @Nonnull
+    @Override
+    public Executor getExecutor() {
+        return nodes.get(0).childCursor.getExecutor();  // Take from the root cursor.
+    }
+
+    @Override
+    public boolean accept(@Nonnull RecordCursorVisitor visitor) {
+        if (visitor.visitEnter(this)) {
+            for (RecursiveNode<T> node : nodes) {
+                RecordCursor<T> childCursor = node.childCursor;
+                if (childCursor != null) {
+                    childCursor.accept(visitor);
+                }
+            }
+        }
+        return visitor.visitLeave(this);
+    }
+
+    /**
+     * A value returned by a recursive descent.
+     * Includes the depth level and whether there are any descendants.
+     * @param <T> the type of elements of the cursors
+     */
+    public static class RecursiveValue<T> {
+        @Nullable
+        private final T value;
+        private final int depth;
+        private final boolean isLeaf;
+
+        public RecursiveValue(@Nullable T value, int depth, boolean isLeaf) {
+            this.value = value;
+            this.depth = depth;
+            this.isLeaf = isLeaf;
+        }
+
+        /**
+         * Get the value for a recursive result.
+         * @return the value associated with the given result
+         */
+        @Nullable
+        public T getValue() {
+            return value;
+        }
+
+        /**
+         * Get the depth for a recursive result.
+         * @return the 1-based depth from the root of the given result
+         */
+        public int getDepth() {
+            return depth;
+        }
+
+        /**
+         * Get whether a recursive result has any descendants.
+         * @return {@code true} if the given result is a leaf, {@code false} if it has any descendants
+         */
+        public boolean isLeaf() {
+            return isLeaf;
+        }
+    }
+
+    /**
+     * Function to generate children of a parent value.
+     * @param <T> the type of elements of the cursors
+     */
+    @FunctionalInterface
+    public interface ChildCursorFunction<T> {
+        /**
+         * Return recursion cursor for this level.
+         * @param value the value at this level
+         * @param depth the 1-based depth of this level
+         * @param continuation an optional continuation
+         * @return a cursor over children of the given value for the given level
+         */
+        RecordCursor<T> apply(@Nullable T value, int depth, @Nullable byte[] continuation);
+    }
+
+    // This implementation keeps a single stack of open cursors and returns in pre-order.
+    // A child is opened and the first element checked in order to be able to include whether a node is a leaf in each result.
+    // It would also be possible to keep a tree of open cursors, opening more children before completing their siblings.
+    // It would also be possible to only have a single open cursor and return in level-order.
+    // These alternatives have even more complicated continuation restoring behavior, though.
+
+    static class RecursiveNode<T> {
+        @Nullable
+        final T value;
+        @Nullable
+        final byte[] checkValue;
+
+        boolean emitPending;
+
+        @Nonnull
+        RecordCursorContinuation childContinuationBefore;
+        @Nonnull
+        RecordCursorContinuation childContinuationAfter;
+        @Nullable
+        RecordCursor<T> childCursor;
+        @Nullable
+        CompletableFuture<RecordCursorResult<T>> childFuture;
+
+        private RecursiveNode(@Nullable T value, @Nullable byte[] checkValue, boolean emitPending,
+                              @Nonnull RecordCursorContinuation childContinuationBefore, @Nullable RecordCursor<T> childCursor) {
+            this.value = value;
+            this.checkValue = checkValue;
+            this.emitPending = emitPending;
+            this.childContinuationAfter = this.childContinuationBefore = childContinuationBefore;
+            this.childCursor = childCursor;
+        }
+
+        static <T> RecursiveNode<T> forRoot(@Nonnull RecordCursorContinuation childContinuationBefore,
+                                            @Nonnull RecordCursor<T> childCursor) {
+            return new RecursiveNode<>(null, null, false, childContinuationBefore, childCursor);
+        }
+
+        static <T> RecursiveNode<T> forValue(@Nullable T value) {
+            return new RecursiveNode<>(value, null, true, RecordCursorStartContinuation.START, null);
+        }
+
+        public static <T> RecursiveNode<T> forContinuation(@Nonnull RecordCursorContinuation childContinuationBefore,
+                                                           @Nullable byte[] checkValue) {
+            return new RecursiveNode<>(null, checkValue, false, childContinuationBefore, null);
+        }
+
+        public RecursiveNode<T> withCheckedValue(@Nullable T value) {
+            return new RecursiveNode<>(value, null, false, childContinuationBefore, null);
+        }
+    }
+
+    /**
+     * Called to advance the recursion.
+     * @return a future that completes to {@code true} if the loop should continue or {@code false} if a result is available
+     */
+    @Nonnull
+    CompletableFuture<Boolean> recursionLoop() {
+        int depth = currentDepth;
+        final RecursiveNode<T> node = nodes.get(depth);
+        if (node.childFuture == null) {
+            if (node.childCursor == null) {
+                node.childCursor = childCursorFunction.apply(node.value, depth, node.childContinuationBefore.toBytes());
+            }
+            node.childFuture = node.childCursor.onNext();
+        }
+        if (!node.childFuture.isDone()) {
+            return node.childFuture.thenApply(rignore -> true);
+        }
+        final RecordCursorResult<T> childResult = node.childFuture.join();
+        node.childFuture = null;
+        if (childResult.hasNext()) {
+            node.childContinuationAfter = childResult.getContinuation();
+            addChildNode(childResult.get());
+            if (node.emitPending) {
+                lastResult = RecordCursorResult.withNextValue(
+                        new RecursiveValue<>(node.value, depth, false),
+                        buildContinuation(depth));
+                node.emitPending = false;
+                return AsyncUtil.READY_FALSE;
+            }
+        } else {
+            final NoNextReason noNextReason = childResult.getNoNextReason();
+            if (noNextReason.isOutOfBand()) {
+                final RecordCursorContinuation continuation;
+                if (node.emitPending) {
+                    // Stop before parent.
+                    continuation = buildContinuation(depth - 1);
+                } else {
+                    // Stop before child.
+                    continuation = buildContinuation(depth);
+                }
+                lastResult = RecordCursorResult.withoutNextValue(continuation, noNextReason);
+                return AsyncUtil.READY_FALSE;
+            }
+            // If the childCursorFunction added a returned row limit, that is not distinguished here.
+            // There is no provision for continuing and adding more descendants at an arbitrary depth.
+            while (nodes.size() > depth) {
+                nodes.remove(depth);
+            }
+            currentDepth = depth - 1;
+            if (depth == 0) {
+                lastResult = RecordCursorResult.exhausted();
+                return AsyncUtil.READY_FALSE;
+            }
+            final RecursiveNode<T> parentNode = nodes.get(depth - 1);
+            parentNode.childContinuationBefore = parentNode.childContinuationAfter;
+            if (node.emitPending) {
+                lastResult = RecordCursorResult.withNextValue(
+                        new RecursiveValue<>(node.value, depth, true),
+                        buildContinuation(depth));
+                node.emitPending = false;
+                return AsyncUtil.READY_FALSE;
+            }
+        }
+        return AsyncUtil.READY_TRUE;
+    }
+
+    private void addChildNode(@Nullable T value) {
+        currentDepth++;
+        if (currentDepth < nodes.size()) {
+            // Have a nested continuation.
+            final RecursiveNode<T> continuationChildNode = nodes.get(currentDepth);
+            boolean addNode = false;
+            if (checkValueFunction != null && continuationChildNode.checkValue != null) {
+                final byte[] actualCheckValue = checkValueFunction.apply(value);
+                if (actualCheckValue != null && !Arrays.equals(continuationChildNode.checkValue, actualCheckValue)) {
+                    // Does not match; discard proposed continuation(s).
+                    while (nodes.size() > currentDepth) {
+                        nodes.remove(currentDepth);
+                    }
+                    addNode = true;
+                }
+            }
+            if (!addNode) {
+                // Replace check value with actual value so can open cursors below there, but using the loaded continuation.
+                nodes.set(currentDepth, continuationChildNode.withCheckedValue(value));
+                return;
+            }
+        }
+        final RecursiveNode<T> childNode = RecursiveNode.forValue(value);
+        nodes.add(childNode);
+    }
+
+    private RecordCursorContinuation buildContinuation(int depth) {
+        final List<RecordCursorContinuation> continuations = new ArrayList<>(depth);
+        final List<byte[]> checkValues = checkValueFunction == null ? null : new ArrayList<>(depth - 1);
+        for (int i = 0; i < depth; i++) {
+            continuations.add(nodes.get(i).childContinuationBefore);
+            if (checkValues != null && i < depth - 1) {
+                checkValues.add(checkValueFunction.apply(nodes.get(i + 1).value));
+            }
+        }
+        return new Continuation(continuations, checkValues);
+    }
+
+    private static class Continuation implements RecordCursorContinuation {
+        @Nonnull
+        private final List<RecordCursorContinuation> continuations;
+        @Nullable
+        private final List<byte[]> checkValues;
+        @Nullable
+        private ByteString cachedByteString;
+        @Nullable
+        private byte[] cachedBytes;
+
+        private Continuation(@Nonnull List<RecordCursorContinuation> continuations, @Nullable List<byte[]> checkValues) {
+            this.continuations = continuations;
+            this.checkValues = checkValues;
+        }
+
+        @Override
+        public boolean isEnd() {
+            for (RecordCursorContinuation continuation : continuations) {
+                if (!continuation.isEnd()) {
+                    return false;
+                }
+            }
+            return true;
+        }
+
+        @Nonnull
+        @Override
+        public ByteString toByteString() {
+            if (cachedByteString == null) {
+                final RecordCursorProto.RecursiveContinuation.Builder builder = RecordCursorProto.RecursiveContinuation.newBuilder();
+                for (int i = 0; i < continuations.size(); i++) {
+                    final RecordCursorProto.RecursiveContinuation.LevelCursor.Builder levelBuilder = builder.addLevelsBuilder();
+                    final RecordCursorContinuation continuation = continuations.get(i);
+                    if (continuation.toBytes() != null) {
+                        levelBuilder.setContinuation(continuation.toByteString());
+                    }
+                    if (checkValues != null && i < checkValues.size()) {
+                        final byte[] checkValue = checkValues.get(i);
+                        if (checkValue != null) {
+                            levelBuilder.setCheckValue(ZeroCopyByteString.wrap(checkValue));
+                        }
+                    }
+                }
+                cachedByteString = builder.build().toByteString();
+            }
+            return cachedByteString;
+        }
+
+        @Nullable
+        @Override
+        public byte[] toBytes() {
+            if (cachedBytes == null) {
+                cachedBytes = toByteString().toByteArray();
+            }
+            return cachedBytes;
+        }
+    }
+}
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/PlanStringRepresentation.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/PlanStringRepresentation.java
index 89c4128c63..1d8ea85042 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/PlanStringRepresentation.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/PlanStringRepresentation.java
@@ -53,6 +53,7 @@
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPlanVisitor;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPredicatesFilterPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryRangePlan;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryRecursivePlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScanPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScoreForRankPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQuerySelectorPlan;
@@ -566,6 +567,16 @@ public PlanStringRepresentation visitSortPlan(@Nonnull RecordQuerySortPlan eleme
                 .append(element.getKey());
     }
 
+    @Nonnull
+    @Override
+    public PlanStringRepresentation visitRecursivePlan(@Nonnull final RecordQueryRecursivePlan recursivePlan) {
+        return append("recursive(")
+                .visit(recursivePlan.getRootQuantifier().getRangesOverPlan())
+                .append(", ")
+                .visit(recursivePlan.getChildQuantifier().getRangesOverPlan())
+                .append(")");
+    }
+
     @Nonnull
     @Override
     public PlanStringRepresentation visitDefault(@Nonnull RecordQueryPlan element) {
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/PlannerRuleSet.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/PlannerRuleSet.java
index b34dc52f95..174a297d59 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/PlannerRuleSet.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/PlannerRuleSet.java
@@ -37,6 +37,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementIntersectionRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementNestedLoopJoinRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementPhysicalScanRule;
+import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementRecursiveRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementSimpleSelectRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementStreamingAggregationRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.ImplementTypeFilterRule;
@@ -66,6 +67,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.rules.PushRequestedOrderingThroughGroupByRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.PushRequestedOrderingThroughInLikeSelectRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.PushRequestedOrderingThroughInsertRule;
+import com.apple.foundationdb.record.query.plan.cascades.rules.PushRequestedOrderingThroughRecursiveRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.PushRequestedOrderingThroughSelectExistentialRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.PushRequestedOrderingThroughSelectRule;
 import com.apple.foundationdb.record.query.plan.cascades.rules.PushRequestedOrderingThroughSortRule;
@@ -130,7 +132,8 @@ public class PlannerRuleSet {
             new PushRequestedOrderingThroughDeleteRule(),
             new PushRequestedOrderingThroughInsertRule(),
             new PushRequestedOrderingThroughUpdateRule(),
-            new PushRequestedOrderingThroughUniqueRule()
+            new PushRequestedOrderingThroughUniqueRule(),
+            new PushRequestedOrderingThroughRecursiveRule()
     );
 
     private static final List<CascadesRule<? extends RelationalExpression>> IMPLEMENTATION_RULES = ImmutableList.of(
@@ -167,7 +170,8 @@ public class PlannerRuleSet {
             new ImplementStreamingAggregationRule(),
             new ImplementDeleteRule(),
             new ImplementInsertRule(),
-            new ImplementUpdateRule()
+            new ImplementUpdateRule(),
+            new ImplementRecursiveRule()
     );
 
     private static final List<CascadesRule<? extends RelationalExpression>> EXPLORATION_RULES =
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/expressions/RecursiveExpression.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/expressions/RecursiveExpression.java
new file mode 100644
index 0000000000..0c2799ec58
--- /dev/null
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/expressions/RecursiveExpression.java
@@ -0,0 +1,144 @@
+/*
+ * RecursiveExpression.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2024 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.query.plan.cascades.expressions;
+
+import com.apple.foundationdb.annotation.API;
+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;
+import com.apple.foundationdb.record.query.plan.cascades.explain.InternalPlannerGraphRewritable;
+import com.apple.foundationdb.record.query.plan.cascades.explain.PlannerGraph;
+import com.apple.foundationdb.record.query.plan.cascades.values.Value;
+import com.apple.foundationdb.record.query.plan.cascades.values.Values;
+import com.apple.foundationdb.record.query.plan.cascades.values.translation.TranslationMap;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+
+import javax.annotation.Nonnull;
+import java.util.List;
+import java.util.Set;
+
+/**
+ * A recursive expression.
+ */
+@API(API.Status.EXPERIMENTAL)
+public class RecursiveExpression implements RelationalExpressionWithChildren, InternalPlannerGraphRewritable {
+    @Nonnull
+    private final Value resultValue;
+    @Nonnull
+    private final Quantifier rootQuantifier;
+    @Nonnull
+    private final Quantifier childQuantifier;
+
+    public RecursiveExpression(@Nonnull Value resultValue,
+                               @Nonnull Quantifier rootQuantifier,
+                               @Nonnull Quantifier childQuantifier) {
+        this.resultValue = resultValue;
+        this.rootQuantifier = rootQuantifier;
+        this.childQuantifier = childQuantifier;
+    }
+
+    @Nonnull
+    @Override
+    public Value getResultValue() {
+        return resultValue;
+    }
+
+    @Nonnull
+    public List<? extends Value> getResultValues() {
+        return Values.deconstructRecord(getResultValue());
+    }
+
+    @Nonnull
+    @Override
+    public List<? extends Quantifier> getQuantifiers() {
+        return List.of(rootQuantifier, childQuantifier);
+    }
+
+    @Override
+    public int getRelationalChildCount() {
+        return 2;
+    }
+
+    @Override
+    public boolean canCorrelate() {
+        return true;
+    }
+
+    @Nonnull
+    @Override
+    public Set<CorrelationIdentifier> getCorrelatedToWithoutChildren() {
+        return resultValue.getCorrelatedTo();
+    }
+
+    @Nonnull
+    @Override
+    public RecursiveExpression translateCorrelations(@Nonnull final TranslationMap translationMap, @Nonnull final List<? extends Quantifier> translatedQuantifiers) {
+        final Value translatedResultValue = resultValue.translateCorrelations(translationMap);
+        return new RecursiveExpression(translatedResultValue, translatedQuantifiers.get(0), translatedQuantifiers.get(1));
+    }
+
+    @SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
+    @Override
+    public boolean equals(final Object other) {
+        return semanticEquals(other);
+    }
+
+    @Override
+    public int hashCode() {
+        return semanticHashCode();
+    }
+
+    @Override
+    @SuppressWarnings({"UnstableApiUsage", "PMD.CompareObjectsWithEquals"})
+    public boolean equalsWithoutChildren(@Nonnull RelationalExpression otherExpression,
+                                         @Nonnull final AliasMap aliasMap) {
+        if (this == otherExpression) {
+            return true;
+        }
+        if (getClass() != otherExpression.getClass()) {
+            return false;
+        }
+
+        return semanticEqualsForResults(otherExpression, aliasMap);
+    }
+
+    @Override
+    public int hashCodeWithoutChildren() {
+        return getResultValue().hashCode();
+    }
+
+    @Nonnull
+    @Override
+    public PlannerGraph rewriteInternalPlannerGraph(@Nonnull final List<? extends PlannerGraph> childGraphs) {
+        return PlannerGraph.fromNodeAndChildGraphs(
+                new PlannerGraph.LogicalOperatorNode(this,
+                        "RECURSIVE " + resultValue,
+                        ImmutableList.of(),
+                        ImmutableMap.of()),
+                childGraphs);
+    }
+
+    @Override
+    public String toString() {
+        return "RECURSIVE " + resultValue;
+    }
+}
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/matching/structure/RelationalExpressionMatchers.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/matching/structure/RelationalExpressionMatchers.java
index 6c114b6541..b7a645815b 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/matching/structure/RelationalExpressionMatchers.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/matching/structure/RelationalExpressionMatchers.java
@@ -35,6 +35,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.expressions.LogicalUnionExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.LogicalUniqueExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.PrimaryScanExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.RecursiveExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.RelationalExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.RelationalExpressionWithPredicates;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.SelectExpression;
@@ -277,4 +278,9 @@ public static BindingMatcher<InsertExpression> insertExpression(@Nonnull final B
     public static BindingMatcher<UpdateExpression> updateExpression(@Nonnull final BindingMatcher<? extends Quantifier> downstream) {
         return ofTypeOwning(UpdateExpression.class, only(downstream));
     }
+
+    @Nonnull
+    public static BindingMatcher<RecursiveExpression> recursiveExpression(@Nonnull final CollectionMatcher<? extends Quantifier> downstream) {
+        return ofTypeOwning(RecursiveExpression.class, downstream);
+    }
 }
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/CardinalitiesProperty.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/CardinalitiesProperty.java
index f07bf9cc86..298fc2873a 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/CardinalitiesProperty.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/CardinalitiesProperty.java
@@ -45,6 +45,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.expressions.LogicalUniqueExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.MatchableSortExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.PrimaryScanExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.RecursiveExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.RelationalExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.SelectExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.UpdateExpression;
@@ -76,6 +77,7 @@
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryMapPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPredicatesFilterPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryRangePlan;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryRecursivePlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScanPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScoreForRankPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQuerySelectorPlan;
@@ -594,6 +596,18 @@ public Cardinalities visitRecordQuerySortPlan(@Nonnull final RecordQuerySortPlan
         return fromChild(querySortPlan);
     }
 
+    @Nonnull
+    @Override
+    public Cardinalities visitRecordQueryRecursivePlan(@Nonnull final RecordQueryRecursivePlan recursivePlan) {
+        return Cardinalities.unknownMaxCardinality();
+    }
+
+    @Nonnull
+    @Override
+    public Cardinalities visitRecursiveExpression(@Nonnull final RecursiveExpression element) {
+        return Cardinalities.unknownMaxCardinality();
+    }
+
     @Nonnull
     @Override
     public Cardinalities evaluateAtExpression(@Nonnull RelationalExpression expression, @Nonnull List<Cardinalities> childResults) {
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DerivationsProperty.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DerivationsProperty.java
index ef793b31b3..1a00f5d7cf 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DerivationsProperty.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DerivationsProperty.java
@@ -68,6 +68,7 @@
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPlanWithComparisons;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPredicatesFilterPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryRangePlan;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryRecursivePlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScanPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScoreForRankPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQuerySelectorPlan;
@@ -715,6 +716,12 @@ public Derivations visitSortPlan(@Nonnull final RecordQuerySortPlan sortPlan) {
             return derivationsFromSingleChild(sortPlan);
         }
 
+        @Nonnull
+        @Override
+        public Derivations visitRecursivePlan(@Nonnull final RecordQueryRecursivePlan recursivePlan) {
+            throw new RecordCoreException("unsupported plan operator");
+        }
+
         @Nonnull
         @Override
         public Derivations visitDefault(@Nonnull final RecordQueryPlan element) {
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DistinctRecordsProperty.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DistinctRecordsProperty.java
index e492a1767c..3d648da4a4 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DistinctRecordsProperty.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/DistinctRecordsProperty.java
@@ -53,6 +53,7 @@
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPlanVisitor;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPredicatesFilterPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryRangePlan;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryRecursivePlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScanPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScoreForRankPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQuerySelectorPlan;
@@ -342,6 +343,12 @@ public Boolean visitSortPlan(@Nonnull final RecordQuerySortPlan sortPlan) {
             return distinctRecordsFromSingleChild(sortPlan);
         }
 
+        @Nonnull
+        @Override
+        public Boolean visitRecursivePlan(@Nonnull final RecordQueryRecursivePlan recursivePlan) {
+            return false;
+        }
+
         @Nonnull
         @Override
         public Boolean visitDefault(@Nonnull final RecordQueryPlan element) {
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/OrderingProperty.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/OrderingProperty.java
index ff7d6dbdc1..97c11fa95e 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/OrderingProperty.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/OrderingProperty.java
@@ -65,6 +65,7 @@
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPlanVisitor;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPredicatesFilterPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryRangePlan;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryRecursivePlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScanPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScoreForRankPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQuerySelectorPlan;
@@ -621,6 +622,12 @@ public Ordering visitSortPlan(@Nonnull final RecordQuerySortPlan element) {
             return Ordering.empty();
         }
 
+        @Nonnull
+        @Override
+        public Ordering visitRecursivePlan(@Nonnull final RecordQueryRecursivePlan recursivePlan) {
+            return Ordering.empty();
+        }
+
         @Nonnull
         @Override
         public Ordering visitDefault(@Nonnull final RecordQueryPlan element) {
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/PrimaryKeyProperty.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/PrimaryKeyProperty.java
index 16fb303936..b3cd2d2eed 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/PrimaryKeyProperty.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/PrimaryKeyProperty.java
@@ -54,6 +54,7 @@
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPlanVisitor;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPredicatesFilterPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryRangePlan;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryRecursivePlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScanPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScoreForRankPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQuerySelectorPlan;
@@ -340,6 +341,15 @@ public Optional<List<Value>> visitSortPlan(@Nonnull final RecordQuerySortPlan so
             return primaryKeyFromSingleChild(sortPlan);
         }
 
+        @Nonnull
+        @Override
+        public Optional<List<Value>> visitRecursivePlan(@Nonnull final RecordQueryRecursivePlan recursivePlan) {
+            if (recursivePlan.isInheritRecordProperties()) {
+                return primaryKeyFromSingleQuantifier(recursivePlan.getChildQuantifier());
+            }
+            return Optional.empty();
+        }
+
         @Nonnull
         @Override
         public Optional<List<Value>> visitDefault(@Nonnull final RecordQueryPlan element) {
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/RecordTypesProperty.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/RecordTypesProperty.java
index 0fc095babf..4537970f77 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/RecordTypesProperty.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/RecordTypesProperty.java
@@ -31,6 +31,7 @@
 import com.apple.foundationdb.record.query.plan.cascades.expressions.FullUnorderedScanExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.LogicalUnionExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.PrimaryScanExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.RecursiveExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.RelationalExpression;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.RelationalExpressionVisitorWithDefaults;
 import com.apple.foundationdb.record.query.plan.cascades.expressions.SelectExpression;
@@ -125,7 +126,8 @@ public Set<String> evaluateAtExpression(@Nonnull RelationalExpression expression
                         expression instanceof RecordQueryUnorderedUnionPlan ||
                         expression instanceof RecordQueryIntersectionPlan ||
                         expression instanceof LogicalUnionExpression ||
-                        expression instanceof SelectExpression) {
+                        expression instanceof SelectExpression ||
+                        expression instanceof RecursiveExpression) {
                     final Set<String> union = new HashSet<>();
                     for (Set<String> childResulSet : childResults) {
                         union.addAll(childResulSet);
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/StoredRecordProperty.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/StoredRecordProperty.java
index 7eb4e59635..96027dbab0 100644
--- a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/StoredRecordProperty.java
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/properties/StoredRecordProperty.java
@@ -51,6 +51,7 @@
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPlanVisitor;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryPredicatesFilterPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryRangePlan;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryRecursivePlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScanPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQueryScoreForRankPlan;
 import com.apple.foundationdb.record.query.plan.plans.RecordQuerySelectorPlan;
@@ -327,6 +328,12 @@ public Boolean visitSortPlan(@Nonnull final RecordQuerySortPlan sortPlan) {
             return storedRecordsFromSingleChild(sortPlan);
         }
 
+        @Nonnull
+        @Override
+        public Boolean visitRecursivePlan(@Nonnull final RecordQueryRecursivePlan recursivePlan) {
+            return false;
+        }
+
         @Nonnull
         @Override
         public Boolean visitDefault(@Nonnull final RecordQueryPlan element) {
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/rules/ImplementRecursiveRule.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/rules/ImplementRecursiveRule.java
new file mode 100644
index 0000000000..2abbeec78b
--- /dev/null
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/rules/ImplementRecursiveRule.java
@@ -0,0 +1,105 @@
+/*
+ * ImplementRecursiveRule.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2024 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.query.plan.cascades.rules;
+
+import com.apple.foundationdb.annotation.API;
+import com.apple.foundationdb.record.logging.KeyValueLogMessage;
+import com.apple.foundationdb.record.query.plan.cascades.CascadesRule;
+import com.apple.foundationdb.record.query.plan.cascades.CascadesRuleCall;
+import com.apple.foundationdb.record.query.plan.cascades.PlanPartition;
+import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
+import com.apple.foundationdb.record.query.plan.cascades.Reference;
+import com.apple.foundationdb.record.query.plan.cascades.debug.Debugger;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.RecursiveExpression;
+import com.apple.foundationdb.record.query.plan.cascades.matching.structure.BindingMatcher;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryRecursivePlan;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.annotation.Nonnull;
+
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.ListMatcher.exactly;
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.MultiMatcher.all;
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.QuantifierMatchers.anyQuantifierOverRef;
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.ReferenceMatchers.anyPlanPartition;
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.ReferenceMatchers.planPartitions;
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.ReferenceMatchers.rollUp;
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.RelationalExpressionMatchers.canBeImplemented;
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.RelationalExpressionMatchers.recursiveExpression;
+
+/**
+ * A rule that implements an existential nested loop join of its (already implemented) children.
+ */
+@API(API.Status.EXPERIMENTAL)
+@SuppressWarnings("PMD.TooManyStaticImports")
+public class ImplementRecursiveRule extends CascadesRule<RecursiveExpression> {
+    @Nonnull
+    private static final Logger logger = LoggerFactory.getLogger(ImplementRecursiveRule.class);
+
+    @Nonnull
+    private static final BindingMatcher<PlanPartition> rootPlanPartitionsMatcher = anyPlanPartition();
+
+    @Nonnull
+    private static final BindingMatcher<Reference> rootReferenceMatcher = planPartitions(rollUp(all(rootPlanPartitionsMatcher)));
+    @Nonnull
+    private static final BindingMatcher<Quantifier> rootQuantifierMatcher = anyQuantifierOverRef(rootReferenceMatcher);
+    @Nonnull
+    private static final BindingMatcher<PlanPartition> childPlanPartitionsMatcher = anyPlanPartition();
+
+    @Nonnull
+    private static final BindingMatcher<Reference> childReferenceMatcher = planPartitions(rollUp(all(childPlanPartitionsMatcher)));
+    @Nonnull
+    private static final BindingMatcher<Quantifier> childQuantifierMatcher = anyQuantifierOverRef(childReferenceMatcher);
+    @Nonnull
+    private static final BindingMatcher<RecursiveExpression> root =
+            recursiveExpression(exactly(rootQuantifierMatcher, childQuantifierMatcher)).where(canBeImplemented());
+
+    public ImplementRecursiveRule() {
+        super(root);
+    }
+
+    @Override
+    public void onMatch(@Nonnull final CascadesRuleCall call) {
+        final var bindings = call.getBindings();
+        final var recursiveExpression = bindings.get(root);
+        Debugger.withDebugger(debugger -> logger.debug(KeyValueLogMessage.of("matched RecursiveExpression", "legs", recursiveExpression.getQuantifiers().size())));
+
+        final var rootQuantifier = bindings.get(rootQuantifierMatcher);
+        final var childQuantifier = bindings.get(childQuantifierMatcher);
+
+        final var rootReference = bindings.get(rootReferenceMatcher);
+        final var childReference = bindings.get(childReferenceMatcher);
+
+        final var rootPartition = bindings.get(rootPlanPartitionsMatcher);
+        final var childPartition = bindings.get(childPlanPartitionsMatcher);
+
+        final var rootAlias = rootQuantifier.getAlias();
+        final var childAlias = childQuantifier.getAlias();
+
+        var rootRef = call.memoizeMemberPlans(rootReference, rootPartition.getPlans());
+        final var newRootQuantifier = Quantifier.physicalBuilder().withAlias(rootAlias).build(rootRef);
+
+        var childRef = call.memoizeMemberPlans(childReference, childPartition.getPlans());
+        final var newChildQuantifier = Quantifier.physicalBuilder().withAlias(childAlias).build(childRef);
+
+        call.yieldExpression(new RecordQueryRecursivePlan(newRootQuantifier, newChildQuantifier, recursiveExpression.getResultValue(), true));
+    }
+}
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/rules/PushRequestedOrderingThroughRecursiveRule.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/rules/PushRequestedOrderingThroughRecursiveRule.java
new file mode 100644
index 0000000000..c7b4ccf998
--- /dev/null
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/rules/PushRequestedOrderingThroughRecursiveRule.java
@@ -0,0 +1,81 @@
+/*
+ * PushRequestedOrderingThroughRecursiveRule.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2024 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.query.plan.cascades.rules;
+
+import com.apple.foundationdb.annotation.API;
+import com.apple.foundationdb.record.query.plan.cascades.CascadesRule;
+import com.apple.foundationdb.record.query.plan.cascades.CascadesRuleCall;
+import com.apple.foundationdb.record.query.plan.cascades.PlannerRule.PreOrderRule;
+import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
+import com.apple.foundationdb.record.query.plan.cascades.Reference;
+import com.apple.foundationdb.record.query.plan.cascades.RequestedOrdering;
+import com.apple.foundationdb.record.query.plan.cascades.RequestedOrderingConstraint;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.RecursiveExpression;
+import com.apple.foundationdb.record.query.plan.cascades.matching.structure.BindingMatcher;
+import com.apple.foundationdb.record.query.plan.cascades.matching.structure.PlannerBindings;
+import com.apple.foundationdb.record.query.plan.cascades.matching.structure.ReferenceMatchers;
+import com.google.common.collect.ImmutableSet;
+
+import javax.annotation.Nonnull;
+import java.util.List;
+import java.util.Optional;
+import java.util.Set;
+
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.MultiMatcher.all;
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.QuantifierMatchers.forEachQuantifierOverRef;
+import static com.apple.foundationdb.record.query.plan.cascades.matching.structure.RelationalExpressionMatchers.recursiveExpression;
+
+/**
+ * A rule that pushes an ordering {@link RequestedOrderingConstraint} through a {@link RecursiveExpression}.
+ */
+@API(API.Status.EXPERIMENTAL)
+public class PushRequestedOrderingThroughRecursiveRule extends CascadesRule<RecursiveExpression> implements PreOrderRule {
+    private static final BindingMatcher<Reference> lowerRefMatcher = ReferenceMatchers.anyRef();
+    private static final BindingMatcher<Quantifier.ForEach> innerQuantifierMatcher = forEachQuantifierOverRef(lowerRefMatcher);
+    private static final BindingMatcher<RecursiveExpression> root =
+            recursiveExpression(all(innerQuantifierMatcher));
+
+    public PushRequestedOrderingThroughRecursiveRule() {
+        super(root, ImmutableSet.of(RequestedOrderingConstraint.REQUESTED_ORDERING));
+    }
+
+    @Override
+    public void onMatch(@Nonnull final CascadesRuleCall call) {
+        final Optional<Set<RequestedOrdering>> requestedOrderingOptional = call.getPlannerConstraint(RequestedOrderingConstraint.REQUESTED_ORDERING);
+        if (requestedOrderingOptional.isEmpty()) {
+            return;
+        }
+
+        // TODO: This isn't right. I think we only can do this if the requested ordering is empty, since the output order will
+        //  be the cursor's pre-order. We do need that case so that the two child quantifiers get plans from the data access rule.
+
+        final PlannerBindings bindings = call.getBindings();
+        final List<? extends Quantifier.ForEach> rangesOverQuantifiers = bindings.getAll(innerQuantifierMatcher);
+
+        rangesOverQuantifiers
+                .stream()
+                .map(Quantifier.ForEach::getRangesOver)
+                .forEach(lowerReference ->
+                        call.pushConstraint(lowerReference,
+                                RequestedOrderingConstraint.REQUESTED_ORDERING,
+                                requestedOrderingOptional.get()));
+    }
+}
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/values/RecursivePriorValue.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/values/RecursivePriorValue.java
new file mode 100644
index 0000000000..790a21490f
--- /dev/null
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/cascades/values/RecursivePriorValue.java
@@ -0,0 +1,162 @@
+/*
+ * RecursivePriorValue.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2024 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.query.plan.cascades.values;
+
+import com.apple.foundationdb.annotation.API;
+import com.apple.foundationdb.annotation.SpotBugsSuppressWarnings;
+import com.apple.foundationdb.record.EvaluationContext;
+import com.apple.foundationdb.record.ObjectPlanHash;
+import com.apple.foundationdb.record.PlanDeserializer;
+import com.apple.foundationdb.record.PlanHashable;
+import com.apple.foundationdb.record.PlanSerializable;
+import com.apple.foundationdb.record.PlanSerializationContext;
+import com.apple.foundationdb.record.planprotos.PRecursivePriorValue;
+import com.apple.foundationdb.record.planprotos.PValue;
+import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase;
+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.Formatter;
+import com.apple.foundationdb.record.query.plan.cascades.typing.Type;
+import com.apple.foundationdb.record.query.plan.plans.QueryResult;
+import com.google.auto.service.AutoService;
+import com.google.protobuf.Message;
+
+import javax.annotation.Nonnull;
+import java.util.List;
+import java.util.Objects;
+
+/**
+ * A value representing the version of a quantifier from the <em>prior</em> iteration of a recursion.
+ */
+@API(API.Status.EXPERIMENTAL)
+public class RecursivePriorValue extends AbstractValue implements LeafValue, PlanSerializable {
+    private static final ObjectPlanHash BASE_HASH = new ObjectPlanHash("Recursive-Prior-Value");
+
+    @Nonnull
+    private final CorrelationIdentifier alias;
+    @Nonnull
+    private final Type resultType;
+
+    private RecursivePriorValue(@Nonnull CorrelationIdentifier alias, @Nonnull Type resultType) {
+        this.alias = alias;
+        this.resultType = resultType;
+    }
+
+    @Nonnull
+    public static RecursivePriorValue of(@Nonnull CorrelationIdentifier alias, @Nonnull Type resultType) {
+        return new RecursivePriorValue(alias, resultType);
+    }
+
+    @Nonnull
+    @Override
+    public Type getResultType() {
+        return resultType;
+    }
+
+    @Nonnull
+    @Override
+    protected Iterable<? extends Value> computeChildren() {
+        return List.of();
+    }
+
+    @Override
+    public <M extends Message> Object eval(@Nonnull final FDBRecordStoreBase<M> store, @Nonnull final EvaluationContext context) {
+        final var binding = (QueryResult)context.getBinding(CorrelationIdentifier.of("prior_" + alias.getId()));
+        if (resultType.isRecord()) {
+            return binding.getDatum() == null ? null : binding.getMessage();
+        } else {
+            return binding.getDatum();
+        }
+    }
+
+    @Override
+    public int hashCodeWithoutChildren() {
+        return PlanHashable.objectsPlanHash(PlanHashable.CURRENT_FOR_CONTINUATION, BASE_HASH);
+    }
+
+    @Override
+    public int planHash(@Nonnull final PlanHashMode mode) {
+        return PlanHashable.objectsPlanHash(mode, BASE_HASH);
+    }
+
+    @Override
+    public String toString() {
+        return "Prior(" + alias + ")";
+    }
+
+    @Nonnull
+    @Override
+    public String explain(@Nonnull final Formatter formatter) {
+        return "PRIOR " + alias;
+    }
+
+    @Override
+    public int hashCode() {
+        return semanticHashCode();
+    }
+
+    @SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
+    @SpotBugsSuppressWarnings("EQ_UNUSUAL")
+    @Override
+    public boolean equals(final Object other) {
+        return semanticEquals(other, AliasMap.emptyMap());
+    }
+
+    @Nonnull
+    @Override
+    public PRecursivePriorValue toProto(@Nonnull final PlanSerializationContext serializationContext) {
+        return PRecursivePriorValue.newBuilder()
+                .setAlias(alias.getId())
+                .setResultType(resultType.toTypeProto(serializationContext))
+                .build();
+    }
+
+    @Nonnull
+    @Override
+    public PValue toValueProto(@Nonnull PlanSerializationContext serializationContext) {
+        return PValue.newBuilder().setRecursivePriorValue(toProto(serializationContext)).build();
+    }
+
+    @Nonnull
+    public static RecursivePriorValue fromProto(@Nonnull final PlanSerializationContext serializationContext, @Nonnull final PRecursivePriorValue recursivePriorValue) {
+        return new RecursivePriorValue(CorrelationIdentifier.of(Objects.requireNonNull(recursivePriorValue.getAlias())),
+                Type.fromTypeProto(serializationContext, Objects.requireNonNull(recursivePriorValue.getResultType())));
+    }
+
+    /**
+     * Deserializer.
+     */
+    @AutoService(PlanDeserializer.class)
+    public static class Deserializer implements PlanDeserializer<PRecursivePriorValue, RecursivePriorValue> {
+        @Nonnull
+        @Override
+        public Class<PRecursivePriorValue> getProtoMessageClass() {
+            return PRecursivePriorValue.class;
+        }
+
+        @Nonnull
+        @Override
+        public RecursivePriorValue fromProto(@Nonnull final PlanSerializationContext serializationContext,
+                                             @Nonnull final PRecursivePriorValue recursivePriorProto) {
+            return RecursivePriorValue.fromProto(serializationContext, recursivePriorProto);
+        }
+    }
+}
diff --git a/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryRecursivePlan.java b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryRecursivePlan.java
new file mode 100644
index 0000000000..8af00febee
--- /dev/null
+++ b/fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/query/plan/plans/RecordQueryRecursivePlan.java
@@ -0,0 +1,306 @@
+/*
+ * RecordQueryRecursivePlan.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2024 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.query.plan.plans;
+
+import com.apple.foundationdb.annotation.API;
+import com.apple.foundationdb.record.EvaluationContext;
+import com.apple.foundationdb.record.ExecuteProperties;
+import com.apple.foundationdb.record.ObjectPlanHash;
+import com.apple.foundationdb.record.PlanDeserializer;
+import com.apple.foundationdb.record.PlanHashable;
+import com.apple.foundationdb.record.PlanSerializationContext;
+import com.apple.foundationdb.record.RecordCursor;
+import com.apple.foundationdb.record.cursors.RecursiveCursor;
+import com.apple.foundationdb.record.planprotos.PRecordQueryPlan;
+import com.apple.foundationdb.record.planprotos.PRecordQueryRecursivePlan;
+import com.apple.foundationdb.record.provider.common.StoreTimer;
+import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase;
+import com.apple.foundationdb.record.query.plan.AvailableFields;
+import com.apple.foundationdb.record.query.plan.PlanStringRepresentation;
+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.Memoizer;
+import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
+import com.apple.foundationdb.record.query.plan.cascades.Quantifiers;
+import com.apple.foundationdb.record.query.plan.cascades.explain.Attribute;
+import com.apple.foundationdb.record.query.plan.cascades.explain.NodeInfo;
+import com.apple.foundationdb.record.query.plan.cascades.explain.PlannerGraph;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.RelationalExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.RelationalExpressionWithChildren;
+import com.apple.foundationdb.record.query.plan.cascades.values.Value;
+import com.apple.foundationdb.record.query.plan.cascades.values.translation.TranslationMap;
+import com.google.auto.service.AutoService;
+import com.google.common.base.Verify;
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.ImmutableMap;
+import com.google.protobuf.Message;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.List;
+import java.util.Objects;
+import java.util.Set;
+
+/**
+ * A query plan that recursively applies a plan to earlier results, starting with a root plan.
+ */
+@API(API.Status.INTERNAL)
+public class RecordQueryRecursivePlan implements RecordQueryPlanWithChildren, RelationalExpressionWithChildren {
+    private static final ObjectPlanHash BASE_HASH = new ObjectPlanHash("Record-Query-Recursive-Plan");
+
+    @Nonnull
+    private final Quantifier.Physical rootQuantifier;
+    @Nonnull
+    private final Quantifier.Physical childQuantifier;
+    @Nonnull
+    private final Value resultValue;
+    private final boolean inheritRecordProperties;
+
+    public RecordQueryRecursivePlan(@Nonnull final Quantifier.Physical rootQuantifier,
+                                    @Nonnull final Quantifier.Physical childQuantifier,
+                                    @Nonnull final Value resultValue,
+                                    final boolean inheritRecordProperties) {
+        this.rootQuantifier = rootQuantifier;
+        this.childQuantifier = childQuantifier;
+        this.resultValue = resultValue;
+        this.inheritRecordProperties = inheritRecordProperties;
+    }
+
+    @Nonnull
+    public Quantifier.Physical getRootQuantifier() {
+        return rootQuantifier;
+    }
+
+    @Nonnull
+    public Quantifier.Physical getChildQuantifier() {
+        return childQuantifier;
+    }
+
+    public boolean isInheritRecordProperties() {
+        return inheritRecordProperties;
+    }
+
+    @SuppressWarnings("resource")
+    @Nonnull
+    @Override
+    public <M extends Message> RecordCursor<QueryResult> executePlan(@Nonnull final FDBRecordStoreBase<M> store,
+                                                                     @Nonnull final EvaluationContext context,
+                                                                     @Nullable final byte[] continuation,
+                                                                     @Nonnull final ExecuteProperties executeProperties) {
+
+        final var nestedExecuteProperties = executeProperties.clearSkipAndLimit();
+        return RecursiveCursor.create(
+                        rootContinuation ->
+                                rootQuantifier.getRangesOverPlan().executePlan(store, context, rootContinuation, nestedExecuteProperties),
+                        (parentResult, depth, innerContinuation) -> {
+                            // TODO: Consider binding depth as well.
+                            final CorrelationIdentifier priorId = CorrelationIdentifier.of("prior_" + childQuantifier.getAlias().getId());
+                            final EvaluationContext childContext = context.withBinding(priorId, parentResult);
+                            return childQuantifier.getRangesOverPlan().executePlan(store, childContext, innerContinuation, nestedExecuteProperties);
+                        },
+                        null,
+                        continuation
+                ).skipThenLimit(executeProperties.getSkip(), executeProperties.getReturnedRowLimit())
+                .map(childResult -> {
+                    // TODO: Consider returning depth and is_leaf as well.
+                    final EvaluationContext childContext = context.withBinding(childQuantifier.getAlias(), childResult.getValue());
+                    final var computed = resultValue.eval(store, childContext);
+                    return inheritRecordProperties ? childResult.getValue().withComputed(computed) : QueryResult.ofComputed(computed);
+                });
+    }
+
+    @Override
+    public int getRelationalChildCount() {
+        return 2;
+    }
+
+    @Override
+    public boolean canCorrelate() {
+        return true;
+    }
+
+    @Nonnull
+    @Override
+    public List<RecordQueryPlan> getChildren() {
+        return ImmutableList.of(rootQuantifier.getRangesOverPlan(), childQuantifier.getRangesOverPlan());
+    }
+
+    @Nonnull
+    @Override
+    public AvailableFields getAvailableFields() {
+        return AvailableFields.NO_FIELDS;
+    }
+
+    @Nonnull
+    @Override
+    public Set<CorrelationIdentifier> getCorrelatedToWithoutChildren() {
+        return resultValue.getCorrelatedTo();
+    }
+
+    @Nonnull
+    @Override
+    public RecordQueryRecursivePlan translateCorrelations(@Nonnull final TranslationMap translationMap, @Nonnull final List<? extends Quantifier> translatedQuantifiers) {
+        Verify.verify(translatedQuantifiers.size() == 2);
+        final Value translatedResultValue = resultValue.translateCorrelations(translationMap);
+        return new RecordQueryRecursivePlan(
+                translatedQuantifiers.get(0).narrow(Quantifier.Physical.class),
+                translatedQuantifiers.get(1).narrow(Quantifier.Physical.class),
+                translatedResultValue,
+                inheritRecordProperties
+        );
+    }
+
+    @Override
+    public boolean isReverse() {
+        return Quantifiers.isReversed(Quantifiers.narrow(Quantifier.Physical.class, getQuantifiers()));
+    }
+
+    @Override
+    public RecordQueryRecursivePlan strictlySorted(@Nonnull Memoizer memoizer) {
+        return this;
+    }
+
+    @Nonnull
+    @Override
+    public Value getResultValue() {
+        return resultValue;
+    }
+
+    @Nonnull
+    @Override
+    public String toString() {
+        return PlanStringRepresentation.toString(this);
+    }
+
+    @Override
+    @SuppressWarnings("PMD.CompareObjectsWithEquals")
+    public boolean equalsWithoutChildren(@Nonnull RelationalExpression otherExpression,
+                                         @Nonnull final AliasMap aliasMap) {
+        if (this == otherExpression) {
+            return true;
+        }
+        if (getClass() != otherExpression.getClass()) {
+            return false;
+        }
+        return semanticEqualsForResults(otherExpression, aliasMap);
+    }
+
+    @SuppressWarnings("EqualsWhichDoesntCheckParameterClass")
+    @Override
+    public boolean equals(final Object other) {
+        return structuralEquals(other);
+    }
+
+    @Override
+    public int hashCode() {
+        return structuralHashCode();
+    }
+
+    @Override
+    public int hashCodeWithoutChildren() {
+        return Objects.hash(getResultValue(), inheritRecordProperties);
+    }
+
+    @Override
+    public void logPlanStructure(StoreTimer timer) {
+    }
+
+    @Override
+    public int getComplexity() {
+        return rootQuantifier.getRangesOverPlan().getComplexity() * childQuantifier.getRangesOverPlan().getComplexity();
+    }
+
+    @Override
+    public int planHash(@Nonnull final PlanHashMode mode) {
+        switch (mode.getKind()) {
+            case LEGACY:
+            case FOR_CONTINUATION:
+                return PlanHashable.objectsPlanHash(mode, BASE_HASH, getChildren(), getResultValue(), inheritRecordProperties);
+            default:
+                throw new UnsupportedOperationException("Hash kind " + mode.name() + " is not supported");
+        }
+    }
+
+    @Nonnull
+    @Override
+    public List<? extends Quantifier> getQuantifiers() {
+        return ImmutableList.of(rootQuantifier, childQuantifier);
+    }
+
+    @Nonnull
+    @Override
+    public PlannerGraph rewritePlannerGraph(@Nonnull final List<? extends PlannerGraph> childGraphs) {
+        return PlannerGraph.fromNodeAndChildGraphs(
+                new PlannerGraph.OperatorNodeWithInfo(this,
+                        NodeInfo.NESTED_LOOP_JOIN_OPERATOR,
+                        ImmutableList.of("RECURSIVE {{expr}}"),
+                        ImmutableMap.of("expr", Attribute.gml(getResultValue().toString()))),
+                childGraphs,
+                getQuantifiers());
+    }
+
+    @Nonnull
+    @Override
+    public PRecordQueryRecursivePlan toProto(@Nonnull final PlanSerializationContext serializationContext) {
+        return PRecordQueryRecursivePlan.newBuilder()
+                .setRootQuantifier(rootQuantifier.toProto(serializationContext))
+                .setChildQuantifier(childQuantifier.toProto(serializationContext))
+                .setResultValue(resultValue.toValueProto(serializationContext))
+                .setInheritRecordProperties(inheritRecordProperties)
+                .build();
+    }
+
+    @Nonnull
+    @Override
+    public PRecordQueryPlan toRecordQueryPlanProto(@Nonnull final PlanSerializationContext serializationContext) {
+        return PRecordQueryPlan.newBuilder().setRecursivePlan(toProto(serializationContext)).build();
+    }
+
+    @Nonnull
+    public static RecordQueryRecursivePlan fromProto(@Nonnull final PlanSerializationContext serializationContext,
+                                                   @Nonnull final PRecordQueryRecursivePlan recordQueryRecursivePlanProto) {
+        return new RecordQueryRecursivePlan(
+                Quantifier.Physical.fromProto(serializationContext, Objects.requireNonNull(recordQueryRecursivePlanProto.getRootQuantifier())),
+                Quantifier.Physical.fromProto(serializationContext, Objects.requireNonNull(recordQueryRecursivePlanProto.getChildQuantifier())),
+                Value.fromValueProto(serializationContext, Objects.requireNonNull(recordQueryRecursivePlanProto.getResultValue())),
+                recordQueryRecursivePlanProto.getInheritRecordProperties()
+        );
+    }
+
+    /**
+     * Deserializer.
+     */
+    @AutoService(PlanDeserializer.class)
+    public static class Deserializer implements PlanDeserializer<PRecordQueryRecursivePlan, RecordQueryRecursivePlan> {
+        @Nonnull
+        @Override
+        public Class<PRecordQueryRecursivePlan> getProtoMessageClass() {
+            return PRecordQueryRecursivePlan.class;
+        }
+
+        @Nonnull
+        @Override
+        public RecordQueryRecursivePlan fromProto(@Nonnull final PlanSerializationContext serializationContext,
+                                                  @Nonnull final PRecordQueryRecursivePlan recordQueryRecursivePlanProto) {
+            return RecordQueryRecursivePlan.fromProto(serializationContext, recordQueryRecursivePlanProto);
+        }
+    }
+}
diff --git a/fdb-record-layer-core/src/main/proto/record_cursor.proto b/fdb-record-layer-core/src/main/proto/record_cursor.proto
index dc9ccff8e0..1135d5465e 100644
--- a/fdb-record-layer-core/src/main/proto/record_cursor.proto
+++ b/fdb-record-layer-core/src/main/proto/record_cursor.proto
@@ -120,3 +120,11 @@ message MultidimensionalIndexScanContinuation {
     optional bytes lastHilbertValue = 1;
     optional bytes lastKey = 2;
 }
+
+message RecursiveContinuation {
+    message LevelCursor {
+        optional bytes continuation = 1;
+        optional bytes check_value = 2;
+    }
+    repeated LevelCursor levels = 1;
+}
diff --git a/fdb-record-layer-core/src/main/proto/record_query_plan.proto b/fdb-record-layer-core/src/main/proto/record_query_plan.proto
index d811e78951..10ee9e8e24 100644
--- a/fdb-record-layer-core/src/main/proto/record_query_plan.proto
+++ b/fdb-record-layer-core/src/main/proto/record_query_plan.proto
@@ -242,6 +242,7 @@ message PValue {
     PFromOrderedBytesValue from_ordered_bytes_value = 43;
     PCollateValue collate_value = 44;
     PNumericAggregationValue.PBitmapConstructAgg numeric_aggregation_value_bitmap_construct_agg = 45;
+    PRecursivePriorValue recursive_prior_value = 46;
   }
 }
 
@@ -1192,6 +1193,7 @@ message PRecordQueryPlan {
     PRecordQueryUnorderedPrimaryKeyDistinctPlan unordered_primary_key_distinct_plan = 30;
     PRecordQueryUnorderedUnionPlan unordered_union_plan = 31;
     PRecordQueryUpdatePlan update_plan = 32;
+    PRecordQueryRecursivePlan recursive_plan = 33;
   }
 }
 
@@ -1724,3 +1726,18 @@ message PRecordQueryUnionPlanBase {
 message PRecordQueryUpdatePlan {
   optional PRecordQueryAbstractDataModificationPlan super = 1;
 }
+
+//
+// PRecordQueryRecursivePlan
+//
+message PRecordQueryRecursivePlan {
+  optional PPhysicalQuantifier root_quantifier = 1;
+  optional PPhysicalQuantifier child_quantifier = 2;
+  optional PValue result_value = 3;
+  optional bool inherit_record_properties = 4;
+}
+
+message PRecursivePriorValue {
+  optional string alias = 1;
+  optional PType result_type = 2;
+}
diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/cursors/RecursiveCursorTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/cursors/RecursiveCursorTest.java
new file mode 100644
index 0000000000..93cf549ac4
--- /dev/null
+++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/cursors/RecursiveCursorTest.java
@@ -0,0 +1,106 @@
+/*
+ * RecursiveCursorTest.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2024 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.cursors;
+
+import com.apple.foundationdb.record.RecordCursor;
+import com.apple.foundationdb.record.RecordCursorResult;
+import com.apple.foundationdb.test.TestExecutors;
+import org.junit.jupiter.api.Test;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.stream.Collectors;
+import java.util.stream.IntStream;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Tests for {@link RecursiveCursor}.
+ */
+public class RecursiveCursorTest {
+    @Nonnull
+    private static RecordCursor<String> numberStrings(int n, @Nullable byte[] continuation) {
+        return new RangeCursor(TestExecutors.defaultThreadPool(), n, continuation).map(i -> Integer.toString(i));
+    }
+
+    @Nonnull
+    private static RecursiveCursor.ChildCursorFunction<String> stringPaths(int nchildren, int maxDepth) {
+        return (value, depth, continuation) -> {
+            if (depth > maxDepth) {
+                return RecordCursor.empty();
+            } else {
+                return numberStrings(nchildren, continuation).map(child -> value + "/" + child);
+            }
+        };
+    }
+
+    @Nonnull
+    private static RecordCursor<String> stringPathsCursor(int nroot, int nchildren, int depth,
+                                                          @Nullable byte[] continuation) {
+        return RecursiveCursor.create(c -> numberStrings(nroot, c), stringPaths(nchildren, depth - 1),
+                        null, continuation)
+                .filter(v -> v.getDepth() == depth)
+                .map(RecursiveCursor.RecursiveValue::getValue);
+    }
+
+    private static String integerStringPath(int n, int size, int radix) {
+        StringBuilder str = new StringBuilder();
+        for (int i = 0; i < size; i++) {
+            if (str.length() > 0) {
+                str.insert(0, '/');
+            }
+            str.insert(0, Character.forDigit(n % radix, radix));
+            n /= radix;
+        }
+        return str.toString();
+    }
+
+    @Test
+    void testStringPaths() {
+        final List<String> expected = IntStream.range(0, 2 * 3 * 3 * 3)
+                .mapToObj(n -> integerStringPath(n, 4, 3))
+                .collect(Collectors.toList());
+        final List<String> actual = stringPathsCursor(2, 3, 4, null).asList().join();
+        assertEquals(expected, actual);
+    }
+
+    @Test
+    void testStringPathContinued() {
+        final List<String> expected = IntStream.range(0, 2 * 3 * 3 * 3)
+                .mapToObj(n -> integerStringPath(n, 4, 3))
+                .collect(Collectors.toList());
+        final List<String> actual = new ArrayList<>();
+        int nsteps = 0;
+        byte[] continuation = null;
+        final AtomicReference<RecordCursorResult<String>> terminatingResultRef = new AtomicReference<>();
+        do {
+            actual.addAll(stringPathsCursor(2, 3, 4, continuation).limitRowsTo(4).asList(terminatingResultRef).join());
+            continuation = terminatingResultRef.get().getContinuation().toBytes();
+            nsteps++;
+        } while (continuation != null);
+        assertEquals(expected, actual);
+        assertEquals(14, nsteps);
+    }
+
+}
diff --git a/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/RecursiveQueryTest.java b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/RecursiveQueryTest.java
new file mode 100644
index 0000000000..1606e3e142
--- /dev/null
+++ b/fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/query/RecursiveQueryTest.java
@@ -0,0 +1,387 @@
+/*
+ * RecursiveQueryTest.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2024 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.query;
+
+import com.apple.foundationdb.record.EvaluationContext;
+import com.apple.foundationdb.record.ExecuteProperties;
+import com.apple.foundationdb.record.PlanSerializationContext;
+import com.apple.foundationdb.record.RecordCursor;
+import com.apple.foundationdb.record.RecordMetaData;
+import com.apple.foundationdb.record.RecordMetaDataBuilder;
+import com.apple.foundationdb.record.TestRecordsParentChildRelationshipProto;
+import com.apple.foundationdb.record.planprotos.PRecordQueryPlan;
+import com.apple.foundationdb.record.provider.common.StoreTimer;
+import com.apple.foundationdb.record.provider.foundationdb.FDBRecordContext;
+import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStoreBase;
+import com.apple.foundationdb.record.query.RecordQuery;
+import com.apple.foundationdb.record.query.expressions.Comparisons;
+import com.apple.foundationdb.record.query.expressions.Query;
+import com.apple.foundationdb.record.query.plan.AvailableFields;
+import com.apple.foundationdb.record.query.plan.QueryPlanner;
+import com.apple.foundationdb.record.query.plan.RecordQueryPlanner;
+import com.apple.foundationdb.record.query.plan.cascades.AliasMap;
+import com.apple.foundationdb.record.query.plan.cascades.CascadesPlanner;
+import com.apple.foundationdb.record.query.plan.cascades.CorrelationIdentifier;
+import com.apple.foundationdb.record.query.plan.cascades.GraphExpansion;
+import com.apple.foundationdb.record.query.plan.cascades.Quantifier;
+import com.apple.foundationdb.record.query.plan.cascades.Reference;
+import com.apple.foundationdb.record.query.plan.cascades.explain.PlannerGraph;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.LogicalSortExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.RecursiveExpression;
+import com.apple.foundationdb.record.query.plan.cascades.expressions.RelationalExpression;
+import com.apple.foundationdb.record.query.plan.cascades.predicates.ValuePredicate;
+import com.apple.foundationdb.record.query.plan.cascades.values.FieldValue;
+import com.apple.foundationdb.record.query.plan.cascades.values.ObjectValue;
+import com.apple.foundationdb.record.query.plan.cascades.values.QuantifiedObjectValue;
+import com.apple.foundationdb.record.query.plan.cascades.values.RecursivePriorValue;
+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.plans.QueryResult;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryPlan;
+import com.apple.foundationdb.record.query.plan.plans.RecordQueryRecursivePlan;
+import com.apple.test.Tags;
+import com.google.protobuf.Message;
+import org.junit.jupiter.api.Tag;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+import java.util.List;
+import java.util.Set;
+
+import static com.apple.foundationdb.record.provider.foundationdb.query.FDBQueryGraphTestHelpers.fullTypeScan;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Recursive query experiments.
+ */
+@Tag(Tags.RequiresFDB)
+public class RecursiveQueryTest extends FDBRecordStoreQueryTestBase {
+
+    @DualPlannerTest()
+    public void testUp() throws Exception {
+        loadRecords();
+
+        final RecordQueryPlan plan;
+        if (planner instanceof RecordQueryPlanner) {
+            plan = buildAncestorPlan((RecordQueryPlanner)planner);
+        } else if (planner instanceof CascadesPlanner) {
+            plan = buildAncestorPlan((CascadesPlanner)planner);
+        } else {
+            plan = null;
+            fail("unsupported planner type");
+        }
+
+        final List<String> actual = queryNames(plan);
+        final List<String> expected = List.of("three", "two", "root");
+        assertEquals(expected, actual);
+    }
+
+    @Nonnull
+    private RecordQueryPlan buildAncestorPlan(@Nonnull RecordQueryPlanner planner) throws Exception {
+        final RecordQuery rootQuery = RecordQuery.newBuilder()
+                .setRecordType("MyChildRecord")
+                .setFilter(Query.field("str_value").equalsValue("three"))
+                .build();
+        final RecordQuery childQuery = RecordQuery.newBuilder()
+                .setRecordType("MyChildRecord")
+                .setFilter(Query.field("rec_no").equalsParameter("parent_rec_no"))
+                .build();
+        return recursiveQuery(planner, rootQuery, childQuery, "parent_rec_no");
+    }
+
+    @Nonnull
+    private RecordQueryPlan buildAncestorPlan(@Nonnull CascadesPlanner cascadesPlanner) {
+        return planGraph(
+                () -> {
+                    var rootQuantifier = fullTypeScan(cascadesPlanner.getRecordMetaData(), "MyChildRecord");
+                    var graphExpansionBuilder = GraphExpansion.builder();
+                    graphExpansionBuilder.addQuantifier(rootQuantifier);
+                    graphExpansionBuilder.addPredicate(
+                            new ValuePredicate(FieldValue.ofFieldName(QuantifiedObjectValue.of(rootQuantifier.getAlias(), rootQuantifier.getFlowedObjectType()), "str_value"),
+                                    new Comparisons.SimpleComparison(Comparisons.Type.EQUALS, "three")));
+                    rootQuantifier = Quantifier.forEach(Reference.of(graphExpansionBuilder.build().buildSimpleSelectOverQuantifier((Quantifier.ForEach)rootQuantifier)));
+
+                    var childQuantifier = fullTypeScan(cascadesPlanner.getRecordMetaData(), "MyChildRecord");
+                    graphExpansionBuilder = GraphExpansion.builder();
+                    graphExpansionBuilder.addQuantifier(childQuantifier);
+                    var aliasForPrior = CorrelationIdentifier.uniqueID();
+                    graphExpansionBuilder.addPredicate(
+                            new ValuePredicate(FieldValue.ofFieldName(QuantifiedObjectValue.of(childQuantifier.getAlias(), childQuantifier.getFlowedObjectType()), "rec_no"),
+                                    new Comparisons.ValueComparison(Comparisons.Type.EQUALS, FieldValue.ofFieldName(RecursivePriorValue.of(aliasForPrior, childQuantifier.getFlowedObjectType()), "parent_rec_no"))));
+                    childQuantifier = Quantifier.forEach(Reference.of(graphExpansionBuilder.build().buildSimpleSelectOverQuantifier((Quantifier.ForEach)childQuantifier)), aliasForPrior);
+
+                    final var resultStrValue = FieldValue.ofFieldName(QuantifiedObjectValue.of(childQuantifier.getAlias(), childQuantifier.getFlowedObjectType()), "str_value");
+                    final var recursive = new RecursiveExpression(resultStrValue, rootQuantifier, childQuantifier);
+                    return Reference.of(LogicalSortExpression.unsorted(Quantifier.forEach(Reference.of(recursive))));
+                });
+    }
+
+    @DualPlannerTest()
+    public void testDown() throws Exception {
+        loadRecords();
+
+        final RecordQueryPlan plan;
+        if (planner instanceof RecordQueryPlanner) {
+            plan = buildDescendantPlan((RecordQueryPlanner)planner);
+        } else if (planner instanceof CascadesPlanner) {
+            plan = buildDescendantPlan((CascadesPlanner)planner);
+        } else {
+            plan = null;
+            fail("unsupported planner type");
+        }
+
+        final List<String> actual = queryNames(plan);
+        final List<String> expected = List.of("root", "two", "three", "five", "four");
+        assertEquals(expected, actual);
+    }
+
+    private static RecordQueryPlan buildDescendantPlan(@Nonnull RecordQueryPlanner planner) throws Exception {
+        final RecordQuery rootQuery = RecordQuery.newBuilder()
+                .setRecordType("MyChildRecord")
+                .setFilter(Query.field("parent_rec_no").isNull())
+                .build();
+        final RecordQuery childQuery = RecordQuery.newBuilder()
+                .setRecordType("MyChildRecord")
+                .setFilter(Query.field("parent_rec_no").equalsParameter("rec_no"))
+                .build();
+
+        return recursiveQuery(planner, rootQuery, childQuery, "rec_no");
+    }
+
+    @Nonnull
+    private RecordQueryPlan buildDescendantPlan(@Nonnull CascadesPlanner cascadesPlanner) {
+        return planGraph(
+                () -> {
+                    var rootQuantifier = fullTypeScan(cascadesPlanner.getRecordMetaData(), "MyChildRecord");
+                    var graphExpansionBuilder = GraphExpansion.builder();
+                    graphExpansionBuilder.addQuantifier(rootQuantifier);
+                    graphExpansionBuilder.addPredicate(
+                            new ValuePredicate(FieldValue.ofFieldName(QuantifiedObjectValue.of(rootQuantifier.getAlias(), rootQuantifier.getFlowedObjectType()), "parent_rec_no"),
+                                    new Comparisons.NullComparison(Comparisons.Type.IS_NULL)));
+                    rootQuantifier = Quantifier.forEach(Reference.of(graphExpansionBuilder.build().buildSimpleSelectOverQuantifier((Quantifier.ForEach)rootQuantifier)));
+
+                    var childQuantifier = fullTypeScan(cascadesPlanner.getRecordMetaData(), "MyChildRecord");
+                    graphExpansionBuilder = GraphExpansion.builder();
+                    graphExpansionBuilder.addQuantifier(childQuantifier);
+                    var aliasForPrior = CorrelationIdentifier.uniqueID();
+                    graphExpansionBuilder.addPredicate(
+                            new ValuePredicate(FieldValue.ofFieldName(QuantifiedObjectValue.of(childQuantifier.getAlias(), childQuantifier.getFlowedObjectType()), "parent_rec_no"),
+                                    new Comparisons.ValueComparison(Comparisons.Type.EQUALS, FieldValue.ofFieldName(RecursivePriorValue.of(aliasForPrior, childQuantifier.getFlowedObjectType()), "rec_no"))));
+                    childQuantifier = Quantifier.forEach(Reference.of(graphExpansionBuilder.build().buildSimpleSelectOverQuantifier((Quantifier.ForEach)childQuantifier)), aliasForPrior);
+
+                    final var resultStrValue = FieldValue.ofFieldName(QuantifiedObjectValue.of(childQuantifier.getAlias(), childQuantifier.getFlowedObjectType()), "str_value");
+                    final var recursive = new RecursiveExpression(resultStrValue, rootQuantifier, childQuantifier);
+                    return Reference.of(LogicalSortExpression.unsorted(Quantifier.forEach(Reference.of(recursive))));
+                });
+    }
+
+    private void openParentChildRecordStore(FDBRecordContext context) throws Exception {
+        final RecordMetaDataBuilder metaDataBuilder = RecordMetaData.newBuilder().setRecords(TestRecordsParentChildRelationshipProto.getDescriptor());
+        createOrOpenRecordStore(context, metaDataBuilder.getRecordMetaData());
+    }
+
+    private void loadRecords() throws Exception {
+        try (FDBRecordContext context = openContext()) {
+            openParentChildRecordStore(context);
+
+            saveRecord(1, -1, "root");
+            saveRecord(2, 1, "two");
+            saveRecord(3, 2, "three");
+            saveRecord(4, 1, "four");
+            saveRecord(5, 2, "five");
+
+            context.commit();
+        }
+    }
+
+    private void saveRecord(int recNo, int parent, String name) {
+        TestRecordsParentChildRelationshipProto.MyChildRecord.Builder builder = TestRecordsParentChildRelationshipProto.MyChildRecord.newBuilder();
+        builder.setRecNo(recNo);
+        if (parent > 0) {
+            builder.setParentRecNo(parent);
+        }
+        builder.setStrValue(name);
+        recordStore.saveRecord(builder.build());
+    }
+
+    private List<String> queryNames(RecordQueryPlan plan) throws Exception {
+        try (FDBRecordContext context = openContext()) {
+            openParentChildRecordStore(context);
+            return plan.execute(recordStore).map(result -> {
+                TestRecordsParentChildRelationshipProto.MyChildRecord record = TestRecordsParentChildRelationshipProto.MyChildRecord.newBuilder()
+                        .mergeFrom(result.getRecord())
+                        .build();
+                return record.getStrValue();
+            }).asList().join();
+        }
+    }
+
+    private static RecordQueryPlan recursiveQuery(@Nonnull QueryPlanner planner,
+                                                  @Nonnull RecordQuery rootQuery, @Nonnull RecordQuery childQuery,
+                                                  @Nonnull String glueField) throws Exception {
+        RecordQueryPlan rootPlan = planner.plan(rootQuery);
+        // Bit of a mess making up for a proper Expression, but also allowing use of old planner.
+        CorrelationIdentifier childId = CorrelationIdentifier.of("child");
+        final CorrelationIdentifier priorId = CorrelationIdentifier.of("prior_" + childId.getId());
+        RecordQueryPlan childPlan = new GluePlan(planner.plan(childQuery), priorId, glueField);
+        final Quantifier.Physical rootQuantifier = Quantifier.physical(Reference.of(rootPlan));
+        final Quantifier.Physical childQuantifier = Quantifier.physical(Reference.of(childPlan), childId);
+        return new RecordQueryRecursivePlan(
+                rootQuantifier,
+                childQuantifier,
+                ObjectValue.of(childQuantifier.getAlias(), childQuantifier.getFlowedObjectType()),
+                true);
+    }
+
+    static class GluePlan implements RecordQueryPlan {
+        private final RecordQueryPlan inner;
+        private final CorrelationIdentifier resultBinding;
+        private final String fieldName;
+
+        public GluePlan(final RecordQueryPlan inner, final CorrelationIdentifier resultBinding, final String fieldName) {
+            this.inner = inner;
+            this.resultBinding = resultBinding;
+            this.fieldName = fieldName;
+        }
+
+        @Nonnull
+        @Override
+        public <M extends Message> RecordCursor<QueryResult> executePlan(@Nonnull final FDBRecordStoreBase<M> store, @Nonnull final EvaluationContext context, @Nullable final byte[] continuation, @Nonnull final ExecuteProperties executeProperties) {
+            final QueryResult result = (QueryResult)context.getBinding(resultBinding);
+            final Message record = result.getQueriedRecord().getRecord();
+            final Object fieldValue = record.getField(record.getDescriptorForType().findFieldByName(fieldName));
+            final EvaluationContext newContext = context.withBinding(fieldName, fieldValue);
+            return inner.executePlan(store, newContext, continuation, executeProperties);
+        }
+
+        @Nonnull
+        @Override
+        public List<RecordQueryPlan> getChildren() {
+            return List.of(inner);
+        }
+
+        @Nonnull
+        @Override
+        public AvailableFields getAvailableFields() {
+            return inner.getAvailableFields();
+        }
+
+        @Nonnull
+        @Override
+        public Message toProto(@Nonnull final PlanSerializationContext serializationContext) {
+            return inner.toProto(serializationContext);
+        }
+
+        @Nonnull
+        @Override
+        public PRecordQueryPlan toRecordQueryPlanProto(@Nonnull final PlanSerializationContext serializationContext) {
+            return inner.toRecordQueryPlanProto(serializationContext);
+        }
+
+        @Nonnull
+        @Override
+        public PlannerGraph rewritePlannerGraph(@Nonnull final List<? extends PlannerGraph> childGraphs) {
+            return inner.rewritePlannerGraph(childGraphs);
+        }
+
+        @Override
+        public boolean isReverse() {
+            return inner.isReverse();
+        }
+
+        @Override
+        public boolean hasRecordScan() {
+            return inner.hasRecordScan();
+        }
+
+        @Override
+        public boolean hasFullRecordScan() {
+            return inner.hasFullRecordScan();
+        }
+
+        @Override
+        public boolean hasIndexScan(@Nonnull final String indexName) {
+            return inner.hasIndexScan(indexName);
+        }
+
+        @Nonnull
+        @Override
+        public Set<String> getUsedIndexes() {
+            return inner.getUsedIndexes();
+        }
+
+        @Override
+        public boolean hasLoadBykeys() {
+            return inner.hasLoadBykeys();
+        }
+
+        @Override
+        public void logPlanStructure(final StoreTimer timer) {
+            inner.logPlanStructure(timer);
+        }
+
+        @Override
+        public int getComplexity() {
+            return inner.getComplexity();
+        }
+
+        @Override
+        public int planHash(@Nonnull final PlanHashMode hashMode) {
+            return inner.planHash(hashMode);
+        }
+
+        @Nonnull
+        @Override
+        public Value getResultValue() {
+            return inner.getResultValue();
+        }
+
+        @Nonnull
+        @Override
+        public List<? extends Quantifier> getQuantifiers() {
+            return inner.getQuantifiers();
+        }
+
+        @Override
+        public boolean equalsWithoutChildren(@Nonnull final RelationalExpression other, @Nonnull final AliasMap equivalences) {
+            return inner.equalsWithoutChildren(other, equivalences);
+        }
+
+        @Override
+        public int hashCodeWithoutChildren() {
+            return inner.hashCodeWithoutChildren();
+        }
+
+        @Nonnull
+        @Override
+        public RelationalExpression translateCorrelations(@Nonnull final TranslationMap translationMap, @Nonnull final List<? extends Quantifier> translatedQuantifiers) {
+            return inner.translateCorrelations(translationMap, translatedQuantifiers);
+        }
+
+        @Nonnull
+        @Override
+        public Set<CorrelationIdentifier> getCorrelatedTo() {
+            return inner.getCorrelatedTo();
+        }
+    }
+
+}