diff --git a/CHANGELOG.md b/CHANGELOG.md
index 9929b546..09be1bd7 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -108,6 +108,27 @@
 
 ### Features
 
+* **SDBQL gains set operations, recursive CTEs, `RETURN DISTINCT`, `COLLECT
+  … KEEP`, the `NONE` quantifier, and standalone `OFFSET`.** Query blocks can
+  be combined with `UNION [ALL]`, `INTERSECT`, and `EXCEPT` (`FOR … RETURN …
+  UNION ALL FOR … RETURN …`; either side may be parenthesized; duplicates are
+  removed except for `UNION ALL`). Chains follow SQL precedence — `INTERSECT`
+  binds tighter than `UNION`/`EXCEPT`, which chain left to right — and rows
+  are compared by value, the same equality `UNION()`/`INTERSECTION()` use.
+  `WITH RECURSIVE name AS (<anchor> UNION ALL <step>)` iterates the step until
+  it stops producing rows — inside the step the CTE name is bound to the rows
+  of the *previous* iteration (hierarchies, org charts, transitive closures),
+  with safety caps at 1,000 iterations / 1M rows. `RETURN DISTINCT expr`
+  deduplicates result rows (first occurrence wins). `COLLECT … INTO g KEEP v1,
+  v2` restricts which in-scope variables are stored in the group arrays (an
+  unknown name is an error). `NONE(x IN arr SATISFIES cond)` — and the
+  function form `NONE(arr, x -> cond)` — is true when no element satisfies.
+  `OFFSET n` works alone or as `LIMIT n OFFSET m`. A CTE declared before a set
+  operation binds in every operand, and operands, CTE bodies and subqueries
+  are all executed as full query blocks — their own `WITH`, pre-`FOR` `LET`s,
+  `SORT`/`LIMIT` and nested set operations all apply. Permission checks, query
+  caching, cache invalidation, the long-running-query gate and
+  `has_mutations()` all see through set-operation operands and CTE bodies.
 * **Query-driven auto-indexes (opt-in).** Collections with `autoIndex: true`
   (or unset + `SOLIDB_AUTO_INDEX=1`) create a persistent `_auto_{field}`
   index the first time a `FOR` + `FILTER` equality/range miss would have used
diff --git a/admin/app/views/query/_explain.html.slv b/admin/app/views/query/_explain.html.slv
index 53e5c604..2fa0ca52 100644
--- a/admin/app/views/query/_explain.html.slv
+++ b/admin/app/views/query/_explain.html.slv
@@ -97,8 +97,13 @@
           </span>
         <% end %>
         <% if !limit_info.nil? %>
+          <%# A standalone OFFSET reports no count %>
+          <% limit_offset = limit_info["offset"] ?? 0 %>
+          <% limit_count = limit_info["count"] %>
+          <% limit_text = limit_count.nil? ? "offset " + str(limit_offset) : "limit " + str(limit_count) %>
+          <% limit_text = limit_text + " offset " + str(limit_offset) if !limit_count.nil? && limit_offset > 0 %>
           <span class="console-label border border-rose-800 px-1.5 py-0.5 text-rose-400">
-            limit <%= limit_info["count"] ?? 0 %><% if (limit_info["offset"] ?? 0) > 0 %> offset <%= limit_info["offset"] %><% end %>
+            <%= limit_text %>
           </span>
         <% end %>
       </div>
diff --git a/doc/app/views/docs/changelog.html.slv b/doc/app/views/docs/changelog.html.slv
index 3a1fe36b..7bd3ab52 100644
--- a/doc/app/views/docs/changelog.html.slv
+++ b/doc/app/views/docs/changelog.html.slv
@@ -17,6 +17,7 @@
         Security hardening, SDBQL function completeness, and <strong class="text-white">one product breaking change:</strong> the client-facing job and cron queue is removed. Application background jobs now live in the Soli framework. <strong class="text-white">Triggers keep working</strong>, on the same machinery as before. Also breaking if you relied on them: default bind is loopback, replication TCP requires a keyfile, <code>LENGTH</code> on a string is character count (not a collection count), installing Lua requires Admin, and API keys must declare a role.
       </p>
       <ul class="space-y-3 text-gray-400 text-sm leading-relaxed">
+        <li><strong class="text-white">New SDBQL syntax: set operations, recursive CTEs, <code>RETURN DISTINCT</code>, <code>COLLECT &hellip; KEEP</code>, <code>NONE</code>, standalone <code>OFFSET</code></strong> &mdash; combine query blocks with <code>UNION [ALL]</code>, <code>INTERSECT</code> and <code>EXCEPT</code> (either side may be parenthesized; duplicates removed except for <code>UNION ALL</code>; chains follow SQL precedence, so <code>INTERSECT</code> binds tighter than <code>UNION</code>/<code>EXCEPT</code>). <code>WITH RECURSIVE name AS (&lt;anchor&gt; UNION ALL &lt;step&gt;)</code> iterates until the step stops producing rows, binding the CTE name inside the step to the rows of the previous iteration (hierarchies, org charts, transitive closures), capped at 1,000 iterations / 1M rows. <code>RETURN DISTINCT expr</code> deduplicates rows. <code>COLLECT &hellip; INTO g KEEP v1, v2</code> restricts which variables land in the group arrays. <code>NONE(x IN arr SATISFIES cond)</code> and <code>NONE(arr, x -&gt; cond)</code> are true when nothing satisfies. <code>OFFSET n</code> works alone or as <code>LIMIT n OFFSET m</code>.</li>
         <li><strong class="text-white">Native HTTPS termination</strong> &mdash; new <code class="text-cyan-400">--tls-cert</code> / <code class="text-cyan-400">--tls-key</code> flags terminate TLS in the server itself (rustls, no OpenSSL). On the multiplexed port the listener sniffs for a TLS handshake and terminates only when the client offers one, so HTTP and the native driver protocol run inside the tunnel for TLS-capable clients while the plaintext-only driver SDKs and sync/cluster peers keep connecting; <code class="text-cyan-400">SOLIDB_TLS_REQUIRE=1</code> refuses plaintext there. Both flags are required together.</li>
         <li><strong class="text-white">Per-client API rate limiting</strong> &mdash; only <code>/auth/login</code> was throttled before; now the whole router has a per-client-IP sliding-window limiter (default 600 requests / 60s) answering <code>429</code> with <code>Retry-After</code> before any handler work. Internal cluster traffic and CORS preflights are exempt. Configure with <code class="text-cyan-400">SOLIDB_API_RATE_LIMIT</code> (0 disables) and <code class="text-cyan-400">SOLIDB_API_RATE_WINDOW_SECS</code>.</li>
         <li><strong class="text-white">Driver queries have a timeout</strong> &mdash; HTTP capped execution at 30s but the binary protocol ran queries unbounded on a runtime thread. Driver <code>Query</code> and <code>Explain</code> now run under the same 30s cap, on the same long-running-query gate HTTP uses so point reads stay inline.</li>
diff --git a/doc/app/views/docs/sdbql-aggregations.html.slv b/doc/app/views/docs/sdbql-aggregations.html.slv
index 36543e25..70d250ec 100644
--- a/doc/app/views/docs/sdbql-aggregations.html.slv
+++ b/doc/app/views/docs/sdbql-aggregations.html.slv
@@ -103,10 +103,33 @@
               </div>
             </div>
 
+            <!-- KEEP restriction -->
+            <div class="bg-gray-800/30 rounded-xl border border-white/5 overflow-hidden">
+              <div class="p-5 border-b border-white/5 bg-white/5">
+                <h3 class="text-lg font-bold text-orange-300 flex items-center">Restricting Group Items with KEEP</h3>
+              </div>
+              <div class="p-5 space-y-4">
+                <p class="text-sm text-gray-400">
+                  With <code>INTO groups</code>, each group array normally stores every variable in
+                  scope. Add <code class="text-orange-300">KEEP v1, v2</code> to store only those
+                  variables — smaller results, no leaked document bodies.
+                </p>
+                <pre><code class="language-sql text-sm">FOR u IN users
+  LET n = u.name
+  COLLECT city = u.city INTO g KEEP n
+  RETURN { city, names: g[*].n }</code></pre>
+                <p class="text-xs text-gray-500">
+                  Without <code>KEEP</code>, each item of <code>g</code> would also carry <code>u</code>.
+                  Every name listed must be a variable in scope at the <code>COLLECT</code> — an
+                  unknown name is an error, not an empty group item.
+                </p>
+              </div>
+            </div>
+
             <!-- Syntax Reference -->
             <div class="bg-gray-950/50 rounded-lg border border-white/5 p-4 mt-8">
               <code class="text-sm text-gray-300 font-mono block whitespace-pre overflow-x-auto">COLLECT <span class="text-orange-400">variable</span> = <span class="text-purple-400">expression</span> [, ...]
-[INTO <span class="text-orange-400">groupVariable</span>]
+[INTO <span class="text-orange-400">groupVariable</span> [KEEP <span class="text-orange-400">var1</span>, <span class="text-orange-400">var2</span>, ...]]
 [WITH COUNT INTO <span class="text-orange-400">countVariable</span>]
 [AGGREGATE <span class="text-orange-400">variable</span> = <span class="text-blue-400">FUNC</span>(<span class="text-purple-400">expression</span>) [, ...]]</code>
             </div>
diff --git a/doc/app/views/docs/sdbql-cte.html.slv b/doc/app/views/docs/sdbql-cte.html.slv
index d429a661..7822cbb2 100644
--- a/doc/app/views/docs/sdbql-cte.html.slv
+++ b/doc/app/views/docs/sdbql-cte.html.slv
@@ -135,10 +135,50 @@ FOR s IN summary
               </div>
             </div>
 
+            <div class="p-6">
+              <h3 class="text-lg font-bold text-white mb-4">Recursive CTE</h3>
+              <p class="text-gray-400 mb-4">
+                <code class="text-orange-400">WITH RECURSIVE</code> walks hierarchical data
+                (org charts, category trees, bill of materials, transitive closures) in pure
+                SDBQL. The body must have the shape
+                <code class="text-orange-400">&lt;anchor query&gt; UNION ALL &lt;recursive step&gt;</code>.
+                The anchor runs once; then the step runs repeatedly until it stops producing rows.
+              </p>
+              <p class="text-gray-400 mb-4">
+                Inside the step, the CTE name is bound to the rows produced by the
+                <strong>previous iteration</strong> — use it to expand one level at a time.
+                Recursion is capped at 1,000 iterations and 1M rows.
+              </p>
+              <div class="bg-gray-950/50 rounded-lg overflow-hidden border border-white/5 mb-4">
+                <div class="bg-black/20 px-4 py-2 border-b border-white/5">
+                  <span class="text-xs text-gray-400">Example: everyone reporting to alice, any depth</span>
+                </div>
+                <pre><code class="language-sql text-xs bg-gray-950/50 p-3 rounded border border-white/5">WITH RECURSIVE reports AS (
+  -- Anchor: the root of the hierarchy
+  FOR e IN employees
+    FILTER e._key == 'alice'
+    RETURN e._key
+  UNION ALL
+  -- Step: find direct reports of the previous level
+  FOR m IN employees
+    FILTER m.manager IN reports
+    RETURN m._key
+)
+FOR x IN reports
+  RETURN x</code></pre>
+              </div>
+              <ul class="text-gray-400 space-y-1 text-sm">
+                <li><code class="text-orange-400">RECURSIVE</code> applies to every CTE in the WITH list (standard SQL semantics)</li>
+                <li>Steps must be joined with <code class="text-orange-400">UNION ALL</code>; other set operators are rejected on a recursive CTE body</li>
+                <li>The anchor and each step are full query blocks: they may have their own <code>LET</code>s, <code>SORT</code>/<code>LIMIT</code>, and nested <code>WITH</code></li>
+                <li>A step that never stops producing new rows ends with an iteration-limit error rather than hanging. Cyclic data (a → b → a) is exactly that case: filter out rows you have already visited in the step, or carry a depth and stop at it</li>
+              </ul>
+            </div>
+
             <div class="p-6">
               <h3 class="text-lg font-bold text-white mb-4">Syntax</h3>
               <div class="bg-gray-950/50 rounded-lg overflow-hidden border border-white/5">
-                <pre><code class="language-sql text-xs bg-gray-950/50 p-3 rounded border border-white/5">WITH
+                <pre><code class="language-sql text-xs bg-gray-950/50 p-3 rounded border border-white/5">WITH [RECURSIVE]
   cte_name [(col1, col2, ...)] AS ( query ),
   cte_name2 [(col1, col2, ...)] AS ( query )
 query_body</code></pre>
diff --git a/doc/app/views/docs/sdbql-operators.html.slv b/doc/app/views/docs/sdbql-operators.html.slv
index bed71f09..ffce91de 100644
--- a/doc/app/views/docs/sdbql-operators.html.slv
+++ b/doc/app/views/docs/sdbql-operators.html.slv
@@ -354,7 +354,10 @@ SATISFIES tag != "deprecated"</code></pre>
                   <span class="text-xs text-gray-400">No matches</span>
                 </div>
                 <pre><code class="language-sql text-xs">FILTER NONE user IN doc.admins
-SATISFIES user.status == "inactive"</code></pre>
+SATISFIES user.status == "inactive"
+
+-- Function form
+RETURN NONE(doc.admins, u -> u.status == "inactive")</code></pre>
                 <div class="mt-3 text-xs text-gray-400">
                   True if <strong>zero</strong> elements satisfy the condition.
                 </div>
diff --git a/doc/app/views/docs/sdbql-syntax.html.slv b/doc/app/views/docs/sdbql-syntax.html.slv
index e6fb51f4..e13e1201 100644
--- a/doc/app/views/docs/sdbql-syntax.html.slv
+++ b/doc/app/views/docs/sdbql-syntax.html.slv
@@ -44,8 +44,8 @@
 FOR variable IN collection
   FILTER expression
   SORT field [ASC|DESC]
-  LIMIT [offset,] count
-  RETURN expression</code></pre>
+  LIMIT [offset,] count | OFFSET n [LIMIT m]
+  RETURN [DISTINCT] expression</code></pre>
             </div>
 
             <div class="grid grid-cols-1 md:grid-cols-2 gap-6">
@@ -75,13 +75,94 @@ FOR variable IN collection
                   <span class="bg-indigo-500/20 p-1.5 rounded mr-2 text-xs">RETURN</span>
                   Projection
                 </h3>
-                <p class="text-sm text-gray-400">Constructs the result documents.</p>
+                <p class="text-sm text-gray-400">Constructs the result documents. Add <code class="text-indigo-300">DISTINCT</code> to remove duplicate rows.</p>
               </div>
             </div>
           </div>
         </div>
       </section>
 
+      <!-- Set Operations -->
+      <section id="set-operations" class="scroll-mt-24">
+        <div class="bg-gray-800/30 rounded-xl border border-white/5 overflow-hidden">
+          <div class="p-5 border-b border-white/5 bg-white/5">
+            <h3 class="text-lg font-bold text-emerald-300 flex items-center">
+              <span class="bg-emerald-500/20 text-emerald-400 text-xs font-mono px-2 py-1 rounded mr-3">UNION / INTERSECT / EXCEPT</span>
+              Set Operations
+            </h3>
+          </div>
+          <div class="p-5 space-y-4">
+            <p class="text-sm text-gray-400">
+              Combine query blocks with <code class="text-emerald-300">UNION [ALL]</code>,
+              <code class="text-emerald-300">INTERSECT</code>, or
+              <code class="text-emerald-300">EXCEPT</code>. Either side may be parenthesized,
+              and rows are compared by value — <code>1</code> and <code>1.0</code> are the
+              same row, like in <code>UNION()</code> and <code>INTERSECTION()</code>.
+            </p>
+            <p class="text-sm text-gray-400">
+              Chains follow SQL precedence: <code class="text-emerald-300">INTERSECT</code>
+              binds tighter than <code class="text-emerald-300">UNION</code> and
+              <code class="text-emerald-300">EXCEPT</code>, which chain left to right. So
+              <code>a UNION b INTERSECT c</code> is <code>a UNION (b INTERSECT c)</code>, and
+              <code>a EXCEPT b EXCEPT c</code> is <code>(a EXCEPT b) EXCEPT c</code>.
+              Parenthesize to group differently.
+            </p>
+            <div class="grid grid-cols-1 gap-4">
+              <div>
+                <h4 class="text-xs font-bold text-gray-400 uppercase tracking-wider mb-2">UNION — both sides, duplicates removed (UNION ALL keeps them)</h4>
+                <pre><code class="language-sql text-sm">FOR u IN users FILTER u.age < 26 RETURN u._key
+UNION
+FOR u IN users FILTER u.city == 'London' RETURN u._key
+
+FOR i IN 1..3 RETURN i UNION ALL FOR i IN 1..3 RETURN i   -- [1, 2, 3, 1, 2, 3]</code></pre>
+              </div>
+              <div>
+                <h4 class="text-xs font-bold text-gray-400 uppercase tracking-wider mb-2">INTERSECT — rows present in both sides; EXCEPT — left rows absent from the right side</h4>
+                <pre><code class="language-sql text-sm">FOR u IN paris_users RETURN u._key
+INTERSECT
+FOR o IN newsletter RETURN o.user_key
+
+FOR u IN users RETURN u._key
+EXCEPT
+FOR b IN banned RETURN b.user_key
+
+-- Explicit grouping on either side
+(FOR u IN eu_users RETURN u._key UNION FOR u IN us_users RETURN u._key)
+EXCEPT
+FOR b IN banned RETURN b.user_key</code></pre>
+              </div>
+            </div>
+          </div>
+        </div>
+      </section>
+
+      <!-- DISTINCT and OFFSET -->
+      <section id="distinct-offset" class="scroll-mt-24">
+        <div class="bg-gray-800/30 rounded-xl border border-white/5 overflow-hidden">
+          <div class="p-5 border-b border-white/5 bg-white/5">
+            <h3 class="text-lg font-bold text-cyan-300 flex items-center">
+              <span class="bg-cyan-500/20 text-cyan-400 text-xs font-mono px-2 py-1 rounded mr-3">DISTINCT / OFFSET</span>
+              Result Shaping
+            </h3>
+          </div>
+          <div class="p-5 space-y-4">
+            <p class="text-sm text-gray-400">
+              <code class="text-cyan-300">RETURN DISTINCT expr</code> removes duplicate result rows
+              (first occurrence wins, order preserved). <code class="text-cyan-300">OFFSET n</code> skips rows —
+              standalone or after a limit as <code class="text-cyan-300">LIMIT count OFFSET n</code>.
+            </p>
+            <pre><code class="language-sql text-sm">-- Unique cities, no COLLECT needed
+FOR u IN users SORT u.city RETURN DISTINCT u.city
+
+-- Second page of 10, SQL style
+FOR d IN docs SORT d.created_at DESC LIMIT 10 OFFSET 10 RETURN d
+
+-- Skip without limiting
+FOR u IN users SORT u.age OFFSET 5 RETURN u._key</code></pre>
+          </div>
+        </div>
+      </section>
+
       <!-- Range Expressions -->
       <section id="range" class="scroll-mt-24">
         <div class="bg-gray-800/30 rounded-xl border border-white/5 overflow-hidden">
diff --git a/doc/public/js/docs-search.js b/doc/public/js/docs-search.js
index 202c043d..0bf5d025 100644
--- a/doc/public/js/docs-search.js
+++ b/doc/public/js/docs-search.js
@@ -248,6 +248,10 @@ var SDBQL_KEYWORDS = [
 ["RETURN","RETURN expression","Shape and return the result of the query.","/docs/sdbql-syntax#syntax"],
 ["SORT","SORT expr ASC | DESC","Order results by one or more expressions.","/docs/sdbql-syntax#syntax"],
 ["LIMIT","LIMIT offset?, count","Restrict the number of results, with an optional offset.","/docs/sdbql-syntax#syntax"],
+["OFFSET","OFFSET n [LIMIT count]","Skip rows before returning — standalone or after a limit (LIMIT 10 OFFSET 20).","/docs/sdbql-syntax#distinct-offset"],
+["UNION","query1 UNION [ALL] query2","Combine two query blocks; duplicates removed unless UNION ALL.","/docs/sdbql-syntax#set-operations"],
+["INTERSECT","query1 INTERSECT query2","Rows present in both sides of a set operation, deduplicated.","/docs/sdbql-syntax#set-operations"],
+["EXCEPT","query1 EXCEPT query2","Rows of the left side not present in the right side.","/docs/sdbql-syntax#set-operations"],
 ["LET","LET name = expression","Bind a variable or a subquery for reuse.","/docs/sdbql-syntax#let-subqueries"],
 ["COLLECT","COLLECT key = expr AGGREGATE …","Group rows and aggregate — SDBQL's GROUP BY.","/docs/sdbql-aggregations"],
 ["AGGREGATE","AGGREGATE total = SUM(x)","Compute aggregates within a COLLECT group.","/docs/sdbql-aggregations"],
@@ -272,7 +276,10 @@ var SDBQL_KEYWORDS = [
 ["WITH","WITH name AS ( subquery )","Define a named CTE (common table expression).","/docs/sdbql-cte"],
 ["OUTBOUND","FOR v IN OUTBOUND start edges","Traverse graph edges in the outbound direction.","/docs/sdbql-graphs"],
 ["INBOUND","FOR v IN INBOUND start edges","Traverse graph edges in the inbound direction.","/docs/sdbql-graphs"],
-["DISTINCT","RETURN DISTINCT expr","Return only unique results.","/docs/sdbql-syntax#syntax"],
+["DISTINCT","RETURN DISTINCT expr","Return only unique results.","/docs/sdbql-syntax#distinct-offset"],
+["RECURSIVE","WITH RECURSIVE t AS (anchor UNION ALL step)","Walk hierarchies: the step re-runs on the previous iteration's rows until empty.","/docs/sdbql-cte"],
+["NONE","NONE x IN arr SATISFIES cond","True when no element satisfies. Function form: NONE(arr, x -> cond).","/docs/sdbql-operators#op-none"],
+["KEEP","COLLECT … INTO g KEEP v1, v2","Restrict which variables are stored in COLLECT group arrays.","/docs/sdbql-aggregations"],
 ["LIKE","FILTER str LIKE \"%foo%\"","Pattern-match a string with % / _ wildcards.","/docs/sdbql-operators"]
 ];
 
diff --git a/src/sdbql/ast.rs b/src/sdbql/ast.rs
index cbd5927d..71ea818b 100644
--- a/src/sdbql/ast.rs
+++ b/src/sdbql/ast.rs
@@ -20,6 +20,33 @@ pub struct WithClause {
     pub ctes: Vec<CteClause>,
 }
 
+/// Set operation combining two query blocks: `a UNION b`, `a INTERSECT c`, ...
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub enum SetOperator {
+    /// UNION - concatenates and removes duplicates
+    Union,
+    /// UNION ALL - concatenates keeping duplicates
+    UnionAll,
+    /// INTERSECT - rows present in both sides, duplicates removed
+    Intersect,
+    /// EXCEPT - rows of the left side not present in the right side, duplicates removed
+    Except,
+}
+
+/// One operand on the right-hand side of a set operation
+#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
+pub struct SetOperationClause {
+    pub op: SetOperator,
+    pub query: Box<Query>,
+}
+
+impl SetOperator {
+    /// True for the `ALL` variants, which keep duplicate rows
+    pub fn is_all(&self) -> bool {
+        matches!(self, SetOperator::UnionAll)
+    }
+}
+
 /// AST node for a complete SDBQL query
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct Query {
@@ -49,6 +76,11 @@ pub struct Query {
     /// Ordered body clauses (FOR, LET, FILTER) preserving declaration order
     /// This enables correlated subqueries where LET can reference outer FOR variables
     pub body_clauses: Vec<BodyClause>,
+
+    /// Set operations applied after this query block: `q1 UNION q2 INTERSECT q3`
+    /// is parsed as `q1` with `set_operations = [UNION q2, INTERSECT q3]`.
+    #[serde(default)]
+    pub set_operations: Vec<SetOperationClause>,
 }
 
 impl Query {
@@ -67,6 +99,14 @@ impl Query {
         }) || self.create_stream_clause.is_some()
             || self.create_materialized_view_clause.is_some()
             || self.refresh_materialized_view_clause.is_some()
+            || self
+                .set_operations
+                .iter()
+                .any(|op| op.query.has_mutations())
+            || self
+                .with_clause
+                .as_ref()
+                .is_some_and(|with| with.ctes.iter().any(|cte| cte.query.has_mutations()))
     }
 }
 
@@ -323,13 +363,17 @@ pub struct JoinClause {
     pub asof: Option<AsofSpec>,
 }
 
-/// COLLECT var = expr [INTO group] [WITH COUNT INTO count] [AGGREGATE ...]
+/// COLLECT var = expr [INTO group [KEEP var1, var2]] [WITH COUNT INTO count] [AGGREGATE ...]
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct CollectClause {
     /// Group variables: (variable_name, expression) pairs
     pub group_vars: Vec<(String, Expression)>,
     /// INTO variable (collects grouped documents into an array)
     pub into_var: Option<String>,
+    /// Optional KEEP restriction on the variables stored in the INTO array.
+    /// Empty = keep every variable currently in scope (default).
+    #[serde(default)]
+    pub keep_vars: Vec<String>,
     /// WITH COUNT INTO variable
     pub count_var: Option<String>,
     /// AGGREGATE expressions
@@ -354,17 +398,23 @@ pub struct SortClause {
     pub fields: Vec<(Expression, bool)>, // (expression, ascending)
 }
 
-/// LIMIT [offset,] count
+/// LIMIT [offset,] count -- or a standalone OFFSET, which has no count
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct LimitClause {
     pub offset: Expression,
-    pub count: Expression,
+    /// Row count. `None` means "no upper bound" (`OFFSET n` without `LIMIT`):
+    /// callers must not substitute a sentinel maximum, because the count is
+    /// pushed down into storage scans and index lookups as an allocation hint.
+    pub count: Option<Expression>,
 }
 
-/// RETURN expression
+/// RETURN [DISTINCT] expression
 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
 pub struct ReturnClause {
     pub expression: Expression,
+    /// RETURN DISTINCT - remove duplicate result rows (first occurrence wins)
+    #[serde(default)]
+    pub distinct: bool,
 }
 
 /// Part of a template string (used in AST after parsing)
@@ -626,11 +676,18 @@ mod tests {
     fn test_limit_clause() {
         let clause = LimitClause {
             offset: Expression::Literal(json!(0)),
-            count: Expression::Literal(json!(10)),
+            count: Some(Expression::Literal(json!(10))),
         };
 
         assert_eq!(clause.offset, Expression::Literal(json!(0)));
-        assert_eq!(clause.count, Expression::Literal(json!(10)));
+        assert_eq!(clause.count, Some(Expression::Literal(json!(10))));
+
+        // A standalone OFFSET has no count at all
+        let unbounded = LimitClause {
+            offset: Expression::Literal(json!(5)),
+            count: None,
+        };
+        assert!(unbounded.count.is_none());
     }
 
     #[test]
@@ -711,6 +768,7 @@ mod tests {
             refresh_materialized_view_clause: None,
             window_clause: None,
             body_clauses: vec![],
+            set_operations: vec![],
         };
 
         assert!(query.for_clauses.is_empty());
@@ -728,6 +786,7 @@ mod tests {
                 ),
             )],
             into_var: Some("items".to_string()),
+            keep_vars: vec![],
             count_var: Some("cnt".to_string()),
             aggregates: vec![],
         };
diff --git a/src/sdbql/executor/execution/clauses.rs b/src/sdbql/executor/execution/clauses.rs
index a477ca30..72153b75 100644
--- a/src/sdbql/executor/execution/clauses.rs
+++ b/src/sdbql/executor/execution/clauses.rs
@@ -1172,12 +1172,36 @@ impl<'a> QueryExecutor<'a> {
                     for (_key, (mut group_ctx, group_docs, count)) in groups {
                         // Add INTO variable if present
                         if let Some(ref into_var) = collect.into_var {
+                            // A KEEP naming a variable that is not in scope
+                            // would silently store `{}` for every group item —
+                            // say so instead.
+                            if let Some(sample) = group_docs.first() {
+                                for keep in &collect.keep_vars {
+                                    if !sample.contains_key(keep) {
+                                        return Err(DbError::ExecutionError(format!(
+                                            "KEEP variable '{}' is not in scope at COLLECT",
+                                            keep
+                                        )));
+                                    }
+                                }
+                            }
+
                             let group_array: Vec<Value> = group_docs
                                 .iter()
                                 .map(|ctx| {
-                                    // Create an object with all variables in the context
-                                    let obj: serde_json::Map<String, Value> =
-                                        ctx.iter().map(|(k, v)| (k.clone(), v.clone())).collect();
+                                    // Create an object with the variables in the
+                                    // context, restricted to KEEP when provided
+                                    let obj: serde_json::Map<String, Value> = if collect
+                                        .keep_vars
+                                        .is_empty()
+                                    {
+                                        ctx.iter().map(|(k, v)| (k.clone(), v.clone())).collect()
+                                    } else {
+                                        ctx.iter()
+                                            .filter(|(k, _)| collect.keep_vars.contains(k))
+                                            .map(|(k, v)| (k.clone(), v.clone()))
+                                            .collect()
+                                    };
                                     Value::Object(obj)
                                 })
                                 .collect();
diff --git a/src/sdbql/executor/execution/entry.rs b/src/sdbql/executor/execution/entry.rs
index 9f1c532c..a802c7dc 100644
--- a/src/sdbql/executor/execution/entry.rs
+++ b/src/sdbql/executor/execution/entry.rs
@@ -10,8 +10,8 @@ use serde_json::Value;
 
 use super::super::types::{Context, MutationStats, QueryExecutionResult};
 use super::super::window::contains_window_functions;
-use super::super::QueryExecutor;
-use crate::error::DbResult;
+use super::super::{QueryExecutor, ValueSet};
+use crate::error::{DbError, DbResult};
 use crate::sdbql::ast::*;
 
 impl<'a> QueryExecutor<'a> {
@@ -33,43 +33,116 @@ impl<'a> QueryExecutor<'a> {
             return self.execute_refresh_materialized_view(clause);
         }
 
-        // First, evaluate initial LET clauses (before any FOR) to create initial binding
-        let mut initial_bindings: Context = HashMap::new();
-
-        // Merge bind variables into initial context
+        // Bind variables are the only bindings a top-level query starts with;
+        // everything else (CTEs, pre-FOR LETs) is part of the query prelude.
+        let mut bindings: Context = HashMap::new();
         for (key, value) in &self.bind_vars {
-            initial_bindings.insert(format!("@{}", key), value.clone());
+            bindings.insert(format!("@{}", key), value.clone());
         }
 
-        // Evaluate CTEs (Common Table Expressions) and store results in initial_bindings
-        // CTEs are evaluated sequentially so later CTEs can reference earlier ones
-        if let Some(ref with_clause) = query.with_clause {
-            for cte in &with_clause.ctes {
-                // Execute CTE query with access to previously computed CTEs
-                let cte_results = self.execute_cte_with_context(&cte.query, &initial_bindings)?;
-                tracing::debug!("CTE '{}' returned {} results", cte.name, cte_results.len());
-                // Store CTE results as an array in the context so FOR ... IN cte_name can iterate
-                initial_bindings.insert(cte.name.clone(), Value::Array(cte_results));
-            }
+        self.execute_query_with_bindings(query, bindings)
+    }
+
+    /// Execute a query block against a set of outer bindings.
+    ///
+    /// This is the single entry point that understands a *whole* query: set
+    /// operations, the `WITH` prelude, pre-FOR `LET`s, and then the execution
+    /// pipeline. Anything that runs a nested query block (CTE bodies, set
+    /// operation operands, recursive steps, correlated subqueries) goes through
+    /// here, so none of them can silently lose a clause.
+    pub(in crate::sdbql::executor) fn execute_query_with_bindings(
+        &self,
+        query: &Query,
+        bindings: Context,
+    ) -> DbResult<QueryExecutionResult> {
+        // Set operations: this query block is combined with further blocks
+        // (`q1 UNION q2`, `q1 INTERSECT q2`, `q1 EXCEPT q2`, ...). Each operand
+        // is executed independently, then combined in list order.
+        if !query.set_operations.is_empty() {
+            return self.execute_set_operations(query, bindings);
         }
 
+        let mut bindings = bindings;
+        self.bind_ctes(query, &mut bindings)?;
+        self.bind_pre_for_lets(query, &mut bindings)?;
+        self.execute_with_initial_bindings(query, bindings)
+    }
+
+    /// Evaluate this block's `WITH` clause into `bindings`.
+    ///
+    /// CTEs are evaluated sequentially so later ones can reference earlier ones.
+    fn bind_ctes(&self, query: &Query, bindings: &mut Context) -> DbResult<()> {
+        let Some(ref with_clause) = query.with_clause else {
+            return Ok(());
+        };
+
+        for cte in &with_clause.ctes {
+            let rows = if cte.recursive {
+                self.execute_recursive_cte(cte, bindings)?
+            } else {
+                self.execute_query_with_bindings(&cte.query, bindings.clone())?
+                    .results
+            };
+            tracing::debug!("CTE '{}' returned {} results", cte.name, rows.len());
+            // Stored as an array so `FOR x IN cte_name` can iterate it
+            bindings.insert(cte.name.clone(), Value::Array(rows));
+        }
+
+        Ok(())
+    }
+
+    /// Evaluate the `LET`s that precede the first `FOR` (evaluated once).
+    fn bind_pre_for_lets(&self, query: &Query, bindings: &mut Context) -> DbResult<()> {
         for let_clause in &query.let_clauses {
-            let value =
-                self.evaluate_expr_with_context(&let_clause.expression, &initial_bindings)?;
-            initial_bindings.insert(let_clause.variable.clone(), value);
+            let value = self.evaluate_expr_with_context(&let_clause.expression, bindings)?;
+            bindings.insert(let_clause.variable.clone(), value);
         }
+        Ok(())
+    }
 
-        self.execute_with_initial_bindings(query, initial_bindings)
+    /// Evaluate a LIMIT clause to `(offset, count)`. A `None` count means the
+    /// query has no upper bound (`OFFSET n` without `LIMIT`) — callers must not
+    /// stand in a maximum, because the count reaches storage as a scan size.
+    pub(in crate::sdbql::executor) fn eval_limit(
+        &self,
+        limit: &LimitClause,
+        ctx: &Context,
+    ) -> (usize, Option<usize>) {
+        let eval = |expr| {
+            self.evaluate_expr_with_context(expr, ctx)
+                .ok()
+                .and_then(|v| v.as_u64())
+                .map(|n| n as usize)
+                .unwrap_or(0)
+        };
+        (eval(&limit.offset), limit.count.as_ref().map(eval))
     }
 
     /// Execute the optimization + body-clause + sort/limit/return pipeline against a
-    /// fully-built initial binding set. Shared by top-level `execute_with_stats` and
-    /// the subquery executor so that subqueries get the same fast paths
-    /// (e.g. the `_key` index-sorted shortcut for SORT+LIMIT).
+    /// fully-built initial binding set, then apply `RETURN DISTINCT`.
+    ///
+    /// Deduplication lives here rather than in the pipeline because the pipeline
+    /// returns from a dozen fast paths, and every one of them must honour
+    /// `DISTINCT` (a columnar collection used to ignore it).
     pub(super) fn execute_with_initial_bindings(
         &self,
         query: &Query,
         initial_bindings: Context,
+    ) -> DbResult<QueryExecutionResult> {
+        let mut result = self.execute_pipeline(query, initial_bindings)?;
+        if query.return_clause.as_ref().is_some_and(|rc| rc.distinct) {
+            result.results = dedupe_values(result.results);
+        }
+        Ok(result)
+    }
+
+    /// The execution pipeline itself. Shared by top-level queries and the
+    /// subquery executor so subqueries get the same fast paths (e.g. the
+    /// `_key` index-sorted shortcut for SORT+LIMIT).
+    fn execute_pipeline(
+        &self,
+        query: &Query,
+        initial_bindings: Context,
     ) -> DbResult<QueryExecutionResult> {
         // Optimization: Streaming bulk INSERT for range-based FOR loops
         // Pattern: FOR i IN start..end INSERT {...} INTO collection [RETURN ...]
@@ -120,29 +193,23 @@ impl<'a> QueryExecutor<'a> {
 
                         let (sort_expr, sort_asc) = &sort.fields[0];
                         if iterates_collection {
-                            // Evaluate limit expressions
-                            let limit_offset = self
-                                .evaluate_expr_with_context(&limit.offset, &initial_bindings)
-                                .ok()
-                                .and_then(|v| v.as_u64())
-                                .map(|n| n as usize)
-                                .unwrap_or(0);
-                            let limit_count = self
-                                .evaluate_expr_with_context(&limit.count, &initial_bindings)
-                                .ok()
-                                .and_then(|v| v.as_u64())
-                                .map(|n| n as usize)
-                                .unwrap_or(0);
+                            // Evaluate limit expressions. An unbounded count
+                            // (`OFFSET n` with no `LIMIT`) fetches everything.
+                            let (limit_offset, limit_count) =
+                                self.eval_limit(limit, &initial_bindings);
 
                             // Check for overflow in limit_offset + limit_count
-                            let max_fetch = match limit_offset.checked_add(limit_count) {
-                                Some(sum) => sum,
-                                None => {
-                                    return Ok(QueryExecutionResult {
-                                        results: vec![],
-                                        mutations: MutationStats::new(),
-                                    });
-                                }
+                            let max_fetch = match limit_count {
+                                Some(count) => match limit_offset.checked_add(count) {
+                                    Some(sum) => Some(sum),
+                                    None => {
+                                        return Ok(QueryExecutionResult {
+                                            results: vec![],
+                                            mutations: MutationStats::new(),
+                                        });
+                                    }
+                                },
+                                None => None,
                             };
 
                             // Check if sort expression is a simple field access on the loop variable
@@ -152,14 +219,14 @@ impl<'a> QueryExecutor<'a> {
                                         if let Ok(collection) =
                                             self.get_collection(&for_clause.collection)
                                         {
-                                            if let Some(docs) = collection.index_sorted(
-                                                field,
-                                                *sort_asc,
-                                                Some(max_fetch),
-                                            ) {
+                                            if let Some(docs) =
+                                                collection.index_sorted(field, *sort_asc, max_fetch)
+                                            {
                                                 let start = limit_offset.min(docs.len());
-                                                let end = match start.checked_add(limit_count) {
-                                                    Some(sum) => sum.min(docs.len()),
+                                                let end = match limit_count {
+                                                    Some(count) => {
+                                                        start.saturating_add(count).min(docs.len())
+                                                    }
                                                     None => docs.len(),
                                                 };
                                                 let docs = &docs[start..end];
@@ -293,24 +360,10 @@ impl<'a> QueryExecutor<'a> {
                                     .unwrap_or(&for_clause.collection);
 
                                 if !initial_bindings.contains_key(source_name) {
-                                    let scan_limit = query.limit_clause.as_ref().map(|l| {
-                                        let offset = self
-                                            .evaluate_expr_with_context(
-                                                &l.offset,
-                                                &initial_bindings,
-                                            )
-                                            .ok()
-                                            .and_then(|v| v.as_u64())
-                                            .map(|n| n as usize)
-                                            .unwrap_or(0);
-                                        let count = self
-                                            .evaluate_expr_with_context(&l.count, &initial_bindings)
-                                            .ok()
-                                            .and_then(|v| v.as_u64())
-                                            .map(|n| n as usize)
-                                            .unwrap_or(0);
-                                        (offset, count)
-                                    });
+                                    let scan_limit = query
+                                        .limit_clause
+                                        .as_ref()
+                                        .map(|l| self.eval_limit(l, &initial_bindings));
 
                                     // This fast path bypasses `get_for_source_docs`,
                                     // so it needs its own columnar check —
@@ -319,7 +372,9 @@ impl<'a> QueryExecutor<'a> {
                                     // that still reports CollectionNotFound.
                                     if let Some(rows) = self.columnar_source_rows(
                                         &for_clause.collection,
-                                        scan_limit.map(|(offset, count)| offset + count),
+                                        scan_limit.and_then(|(offset, count)| {
+                                            count.map(|count| offset.saturating_add(count))
+                                        }),
                                     )? {
                                         let results = match scan_limit {
                                             Some((offset, _)) => {
@@ -334,10 +389,11 @@ impl<'a> QueryExecutor<'a> {
                                     }
 
                                     let collection = self.get_collection(&for_clause.collection)?;
-                                    let results = if let Some((offset, count)) = scan_limit {
-                                        collection.scan_values_range(offset, Some(count))
-                                    } else {
-                                        collection.scan_values(None)
+                                    let results = match scan_limit {
+                                        Some((offset, count)) => {
+                                            collection.scan_values_range(offset, count)
+                                        }
+                                        None => collection.scan_values(None),
                                     };
 
                                     return Ok(QueryExecutionResult {
@@ -367,19 +423,9 @@ impl<'a> QueryExecutor<'a> {
 
             if for_count == 1 && filter_count == 0 {
                 query.limit_clause.as_ref().and_then(|l| {
-                    let offset = self
-                        .evaluate_expr_with_context(&l.offset, &initial_bindings)
-                        .ok()
-                        .and_then(|v| v.as_u64())
-                        .map(|n| n as usize)
-                        .unwrap_or(0);
-                    let count = self
-                        .evaluate_expr_with_context(&l.count, &initial_bindings)
-                        .ok()
-                        .and_then(|v| v.as_u64())
-                        .map(|n| n as usize)
-                        .unwrap_or(0);
-                    offset.checked_add(count)
+                    let (offset, count) = self.eval_limit(l, &initial_bindings);
+                    // No count means no upper bound: nothing to push down.
+                    offset.checked_add(count?)
                 })
             } else {
                 None
@@ -403,20 +449,9 @@ impl<'a> QueryExecutor<'a> {
                 .is_none_or(|rc| !contains_window_functions(&rc.expression))
         {
             query.limit_clause.as_ref().and_then(|l| {
-                let offset = self
-                    .evaluate_expr_with_context(&l.offset, &initial_bindings)
-                    .ok()
-                    .and_then(|v| v.as_u64())
-                    .map(|n| n as usize)
-                    .unwrap_or(0);
-                let count = self
-                    .evaluate_expr_with_context(&l.count, &initial_bindings)
-                    .ok()
-                    .and_then(|v| v.as_u64())
-                    .map(|n| n as usize)
-                    .unwrap_or(0);
+                let (offset, count) = self.eval_limit(l, &initial_bindings);
                 // Fetch offset+count — the LIMIT below still applies the offset
-                offset.checked_add(count)
+                offset.checked_add(count?)
             })
         } else {
             None
@@ -459,19 +494,9 @@ impl<'a> QueryExecutor<'a> {
                 if !no_windows {
                     return None;
                 }
-                let offset = self
-                    .evaluate_expr_with_context(&limit.offset, &initial_bindings)
-                    .ok()
-                    .and_then(|v| v.as_u64())
-                    .map(|n| n as usize)
-                    .unwrap_or(0);
-                let count = self
-                    .evaluate_expr_with_context(&limit.count, &initial_bindings)
-                    .ok()
-                    .and_then(|v| v.as_u64())
-                    .map(|n| n as usize)
-                    .unwrap_or(0);
-                let k = offset.checked_add(count)?;
+                let (offset, count) = self.eval_limit(limit, &initial_bindings);
+                // Unbounded: every row survives, so top-k buys nothing.
+                let k = offset.checked_add(count?)?;
                 (k.saturating_mul(4) < rows.len()).then_some(k)
             });
             rows = match top_k {
@@ -487,27 +512,17 @@ impl<'a> QueryExecutor<'a> {
             }
         }
 
-        // Apply LIMIT
+        // Apply LIMIT (and/or a standalone OFFSET, which has no count)
         if let Some(limit) = &query.limit_clause {
-            let offset = self
-                .evaluate_expr_with_context(&limit.offset, &initial_bindings)
-                .ok()
-                .and_then(|v| v.as_u64())
-                .map(|n| n as usize)
-                .unwrap_or(0);
-            let count = self
-                .evaluate_expr_with_context(&limit.count, &initial_bindings)
-                .ok()
-                .and_then(|v| v.as_u64())
-                .map(|n| n as usize)
-                .unwrap_or(0);
+            let (offset, count) = self.eval_limit(limit, &initial_bindings);
 
             let start = offset.min(rows.len());
-            let end = (start + count).min(rows.len());
             if start > 0 {
                 rows.drain(0..start);
             }
-            rows.truncate(end - start);
+            if let Some(count) = count {
+                rows.truncate(count);
+            }
         }
 
         // Apply RETURN projection (if present)
@@ -528,34 +543,154 @@ impl<'a> QueryExecutor<'a> {
         })
     }
 
-    /// Execute a CTE query with access to a provided context (for CTE chaining)
-    fn execute_cte_with_context(
+    /// Combine this query block with its set-operation operands
+    /// (`UNION [ALL]` / `INTERSECT` / `EXCEPT`).
+    ///
+    /// The operand list is applied left to right; SQL's tighter binding for
+    /// `INTERSECT` is already expressed by the parser, which nests such an
+    /// operand inside the one before it.
+    fn execute_set_operations(
         &self,
         query: &Query,
+        bindings: Context,
+    ) -> DbResult<QueryExecutionResult> {
+        let mut left = query.clone();
+        let ops = std::mem::take(&mut left.set_operations);
+
+        // The `WITH` clause sits on the left block syntactically but binds for
+        // the whole combined query, so evaluate it once here and share it with
+        // every operand (evaluating it per operand would re-run a CTE body,
+        // and re-apply its mutations).
+        let mut shared = bindings;
+        self.bind_ctes(&left, &mut shared)?;
+
+        let mut left_bindings = shared.clone();
+        self.bind_pre_for_lets(&left, &mut left_bindings)?;
+        let mut acc = self.execute_with_initial_bindings(&left, left_bindings)?;
+
+        for op in &ops {
+            let rhs = self.execute_query_with_bindings(&op.query, shared.clone())?;
+            acc.mutations.documents_inserted += rhs.mutations.documents_inserted;
+            acc.mutations.documents_updated += rhs.mutations.documents_updated;
+            acc.mutations.documents_removed += rhs.mutations.documents_removed;
+
+            match op.op {
+                SetOperator::UnionAll => acc.results.extend(rhs.results),
+                SetOperator::Union => {
+                    // Duplicates go, wherever they came from: the left side is
+                    // deduplicated too, not just the incoming rows.
+                    let mut seen = ValueSet::with_capacity(acc.results.len() + rhs.results.len());
+                    acc.results.retain(|value| seen.insert(value));
+                    for value in rhs.results {
+                        if seen.insert(&value) {
+                            acc.results.push(value);
+                        }
+                    }
+                }
+                SetOperator::Intersect | SetOperator::Except => {
+                    let keep_when_present = op.op == SetOperator::Intersect;
+                    let mut right = ValueSet::with_capacity(rhs.results.len());
+                    for value in &rhs.results {
+                        right.insert(value);
+                    }
+                    let mut seen = ValueSet::with_capacity(acc.results.len());
+                    acc.results.retain(|value| {
+                        right.contains(value) == keep_when_present && seen.insert(value)
+                    });
+                }
+            }
+        }
+
+        Ok(acc)
+    }
+
+    /// Execute a recursive CTE: `WITH RECURSIVE t AS (<anchor> UNION ALL <step>)`.
+    ///
+    /// The anchor runs once. Then the step queries run repeatedly; inside them
+    /// the CTE name is bound to the rows produced by the *previous* iteration
+    /// (`FOR x IN t ...` sees the last batch). Iteration stops when a batch is
+    /// empty or the safety limits are hit.
+    pub(super) fn execute_recursive_cte(
+        &self,
+        cte: &CteClause,
         initial_bindings: &Context,
     ) -> DbResult<Vec<Value>> {
-        // CTEs don't support nested CTEs, so we just use the provided context
-        // Evaluate LET clauses first
-        let mut ctx = initial_bindings.clone();
-        for let_clause in &query.let_clauses {
-            let value = self.evaluate_expr_with_context(&let_clause.expression, &ctx)?;
-            ctx.insert(let_clause.variable.clone(), value);
+        const MAX_ITERATIONS: usize = 1000;
+        const MAX_ROWS: usize = 1_000_000;
+
+        if cte.query.set_operations.is_empty() {
+            return Err(DbError::ExecutionError(format!(
+                "Recursive CTE '{}' requires a body of the form \
+                 `<anchor query> UNION ALL <recursive step>`",
+                cte.name
+            )));
         }
 
-        // Execute body clauses
-        let (rows, _) = self.execute_body_clauses(&query.body_clauses, &ctx, None, None)?;
+        // Split into the anchor (the CTE body without set operations) and the
+        // iterative steps (every UNION ALL operand).
+        let mut anchor = (*cte.query).clone();
+        let steps = std::mem::take(&mut anchor.set_operations);
+        for op in &steps {
+            if op.op != SetOperator::UnionAll {
+                return Err(DbError::ExecutionError(format!(
+                    "Recursive CTE '{}' only supports UNION ALL between anchor and recursive steps",
+                    cte.name
+                )));
+            }
+        }
 
-        // Apply RETURN projection
-        let results = if let Some(ref return_clause) = query.return_clause {
-            let results: DbResult<Vec<Value>> = rows
-                .iter()
-                .map(|r| self.evaluate_expr_with_context(&return_clause.expression, r))
-                .collect();
-            results?
-        } else {
-            vec![]
-        };
+        // The anchor and the steps are full query blocks: they may carry their
+        // own `WITH`, pre-FOR `LET`s and nested set operations.
+        let mut accumulated = self
+            .execute_query_with_bindings(&anchor, initial_bindings.clone())?
+            .results;
+
+        let mut batch = accumulated.clone();
+        let mut iterations = 0usize;
+        while !batch.is_empty() {
+            iterations += 1;
+            if iterations > MAX_ITERATIONS {
+                return Err(DbError::ExecutionError(format!(
+                    "Recursive CTE '{}' exceeded {} iterations (possible infinite recursion)",
+                    cte.name, MAX_ITERATIONS
+                )));
+            }
 
-        Ok(results)
+            // Bind the CTE name to the previous batch for this iteration. The
+            // batch is moved in, not cloned — it is not needed again.
+            let mut ctx = initial_bindings.clone();
+            ctx.insert(cte.name.clone(), Value::Array(std::mem::take(&mut batch)));
+
+            let mut next = Vec::new();
+            for step in &steps {
+                let result = self.execute_query_with_bindings(&step.query, ctx.clone())?;
+                next.extend(result.results);
+            }
+
+            if next.is_empty() {
+                break;
+            }
+            if accumulated.len().saturating_add(next.len()) > MAX_ROWS {
+                return Err(DbError::ExecutionError(format!(
+                    "Recursive CTE '{}' produced more than {} rows",
+                    cte.name, MAX_ROWS
+                )));
+            }
+
+            // Continue iterating with the newly produced rows
+            accumulated.extend(next.iter().cloned());
+            batch = next;
+        }
+
+        Ok(accumulated)
     }
 }
+
+/// Remove duplicate values preserving first-occurrence order.
+///
+/// Row identity is `ValueSet`'s — the same value equality the `UNION()` and
+/// `INTERSECTION()` array builtins use, so `1` and `1.0` are one row here too.
+pub(super) fn dedupe_values(values: Vec<Value>) -> Vec<Value> {
+    let mut seen = ValueSet::with_capacity(values.len());
+    values.into_iter().filter(|v| seen.insert(v)).collect()
+}
diff --git a/src/sdbql/executor/execution/subquery.rs b/src/sdbql/executor/execution/subquery.rs
index 5f54e267..554deea4 100644
--- a/src/sdbql/executor/execution/subquery.rs
+++ b/src/sdbql/executor/execution/subquery.rs
@@ -12,11 +12,11 @@ use crate::sdbql::ast::*;
 impl<'a> QueryExecutor<'a> {
     /// Execute query with parent context for correlated subqueries.
     ///
-    /// Builds an initial binding set seeded by the parent context (for correlation),
-    /// adds bind variables, evaluates pre-FOR LET clauses, then dispatches to the
-    /// shared `execute_with_initial_bindings` pipeline so the subquery benefits from
-    /// the same optimizations as a top-level query (index-sorted SORT+LIMIT,
-    /// LIMIT pushdown, direct-scan, etc.).
+    /// Seeds the bindings with the parent context (for correlation) and the bind
+    /// variables, then dispatches to `execute_query_with_bindings` so a subquery
+    /// is executed exactly like a top-level query: its `WITH` prelude, pre-FOR
+    /// `LET`s and set operations all apply, and it gets the same optimizations
+    /// (index-sorted SORT+LIMIT, LIMIT pushdown, direct-scan, ...).
     pub(in crate::sdbql::executor) fn execute_with_parent_context(
         &self,
         query: &Query,
@@ -30,15 +30,8 @@ impl<'a> QueryExecutor<'a> {
             initial_bindings.insert(format!("@{}", key), value.clone());
         }
 
-        // Evaluate initial LET clauses (before FOR)
-        for let_clause in &query.let_clauses {
-            let value =
-                self.evaluate_expr_with_context(&let_clause.expression, &initial_bindings)?;
-            initial_bindings.insert(let_clause.variable.clone(), value);
-        }
-
         Ok(self
-            .execute_with_initial_bindings(query, initial_bindings)?
+            .execute_query_with_bindings(query, initial_bindings)?
             .results)
     }
 }
diff --git a/src/sdbql/executor/explain.rs b/src/sdbql/executor/explain.rs
index 7d6ac484..6cc30bae 100644
--- a/src/sdbql/executor/explain.rs
+++ b/src/sdbql/executor/explain.rs
@@ -74,20 +74,10 @@ impl<'a> QueryExecutor<'a> {
                 .count();
 
             if for_count == 1 && filter_count == 0 {
-                query.limit_clause.as_ref().map(|l| {
-                    let offset = self
-                        .evaluate_expr_with_context(&l.offset, &initial_bindings)
-                        .ok()
-                        .and_then(|v| v.as_u64())
-                        .map(|n| n as usize)
-                        .unwrap_or(0);
-                    let count = self
-                        .evaluate_expr_with_context(&l.count, &initial_bindings)
-                        .ok()
-                        .and_then(|v| v.as_u64())
-                        .map(|n| n as usize)
-                        .unwrap_or(0);
-                    offset + count
+                query.limit_clause.as_ref().and_then(|l| {
+                    let (offset, count) = self.eval_limit(l, &initial_bindings);
+                    // No count means no upper bound: nothing to push down.
+                    offset.checked_add(count?)
                 })
             } else {
                 None
@@ -263,24 +253,18 @@ impl<'a> QueryExecutor<'a> {
         // Apply LIMIT
         let mut documents_returned = rows.len();
         let mut limit_offset_val: usize = 0;
-        let mut limit_count_val: usize = 0;
+        let mut limit_count_val: Option<usize> = None;
         if let Some(limit) = &query.limit_clause {
             let limit_start = Instant::now();
-            limit_offset_val = self
-                .evaluate_expr_with_context(&limit.offset, &initial_bindings)
-                .ok()
-                .and_then(|v| v.as_u64())
-                .map(|n| n as usize)
-                .unwrap_or(0);
-            limit_count_val = self
-                .evaluate_expr_with_context(&limit.count, &initial_bindings)
-                .ok()
-                .and_then(|v| v.as_u64())
-                .map(|n| n as usize)
-                .unwrap_or(0);
+            let (offset, count) = self.eval_limit(limit, &initial_bindings);
+            limit_offset_val = offset;
+            limit_count_val = count;
 
             let start = limit_offset_val.min(rows.len());
-            let end = (start + limit_count_val).min(rows.len());
+            let end = match limit_count_val {
+                Some(count) => start.saturating_add(count).min(rows.len()),
+                None => rows.len(),
+            };
             rows = rows[start..end].to_vec();
             documents_returned = rows.len();
             limit_us = limit_start.elapsed().as_micros() as u64;
diff --git a/src/sdbql/executor/expression.rs b/src/sdbql/executor/expression.rs
index c964c6dd..85d9acee 100644
--- a/src/sdbql/executor/expression.rs
+++ b/src/sdbql/executor/expression.rs
@@ -898,6 +898,22 @@ impl<'a> QueryExecutor<'a> {
                 }
                 Ok(Value::Bool(false))
             }
+            "NONE" => {
+                for item in arr {
+                    let mut lambda_ctx = ctx.clone();
+                    if let Some(param) = params.first() {
+                        lambda_ctx.insert(param.clone(), item.clone());
+                    }
+                    if self
+                        .evaluate_expr_with_context(&body, &lambda_ctx)
+                        .map(|v| to_bool(&v))
+                        .unwrap_or(false)
+                    {
+                        return Ok(Value::Bool(false));
+                    }
+                }
+                Ok(Value::Bool(true))
+            }
             "REDUCE" => {
                 // REDUCE needs initial value - find non-lambda arg in original_args
                 let initial = original_args
diff --git a/src/sdbql/executor/types.rs b/src/sdbql/executor/types.rs
index a9952028..80522caf 100644
--- a/src/sdbql/executor/types.rs
+++ b/src/sdbql/executor/types.rs
@@ -112,7 +112,8 @@ pub struct SortInfo {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 pub struct LimitInfo {
     pub offset: usize,
-    pub count: usize,
+    /// Row count, or `null` for a standalone `OFFSET` (no upper bound)
+    pub count: Option<usize>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize)]
diff --git a/src/sdbql/parser/clauses.rs b/src/sdbql/parser/clauses.rs
index e1b6867a..c5e55728 100644
--- a/src/sdbql/parser/clauses.rs
+++ b/src/sdbql/parser/clauses.rs
@@ -442,6 +442,7 @@ impl Parser {
 
         let mut group_vars = Vec::new();
         let mut into_var = None;
+        let mut keep_vars = Vec::new();
         let mut count_var = None;
         let mut aggregates = Vec::new();
 
@@ -481,7 +482,7 @@ impl Parser {
             }
         }
 
-        // Parse optional INTO var
+        // Parse optional INTO var [KEEP var1, var2, ...]
         if matches!(self.current_token(), Token::Into) {
             self.advance(); // consume INTO
             if let Token::Identifier(var_name) = self.current_token() {
@@ -492,6 +493,27 @@ impl Parser {
                     "Expected variable name after INTO".to_string(),
                 ));
             }
+
+            // Optional KEEP restriction: only listed variables are stored in
+            // the group arrays. Must come before WITH COUNT / AGGREGATE.
+            if self.ident_eq("KEEP") {
+                self.advance();
+                loop {
+                    if let Token::Identifier(var_name) = self.current_token() {
+                        keep_vars.push(var_name.clone());
+                        self.advance();
+                    } else {
+                        return Err(DbError::ParseError(
+                            "Expected variable name after KEEP".to_string(),
+                        ));
+                    }
+                    if matches!(self.current_token(), Token::Comma) {
+                        self.advance();
+                    } else {
+                        break;
+                    }
+                }
+            }
         }
 
         // Parse optional WITH COUNT INTO var
@@ -582,16 +604,29 @@ impl Parser {
         Ok(CollectClause {
             group_vars,
             into_var,
+            keep_vars,
             count_var,
             aggregates,
         })
     }
 
-    /// Parse WITH clause for CTEs: WITH cte_name [(col1, col2, ...)] AS (query) [, cte_name AS (query)]*
-    /// Example: WITH temp AS (SELECT 1), temp2 AS (SELECT 2) SELECT * FROM temp
+    /// Parse WITH clause for CTEs:
+    /// `WITH [RECURSIVE] cte_name [(col1, col2, ...)] AS (query) [, cte_name AS (query)]*`
+    ///
+    /// When RECURSIVE is present it applies to every CTE in the list (standard SQL
+    /// semantics). A recursive CTE body must be `anchor UNION ALL step` where the
+    /// step query references the CTE name to see the rows produced by the previous
+    /// iteration.
     pub(crate) fn parse_with_clause(&mut self) -> DbResult<WithClause> {
         self.expect(Token::With)?;
 
+        let recursive = if matches!(self.current_token(), Token::Recursive) {
+            self.advance();
+            true
+        } else {
+            false
+        };
+
         let mut ctes = Vec::new();
 
         loop {
@@ -638,7 +673,7 @@ impl Parser {
             ctes.push(CteClause {
                 name: cte_name,
                 columns,
-                recursive: false, // RECURSIVE is parsed at the WITH level, not per-CTE
+                recursive,
                 query: Box::new(cte_query),
             });
 
@@ -1090,19 +1125,31 @@ impl Parser {
 
             Ok(LimitClause {
                 offset: first,
-                count,
+                count: Some(count),
             })
         } else {
             Ok(LimitClause {
                 offset: Expression::Literal(Value::Number(serde_json::Number::from(0))),
-                count: first,
+                count: Some(first),
             })
         }
     }
 
     pub(crate) fn parse_return_clause(&mut self) -> DbResult<ReturnClause> {
         self.expect(Token::Return)?;
+
+        // Optional DISTINCT: RETURN DISTINCT expr
+        let distinct = if self.ident_eq("DISTINCT") {
+            self.advance();
+            true
+        } else {
+            false
+        };
+
         let expression = self.parse_expression()?;
-        Ok(ReturnClause { expression })
+        Ok(ReturnClause {
+            expression,
+            distinct,
+        })
     }
 }
diff --git a/src/sdbql/parser/expressions/primary.rs b/src/sdbql/parser/expressions/primary.rs
index 6420b659..2df2bece 100644
--- a/src/sdbql/parser/expressions/primary.rs
+++ b/src/sdbql/parser/expressions/primary.rs
@@ -135,7 +135,19 @@ impl Parser {
     /// Parse primary expression (highest precedence)
     pub(super) fn parse_primary_expression(&mut self) -> DbResult<Expression> {
         match self.current_token() {
-            Token::Identifier(name) => self.parse_identifier_expression(name.clone()),
+            Token::Identifier(ref name) => {
+                // Quantifier: NONE x IN array SATISFIES condition
+                // (also accepts the parenthesized form NONE(x IN ... SATISFIES ...))
+                if name.eq_ignore_ascii_case("NONE") && self.is_none_quantifier() {
+                    let parenthesized = matches!(self.peek_token(1), Token::LeftParen);
+                    self.advance(); // consume NONE
+                    if parenthesized {
+                        self.advance(); // consume (
+                    }
+                    return self.parse_quantifier_expression_after_keyword("NONE", parenthesized);
+                }
+                self.parse_identifier_expression(name.clone())
+            }
             Token::Any => self.parse_quantifier_expression("ANY"),
             Token::Count => self.parse_keyword_as_function("COUNT"),
             Token::Left => self.parse_keyword_as_function_no_window(
@@ -186,6 +198,22 @@ impl Parser {
         }
     }
 
+    /// Detect the NONE quantifier form: `NONE x IN ...` or `NONE(x IN ...)`.
+    /// A plain function call like `NONE(arr)` or `NONE(arr, x -> cond)` does
+    /// not match (the token after the variable must be IN).
+    fn is_none_quantifier(&self) -> bool {
+        // NONE x IN ...
+        if matches!(self.peek_token(1), Token::Identifier(_))
+            && matches!(self.peek_token(2), Token::In)
+        {
+            return true;
+        }
+        // NONE(x IN ...)
+        matches!(self.peek_token(1), Token::LeftParen)
+            && matches!(self.peek_token(2), Token::Identifier(_))
+            && matches!(self.peek_token(3), Token::In)
+    }
+
     /// Parse identifier: variable or function call
     fn parse_identifier_expression(&mut self, name: String) -> DbResult<Expression> {
         // Check for lambda: x -> expr
diff --git a/src/sdbql/parser/expressions/special.rs b/src/sdbql/parser/expressions/special.rs
index f01f18b8..9487beda 100644
--- a/src/sdbql/parser/expressions/special.rs
+++ b/src/sdbql/parser/expressions/special.rs
@@ -278,9 +278,15 @@ impl Parser {
     // ========================================================================
 
     /// Parse quantifier expression: ANY x IN array SATISFIES condition
-    pub(super) fn parse_quantifier_expression(&mut self, name: &str) -> DbResult<Expression> {
-        self.advance(); // consume ANY/SOME/ALL
-
+    ///
+    /// The caller must have consumed the quantifier keyword (and the opening
+    /// parenthesis for the optional parenthesized form); `parenthesized`
+    /// selects whether a trailing ')' is consumed.
+    pub(super) fn parse_quantifier_expression_after_keyword(
+        &mut self,
+        name: &str,
+        parenthesized: bool,
+    ) -> DbResult<Expression> {
         let variable = if let Token::Identifier(v) = self.current_token() {
             v.clone()
         } else {
@@ -301,6 +307,10 @@ impl Parser {
             Expression::Literal(Value::Bool(true))
         };
 
+        if parenthesized {
+            self.expect(Token::RightParen)?;
+        }
+
         // Construct desugared ANY(array, x -> condition)
         let lambda = Expression::Lambda {
             params: vec![variable],
@@ -313,6 +323,12 @@ impl Parser {
         })
     }
 
+    /// Parse quantifier expression: ANY x IN array SATISFIES condition
+    pub(super) fn parse_quantifier_expression(&mut self, name: &str) -> DbResult<Expression> {
+        self.advance(); // consume ANY/SOME/ALL/NONE
+        self.parse_quantifier_expression_after_keyword(name, false)
+    }
+
     // ========================================================================
     // Template strings
     // ========================================================================
diff --git a/src/sdbql/parser/mod.rs b/src/sdbql/parser/mod.rs
index d8cc2940..88c5d415 100644
--- a/src/sdbql/parser/mod.rs
+++ b/src/sdbql/parser/mod.rs
@@ -96,13 +96,40 @@ impl Parser {
 
     /// Parse a query, optionally checking for trailing tokens (false for subqueries)
     pub(crate) fn parse_query(&mut self, check_trailing: bool) -> DbResult<Query> {
+        self.parse_query_with(check_trailing, true)
+    }
+
+    /// Parse a single query block, stopping *before* any set operator.
+    ///
+    /// Used for the bare (unparenthesized) operand of a set operation: if the
+    /// operand consumed the following operator itself, `a EXCEPT b EXCEPT c`
+    /// would nest to the right and mean `a EXCEPT (b EXCEPT c)`.
+    fn parse_query_block(&mut self) -> DbResult<Query> {
+        self.parse_query_with(false, false)
+    }
+
+    fn parse_query_with(&mut self, check_trailing: bool, parse_set_ops: bool) -> DbResult<Query> {
         self.check_depth()?;
-        let result = self.parse_query_inner(check_trailing);
+        let result = self.parse_query_inner(check_trailing, parse_set_ops);
         self.leave_depth();
         result
     }
 
-    pub(crate) fn parse_query_inner(&mut self, check_trailing: bool) -> DbResult<Query> {
+    pub(crate) fn parse_query_inner(
+        &mut self,
+        check_trailing: bool,
+        parse_set_ops: bool,
+    ) -> DbResult<Query> {
+        // A parenthesized leading block: `(FOR ...) UNION (FOR ...)`. The
+        // parentheses only group, so the inner block *is* this query and any
+        // following set operations hang off it.
+        if parse_set_ops
+            && matches!(self.current_token(), Token::LeftParen)
+            && starts_query(self.peek_token(1))
+        {
+            return self.parse_parenthesized_query(check_trailing);
+        }
+
         // Parse optional CREATE STREAM or CREATE MATERIALIZED VIEW
         let (create_stream_clause, create_mv_clause) =
             if matches!(self.current_token(), Token::Create) {
@@ -225,12 +252,39 @@ impl Parser {
             None
         };
 
-        let limit_clause = if matches!(self.current_token(), Token::Limit) {
+        // Optional standalone OFFSET before LIMIT: OFFSET n LIMIT m
+        let mut offset_expr: Option<Expression> = None;
+        if self.ident_eq("OFFSET") {
+            self.advance();
+            offset_expr = Some(self.parse_expression()?);
+        }
+
+        let mut limit_clause = if matches!(self.current_token(), Token::Limit) {
             Some(self.parse_limit_clause()?)
         } else {
             None
         };
 
+        // OFFSET after LIMIT: LIMIT n OFFSET m
+        if self.ident_eq("OFFSET") {
+            if offset_expr.is_some() {
+                return Err(DbError::ParseError("Duplicate OFFSET clause".to_string()));
+            }
+            self.advance();
+            offset_expr = Some(self.parse_expression()?);
+        }
+
+        // Merge a standalone OFFSET into the limit clause. Without LIMIT the
+        // offset applies alone and the row count stays unbounded (`None`) --
+        // never a sentinel maximum, which would reach storage as an
+        // allocation hint.
+        if let Some(offset) = offset_expr {
+            limit_clause = Some(LimitClause {
+                offset,
+                count: limit_clause.and_then(|limit| limit.count),
+            });
+        }
+
         // RETURN clause is optional - mutations (INSERT/UPDATE/REMOVE) don't require it
         let return_clause = if matches!(self.current_token(), Token::Return) {
             Some(self.parse_return_clause()?)
@@ -238,6 +292,16 @@ impl Parser {
             None
         };
 
+        // Set operations combining this query block with more blocks:
+        // q1 UNION [ALL] q2 INTERSECT q3 EXCEPT q4
+        let mut set_operations = Vec::new();
+        if parse_set_ops {
+            while let Some(op) = self.try_parse_set_operator() {
+                let operand = self.parse_set_operation_operand()?;
+                push_set_operation(&mut set_operations, op, operand);
+            }
+        }
+
         // Only validate at top-level, not for subqueries
         if check_trailing {
             // Validate that we have a valid query structure
@@ -309,8 +373,127 @@ impl Parser {
             return_clause,
             window_clause,
             body_clauses,
+            set_operations,
         })
     }
+
+    /// Try to parse a set-operation keyword (UNION [ALL] / INTERSECT / EXCEPT)
+    /// at the current position. Consumes the keyword(s) when found.
+    fn try_parse_set_operator(&mut self) -> Option<SetOperator> {
+        if self.ident_eq("UNION") {
+            self.advance();
+            if matches!(self.current_token(), Token::Identifier(id) if id.eq_ignore_ascii_case("ALL"))
+            {
+                self.advance();
+                Some(SetOperator::UnionAll)
+            } else {
+                Some(SetOperator::Union)
+            }
+        } else if self.ident_eq("INTERSECT") {
+            self.advance();
+            Some(SetOperator::Intersect)
+        } else if self.ident_eq("EXCEPT") {
+            self.advance();
+            Some(SetOperator::Except)
+        } else {
+            None
+        }
+    }
+
+    /// Parse a query whose left-hand side is parenthesized:
+    /// `(<query>) [UNION|INTERSECT|EXCEPT <operand>]*`.
+    fn parse_parenthesized_query(&mut self, check_trailing: bool) -> DbResult<Query> {
+        self.advance(); // consume (
+        let mut query = self.parse_query(false)?;
+        self.expect(Token::RightParen)?;
+
+        // Operations that follow apply to the whole parenthesized group, so the
+        // first one never re-groups: `(a UNION b) INTERSECT c` intersects the
+        // union, it does not intersect `b` alone.
+        let mut chain: Vec<SetOperationClause> = Vec::new();
+        while let Some(op) = self.try_parse_set_operator() {
+            let operand = self.parse_set_operation_operand()?;
+            if chain.is_empty() {
+                chain.push(SetOperationClause {
+                    op,
+                    query: Box::new(operand),
+                });
+            } else {
+                push_set_operation(&mut chain, op, operand);
+            }
+        }
+        query.set_operations.extend(chain);
+
+        if check_trailing && !matches!(self.current_token(), Token::Eof) {
+            return Err(DbError::ParseError(format!(
+                "Unexpected token after query: {:?}",
+                self.current_token()
+            )));
+        }
+
+        Ok(query)
+    }
+
+    /// Parse the right-hand operand of a set operation: either a parenthesized
+    /// query `(FOR ... RETURN ...)` or a plain query block starting with a
+    /// query keyword (FOR/LET/WITH/RETURN/mutation/DDL).
+    fn parse_set_operation_operand(&mut self) -> DbResult<Query> {
+        if matches!(self.current_token(), Token::LeftParen) && starts_query(self.peek_token(1)) {
+            self.advance();
+            let query = self.parse_query(false)?;
+            self.expect(Token::RightParen)?;
+            Ok(query)
+        } else if starts_query(self.current_token()) {
+            // A bare operand is one block only: the operator after it belongs
+            // to the enclosing chain, not to this operand.
+            self.parse_query_block()
+        } else {
+            Err(DbError::ParseError(
+                "Expected a query after UNION/INTERSECT/EXCEPT, e.g. UNION (FOR ...)".to_string(),
+            ))
+        }
+    }
+}
+
+/// True when this token can open a query block.
+fn starts_query(token: &Token) -> bool {
+    matches!(
+        token,
+        Token::For
+            | Token::Let
+            | Token::Return
+            | Token::Insert
+            | Token::Update
+            | Token::Remove
+            | Token::Upsert
+            | Token::With
+            | Token::Create
+            | Token::Refresh
+    )
+}
+
+/// Append one set operation to a chain, honouring SQL precedence.
+///
+/// `INTERSECT` binds tighter than `UNION` / `EXCEPT`, so it groups with the
+/// operand immediately before it rather than with everything accumulated so
+/// far: `a UNION b INTERSECT c` becomes `a UNION (b INTERSECT c)`. Operators of
+/// equal precedence chain left to right, which is what a flat list means to the
+/// executor.
+fn push_set_operation(ops: &mut Vec<SetOperationClause>, op: SetOperator, operand: Query) {
+    let is_intersect = op == SetOperator::Intersect;
+    let clause = SetOperationClause {
+        op,
+        query: Box::new(operand),
+    };
+
+    if is_intersect {
+        if let Some(previous) = ops.last_mut() {
+            previous.query.set_operations.push(clause);
+            return;
+        }
+    }
+
+    ops.push(clause);
 }
 
 /// Result of parsing a FOR clause - could be regular FOR or graph traversal
diff --git a/src/sdbql/parser/tests.rs b/src/sdbql/parser/tests.rs
index 91136df7..1669dfbe 100644
--- a/src/sdbql/parser/tests.rs
+++ b/src/sdbql/parser/tests.rs
@@ -233,6 +233,163 @@ fn test_no_cte() {
     assert!(query.with_clause.is_none());
 }
 
+#[test]
+fn test_recursive_cte() {
+    let query = parse(
+        "WITH RECURSIVE tree AS (FOR d IN nodes FILTER d._key == @root RETURN d._key \
+         UNION ALL FOR n IN nodes FILTER n.parent IN tree RETURN n._key) \
+         FOR x IN tree RETURN x",
+    );
+    assert!(
+        query.is_ok(),
+        "Failed to parse recursive CTE: {:?}",
+        query.err()
+    );
+    let query = query.unwrap();
+    let with = query.with_clause.unwrap();
+    assert_eq!(with.ctes.len(), 1);
+    assert!(with.ctes[0].recursive);
+    // Body must be anchor UNION ALL step
+    assert_eq!(with.ctes[0].query.set_operations.len(), 1);
+}
+
+#[test]
+fn test_return_distinct() {
+    let query = parse("FOR doc IN coll RETURN DISTINCT doc.city").unwrap();
+    let rc = query.return_clause.unwrap();
+    assert!(rc.distinct);
+
+    // Plain RETURN must not set the flag
+    let query = parse("FOR doc IN coll RETURN doc.city").unwrap();
+    assert!(!query.return_clause.unwrap().distinct);
+}
+
+#[test]
+fn test_set_operations() {
+    for (sql, expected) in [
+        (
+            "FOR a IN c1 RETURN a.x UNION FOR b IN c2 RETURN b.y",
+            "Union",
+        ),
+        (
+            "FOR a IN c1 RETURN a.x UNION ALL FOR b IN c2 RETURN b.y",
+            "UnionAll",
+        ),
+        (
+            "FOR a IN c1 RETURN a.x INTERSECT FOR b IN c2 RETURN b.y",
+            "Intersect",
+        ),
+        (
+            "FOR a IN c1 RETURN a.x EXCEPT FOR b IN c2 RETURN b.y",
+            "Except",
+        ),
+    ] {
+        let query = parse(sql).unwrap_or_else(|e| panic!("Failed to parse {sql}: {e:?}"));
+        assert_eq!(query.set_operations.len(), 1, "{sql}");
+        let op_name = format!("{:?}", query.set_operations[0].op);
+        assert_eq!(op_name, expected, "{sql}");
+    }
+}
+
+#[test]
+fn test_set_operation_chain_is_flat_and_left_to_right() {
+    // `a EXCEPT b EXCEPT c` must be one flat chain — nesting it to the right
+    // would mean `a EXCEPT (b EXCEPT c)`.
+    let query =
+        parse("FOR a IN c1 RETURN a.x EXCEPT FOR b IN c2 RETURN b.x EXCEPT FOR c IN c3 RETURN c.x")
+            .unwrap();
+    assert_eq!(query.set_operations.len(), 2);
+    assert!(query.set_operations[0].query.set_operations.is_empty());
+}
+
+#[test]
+fn test_intersect_binds_tighter_than_union() {
+    // `a UNION b INTERSECT c` groups as `a UNION (b INTERSECT c)`
+    let query = parse(
+        "FOR a IN c1 RETURN a.x UNION FOR b IN c2 RETURN b.x INTERSECT FOR c IN c3 RETURN c.x",
+    )
+    .unwrap();
+    assert_eq!(query.set_operations.len(), 1);
+    assert!(matches!(query.set_operations[0].op, SetOperator::Union));
+    let nested = &query.set_operations[0].query.set_operations;
+    assert_eq!(nested.len(), 1);
+    assert!(matches!(nested[0].op, SetOperator::Intersect));
+}
+
+#[test]
+fn test_parenthesized_left_operand() {
+    // Explicit grouping on the left: `(a UNION b) INTERSECT c` intersects the
+    // union, so the chain stays flat instead of nesting under `b`.
+    let query = parse(
+        "(FOR a IN c1 RETURN a.x UNION FOR b IN c2 RETURN b.x)          INTERSECT FOR c IN c3 RETURN c.x",
+    )
+    .unwrap();
+    assert_eq!(query.set_operations.len(), 2);
+    assert!(matches!(query.set_operations[0].op, SetOperator::Union));
+    assert!(matches!(query.set_operations[1].op, SetOperator::Intersect));
+    assert!(query.set_operations[0].query.set_operations.is_empty());
+}
+
+#[test]
+fn test_offset_without_limit_has_no_count() {
+    // A standalone OFFSET must not invent a count: the count is pushed into
+    // storage scans as an allocation size.
+    let query = parse("FOR d IN coll OFFSET 5 RETURN d").unwrap();
+    let limit = query.limit_clause.expect("OFFSET produces a limit clause");
+    assert!(limit.count.is_none());
+
+    let query = parse("FOR d IN coll LIMIT 10 OFFSET 5 RETURN d").unwrap();
+    let limit = query.limit_clause.expect("limit clause");
+    assert_eq!(
+        limit.count,
+        Some(Expression::Literal(serde_json::json!(10)))
+    );
+    assert_eq!(limit.offset, Expression::Literal(serde_json::json!(5)));
+}
+
+#[test]
+fn test_has_mutations_sees_nested_blocks() {
+    // The HTTP handler decides caching, transaction handling and write
+    // permission from this: a mutation hidden in an operand or a CTE body must
+    // not read as a read-only query.
+    let query =
+        parse("FOR a IN c1 RETURN a.x UNION FOR d IN c2 REMOVE d IN c2 RETURN d._key").unwrap();
+    assert!(query.has_mutations());
+
+    let query =
+        parse("WITH gone AS (FOR d IN c2 REMOVE d IN c2 RETURN d._key) FOR x IN gone RETURN x")
+            .unwrap();
+    assert!(query.has_mutations());
+
+    let query = parse("FOR a IN c1 RETURN a.x UNION FOR b IN c2 RETURN b.x").unwrap();
+    assert!(!query.has_mutations());
+}
+
+#[test]
+fn test_set_operations_parenthesized_operand() {
+    let query =
+        parse("FOR a IN c1 RETURN a.x UNION (FOR b IN c2 FILTER b.z > 1 RETURN b.y)").unwrap();
+    assert_eq!(query.set_operations.len(), 1);
+    assert!(matches!(query.set_operations[0].op, SetOperator::Union));
+}
+
+#[test]
+fn test_collect_keep() {
+    let query = parse(
+        "FOR u IN users COLLECT city = u.city INTO groups KEEP name, age SORT city RETURN city",
+    )
+    .unwrap();
+    let collect = query
+        .body_clauses
+        .iter()
+        .find_map(|c| match c {
+            BodyClause::Collect(cc) => Some(cc.clone()),
+            _ => None,
+        })
+        .expect("COLLECT clause");
+    assert_eq!(collect.keep_vars, vec!["name", "age"]);
+}
+
 #[test]
 fn test_parse_collect_with_aggregate_count() {
     let query =
diff --git a/src/server/handlers/query.rs b/src/server/handlers/query.rs
index 312685c1..96387e28 100644
--- a/src/server/handlers/query.rs
+++ b/src/server/handlers/query.rs
@@ -68,9 +68,11 @@ pub struct ExecuteQueryResponse {
 // ==================== Helper Functions ====================
 
 /// Check if a query is potentially long-running (contains mutations or range iterations)
-#[inline]
+///
+/// Looks into set-operation operands and CTE bodies too: an operand's `FOR`
+/// scan is just as blocking as a top-level one.
 pub(crate) fn is_long_running_query(query: &Query) -> bool {
-    query.body_clauses.iter().any(|clause| match clause {
+    let own = query.body_clauses.iter().any(|clause| match clause {
         BodyClause::Insert(_)
         | BodyClause::Update(_)
         | BodyClause::Remove(_)
@@ -80,7 +82,17 @@ pub(crate) fn is_long_running_query(query: &Query) -> bool {
         // 2. Collection scans might trigger scatter-gather with blocking HTTP calls
         BodyClause::For(_) => true,
         _ => false,
-    })
+    });
+
+    own || query
+        .set_operations
+        .iter()
+        .any(|op| is_long_running_query(&op.query))
+        || query.with_clause.as_ref().is_some_and(|with| {
+            with.ctes
+                .iter()
+                .any(|cte| is_long_running_query(&cte.query))
+        })
 }
 
 /// Get collection names affected by mutation clauses for targeted cache invalidation.
@@ -102,7 +114,7 @@ pub(crate) fn invalidate_collections(collections: &[String]) {
 }
 
 pub(crate) fn mutated_collections(query: &Query) -> std::collections::HashSet<&str> {
-    query
+    let mut collections: std::collections::HashSet<&str> = query
         .body_clauses
         .iter()
         .filter_map(|clause| match clause {
@@ -112,7 +124,19 @@ pub(crate) fn mutated_collections(query: &Query) -> std::collections::HashSet<&s
             BodyClause::Upsert(c) => Some(c.collection.as_str()),
             _ => None,
         })
-        .collect()
+        .collect();
+
+    // Set-operation operands and CTE bodies can carry their own mutations
+    for operand in query.set_operations.iter().map(|op| op.query.as_ref()) {
+        collections.extend(mutated_collections(operand));
+    }
+    if let Some(with) = &query.with_clause {
+        for cte in &with.ctes {
+            collections.extend(mutated_collections(&cte.query));
+        }
+    }
+
+    collections
 }
 
 /// Log slow query to _slow_queries collection (async, non-blocking)
@@ -274,13 +298,11 @@ pub async fn execute_query(
         let wal = tx_manager.wal().clone();
         let lock_manager = tx_manager.lock_manager().clone();
 
-        // Check if query contains mutation operations
-        let has_mutations = query.body_clauses.iter().any(|clause| {
-            matches!(
-                clause,
-                BodyClause::Insert(_) | BodyClause::Update(_) | BodyClause::Remove(_)
-            )
-        });
+        // Check if query contains mutation operations. `has_mutations()` also
+        // sees UPSERT and mutations nested in set-operation operands or CTE
+        // bodies — those must not take the read-only path, which bypasses the
+        // transaction's WAL and locks.
+        let has_mutations = query.has_mutations();
 
         if !has_mutations {
             // No mutations - just execute normally (read operations)
@@ -523,16 +545,11 @@ pub async fn execute_query(
     let prepared = crate::sdbql::get_prepared_statement_cache().parse_if_needed(&req.query)?;
     let query = prepared.query.as_ref();
 
-    // Check if query is cacheable (read-only with no mutations)
-    let is_read_only = !query.body_clauses.iter().any(|clause| {
-        matches!(
-            clause,
-            BodyClause::Insert(_)
-                | BodyClause::Update(_)
-                | BodyClause::Remove(_)
-                | BodyClause::Upsert(_)
-        )
-    });
+    // Check if query is cacheable (read-only with no mutations). This has to be
+    // the deep check: caching a query whose mutation hides in a set-operation
+    // operand or a CTE body would serve the cached rows on the next identical
+    // request and never run the mutation at all.
+    let is_read_only = !query.has_mutations();
 
     // Try to get cached result for read-only queries (unless cache is disabled).
     // Include db_name so queries on different databases don't share cache entries.
diff --git a/tests/columnar_sdbql_tests.rs b/tests/columnar_sdbql_tests.rs
index 95c94a40..9f1f6985 100644
--- a/tests/columnar_sdbql_tests.rs
+++ b/tests/columnar_sdbql_tests.rs
@@ -117,6 +117,27 @@ fn limit_with_offset_works() {
     assert_eq!(rows.len(), 2, "offset+count should skip the first row");
 }
 
+/// A standalone OFFSET has no count; the scan must stay bounded by the data,
+/// not by a sentinel maximum.
+#[test]
+fn offset_without_limit_works() {
+    let (storage, _d) = fixture();
+    let rows = run(&storage, "FOR m IN metrics OFFSET 1 RETURN m");
+    assert_eq!(rows.len(), 2);
+
+    let rows = run(&storage, "FOR m IN metrics OFFSET 1 RETURN m.value");
+    assert_eq!(rows.len(), 2);
+}
+
+/// The columnar scan is its own fast path in the executor and used to return
+/// before DISTINCT was applied.
+#[test]
+fn return_distinct_applies_to_columnar_rows() {
+    let (storage, _d) = fixture();
+    let rows = run(&storage, "FOR m IN metrics RETURN DISTINCT m.host");
+    assert_eq!(rows.len(), 2, "hosts a and b, once each: {rows:?}");
+}
+
 /// The point of making columnar a real data source: it composes with the rest
 /// of the language instead of living behind a separate API.
 #[test]
diff --git a/tests/sdbql_executor_coverage_tests.rs b/tests/sdbql_executor_coverage_tests.rs
index 92ab8bc5..c1cf03d4 100644
--- a/tests/sdbql_executor_coverage_tests.rs
+++ b/tests/sdbql_executor_coverage_tests.rs
@@ -462,7 +462,13 @@ fn test_explain_with_limit() {
 
     assert!(explain.limit.is_some());
     let limit_info = explain.limit.unwrap();
-    assert_eq!(limit_info.count, 3);
+    assert_eq!(limit_info.count, Some(3));
+
+    // A standalone OFFSET has no count to report
+    let explain = explain_query(&engine, "FOR u IN users OFFSET 1 RETURN u.name");
+    let limit_info = explain.limit.expect("OFFSET is reported as a limit clause");
+    assert_eq!(limit_info.offset, 1);
+    assert_eq!(limit_info.count, None);
 }
 
 #[test]
diff --git a/tests/sdbql_new_syntax_tests.rs b/tests/sdbql_new_syntax_tests.rs
new file mode 100644
index 00000000..49a94299
--- /dev/null
+++ b/tests/sdbql_new_syntax_tests.rs
@@ -0,0 +1,557 @@
+//! Tests for newly added SDBQL syntax:
+//! - Set operations: UNION [ALL] / INTERSECT / EXCEPT
+//! - WITH RECURSIVE (recursive CTEs)
+//! - RETURN DISTINCT
+//! - COLLECT ... INTO ... KEEP
+//! - NONE quantifier (`NONE x IN arr SATISFIES cond` and `NONE(arr, x -> cond)`)
+//! - Standalone OFFSET clause (with and without LIMIT)
+//!
+//! Regression tests for the bugs found reviewing the above live at the bottom.
+
+use serde_json::json;
+use solidb::storage::StorageEngine;
+use solidb::{parse, QueryExecutor};
+use tempfile::TempDir;
+
+fn execute_query(engine: &StorageEngine, query_str: &str) -> Vec<serde_json::Value> {
+    let query = parse(query_str).unwrap_or_else(|_| panic!("Failed to parse: {}", query_str));
+    let executor = QueryExecutor::new(engine);
+    executor
+        .execute(&query)
+        .unwrap_or_else(|_| panic!("Query failed: {}", query_str))
+}
+
+fn create_seeded_engine() -> (StorageEngine, TempDir) {
+    let tmp_dir = TempDir::new().expect("Failed to create temp dir");
+    let engine = StorageEngine::new(tmp_dir.path().to_str().unwrap())
+        .expect("Failed to create storage engine");
+
+    engine.create_collection("users".to_string(), None).unwrap();
+    let users = engine.get_collection("users").unwrap();
+    for doc in [
+        json!({"_key": "alice", "name": "Alice", "age": 30, "city": "Paris"}),
+        json!({"_key": "bob", "name": "Bob", "age": 25, "city": "London"}),
+        json!({"_key": "carol", "name": "Carol", "age": 35, "city": "Paris"}),
+        json!({"_key": "dave", "name": "Dave", "age": 28, "city": "Berlin"}),
+        json!({"_key": "eve", "name": "Eve", "age": 32, "city": "London"}),
+    ] {
+        users.insert(doc).unwrap();
+    }
+
+    // Org chart: alice manages bob; bob manages carol and dave
+    engine
+        .create_collection("employees".to_string(), None)
+        .unwrap();
+    let employees = engine.get_collection("employees").unwrap();
+    for doc in [
+        json!({"_key": "alice", "manager": null}),
+        json!({"_key": "bob", "manager": "alice"}),
+        json!({"_key": "carol", "manager": "bob"}),
+        json!({"_key": "dave", "manager": "bob"}),
+        json!({"_key": "zoe", "manager": null}),
+    ] {
+        employees.insert(doc).unwrap();
+    }
+
+    (engine, tmp_dir)
+}
+
+fn keys(mut results: Vec<serde_json::Value>) -> Vec<String> {
+    results.sort_by_key(|v| v.to_string());
+    results
+        .into_iter()
+        .map(|v| v.as_str().unwrap().to_string())
+        .collect()
+}
+
+// ============================================================================
+// UNION / INTERSECT / EXCEPT
+// ============================================================================
+
+#[test]
+fn test_union_dedupes() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "FOR u IN users FILTER u.age < 26 RETURN u._key \
+         UNION FOR u IN users FILTER u.city == 'London' RETURN u._key",
+    );
+    // {bob} U {bob, eve} = {bob, eve}
+    assert_eq!(keys(results), vec!["bob".to_string(), "eve".to_string()]);
+}
+
+#[test]
+fn test_union_all_keeps_duplicates() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "FOR u IN users FILTER u.age < 26 RETURN u._key \
+         UNION ALL FOR u IN users FILTER u.city == 'London' RETURN u._key",
+    );
+    assert_eq!(
+        keys(results),
+        vec!["bob".to_string(), "bob".to_string(), "eve".to_string()]
+    );
+}
+
+#[test]
+fn test_intersect() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "FOR u IN users FILTER u.age < 26 RETURN u._key \
+         INTERSECT FOR u IN users FILTER u.city == 'London' RETURN u._key",
+    );
+    assert_eq!(keys(results), vec!["bob".to_string()]);
+}
+
+#[test]
+fn test_except() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "FOR u IN users FILTER u.city == 'Paris' OR u.city == 'London' RETURN u._key \
+         EXCEPT FOR u IN users FILTER u.city == 'London' RETURN u._key",
+    );
+    assert_eq!(
+        keys(results),
+        vec!["alice".to_string(), "carol".to_string()]
+    );
+}
+
+#[test]
+fn test_union_parenthesized_operand() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "FOR u IN users FILTER u.age < 26 RETURN u._key \
+         UNION (FOR u IN users FILTER u.city == 'Berlin' RETURN u._key)",
+    );
+    assert_eq!(keys(results), vec!["bob".to_string(), "dave".to_string()]);
+}
+
+// ============================================================================
+// WITH RECURSIVE
+// ============================================================================
+
+#[test]
+fn test_recursive_cte_hierarchy() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    // Everyone in alice's reporting chain, found level by level
+    let results = execute_query(
+        &engine,
+        "WITH RECURSIVE reports AS (\
+             FOR e IN employees FILTER e._key == 'alice' RETURN e._key \
+             UNION ALL \
+             FOR m IN employees FILTER m.manager IN reports RETURN m._key \
+         ) FOR x IN reports RETURN x",
+    );
+    assert_eq!(
+        keys(results),
+        vec![
+            "alice".to_string(),
+            "bob".to_string(),
+            "carol".to_string(),
+            "dave".to_string()
+        ]
+    );
+}
+
+#[test]
+fn test_recursive_cte_requires_union_all() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let query = parse("WITH RECURSIVE t AS (FOR e IN employees RETURN e._key) FOR x IN t RETURN x")
+        .unwrap();
+    let executor = QueryExecutor::new(&engine);
+    let err = executor.execute(&query).unwrap_err().to_string();
+    assert!(err.contains("UNION ALL"), "unexpected error: {}", err);
+}
+
+// ============================================================================
+// RETURN DISTINCT
+// ============================================================================
+
+#[test]
+fn test_return_distinct() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(&engine, "FOR u IN users RETURN DISTINCT u.city");
+    assert_eq!(
+        keys(results),
+        vec![
+            "Berlin".to_string(),
+            "London".to_string(),
+            "Paris".to_string()
+        ]
+    );
+
+    // Without DISTINCT all five rows come back
+    let results = execute_query(&engine, "FOR u IN users RETURN u.city");
+    assert_eq!(results.len(), 5);
+}
+
+#[test]
+fn test_return_distinct_with_sort_limit() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "FOR u IN users SORT u.age DESC LIMIT 4 RETURN DISTINCT u.city",
+    );
+    // ages desc: carol(35), eve(32), alice(30), dave(28) => Paris, London, Paris, Berlin
+    assert_eq!(
+        keys(results),
+        vec![
+            "Berlin".to_string(),
+            "London".to_string(),
+            "Paris".to_string()
+        ]
+    );
+}
+
+// ============================================================================
+// COLLECT ... KEEP
+// ============================================================================
+
+#[test]
+fn test_collect_keep_restricts_variables() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "FOR u IN users LET n = u.name \
+         COLLECT city = u.city INTO g KEEP n SORT city LIMIT 1 \
+         RETURN {city: city, names: g[*].n}",
+    );
+    assert_eq!(results.len(), 1);
+    assert_eq!(results[0]["city"], json!("Berlin"));
+    assert_eq!(results[0]["names"], json!(["Dave"]));
+
+    // The group items must not contain other variables (u, n is kept)
+    let raw = execute_query(
+        &engine,
+        "FOR u IN users LET n = u.name \
+         COLLECT city = u.city INTO g KEEP n SORT city LIMIT 1 \
+         RETURN FIRST(g)",
+    );
+    let item = raw[0].as_object().unwrap();
+    assert!(
+        !item.contains_key("u") && !item.contains_key("doc"),
+        "KEEP should restrict stored variables: {:?}",
+        item.keys().collect::<Vec<_>>()
+    );
+}
+
+// ============================================================================
+// NONE quantifier
+// ============================================================================
+
+#[test]
+fn test_none_quantifier_satisfies() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(&engine, "RETURN NONE(x IN [1, 2, 3] SATISFIES x > 5)");
+    assert_eq!(results[0], json!(true));
+
+    let results = execute_query(&engine, "RETURN NONE(x IN [1, 2, 3] SATISFIES x > 2)");
+    assert_eq!(results[0], json!(false));
+
+    // Function form still works
+    let results = execute_query(&engine, "RETURN NONE([1, 2, 3], x -> x > 5)");
+    assert_eq!(results[0], json!(true));
+}
+
+#[test]
+fn test_none_in_filter() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    // Cities with no resident younger than 26
+    let results = execute_query(
+        &engine,
+        "FOR c IN ['Paris', 'London', 'Berlin'] \
+         FILTER NONE(u IN (FOR p IN users FILTER p.city == c RETURN p.age) SATISFIES u < 26) \
+         SORT c \
+         RETURN c",
+    );
+    assert_eq!(
+        keys(results),
+        vec!["Berlin".to_string(), "Paris".to_string()]
+    );
+}
+
+// ============================================================================
+// OFFSET
+// ============================================================================
+
+#[test]
+fn test_limit_offset() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "FOR u IN users SORT u.age ASC LIMIT 2 OFFSET 1 RETURN u._key",
+    );
+    // ages asc: bob(25), dave(28), alice(30), eve(32), carol(35); skip bob take 2
+    assert_eq!(keys(results), vec!["alice".to_string(), "dave".to_string()]);
+}
+
+#[test]
+fn test_standalone_offset_without_limit() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "FOR u IN users SORT u.age ASC OFFSET 3 RETURN u._key",
+    );
+    assert_eq!(keys(results), vec!["carol".to_string(), "eve".to_string()]);
+}
+
+// ============================================================================
+// Regressions
+// ============================================================================
+
+/// A standalone OFFSET used to reach storage as a sentinel maximum count,
+/// which `scan_values_range` turned into `Vec::with_capacity(i64::MAX)`.
+/// Every shape below skips the SORT fast path, so they all hit that scan.
+#[test]
+fn test_standalone_offset_without_sort() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    // Direct-scan fast path: FOR ... OFFSET n RETURN var
+    let results = execute_query(&engine, "FOR u IN users OFFSET 2 RETURN u");
+    assert_eq!(results.len(), 3);
+
+    // Projection path
+    let results = execute_query(&engine, "FOR u IN users OFFSET 2 RETURN u._key");
+    assert_eq!(results.len(), 3);
+
+    // Offset past the end is empty, not an error
+    let results = execute_query(&engine, "FOR u IN users OFFSET 99 RETURN u._key");
+    assert!(results.is_empty());
+
+    // Inside a subquery
+    let results = execute_query(
+        &engine,
+        "LET rest = (FOR u IN users OFFSET 4 RETURN u._key) RETURN LENGTH(rest)",
+    );
+    assert_eq!(results[0], json!(1));
+
+    // Inside a set-operation operand
+    let results = execute_query(
+        &engine,
+        "FOR u IN users FILTER u.city == 'Berlin' RETURN u._key \
+         UNION ALL FOR v IN users OFFSET 4 RETURN v._key",
+    );
+    assert_eq!(results.len(), 2);
+
+    // With a FILTER (index/filter pushdown path)
+    let results = execute_query(
+        &engine,
+        "FOR u IN users FILTER u.city == 'Paris' OFFSET 1 RETURN u._key",
+    );
+    assert_eq!(results.len(), 1);
+}
+
+/// UNION only deduplicated the incoming side, so duplicates already present
+/// in the left operand survived.
+#[test]
+fn test_union_dedupes_left_side_too() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    // Paris and London each have two users
+    let results = execute_query(
+        &engine,
+        "FOR u IN users RETURN u.city \
+         UNION FOR v IN users FILTER v.city == 'Rome' RETURN v.city",
+    );
+    assert_eq!(
+        keys(results),
+        vec![
+            "Berlin".to_string(),
+            "London".to_string(),
+            "Paris".to_string()
+        ]
+    );
+
+    // ... and both sides at once
+    let results = execute_query(
+        &engine,
+        "FOR u IN users RETURN u.city UNION FOR v IN users RETURN v.city",
+    );
+    assert_eq!(results.len(), 3);
+
+    // UNION ALL still keeps every row
+    let results = execute_query(
+        &engine,
+        "FOR u IN users RETURN u.city UNION ALL FOR v IN users RETURN v.city",
+    );
+    assert_eq!(results.len(), 10);
+}
+
+/// Set operations compare rows by value, like the UNION()/INTERSECTION()
+/// array builtins — 1 and 1.0 are the same row.
+#[test]
+fn test_set_operations_use_value_equality() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(&engine, "RETURN 1 UNION RETURN 1.0");
+    assert_eq!(results.len(), 1);
+
+    let results = execute_query(&engine, "RETURN 1 INTERSECT RETURN 1.0");
+    assert_eq!(results.len(), 1);
+}
+
+/// A recursive CTE body is a full query block: its pre-FOR LETs used to be
+/// dropped, which silently produced an empty result instead of the hierarchy.
+#[test]
+fn test_recursive_cte_body_honours_let() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "WITH RECURSIVE reports AS (\
+             LET root = 'alice' \
+             FOR e IN employees FILTER e._key == root RETURN e._key \
+             UNION ALL \
+             LET previous = reports \
+             FOR m IN employees FILTER m.manager IN previous RETURN m._key \
+         ) FOR x IN reports RETURN x",
+    );
+    assert_eq!(
+        keys(results),
+        vec![
+            "alice".to_string(),
+            "bob".to_string(),
+            "carol".to_string(),
+            "dave".to_string()
+        ]
+    );
+}
+
+/// A CTE declared before a set operation binds for every operand, not just
+/// the left one (it used to fail with "Collection 'x' not found").
+#[test]
+fn test_cte_visible_in_set_operation_operands() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "WITH parisians AS (FOR u IN users FILTER u.city == 'Paris' RETURN u._key) \
+         FOR a IN parisians FILTER a == 'alice' RETURN a \
+         UNION FOR b IN parisians FILTER b == 'carol' RETURN b",
+    );
+    assert_eq!(
+        keys(results),
+        vec!["alice".to_string(), "carol".to_string()]
+    );
+}
+
+/// Non-recursive CTE bodies went through a reduced pipeline that ignored
+/// SORT/LIMIT and set operations.
+#[test]
+fn test_cte_body_honours_sort_limit_and_set_operations() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "WITH oldest AS (FOR u IN users SORT u.age DESC LIMIT 2 RETURN u._key) \
+         FOR x IN oldest RETURN x",
+    );
+    assert_eq!(results, vec![json!("carol"), json!("eve")]);
+
+    let results = execute_query(
+        &engine,
+        "WITH both AS (\
+             FOR u IN users FILTER u.city == 'Berlin' RETURN u._key \
+             UNION FOR v IN users FILTER v.city == 'London' RETURN v._key\
+         ) FOR x IN both RETURN x",
+    );
+    assert_eq!(
+        keys(results),
+        vec!["bob".to_string(), "dave".to_string(), "eve".to_string()]
+    );
+}
+
+/// Set operations inside a subquery used to be dropped silently.
+#[test]
+fn test_set_operation_in_subquery() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let results = execute_query(
+        &engine,
+        "LET combined = (\
+             FOR u IN users FILTER u.city == 'Berlin' RETURN u._key \
+             UNION FOR v IN users FILTER v.city == 'London' RETURN v._key\
+         ) RETURN LENGTH(combined)",
+    );
+    assert_eq!(results[0], json!(3));
+}
+
+/// Chained set operators follow SQL precedence: INTERSECT binds tighter than
+/// UNION / EXCEPT, and same-precedence operators chain left to right. A bare
+/// operand used to swallow the next operator, which nested chains to the right
+/// and made `a EXCEPT b EXCEPT c` mean `a EXCEPT (b EXCEPT c)`.
+#[test]
+fn test_set_operation_precedence() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    // Berlin UNION (London INTERSECT London) = {dave} U {bob, eve}
+    let results = execute_query(
+        &engine,
+        "FOR u IN users FILTER u.city == 'Berlin' RETURN u._key \
+         UNION FOR v IN users FILTER v.city == 'London' RETURN v._key \
+         INTERSECT FOR w IN users FILTER w.city == 'London' RETURN w._key",
+    );
+    assert_eq!(
+        keys(results),
+        vec!["bob".to_string(), "dave".to_string(), "eve".to_string()]
+    );
+
+    // ((all EXCEPT Paris) EXCEPT London) = {dave}, not all EXCEPT (Paris EXCEPT London)
+    let results = execute_query(
+        &engine,
+        "FOR u IN users RETURN u._key \
+         EXCEPT FOR v IN users FILTER v.city == 'Paris' RETURN v._key \
+         EXCEPT FOR w IN users FILTER w.city == 'London' RETURN w._key",
+    );
+    assert_eq!(keys(results), vec!["dave".to_string()]);
+
+    // Parentheses override: Berlin UNION (London INTERSECT Paris) = {dave}
+    let results = execute_query(
+        &engine,
+        "FOR u IN users FILTER u.city == 'Berlin' RETURN u._key \
+         UNION (FOR v IN users FILTER v.city == 'London' RETURN v._key \
+                INTERSECT FOR w IN users FILTER w.city == 'Paris' RETURN w._key)",
+    );
+    assert_eq!(keys(results), vec!["dave".to_string()]);
+
+    // (Paris UNION London) EXCEPT London = Paris
+    let results = execute_query(
+        &engine,
+        "(FOR u IN users FILTER u.city == 'Paris' RETURN u._key \
+           UNION FOR v IN users FILTER v.city == 'London' RETURN v._key) \
+         EXCEPT FOR w IN users FILTER w.city == 'London' RETURN w._key",
+    );
+    assert_eq!(
+        keys(results),
+        vec!["alice".to_string(), "carol".to_string()]
+    );
+}
+
+/// KEEP naming a variable that is not in scope used to store `{}` per item.
+#[test]
+fn test_collect_keep_unknown_variable_errors() {
+    let (engine, _tmp) = create_seeded_engine();
+
+    let query = parse("FOR u IN users COLLECT city = u.city INTO g KEEP nosuch RETURN g").unwrap();
+    let executor = QueryExecutor::new(&engine);
+    let err = executor.execute(&query).unwrap_err().to_string();
+    assert!(err.contains("nosuch"), "unexpected error: {}", err);
+}