diff --git a/lib/efsql.ex b/lib/efsql.ex
index 109ca7e..cfb22a5 100644
--- a/lib/efsql.ex
+++ b/lib/efsql.ex
@@ -5,33 +5,33 @@ defmodule Efsql do
   A statement flows through the textbook pipeline:
 
       SQL text
-        |> Efsql.Parser.to_logical()     # parse tree -> Efsql.Logical.Select
+        |> Efsql.Parser.sql_to_logical() # parse tree -> Efsql.Logical.Select
         |> resolve tenant
         |> Efsql.Rewrite.normalize()     # rewrite passes
         |> Efsql.Planner.plan()          # access-path selection -> Efsql.Physical.Plan
         |> Efsql.Executor.run()          # adapter pull + operator pipeline
-  """
-
-  import Ecto.Query
-
-  def hello() do
-    tenant = EctoFoundationDB.Tenant.open!(Efsql.Repo, "localhost")
 
-    query = from(s in "secrets", select: [id: s.id, iv: s.iv])
-    r1 = Efsql.Repo.all(query, prefix: tenant)
+  A query across tenants (`*.table`) is planned by `Efsql.Fanout` instead,
+  which plans each tenant's read with `Efsql.Planner`.
 
-    r2 = all("select id, iv from localhost.secrets;")
-
-    {r1, r2}
-  end
+  The CLI and the TUI run statements in an `Efsql.Session` (open tenants,
+  settings, the active tenant) and show the `Efsql.Result` it returns.
+  `qall/3` and `all/2` run one outside any session.
+  """
 
   def all(sql, options \\ []) do
     {_, result, _tenants} = qall(sql, options)
     result
   end
 
+  @doc """
+  Runs one statement outside any session, reusing and returning the
+  `tenants` cache: `{plan, rows, tenants}`. See `Efsql.Session` for the
+  CLI's and TUI's way in.
+  """
   def qall(sql, options \\ [], tenants \\ %{}) do
-    sql |> Efsql.Parser.sql_to_logical() |> run_logical(options, tenants)
+    {result, session} = Efsql.Session.run(%Efsql.Session{tenants: tenants}, sql, options)
+    {result.plan, result.rows, session.tenants}
   end
 
   @doc """
@@ -52,27 +52,12 @@ defmodule Efsql do
     {plan, Efsql.Executor.run(plan), tenants}
   end
 
-  def stream(sql) do
-    {logical, _tenants} = sql_to_logical(sql)
-    query = logical |> Efsql.Rewrite.normalize() |> Efsql.Planner.to_ecto_query()
-    {query, Efsql.Repo.stream(query)}
-  end
-
-  def sql_to_logical(sql, tenants \\ %{}) do
-    logical = %Efsql.Logical.Select{} = Efsql.Parser.sql_to_logical(sql)
-    resolve_tenant(logical, tenants)
-  end
-
-  def resolve_tenant(%Efsql.Logical.Select{prefix: {:all_tenants, _}}, _tenants) do
-    raise Efsql.Exception.Unsupported, "a query across tenants can't be streamed"
-  end
-
   # Already resolved, e.g. to the TUI session's active tenant.
-  def resolve_tenant(%Efsql.Logical.Select{prefix: nil, tenant: tenant} = logical, tenants)
-      when tenant != nil,
-      do: {logical, tenants}
+  defp resolve_tenant(%Efsql.Logical.Select{prefix: nil, tenant: tenant} = logical, tenants)
+       when tenant != nil,
+       do: {logical, tenants}
 
-  def resolve_tenant(%Efsql.Logical.Select{} = logical, tenants) do
+  defp resolve_tenant(%Efsql.Logical.Select{} = logical, tenants) do
     {tenant_name, storage_id} =
       case logical.prefix do
         {storage_id, tenant_name} -> {tenant_name, storage_id}
@@ -80,30 +65,42 @@ defmodule Efsql do
         tenant_name -> {tenant_name, nil}
       end
 
-    if not Map.has_key?(tenants, {tenant_name, storage_id}) and
-         not EctoFoundationDB.Tenant.exists?(Efsql.Repo, tenant_name) do
-      raise Efsql.Exception.Unsupported, "Tenant '#{tenant_name}' does not exist"
-    end
-
-    {tenant, tenants} = open_tenant(tenants, tenant_name, storage_id)
+    {tenant, tenants} = open_tenant(tenants, tenant_name, storage_id, check_exists: true)
     {%Efsql.Logical.Select{logical | tenant: tenant}, tenants}
   end
 
   @doc """
   Opens a tenant through the session's cache of open tenants, keyed by
   name and storage id (nil for the Repo's own).
+
+  Read-only: the tenant is opened with `migrate: false`, and
+  `Tenant.open/3` only opens (never creates), so efsql never writes to the
+  database it is exploring. A storage id other than the Repo's gets its
+  tenant cache started first. With `check_exists: true`, a tenant that
+  doesn't exist in that storage id is an `Efsql.Exception.Unsupported`
+  error; tenants just listed from the storage id skip the check.
   """
-  def open_tenant(tenants, tenant_name, storage_id) do
+  def open_tenant(tenants, tenant_name, storage_id, options \\ []) do
     case Map.fetch(tenants, {tenant_name, storage_id}) do
       {:ok, tenant} ->
         {tenant, tenants}
 
       :error ->
+        if storage_id, do: Efsql.Discover.ensure_storage_cache(storage_id)
+        config = Efsql.Repo.config()
+        config = if storage_id, do: Keyword.put(config, :storage_id, storage_id), else: config
+
+        if Keyword.get(options, :check_exists, false) and
+             not EctoFoundationDB.Tenant.Backend.exists?(
+               Ecto.Adapters.FoundationDB.db(Efsql.Repo),
+               tenant_name,
+               config
+             ) do
+          raise Efsql.Exception.Unsupported, "Tenant '#{tenant_name}' does not exist"
+        end
+
         open_opts = if storage_id, do: [storage_id: storage_id], else: []
 
-        # migrate: false keeps this read-only. Tenant.open/3 already requires
-        # the tenant to exist (only open!/3 creates), so with the migration
-        # step skipped efsql never writes to the database it is exploring.
         tenant =
           EctoFoundationDB.Tenant.open(Efsql.Repo, tenant_name, [migrate: false] ++ open_opts)
 
diff --git a/lib/efsql/aggregate.ex b/lib/efsql/aggregate.ex
index 3ced7db..23b17b4 100644
--- a/lib/efsql/aggregate.ex
+++ b/lib/efsql/aggregate.ex
@@ -25,7 +25,7 @@ defmodule Efsql.Aggregate do
   @spec run([map], [atom], [aggregate]) :: [map]
   def run(rows, group_by, aggregates) do
     rows
-    |> Enum.group_by(fn row -> Enum.map(group_by, &key(Map.get(row, &1))) end)
+    |> Enum.group_by(fn row -> Enum.map(group_by, &Types.equality_key(Map.get(row, &1))) end)
     |> ensure_one_group(group_by)
     |> Enum.map(fn {_key, rows} ->
       first = List.first(rows, %{})
@@ -35,20 +35,13 @@ defmodule Efsql.Aggregate do
         Map.put(acc, name, compute(function, arg, rows))
       end)
     end)
-    |> Enum.sort(&(compare_groups(&1, &2, group_by) != :gt))
+    |> Efsql.Executor.sort(Enum.map(group_by, &{:asc, &1}))
   end
 
   # A whole-table aggregate answers even when no row matched.
   defp ensure_one_group(groups, []) when groups == %{}, do: [{[], []}]
   defp ensure_one_group(groups, _group_by), do: Enum.to_list(groups)
 
-  # A term that is equal for values Types.compare/2 calls equal.
-  defp key(%Decimal{} = d), do: Decimal.normalize(d)
-  defp key(%NaiveDateTime{microsecond: {us, _}} = t), do: %{t | microsecond: {us, 6}}
-  defp key(%DateTime{microsecond: {us, _}} = t), do: %{t | microsecond: {us, 6}}
-  defp key(%Time{microsecond: {us, _}} = t), do: %{t | microsecond: {us, 6}}
-  defp key(value), do: value
-
   # -- aggregates --
 
   defp compute(:count, :star, rows), do: length(rows)
@@ -105,28 +98,4 @@ defmodule Efsql.Aggregate do
   defp decimal(%Decimal{} = d), do: d
   defp decimal(n) when is_integer(n), do: Decimal.new(n)
   defp decimal(n) when is_float(n), do: Decimal.from_float(n)
-
-  # -- group order --
-
-  # By key, NULLs last, like an ascending ORDER BY.
-  defp compare_groups(_a, _b, []), do: :eq
-
-  defp compare_groups(a, b, [field | rest]) do
-    case {Map.get(a, field), Map.get(b, field)} do
-      {nil, nil} ->
-        compare_groups(a, b, rest)
-
-      {nil, _} ->
-        :gt
-
-      {_, nil} ->
-        :lt
-
-      {va, vb} ->
-        case Types.compare(va, vb) do
-          :eq -> compare_groups(a, b, rest)
-          cmp -> cmp
-        end
-    end
-  end
 end
diff --git a/lib/efsql/cli.ex b/lib/efsql/cli.ex
index ddaab66..fa1b4ef 100644
--- a/lib/efsql/cli.ex
+++ b/lib/efsql/cli.ex
@@ -1,5 +1,5 @@
 defmodule Efsql.Cli do
-  defstruct args: [], history: [], debug: false, tenants: %{}, limit: 15, tenant_batch: nil
+  defstruct args: [], history: [], debug: false, session: %Efsql.Session{}
 
   use GenServer
 
@@ -92,9 +92,7 @@ defmodule Efsql.Cli do
         """
         Meta-commands:
           \\tenants [storage_id]  list tenants (optionally for a specific storage id)
-          \\set limit N           set the default row limit (currently #{state.limit})
-          \\set tenant_batch N|off  read *.table queries N tenants per transaction
-                                 (currently #{state.tenant_batch || "off"})
+        #{settings_help(state.session.settings)}
           \\?                     show this help
         """,
         :light_black
@@ -140,57 +138,41 @@ defmodule Efsql.Cli do
     state
   end
 
-  defp handle_input("\\set limit " <> rest, state = %__MODULE__{}) do
-    case Integer.parse(String.trim(rest)) do
-      {n, ""} when n > 0 ->
-        Owl.IO.puts(Owl.Data.tag("limit set to #{n}", :light_black))
-        %__MODULE__{state | limit: n}
+  defp handle_input("\\set " <> rest, state = %__MODULE__{}) do
+    case Efsql.Settings.set(state.session.settings, rest) do
+      {:ok, settings, message} ->
+        Owl.IO.puts(Owl.Data.tag(message, :light_black))
+        %__MODULE__{state | session: %{state.session | settings: settings}}
 
-      _ ->
-        print_error("Usage: \\set limit <positive integer>")
-        state
-    end
-  end
-
-  defp handle_input("\\set tenant_batch " <> rest, state = %__MODULE__{}) do
-    case {String.trim(rest), Integer.parse(String.trim(rest))} do
-      {"off", _} ->
-        Owl.IO.puts(Owl.Data.tag("tenant_batch off: one transaction", :light_black))
-        %__MODULE__{state | tenant_batch: nil}
-
-      {_, {n, ""}} when n > 0 ->
-        Owl.IO.puts(Owl.Data.tag("tenant_batch set to #{n}", :light_black))
-        %__MODULE__{state | tenant_batch: n}
-
-      _ ->
-        print_error("Usage: \\set tenant_batch <positive integer> | off")
+      {:error, usage} ->
+        print_error(usage)
         state
     end
   end
 
   defp handle_input(data, state = %__MODULE__{}) do
-    limit_sql = "limit #{state.limit + 1}"
+    limit = state.session.settings.limit
+    limit_sql = "limit #{limit + 1}"
 
-    {tenants} =
+    session =
       try do
         {sql, display_limit} =
           if String.match?(data, ~r/\blimit\b/i),
             do: {data, :all},
-            else: {String.replace(data, ~r/;\s*$/, " #{limit_sql};"), state.limit}
-
-        options = if state.tenant_batch, do: [tenant_batch: state.tenant_batch], else: []
-        {call, rows, tenants} = Efsql.qall(sql, options, state.tenants)
-        if state.debug, do: print_debug(call)
-        print_table(rows, display_limit, call.columns)
-        print_snapshots(Efsql.Fanout.transactions(call))
-        {tenants}
+            else: {String.replace(data, ~r/;\s*$/, " #{limit_sql};"), limit}
+
+        {result, session} = Efsql.Session.run(state.session, sql)
+        if state.debug, do: print_debug(result.plan)
+        print_table(result.rows, display_limit, result.columns)
+        print_transactions(result.transactions)
+        session
       rescue
         e ->
           print_error(e)
-          {state.tenants}
+          state.session
       end
 
-    %__MODULE__{state | history: [data | state.history], tenants: tenants}
+    %__MODULE__{state | history: [data | state.history], session: session}
   end
 
   def init_ecto_foundationdb!(args) do
@@ -258,12 +240,10 @@ defmodule Efsql.Cli do
     print_rows(display_rows, more?, columns)
   end
 
-  # Columns in select-list order; `select *` (no columns) sorts them.
   defp print_rows(rows, more?, columns) do
     rows
     |> Enum.map(fn row ->
-      row = if columns, do: Map.new(columns, &{&1, Map.get(row, &1)}), else: row
-      Map.new(row, fn {k, v} -> {to_string(k), format_value(v)} end)
+      Map.new(columns, &{to_string(&1), format_value(Map.get(row, &1))})
     end)
     |> Owl.Table.new(
       border_style: :solid_rounded,
@@ -277,14 +257,19 @@ defmodule Efsql.Cli do
     Owl.IO.puts(Owl.Data.tag(label, :light_black))
   end
 
-  defp print_snapshots(1), do: :ok
+  defp settings_help(settings) do
+    Enum.map_join(Efsql.Settings.describe(settings), "\n", fn {command, what} ->
+      "  " <> String.pad_trailing(command, 22) <> " " <> what
+    end)
+  end
 
-  defp print_snapshots(n) do
+  defp print_transactions(1), do: :ok
+
+  defp print_transactions(n) do
     Owl.IO.puts(Owl.Data.tag("(read in #{n} transactions)", :yellow))
   end
 
-  defp column_sorter(nil), do: :asc
-
+  # Owl sorts a table's columns; keep them in the result's order.
   defp column_sorter(columns) do
     position = columns |> Enum.map(&to_string/1) |> Enum.with_index() |> Map.new()
     &(Map.fetch!(position, &1) <= Map.fetch!(position, &2))
diff --git a/lib/efsql/discover.ex b/lib/efsql/discover.ex
index 20a0be3..5d7b7b8 100644
--- a/lib/efsql/discover.ex
+++ b/lib/efsql/discover.ex
@@ -171,9 +171,7 @@ defmodule Efsql.Discover do
   end
 
   def indexes(tenant, source) do
-    EctoFoundationDB.Layer.Metadata.transactional(tenant, source, fn _tx, metadata ->
-      for idx <- metadata.indexes, do: %{name: idx[:id], fields: idx[:fields]}
-    end)
+    for idx <- Planner.indexes(tenant, source), do: %{name: idx[:id], fields: idx[:fields]}
   end
 
   # -- schema cache --
diff --git a/lib/efsql/executor.ex b/lib/efsql/executor.ex
index ff36a6d..242f67f 100644
--- a/lib/efsql/executor.ex
+++ b/lib/efsql/executor.ex
@@ -17,20 +17,12 @@ defmodule Efsql.Executor do
 
   alias Efsql.Exception.Unsupported
   alias Efsql.Physical.Plan
+  alias Efsql.Predicate
   alias Efsql.Types
 
   # transaction_too_old (retryable) and transaction_timed_out
   @too_long [1007, 1031]
 
-  @cmp_ops ~w[== > >= < <=]a
-  @cmp_results %{
-    ==: [:eq],
-    >: [:gt],
-    >=: [:gt, :eq],
-    <: [:lt],
-    <=: [:lt, :eq]
-  }
-
   def run(%Plan{access: access, ops: ops}) do
     Enum.reduce(ops, fetch(access), &apply_op/2)
   end
@@ -68,23 +60,8 @@ defmodule Efsql.Executor do
     end
   end
 
-  @doc "Whether `row` satisfies every one of `predicates`, with SQL NULL semantics."
-  def matches?(row, predicates), do: Enum.all?(predicates, &eval(&1, row))
-
   # -- access nodes --
 
-  defp fetch({:pk_range, query, id_start, id_end, options}) do
-    Efsql.Repo.all_range(query, id_start, id_end, options)
-  end
-
-  defp fetch({:index_scan, query, options}) do
-    Efsql.Repo.all(query, options)
-  end
-
-  defp fetch({:all_from_source, query, options}) do
-    Efsql.Repo.all_from_source(query, options)
-  end
-
   # One transaction per batch, up to `concurrency` at a time.
   defp fetch({:batches, batches, concurrency}),
     do: batches |> map_concurrently(concurrency, &fetch/1) |> Enum.concat()
@@ -96,12 +73,9 @@ defmodule Efsql.Executor do
     Ecto.Adapters.FoundationDB.transactional(db, fn -> read_tenants(tenant_plans) end)
   end
 
-  defp fetch({:union, nodes}) do
-    nodes
-    |> Enum.map(&async_fetch/1)
-    |> Efsql.Repo.await()
-    |> List.flatten()
-  end
+  # Every other node, a union included, is one or more adapter reads:
+  # start them all, then await them together, so they are pipelined.
+  defp fetch(access), do: access |> start() |> Efsql.Repo.await() |> List.flatten()
 
   # Runs inside the transaction. A read too big for one transaction fails
   # with transaction_too_old, which erlfdb's transactional loop would retry,
@@ -160,17 +134,14 @@ defmodule Efsql.Executor do
 
   # -- operators --
 
-  defp apply_op({:filter, predicates}, rows) do
-    Enum.filter(rows, fn row -> Enum.all?(predicates, &eval(&1, row)) end)
-  end
+  defp apply_op({:filter, predicates}, rows),
+    do: Enum.filter(rows, &Predicate.matches?(&1, predicates))
 
   defp apply_op({:aggregate, group_by, aggregates}, rows) do
     Efsql.Aggregate.run(rows, group_by, aggregates)
   end
 
-  defp apply_op({:sort, sort}, rows) do
-    Enum.sort(rows, fn a, b -> compare(a, b, sort) != :gt end)
-  end
+  defp apply_op({:sort, order}, rows), do: sort(rows, order)
 
   defp apply_op({:limit, n}, rows) do
     Enum.take(rows, n)
@@ -180,55 +151,14 @@ defmodule Efsql.Executor do
     Enum.map(rows, &Map.take(&1, fields))
   end
 
-  # -- predicate evaluation --
-
-  defp eval({:cmp, op, field, param}, row) when op in @cmp_ops do
-    case Map.get(row, field) do
-      nil -> false
-      value -> Types.compare(value, param) in Map.fetch!(@cmp_results, op)
-    end
-  end
-
-  defp eval({:range, field, {lower_op, lower}, {upper_op, upper}}, row) do
-    eval({:cmp, lower_op, field, lower}, row) and eval({:cmp, upper_op, field, upper}, row)
-  end
-
-  defp eval({:in, field, values}, row) do
-    case Map.get(row, field) do
-      nil -> false
-      value -> Enum.any?(values, &(Types.compare(value, &1) == :eq))
-    end
-  end
-
-  defp eval({:is_null, field}, row), do: Map.get(row, field) == nil
-  defp eval({:not_null, field}, row), do: Map.get(row, field) != nil
-
-  defp eval({:like, field, pattern}, row) do
-    case Map.get(row, field) do
-      nil -> false
-      value -> Regex.match?(like_regex(pattern), value)
-    end
-  end
-
-  defp eval({:not_like, field, pattern}, row) do
-    case Map.get(row, field) do
-      nil -> false
-      value -> not Regex.match?(like_regex(pattern), value)
-    end
-  end
-
-  defp like_regex(pattern) do
-    source =
-      pattern
-      |> Regex.escape()
-      |> String.replace("%", ".*")
-      |> String.replace("_", ".")
-
-    Regex.compile!("\\A" <> source <> "\\z", "s")
-  end
-
   # -- sorting --
 
+  @doc """
+  Sorts rows by `[{:asc | :desc, field}]`: NULLs last ascending and first
+  descending, as PostgreSQL does, and values by `Efsql.Types.compare/2`.
+  """
+  def sort(rows, order), do: Enum.sort(rows, fn a, b -> compare(a, b, order) != :gt end)
+
   defp compare(_a, _b, []), do: :eq
 
   defp compare(a, b, [{dir, field} | rest]) do
diff --git a/lib/efsql/fanout.ex b/lib/efsql/fanout.ex
index cc6eda9..01bbc4d 100644
--- a/lib/efsql/fanout.ex
+++ b/lib/efsql/fanout.ex
@@ -28,10 +28,10 @@ defmodule Efsql.Fanout do
   """
 
   alias Efsql.Exception.Unsupported
-  alias Efsql.Executor
   alias Efsql.Logical
   alias Efsql.Physical.Plan
   alias Efsql.Planner
+  alias Efsql.Predicate
 
   @default_max_tenants 100
   @default_batch_concurrency 4
@@ -70,12 +70,12 @@ defmodule Efsql.Fanout do
     end
 
     {tenant_preds, row_preds} =
-      Enum.split_with(logical.predicates, &(Logical.predicate_field(&1) == :_tenant))
+      Enum.split_with(logical.predicates, &(Predicate.field(&1) == :_tenant))
 
     names =
       storage_id
       |> list_tenants()
-      |> Enum.filter(&Executor.matches?(%{_tenant: &1}, tenant_preds))
+      |> Enum.filter(&Predicate.matches?(%{_tenant: &1}, tenant_preds))
       |> Enum.sort()
 
     if batch == nil and length(names) > max_tenants do
diff --git a/lib/efsql/logical.ex b/lib/efsql/logical.ex
index 474227e..9c34b0b 100644
--- a/lib/efsql/logical.ex
+++ b/lib/efsql/logical.ex
@@ -3,14 +3,7 @@ defmodule Efsql.Logical do
   The logical query representation: what the SQL statement asks for,
   independent of how it will be executed.
 
-  Predicates are an implicitly AND-ed list of:
-
-    * `{:cmp, op, field, value}` — op in `:== :> :>= :< :<=`
-    * `{:range, field, {lower_op, value}, {upper_op, value}}` — a two-sided
-      bound, `lower_op` in `:> :>=`, `upper_op` in `:< :<=`
-    * `{:like, field, pattern}` / `{:not_like, field, pattern}`
-    * `{:in, field, values}`
-    * `{:is_null, field}` / `{:not_null, field}`
+  Its predicates are an implicitly AND-ed list of `Efsql.Predicate`s.
 
   The primary key is the pseudo-field `:_`. A query across every tenant of
   a storage id (`*.table`) also has `:_tenant`, the tenant's name, which
@@ -46,12 +39,4 @@ defmodule Efsql.Logical do
               # [{output_name :: atom, function :: atom, field :: atom | :star}]
               aggregates: []
   end
-
-  def predicate_field({:cmp, _op, field, _value}), do: field
-  def predicate_field({:range, field, _lower, _upper}), do: field
-  def predicate_field({:like, field, _pattern}), do: field
-  def predicate_field({:not_like, field, _pattern}), do: field
-  def predicate_field({:in, field, _values}), do: field
-  def predicate_field({:is_null, field}), do: field
-  def predicate_field({:not_null, field}), do: field
 end
diff --git a/lib/efsql/parser.ex b/lib/efsql/parser.ex
index 3d4637b..71fe778 100644
--- a/lib/efsql/parser.ex
+++ b/lib/efsql/parser.ex
@@ -49,7 +49,7 @@ defmodule Efsql.Parser do
   defp check_tenant_field(logical) do
     used =
       if(is_list(logical.projection), do: logical.projection, else: []) ++
-        Enum.map(logical.predicates, &Logical.predicate_field/1) ++
+        Enum.map(logical.predicates, &Efsql.Predicate.field/1) ++
         Enum.map(logical.order, &elem(&1, 1)) ++
         (logical.group_by || []) ++ Enum.map(logical.aggregates, &elem(&1, 2))
 
diff --git a/lib/efsql/physical.ex b/lib/efsql/physical.ex
index 9964c05..02e6829 100644
--- a/lib/efsql/physical.ex
+++ b/lib/efsql/physical.ex
@@ -31,5 +31,6 @@ defmodule Efsql.Physical do
 
   defmodule Plan do
     defstruct access: nil, ops: [], columns: nil
+    @type t :: %__MODULE__{access: tuple(), ops: [tuple()], columns: [atom()] | nil}
   end
 end
diff --git a/lib/efsql/planner.ex b/lib/efsql/planner.ex
index a18923b..234ea34 100644
--- a/lib/efsql/planner.ex
+++ b/lib/efsql/planner.ex
@@ -26,7 +26,7 @@ defmodule Efsql.Planner do
   alias Efsql.Exception.Unsupported
   alias Efsql.Logical
   alias Efsql.Physical.Plan
-  alias Efsql.Types
+  alias Efsql.Predicate
   alias EctoFoundationDB.Layer.Metadata
 
   @pk_field :_
@@ -130,23 +130,6 @@ defmodule Efsql.Planner do
     end
   end
 
-  @doc """
-  A plain Ecto query with every predicate as a where clause, for callers
-  that hand the query to the adapter unplanned (e.g. `Repo.stream`).
-  """
-  def to_ecto_query(%Logical.Select{group_by: group_by}) when is_list(group_by),
-    do: raise(Unsupported, "GROUP BY and aggregates can't be streamed")
-
-  def to_ecto_query(%Logical.Select{} = logical) do
-    take = if logical.projection == :star, do: nil, else: logical.projection
-
-    base_query(logical)
-    |> put_select(take)
-    |> put_wheres(logical.predicates)
-    |> put_order(logical.order)
-    |> put_limit(logical.limit)
-  end
-
   # -- SELECT planning --
 
   defp plan_select(logical, options, pushables, residuals, indexes, star?) do
@@ -186,7 +169,7 @@ defmodule Efsql.Planner do
     if idx == nil do
       field == :id or Enum.any?(indexes, fn ix -> List.first(ix[:fields]) == field end)
     else
-      equal_fields = for {:cmp, :==, f, _v} <- pushed, do: f
+      equal_fields = for p <- pushed, Predicate.equality?(p), do: Predicate.field(p)
       List.first(idx[:fields] -- equal_fields) == field
     end
   end
@@ -221,7 +204,7 @@ defmodule Efsql.Planner do
     |> Enum.filter(fn {_idx, pushed, _rest} -> pushed != [] end)
     |> Enum.max_by(
       fn {_idx, pushed, _rest} ->
-        {length(pushed), Enum.count(pushed, &match?({:cmp, :==, _, _}, &1))}
+        {length(pushed), Enum.count(pushed, &Predicate.equality?/1)}
       end,
       fn -> {nil, [], pushables} end
     )
@@ -232,58 +215,38 @@ defmodule Efsql.Planner do
   defp match_index([], pushables, acc), do: {Enum.reverse(acc), pushables}
 
   defp match_index([field | rest_fields], pushables, acc) do
-    case take_pred(pushables, field, &match?({:cmp, :==, _, _}, &1)) do
+    case take_pred(pushables, field, &Predicate.equality?/1) do
       {eq, rest} when eq != nil ->
         match_index(rest_fields, rest, [eq | acc])
 
       {nil, _} ->
-        case take_pred(pushables, field, &range?/1) do
+        case take_pred(pushables, field, &Predicate.range?/1) do
           {range, rest} when range != nil -> {Enum.reverse([range | acc]), rest}
           {nil, _} -> {Enum.reverse(acc), pushables}
         end
     end
   end
 
-  defp take_pred(preds, field, pred_fun) do
-    case Enum.split_while(preds, fn p ->
-           not (Logical.predicate_field(p) == field and pred_fun.(p))
-         end) do
-      {_before, []} -> {nil, preds}
-      {before, [match | rest]} -> {match, before ++ rest}
-    end
-  end
+  defp take_pred(preds, field, pred_fun),
+    do: Predicate.take_first(preds, &(Predicate.field(&1) == field and pred_fun.(&1)))
 
-  defp range?({:range, _field, _lower, _upper}), do: true
-  defp range?({:cmp, op, _field, _value}) when op in ~w[> >= < <=]a, do: true
-  defp range?(_), do: false
+  defp load_indexes(%Logical.Select{tenant: tenant, source: source}),
+    do: indexes(tenant, source)
 
-  defp load_indexes(%Logical.Select{tenant: tenant, source: source}) do
+  @doc "A source's indexes, from the tenant's metadata, each with its `:id` and `:fields`."
+  def indexes(tenant, source) do
     Metadata.transactional(tenant, source, fn _tx, metadata -> metadata.indexes end)
   end
 
   # -- classification --
 
+  # Grouped by how they can be served (see Predicate.pushdown/1), each
+  # group in its original order.
   defp classify(preds) do
-    Enum.reduce(preds, {[], [], [], []}, fn pred, {pks, ins, pushables, residuals} ->
-      case kind(pred) do
-        :pk -> {pks ++ [pred], ins, pushables, residuals}
-        :in -> {pks, ins ++ [pred], pushables, residuals}
-        :pushable -> {pks, ins, pushables ++ [pred], residuals}
-        :residual -> {pks, ins, pushables, residuals ++ [pred]}
-      end
-    end)
+    groups = Enum.group_by(preds, &Predicate.pushdown/1)
+    {groups[:key] || [], groups[:in] || [], groups[:index] || [], groups[:filter] || []}
   end
 
-  defp kind({:in, _field, _values}), do: :in
-  defp kind({:like, _field, _pattern}), do: :residual
-  defp kind({:not_like, _field, _pattern}), do: :residual
-  defp kind({:is_null, _field}), do: :residual
-  defp kind({:not_null, _field}), do: :residual
-  defp kind({:cmp, _op, @pk_field, _value}), do: :pk
-  defp kind({:range, @pk_field, _lower, _upper}), do: :pk
-  defp kind({:cmp, _op, _field, _value}), do: :pushable
-  defp kind({:range, _field, _lower, _upper}), do: :pushable
-
   # -- IN planning --
 
   defp plan_in(
@@ -394,7 +357,7 @@ defmodule Efsql.Planner do
 
   defp takes(fields, residual, sort) do
     needed =
-      (Enum.map(residual, &Logical.predicate_field/1) ++ Enum.map(sort, fn {_dir, f} -> f end))
+      (Enum.map(residual, &Predicate.field/1) ++ Enum.map(sort, fn {_dir, f} -> f end))
       |> Enum.uniq()
 
     case needed -- fields do
@@ -404,7 +367,7 @@ defmodule Efsql.Planner do
   end
 
   defp ensure_residual_evaluable!(pred) do
-    if Logical.predicate_field(pred) == @pk_field do
+    if Predicate.field(pred) == @pk_field do
       raise Unsupported,
             "a constraint on the primary key '_' cannot be combined with this query shape"
     end
@@ -444,7 +407,12 @@ defmodule Efsql.Planner do
   defp put_wheres(%Ecto.Query{} = q, preds) do
     wheres =
       Enum.map(preds, fn pred ->
-        %Ecto.Query.BooleanExpr{op: :and, expr: pred_to_expr(pred), params: [], subqueries: []}
+        %Ecto.Query.BooleanExpr{
+          op: :and,
+          expr: Predicate.to_ecto(pred),
+          params: [],
+          subqueries: []
+        }
       end)
 
     %Ecto.Query{q | wheres: wheres}
@@ -453,7 +421,7 @@ defmodule Efsql.Planner do
   defp put_order(%Ecto.Query{} = q, []), do: q
 
   defp put_order(%Ecto.Query{} = q, order) do
-    expr = Enum.map(order, fn {dir, field} -> {dir, field_ref(field)} end)
+    expr = Enum.map(order, fn {dir, field} -> {dir, Predicate.field_ref(field)} end)
     %Ecto.Query{q | order_bys: [%Ecto.Query.ByExpr{expr: expr, params: [], subqueries: []}]}
   end
 
@@ -462,21 +430,4 @@ defmodule Efsql.Planner do
   defp put_limit(%Ecto.Query{} = q, n) do
     %Ecto.Query{q | limit: %Ecto.Query.LimitExpr{expr: n, with_ties: false, params: []}}
   end
-
-  defp pred_to_expr({:cmp, op, field, value}) do
-    {op, [], [field_ref(field), Types.index_key(value)]}
-  end
-
-  defp pred_to_expr({:range, field, {lower_op, lower}, {upper_op, upper}}) do
-    {{lower_op, [], [field_ref(field), Types.index_key(lower)]},
-     {upper_op, [], [field_ref(field), Types.index_key(upper)]}}
-  end
-
-  defp pred_to_expr(pred) do
-    raise Unsupported, "predicate #{inspect(pred)} cannot be pushed to the adapter"
-  end
-
-  defp field_ref(field) do
-    {{:., [], [{:&, [], [0]}, field]}, [], []}
-  end
 end
diff --git a/lib/efsql/predicate.ex b/lib/efsql/predicate.ex
new file mode 100644
index 0000000..eb42d73
--- /dev/null
+++ b/lib/efsql/predicate.ex
@@ -0,0 +1,158 @@
+defmodule Efsql.Predicate do
+  @moduledoc """
+  A condition from a `WHERE` clause, and everything it means: which field
+  it's on, whether a row satisfies it, how the planner can serve it, and
+  the Ecto expression that pushes it to the adapter. A query's predicates
+  are an implicitly AND-ed list of:
+
+    * `{:cmp, op, field, value}` — op in `:== :> :>= :< :<=`
+    * `{:range, field, {lower_op, value}, {upper_op, value}}` — a two-sided
+      bound, `lower_op` in `:> :>=`, `upper_op` in `:< :<=`
+    * `{:like, field, pattern}` / `{:not_like, field, pattern}`
+    * `{:in, field, values}`
+    * `{:is_null, field}` / `{:not_null, field}`
+
+  The primary key is the pseudo-field `:_`.
+
+  A new kind of predicate is added here, clause by clause, plus its
+  translation in `Efsql.Parser` and any rewrite in `Efsql.Rewrite`.
+  """
+
+  alias Efsql.Exception.Unsupported
+  alias Efsql.Types
+
+  @type field :: atom()
+  @type t ::
+          {:cmp, :== | :> | :>= | :< | :<=, field(), term()}
+          | {:range, field(), {:> | :>=, term()}, {:< | :<=, term()}}
+          | {:like | :not_like, field(), String.t()}
+          | {:in, field(), [term()]}
+          | {:is_null | :not_null, field()}
+
+  @pk_field :_
+
+  @cmp_results %{
+    ==: [:eq],
+    >: [:gt],
+    >=: [:gt, :eq],
+    <: [:lt],
+    <=: [:lt, :eq]
+  }
+
+  @doc "The field a predicate constrains."
+  @spec field(t()) :: field()
+  def field({:cmp, _op, field, _value}), do: field
+  def field({:range, field, _lower, _upper}), do: field
+  def field({:like, field, _pattern}), do: field
+  def field({:not_like, field, _pattern}), do: field
+  def field({:in, field, _values}), do: field
+  def field({:is_null, field}), do: field
+  def field({:not_null, field}), do: field
+
+  # -- evaluation --
+
+  @doc """
+  Whether `row` satisfies every one of `predicates`, with SQL NULL
+  semantics: a NULL (nil or absent) field matches no comparison, LIKE or
+  IN, NOT LIKE included, and IS NULL matches it. Values compare with
+  `Efsql.Types.compare/2`.
+  """
+  @spec matches?(map(), [t()]) :: boolean()
+  def matches?(row, predicates), do: Enum.all?(predicates, &eval(&1, row))
+
+  defp eval({:cmp, op, field, param}, row),
+    do: on_value(row, field, &(Types.compare(&1, param) in Map.fetch!(@cmp_results, op)))
+
+  defp eval({:range, field, {lower_op, lower}, {upper_op, upper}}, row),
+    do: eval({:cmp, lower_op, field, lower}, row) and eval({:cmp, upper_op, field, upper}, row)
+
+  defp eval({:in, field, values}, row),
+    do:
+      on_value(row, field, fn value -> Enum.any?(values, &(Types.compare(value, &1) == :eq)) end)
+
+  defp eval({:is_null, field}, row), do: Map.get(row, field) == nil
+  defp eval({:not_null, field}, row), do: Map.get(row, field) != nil
+
+  defp eval({:like, field, pattern}, row),
+    do: on_value(row, field, &Regex.match?(like_regex(pattern), &1))
+
+  defp eval({:not_like, field, pattern}, row),
+    do: on_value(row, field, &(not Regex.match?(like_regex(pattern), &1)))
+
+  defp on_value(row, field, fun) do
+    case Map.get(row, field) do
+      nil -> false
+      value -> fun.(value)
+    end
+  end
+
+  defp like_regex(pattern) do
+    source =
+      pattern
+      |> Regex.escape()
+      |> String.replace("%", ".*")
+      |> String.replace("_", ".")
+
+    Regex.compile!("\\A" <> source <> "\\z", "s")
+  end
+
+  # -- planning --
+
+  @doc """
+  How the planner can serve a predicate:
+
+    * `:key` - a constraint on the primary key, read as a key range
+    * `:in` - an `IN` list, read as one lookup per value when an index or
+      the key can serve them
+    * `:index` - an equality or range an index may serve
+    * `:filter` - only checked on the rows read
+  """
+  @spec pushdown(t()) :: :key | :in | :index | :filter
+  def pushdown({:in, _field, _values}), do: :in
+  def pushdown({:cmp, _op, @pk_field, _value}), do: :key
+  def pushdown({:range, @pk_field, _lower, _upper}), do: :key
+  def pushdown({:cmp, _op, _field, _value}), do: :index
+  def pushdown({:range, _field, _lower, _upper}), do: :index
+  def pushdown(_predicate), do: :filter
+
+  @doc "Whether a predicate pins its field to one value."
+  def equality?({:cmp, :==, _field, _value}), do: true
+  def equality?(_predicate), do: false
+
+  @doc "Whether a predicate bounds its field on one or both sides."
+  def range?({:range, _field, _lower, _upper}), do: true
+  def range?({:cmp, op, _field, _value}) when op in ~w[> >= < <=]a, do: true
+  def range?(_predicate), do: false
+
+  @doc """
+  Takes the first predicate `fun` accepts out of `predicates`:
+  `{predicate, rest}`, or `{nil, predicates}` when none does.
+  """
+  def take_first(predicates, fun) do
+    case Enum.split_while(predicates, &(not fun.(&1))) do
+      {_before, []} -> {nil, predicates}
+      {before, [match | rest]} -> {match, before ++ rest}
+    end
+  end
+
+  # -- pushdown --
+
+  @doc """
+  The Ecto `where` expression that pushes an `:index` predicate to the
+  adapter, its values encoded as the adapter's indexes store them (see
+  `Efsql.Types.index_key/1`).
+  """
+  def to_ecto({:cmp, op, field, value}),
+    do: {op, [], [field_ref(field), Types.index_key(value)]}
+
+  def to_ecto({:range, field, {lower_op, lower}, {upper_op, upper}}) do
+    {{lower_op, [], [field_ref(field), Types.index_key(lower)]},
+     {upper_op, [], [field_ref(field), Types.index_key(upper)]}}
+  end
+
+  def to_ecto(predicate),
+    do: raise(Unsupported, "predicate #{inspect(predicate)} cannot be pushed to the adapter")
+
+  @doc "An Ecto reference to a field of the query's (only) source."
+  def field_ref(field), do: {{:., [], [{:&, [], [0]}, field]}, [], []}
+end
diff --git a/lib/efsql/render.ex b/lib/efsql/render.ex
index 79a1116..f55cb16 100644
--- a/lib/efsql/render.ex
+++ b/lib/efsql/render.ex
@@ -277,6 +277,19 @@ defmodule Efsql.Render do
     end
   end
 
+  @doc """
+  A result's columns, in order: the select list's, from the plan, or for
+  `select *` (which has none) the rows' fields, `id` first and the rest
+  sorted.
+  """
+  def columns(_plan, []), do: []
+  def columns(%Efsql.Physical.Plan{columns: [_ | _] = columns}, _rows), do: columns
+
+  def columns(_plan, rows) do
+    keys = rows |> Enum.flat_map(&Map.keys/1) |> Enum.uniq() |> Enum.sort()
+    if :id in keys, do: [:id | List.delete(keys, :id)], else: keys
+  end
+
   @doc "Short name for a value's type, for schema discovery displays."
   def type_of(nil), do: :null
   def type_of(v) when is_boolean(v), do: :boolean
diff --git a/lib/efsql/result.ex b/lib/efsql/result.ex
new file mode 100644
index 0000000..8baadc1
--- /dev/null
+++ b/lib/efsql/result.ex
@@ -0,0 +1,31 @@
+defmodule Efsql.Result do
+  @moduledoc """
+  What a query returns, ready to show: its rows, their columns in order
+  (see `Efsql.Render.columns/2`), how many transactions they were read in
+  (more than one only for a query across tenants read in batches), the
+  plan that produced them, and how long it took.
+  """
+
+  alias Efsql.Physical.Plan
+
+  defstruct rows: [], columns: [], transactions: 1, plan: nil, elapsed_ms: 0
+
+  @type t :: %__MODULE__{
+          rows: [map()],
+          columns: [atom()],
+          transactions: pos_integer(),
+          plan: Plan.t(),
+          elapsed_ms: non_neg_integer()
+        }
+
+  @spec new(Plan.t(), [map()], non_neg_integer()) :: t()
+  def new(%Plan{} = plan, rows, elapsed_ms \\ 0) do
+    %__MODULE__{
+      rows: rows,
+      columns: Efsql.Render.columns(plan, rows),
+      transactions: Efsql.Fanout.transactions(plan),
+      plan: plan,
+      elapsed_ms: elapsed_ms
+    }
+  end
+end
diff --git a/lib/efsql/rewrite.ex b/lib/efsql/rewrite.ex
index 28b58c9..22bcb13 100644
--- a/lib/efsql/rewrite.ex
+++ b/lib/efsql/rewrite.ex
@@ -7,6 +7,7 @@ defmodule Efsql.Rewrite do
   """
 
   alias Efsql.Logical
+  alias Efsql.Predicate
 
   @lower_ops ~w[> >=]a
   @upper_ops ~w[< <=]a
@@ -28,7 +29,7 @@ defmodule Efsql.Rewrite do
 
   defp merge_ranges([{:cmp, op, field, value} | rest], acc) when op in @lower_ops do
     case take_bound(rest, field, @upper_ops) do
-      nil ->
+      {nil, _} ->
         merge_ranges(rest, [{:cmp, op, field, value} | acc])
 
       {{:cmp, op2, _f, value2}, rest2} ->
@@ -38,7 +39,7 @@ defmodule Efsql.Rewrite do
 
   defp merge_ranges([{:cmp, op, field, value} | rest], acc) when op in @upper_ops do
     case take_bound(rest, field, @lower_ops) do
-      nil ->
+      {nil, _} ->
         merge_ranges(rest, [{:cmp, op, field, value} | acc])
 
       {{:cmp, op2, _f, value2}, rest2} ->
@@ -49,15 +50,10 @@ defmodule Efsql.Rewrite do
   defp merge_ranges([pred | rest], acc), do: merge_ranges(rest, [pred | acc])
 
   defp take_bound(preds, field, ops) do
-    match? = fn
+    Predicate.take_first(preds, fn
       {:cmp, op, f, _v} -> op in ops and f == field
       _ -> false
-    end
-
-    case Enum.split_while(preds, &(not match?.(&1))) do
-      {_before, []} -> nil
-      {before, [match | rest]} -> {match, before ++ rest}
-    end
+    end)
   end
 
   # `field like 'prefix%'` is exactly a key range. A pattern with a wildcard
diff --git a/lib/efsql/session.ex b/lib/efsql/session.ex
new file mode 100644
index 0000000..66b7671
--- /dev/null
+++ b/lib/efsql/session.ex
@@ -0,0 +1,86 @@
+defmodule Efsql.Session do
+  @moduledoc """
+  What a query runs in, for the line CLI and the TUI alike: the tenants
+  opened so far (a cache, keyed by name and storage id), the `\\set`
+  settings, and the active tenant and storage id, if any.
+
+  A statement's scope comes from its table name and the session:
+
+    * `tenant.table` or `storage_id.tenant.table` - that tenant
+    * `table` - the active tenant (see `activate/3`)
+    * `*.table` - every tenant of the active storage id, or of the Repo's
+      when there is none; `storage_id.*.table` names one
+
+  `run/3` returns an `Efsql.Result` and the session, whose tenant cache may
+  have grown.
+  """
+
+  alias Efsql.Exception.Unsupported
+  alias Efsql.Logical
+  alias Efsql.Result
+  alias Efsql.Settings
+
+  defstruct tenants: %{},
+            settings: %Settings{},
+            tenant: nil,
+            tenant_name: nil,
+            storage_id: nil
+
+  @type t :: %__MODULE__{
+          tenants: map(),
+          settings: Settings.t(),
+          tenant: term() | nil,
+          tenant_name: String.t() | nil,
+          storage_id: String.t() | nil
+        }
+
+  @doc """
+  Runs one statement. `options` go to the query on top of the settings'
+  (see `Efsql.Fanout.plan/3` for the ones a query across tenants takes).
+  """
+  @spec run(t(), String.t(), keyword()) :: {Result.t(), t()}
+  def run(%__MODULE__{} = session, sql, options \\ []) do
+    started = System.monotonic_time(:millisecond)
+    options = Keyword.merge(Settings.query_options(session.settings), options)
+
+    {plan, rows, tenants} =
+      sql
+      |> Efsql.Parser.sql_to_logical()
+      |> scope(session)
+      |> Efsql.run_logical(options, session.tenants)
+
+    elapsed = System.monotonic_time(:millisecond) - started
+    {Result.new(plan, rows, elapsed), %__MODULE__{session | tenants: tenants}}
+  end
+
+  @doc """
+  Makes a tenant the active one, for statements that don't name one. The
+  tenant must exist in the storage id.
+  """
+  @spec activate(t(), String.t() | nil, String.t()) :: t()
+  def activate(%__MODULE__{} = session, storage_id, tenant_name) do
+    {tenant, tenants} =
+      Efsql.open_tenant(session.tenants, tenant_name, storage_id, check_exists: true)
+
+    %__MODULE__{
+      session
+      | tenant: tenant,
+        tenant_name: tenant_name,
+        storage_id: storage_id,
+        tenants: tenants
+    }
+  end
+
+  defp scope(%Logical.Select{prefix: {:all_tenants, nil}} = logical, session),
+    do: %Logical.Select{logical | prefix: {:all_tenants, session.storage_id}}
+
+  defp scope(%Logical.Select{prefix: nil}, %__MODULE__{tenant: nil}) do
+    raise Unsupported,
+          "no active tenant — qualify the table (tenant.table) or pick a tenant in the navigator"
+  end
+
+  defp scope(%Logical.Select{prefix: nil} = logical, session),
+    do: %Logical.Select{logical | tenant: session.tenant}
+
+  defp scope(logical, _session), do: logical
+end
diff --git a/lib/efsql/settings.ex b/lib/efsql/settings.ex
new file mode 100644
index 0000000..5c20b3f
--- /dev/null
+++ b/lib/efsql/settings.ex
@@ -0,0 +1,60 @@
+defmodule Efsql.Settings do
+  @moduledoc """
+  The session settings `\\set` changes, shared by the line CLI and the TUI:
+
+    * `limit` - the default row limit (15)
+    * `tenant_batch` - how many tenants a `*.table` query reads per
+      transaction; `off` (the default) reads them all in one
+  """
+
+  defstruct limit: 15, tenant_batch: nil
+
+  @type t :: %__MODULE__{limit: pos_integer(), tenant_batch: pos_integer() | nil}
+
+  @doc """
+  Applies the text after `\\set `: `{:ok, settings, message}`, or
+  `{:error, usage}` for an unknown setting or a bad value.
+  """
+  @spec set(t(), String.t()) :: {:ok, t(), String.t()} | {:error, String.t()}
+  def set(%__MODULE__{} = settings, text) do
+    case String.split(String.trim(text), ~r/\s+/, parts: 2) do
+      ["limit", value] ->
+        with {:ok, n} <- positive(value, "\\set limit N") do
+          {:ok, %__MODULE__{settings | limit: n}, "limit set to #{n}"}
+        end
+
+      ["tenant_batch", "off"] ->
+        {:ok, %__MODULE__{settings | tenant_batch: nil}, "tenant_batch off: one transaction"}
+
+      ["tenant_batch", value] ->
+        with {:ok, n} <- positive(value, "\\set tenant_batch N|off") do
+          {:ok, %__MODULE__{settings | tenant_batch: n}, "tenant_batch set to #{n}"}
+        end
+
+      _ ->
+        {:error, "usage: \\set limit N, or \\set tenant_batch N|off"}
+    end
+  end
+
+  @doc "Each setting and its current value, for help text."
+  @spec describe(t()) :: [{String.t(), String.t()}]
+  def describe(%__MODULE__{} = settings) do
+    [
+      {"\\set limit N", "default row limit (#{settings.limit})"},
+      {"\\set tenant_batch N|off",
+       "tenants per transaction for *.table (#{settings.tenant_batch || "off"})"}
+    ]
+  end
+
+  @doc "The query options for these settings (see `Efsql.Session.run/3`)."
+  @spec query_options(t()) :: keyword()
+  def query_options(%__MODULE__{tenant_batch: nil}), do: []
+  def query_options(%__MODULE__{tenant_batch: n}), do: [tenant_batch: n]
+
+  defp positive(value, usage) do
+    case Integer.parse(value) do
+      {n, ""} when n > 0 -> {:ok, n}
+      _ -> {:error, "usage: #{usage}"}
+    end
+  end
+end
diff --git a/lib/efsql/tui/app.ex b/lib/efsql/tui/app.ex
index 42b1f95..6cc0191 100644
--- a/lib/efsql/tui/app.ex
+++ b/lib/efsql/tui/app.ex
@@ -26,9 +26,6 @@ defmodule Efsql.Tui.App do
               mode: :navigator,
               flash: nil,
               busy: nil,
-              limit: 15,
-              # tenants per transaction for *.table queries; nil is one transaction
-              tenant_batch: nil,
               cluster_file: nil,
               # navigator
               nav_path: [],
@@ -38,11 +35,8 @@ defmodule Efsql.Tui.App do
               # entries matching the filter, recomputed only when either
               # changes (see `refilter/1`), so a held arrow key stays O(1)
               nav_visible: [],
-              # session
-              storage_id: nil,
-              tenant_id: nil,
-              tenant: nil,
-              tenants: %{},
+              # the active tenant, open tenants and settings (see Efsql.Session)
+              session: %Efsql.Session{},
               # schema browser
               sources: nil,
               src_cursor: 0,
@@ -55,16 +49,14 @@ defmodule Efsql.Tui.App do
               history: [],
               hist_ix: nil,
               saved_input: "",
-              rows: nil,
-              columns: [],
+              # the last query's Efsql.Result, nil before the first
+              result: nil,
               # rows rendered once per query result (see `done/3`): a list of
               # cell strings per row, and each column's natural width
               cells: [],
               col_widths: [],
               col_align: [],
-              plan: nil,
               qerror: nil,
-              elapsed_ms: nil,
               row_cursor: 0,
               row_scroll: 0,
               qfocus: :input,
@@ -179,7 +171,8 @@ defmodule Efsql.Tui.App do
     {model, [load_nav(model)]}
   end
 
-  defp navigator(%Model{tenant: tenant} = model, {:key, :esc}) when tenant != nil do
+  defp navigator(%Model{session: %Efsql.Session{tenant: tenant}} = model, {:key, :esc})
+       when tenant != nil do
     {%{model | mode: :schema}, []}
   end
 
@@ -233,24 +226,8 @@ defmodule Efsql.Tui.App do
   end
 
   defp activate(model, storage_id, tenant_id) do
-    fun = fn ->
-      Discover.ensure_storage_cache(storage_id)
-      db = Ecto.Adapters.FoundationDB.db(Efsql.Repo)
-      config = Keyword.put(Efsql.Repo.config(), :storage_id, storage_id)
-
-      unless EctoFoundationDB.Tenant.Backend.exists?(db, tenant_id, config) do
-        raise "Tenant '#{tenant_id}' does not exist"
-      end
-
-      # migrate: false — see Efsql.resolve_tenant/2; the TUI never writes.
-      tenant =
-        EctoFoundationDB.Tenant.open(Efsql.Repo, tenant_id,
-          storage_id: storage_id,
-          migrate: false
-        )
-
-      {storage_id, tenant_id, tenant}
-    end
+    session = model.session
+    fun = fn -> Efsql.Session.activate(session, storage_id, tenant_id) end
 
     {%{model | busy: "opening #{tenant_id}"}, [{:task, :activate, fun}]}
   end
@@ -286,7 +263,7 @@ defmodule Efsql.Tui.App do
     with source when source != nil <- current_source(model),
          %Discover.Schema{fields: fields} <- model.schemas[source],
          %{name: field} <- Enum.at(fields, model.field_cursor) do
-      input = "select #{field} from #{source} limit #{model.limit};"
+      input = "select #{field} from #{source} limit #{model.session.settings.limit};"
       {%{model | mode: :query, input: input, qcursor: String.length(input), qfocus: :input}, []}
     else
       _ -> {model, []}
@@ -296,7 +273,7 @@ defmodule Efsql.Tui.App do
   defp schema(model, {:char, "q"}), do: {%{model | mode: :query}, []}
   defp schema(model, {:char, "t"}), do: {%{model | mode: :navigator}, []}
 
-  defp schema(%Model{tenant: t} = model, {:key, :esc}) when t != nil,
+  defp schema(%Model{session: %Efsql.Session{tenant: t}} = model, {:key, :esc}) when t != nil,
     do: {%{model | mode: :navigator}, []}
 
   defp schema(model, {:char, "r"}) do
@@ -354,7 +331,7 @@ defmodule Efsql.Tui.App do
   def current_source(_), do: nil
 
   defp sample_task(model, source) do
-    tenant = model.tenant
+    tenant = model.session.tenant
     key = schema_cache_key(model, source)
 
     {:task, {:schema, source},
@@ -366,7 +343,7 @@ defmodule Efsql.Tui.App do
   end
 
   defp schema_cache_key(model, source) do
-    {model.cluster_file, model.storage_id, model.tenant_id, source}
+    {model.cluster_file, model.session.storage_id, model.session.tenant_name, source}
   end
 
   defp put_schema(model, source, schema) do
@@ -414,13 +391,13 @@ defmodule Efsql.Tui.App do
   defp query(model, {:key, :up}), do: {history(model, 1), []}
   defp query(model, {:key, :down}), do: {history(model, -1), []}
 
-  defp query(%Model{input: "", rows: rows} = model, {:key, :tab}) when is_list(rows) do
+  defp query(%Model{input: "", result: %Efsql.Result{}} = model, {:key, :tab}) do
     {%{model | qfocus: :results}, []}
   end
 
   defp query(model, {:key, :tab}), do: {complete(model), []}
 
-  defp query(%Model{tenant: t} = model, {:key, :esc}) when t != nil,
+  defp query(%Model{session: %Efsql.Session{tenant: t}} = model, {:key, :esc}) when t != nil,
     do: {%{model | mode: :schema}, []}
 
   defp query(model, {:key, :esc}), do: {%{model | mode: :navigator}, []}
@@ -436,11 +413,8 @@ defmodule Efsql.Tui.App do
       "\\plan" ->
         {%{model | show_plan?: not model.show_plan?, input: "", qcursor: 0}, []}
 
-      "\\set limit " <> n ->
-        set_limit(model, n)
-
-      "\\set tenant_batch " <> n ->
-        set_tenant_batch(model, n)
+      "\\set " <> text ->
+        set(model, text)
 
       sql ->
         run_query(model, sql)
@@ -449,47 +423,24 @@ defmodule Efsql.Tui.App do
 
   defp query(model, _msg), do: {model, []}
 
-  defp set_limit(model, n) do
-    case Integer.parse(String.trim(n)) do
-      {n, ""} when n > 0 ->
-        {%{model | limit: n, input: "", qcursor: 0, flash: {:info, "limit set to #{n}"}}, []}
-
-      _ ->
-        {%{model | flash: {:error, "usage: \\set limit N"}, input: "", qcursor: 0}, []}
-    end
-  end
-
-  defp set_tenant_batch(model, n) do
-    case {String.trim(n), Integer.parse(String.trim(n))} do
-      {"off", _} ->
-        {%{model | tenant_batch: nil, input: "", qcursor: 0, flash: {:info, "tenant_batch off"}},
-         []}
+  defp set(model, text) do
+    model = %{model | input: "", qcursor: 0}
 
-      {_, {n, ""}} when n > 0 ->
-        flash = {:info, "tenant_batch set to #{n}"}
-        {%{model | tenant_batch: n, input: "", qcursor: 0, flash: flash}, []}
+    case Efsql.Settings.set(model.session.settings, text) do
+      {:ok, settings, message} ->
+        session = %{model.session | settings: settings}
+        {%{model | session: session, flash: {:info, message}}, []}
 
-      _ ->
-        flash = {:error, "usage: \\set tenant_batch N|off"}
-        {%{model | flash: flash, input: "", qcursor: 0}, []}
+      {:error, usage} ->
+        {%{model | flash: {:error, usage}}, []}
     end
   end
 
   defp run_query(model, sql) do
     sql = if String.ends_with?(sql, ";"), do: sql, else: sql <> ";"
 
-    session = %{
-      tenant: model.tenant,
-      tenants: model.tenants,
-      storage_id: model.storage_id,
-      tenant_batch: model.tenant_batch
-    }
-
-    fun = fn ->
-      started = System.monotonic_time(:millisecond)
-      {plan, rows, tenants} = Efsql.Tui.Session.qall(sql, session)
-      {plan, rows, tenants, System.monotonic_time(:millisecond) - started}
-    end
+    session = model.session
+    fun = fn -> Efsql.Session.run(session, sql) end
 
     model = %{
       model
@@ -593,8 +544,8 @@ defmodule Efsql.Tui.App do
     {move_row(model, if(c == "j", do: 1, else: -1)), []}
   end
 
-  defp results(%Model{rows: rows} = model, {:key, :enter}) do
-    case Enum.at(rows || [], model.row_cursor) do
+  defp results(model, {:key, :enter}) do
+    case Enum.at(result_rows(model), model.row_cursor) do
       nil ->
         {model, []}
 
@@ -616,10 +567,13 @@ defmodule Efsql.Tui.App do
   defp results(model, _msg), do: {model, []}
 
   defp move_row(model, delta) do
-    count = length(model.rows || [])
+    count = length(result_rows(model))
     %{model | row_cursor: clamp(model.row_cursor + delta, count)}
   end
 
+  defp result_rows(%Model{result: %Efsql.Result{rows: rows}}), do: rows
+  defp result_rows(_model), do: []
+
   # -- inspector --
 
   defp inspector(model, {:key, key}) when key in [:esc, :enter] do
@@ -722,7 +676,10 @@ defmodule Efsql.Tui.App do
   The inspected row's fields: in select-list order when the query has one,
   else sorted.
   """
-  def inspector_fields(%Model{irow: row, plan: %Efsql.Physical.Plan{columns: [_ | _] = columns}}) do
+  def inspector_fields(%Model{
+        irow: row,
+        result: %Efsql.Result{plan: %Efsql.Physical.Plan{columns: [_ | _] = columns}}
+      }) do
     listed = Enum.filter(columns, &Map.has_key?(row, &1))
     listed ++ Enum.sort(Map.keys(row) -- listed)
   end
@@ -759,20 +716,17 @@ defmodule Efsql.Tui.App do
     {refilter(%{model | busy: nil, nav_entries: Enum.sort(entries)}), []}
   end
 
-  defp done(model, :activate, {:ok, {storage_id, tenant_id, tenant}}) do
+  defp done(model, :activate, {:ok, %Efsql.Session{tenant: tenant} = session}) do
     model = %{
       model
       | busy: "listing sources",
-        storage_id: storage_id,
-        tenant_id: tenant_id,
-        tenant: tenant,
+        session: session,
         mode: :schema,
         sources: nil,
         src_cursor: 0,
         schemas: %{},
         focus: :sources,
-        rows: nil,
-        plan: nil,
+        result: nil,
         qerror: nil
     }
 
@@ -787,24 +741,21 @@ defmodule Efsql.Tui.App do
     {%{put_schema(model, source, schema) | busy: nil}, []}
   end
 
-  defp done(model, :query, {:ok, {plan, rows, tenants, elapsed}}) do
-    columns = columns(plan, rows)
+  defp done(model, :query, {:ok, {%Efsql.Result{} = result, %Efsql.Session{} = session}}) do
+    %Efsql.Result{rows: rows, columns: columns} = result
     cells = render_cells(rows, columns)
 
     model = %{
       model
       | busy: nil,
-        rows: rows,
-        columns: columns,
+        result: result,
+        session: session,
         cells: cells,
         col_widths: col_widths(columns, cells),
         col_align: col_aligns(rows, columns),
-        plan: plan,
         qerror: nil,
-        elapsed_ms: elapsed,
         row_cursor: 0,
-        row_scroll: 0,
-        tenants: tenants
+        row_scroll: 0
     }
 
     {model, []}
@@ -812,17 +763,6 @@ defmodule Efsql.Tui.App do
 
   defp done(model, _tag, _result), do: {%{model | busy: nil}, []}
 
-  defp columns(_plan, []), do: []
-
-  # In select-list order; `select *` has no list, so its fields are sorted
-  # with id first.
-  defp columns(%Efsql.Physical.Plan{columns: [_ | _] = columns}, _rows), do: columns
-
-  defp columns(_plan, rows) do
-    keys = rows |> Enum.flat_map(&Map.keys/1) |> Enum.uniq() |> Enum.sort()
-    if :id in keys, do: [:id | List.delete(keys, :id)], else: keys
-  end
-
   # Cells are rendered once here rather than per frame: a frame is painted on
   # every key press, and inspecting every visible value each time is what
   # made the editor lag on wide or large rows.
diff --git a/lib/efsql/tui/session.ex b/lib/efsql/tui/session.ex
deleted file mode 100644
index 95e0960..0000000
--- a/lib/efsql/tui/session.ex
+++ /dev/null
@@ -1,34 +0,0 @@
-defmodule Efsql.Tui.Session do
-  @moduledoc """
-  Session-aware query execution: statements without a tenant qualifier run
-  against the session's active tenant, so `select id from users;` works
-  once a tenant is activated in the Navigator, and `*.users` reads every
-  tenant of the active storage id. Tenant-qualified statements behave
-  exactly as in the line CLI.
-  """
-
-  alias Efsql.Logical
-
-  def qall(sql, session) do
-    sql
-    |> Efsql.Parser.sql_to_logical()
-    |> in_session(session)
-    |> Efsql.run_logical(options(session), session.tenants)
-  end
-
-  defp options(%{tenant_batch: n}) when is_integer(n), do: [tenant_batch: n]
-  defp options(_session), do: []
-
-  defp in_session(%Logical.Select{prefix: {:all_tenants, nil}} = logical, session),
-    do: %Logical.Select{logical | prefix: {:all_tenants, session[:storage_id]}}
-
-  defp in_session(%Logical.Select{prefix: nil}, %{tenant: nil}) do
-    raise Efsql.Exception.Unsupported,
-          "no active tenant — qualify the table (tenant.table) or pick a tenant in the navigator"
-  end
-
-  defp in_session(%Logical.Select{prefix: nil} = logical, %{tenant: tenant}),
-    do: %Logical.Select{logical | tenant: tenant}
-
-  defp in_session(logical, _session), do: logical
-end
diff --git a/lib/efsql/tui/view.ex b/lib/efsql/tui/view.ex
index 033dd37..48ad04d 100644
--- a/lib/efsql/tui/view.ex
+++ b/lib/efsql/tui/view.ex
@@ -57,9 +57,9 @@ defmodule Efsql.Tui.View do
 
   defp title_bar(model, _cols) do
     session =
-      case model.tenant_id do
+      case model.session.tenant_name do
         nil -> "no tenant"
-        tenant_id -> "#{model.storage_id} / #{tenant_id}"
+        tenant_name -> "#{model.session.storage_id} / #{tenant_name}"
       end
 
     [
@@ -263,22 +263,18 @@ defmodule Efsql.Tui.View do
           []
       end
 
+    show_plan? = model.show_plan?
+
     plan_lines =
-      if model.show_plan? and model.plan do
-        [
-          [
-            {:dim,
-             Render.truncate(" plan: " <> inspect(model.plan.access, width: :infinity), cols)}
-          ]
-        ] ++
+      case model.result do
+        %Efsql.Result{plan: plan} when show_plan? ->
           [
-            [
-              {:dim,
-               Render.truncate(" ops:  " <> inspect(model.plan.ops, width: :infinity), cols)}
-            ]
+            [{:dim, Render.truncate(" plan: " <> inspect(plan.access, width: :infinity), cols)}],
+            [{:dim, Render.truncate(" ops:  " <> inspect(plan.ops, width: :infinity), cols)}]
           ]
-      else
-        []
+
+        _ ->
+          []
       end
 
     used = 1 + length(completion) + length(plan_lines)
@@ -287,19 +283,19 @@ defmodule Efsql.Tui.View do
     {[input_line] ++ completion ++ plan_lines ++ [[]] ++ results, cursor}
   end
 
-  defp results_lines(%Model{rows: nil}, _height, _cols) do
+  defp results_lines(%Model{result: nil}, _height, _cols) do
     [[{:dim, " no results yet — run a statement, or Esc for the schema browser"}]]
   end
 
-  defp results_lines(%Model{rows: []} = model, _height, _cols) do
-    [[{:dim, " (0 rows, #{model.elapsed_ms} ms)" <> snapshots_note(model)}]]
+  defp results_lines(%Model{result: %Efsql.Result{rows: []} = result}, _height, _cols) do
+    [[{:dim, " (0 rows, #{result.elapsed_ms} ms)" <> transactions_note(result)}]]
   end
 
-  defp results_lines(model, height, cols) do
+  defp results_lines(%Model{result: result} = model, height, cols) do
     {widths, column_map} = Columns.layout(model.col_widths, cols)
     aligns = Enum.map(column_map, &align_of(model, &1))
     visible = max(height - 2, 1)
-    count = length(model.rows)
+    count = length(result.rows)
     start = if model.row_cursor >= visible, do: model.row_cursor - visible + 1, else: 0
 
     names = Enum.map(column_map, &header_of(model, &1))
@@ -317,25 +313,20 @@ defmodule Efsql.Tui.View do
     footer =
       [
         {:dim,
-         " (#{count} rows, #{model.elapsed_ms} ms)" <>
-           snapshots_note(model) <> hidden_note(model, column_map) <> browse_hint(model, count)}
+         " (#{count} rows, #{result.elapsed_ms} ms)" <>
+           transactions_note(result) <>
+           hidden_note(result, column_map) <> browse_hint(model, count)}
       ]
 
     [header] ++ rows ++ [footer]
   end
 
   # A query across tenants read in batches says how many transactions it took.
-  defp snapshots_note(%Model{plan: %Efsql.Physical.Plan{} = plan}) do
-    case Efsql.Fanout.transactions(plan) do
-      1 -> ""
-      n -> " · #{n} transactions"
-    end
-  end
-
-  defp snapshots_note(_model), do: ""
+  defp transactions_note(%Efsql.Result{transactions: 1}), do: ""
+  defp transactions_note(%Efsql.Result{transactions: n}), do: " · #{n} transactions"
 
   defp header_of(_model, :split), do: "…"
-  defp header_of(model, {:col, ix}), do: model.columns |> Enum.at(ix) |> to_string()
+  defp header_of(model, {:col, ix}), do: model.result.columns |> Enum.at(ix) |> to_string()
 
   defp align_of(_model, :split), do: :left
   defp align_of(model, {:col, ix}), do: Enum.at(model.col_align, ix, :left)
@@ -351,8 +342,8 @@ defmodule Efsql.Tui.View do
 
   # Dropped columns are reported the way DuckDB reports them: a bare `…`
   # column otherwise reads as truncated data rather than a layout decision.
-  defp hidden_note(model, column_map) do
-    total = length(model.columns)
+  defp hidden_note(result, column_map) do
+    total = length(result.columns)
     shown = Enum.count(column_map, &match?({:col, _}, &1))
 
     if shown < total, do: " — #{total} columns (#{shown} shown)", else: ""
diff --git a/lib/efsql/types.ex b/lib/efsql/types.ex
index e1ea03b..7078238 100644
--- a/lib/efsql/types.ex
+++ b/lib/efsql/types.ex
@@ -102,6 +102,17 @@ defmodule Efsql.Types do
     end
   end
 
+  @doc """
+  A term that is equal exactly when `compare/2` calls two values equal,
+  for grouping: a `Decimal` 1.0 and 1.00 have the same key, as do
+  datetimes stored at different precisions.
+  """
+  def equality_key(%Decimal{} = d), do: Decimal.normalize(d)
+  def equality_key(%NaiveDateTime{microsecond: {us, _}} = t), do: %{t | microsecond: {us, 6}}
+  def equality_key(%DateTime{microsecond: {us, _}} = t), do: %{t | microsecond: {us, 6}}
+  def equality_key(%Time{microsecond: {us, _}} = t), do: %{t | microsecond: {us, 6}}
+  def equality_key(value), do: value
+
   @doc """
   Orders two field values, `:lt`, `:eq` or `:gt`. Dates and times compare
   chronologically (term order would compare their struct fields, and term
diff --git a/test/discover_test.exs b/test/discover_test.exs
index 86a64e5..9f48998 100644
--- a/test/discover_test.exs
+++ b/test/discover_test.exs
@@ -38,10 +38,10 @@ defmodule EfsqlTest.Integration.Discover do
   test "session qall runs unqualified statements against the active tenant", context do
     tenant = context[:tenant]
 
-    assert {_plan, [%{name: "Alice"}], _tenants} =
-             Efsql.Tui.Session.qall(
-               "select id, name from users where name = 'Alice';",
-               %{tenant: tenant, tenants: %{}}
+    assert {%Efsql.Result{rows: [%{name: "Alice"}]}, %Efsql.Session{}} =
+             Efsql.Session.run(
+               %Efsql.Session{tenant: tenant},
+               "select id, name from users where name = 'Alice';"
              )
   end
 end
diff --git a/test/planner_split_test.exs b/test/planner_split_test.exs
index a92fbfc..1b4463f 100644
--- a/test/planner_split_test.exs
+++ b/test/planner_split_test.exs
@@ -1,7 +1,7 @@
 defmodule Efsql.PlannerSplitTest do
   use ExUnit.Case, async: true
 
-  alias Efsql.Executor
+  alias Efsql.Predicate
   alias Efsql.Logical
   alias Efsql.Planner
 
@@ -45,8 +45,8 @@ defmodule Efsql.PlannerSplitTest do
 
   test "tenant names are chosen with the usual predicate semantics" do
     preds = [{:like, :_tenant, "acme%"}]
-    assert Executor.matches?(%{_tenant: "acme-eu"}, preds)
-    refute Executor.matches?(%{_tenant: "globex"}, preds)
-    assert Executor.matches?(%{_tenant: "anything"}, [])
+    assert Predicate.matches?(%{_tenant: "acme-eu"}, preds)
+    refute Predicate.matches?(%{_tenant: "globex"}, preds)
+    assert Predicate.matches?(%{_tenant: "anything"}, [])
   end
 end
diff --git a/test/predicate_test.exs b/test/predicate_test.exs
new file mode 100644
index 0000000..7256f6d
--- /dev/null
+++ b/test/predicate_test.exs
@@ -0,0 +1,96 @@
+defmodule Efsql.PredicateTest do
+  use ExUnit.Case, async: true
+
+  alias Efsql.Predicate
+
+  @row %{name: "Alice", age: 30, notes: nil, at: ~D[2024-03-01], price: Decimal.new("9.50")}
+
+  defp holds?(predicate, row \\ @row), do: Predicate.matches?(row, [predicate])
+
+  test "comparisons use Types.compare, so dates and Decimals compare by value" do
+    assert holds?({:cmp, :==, :age, 30})
+    assert holds?({:cmp, :>=, :at, ~D[2024-01-01]})
+    refute holds?({:cmp, :<, :at, ~D[2024-01-01]})
+    assert holds?({:cmp, :<, :price, 10})
+    assert holds?({:range, :age, {:>, 20}, {:<=, 30}})
+    refute holds?({:range, :age, {:>, 30}, {:<=, 40}})
+  end
+
+  test "LIKE and IN" do
+    assert holds?({:like, :name, "Al%"})
+    assert holds?({:like, :name, "_lice"})
+    refute holds?({:like, :name, "al%"})
+    assert holds?({:not_like, :name, "B%"})
+    assert holds?({:in, :age, [10, 30]})
+    refute holds?({:in, :age, [10, 20]})
+  end
+
+  test "a NULL field matches no comparison, LIKE or IN, not even NOT LIKE" do
+    for predicate <- [
+          {:cmp, :==, :notes, "x"},
+          {:range, :notes, {:>, "a"}, {:<, "z"}},
+          {:like, :notes, "%"},
+          {:not_like, :notes, "x"},
+          {:in, :notes, ["x"]},
+          {:cmp, :==, :missing, 1}
+        ] do
+      refute holds?(predicate), inspect(predicate)
+    end
+
+    assert holds?({:is_null, :notes})
+    assert holds?({:is_null, :missing})
+    refute holds?({:not_null, :notes})
+    assert holds?({:not_null, :name})
+  end
+
+  test "every predicate must hold" do
+    assert Predicate.matches?(@row, [])
+    refute Predicate.matches?(@row, [{:cmp, :==, :age, 30}, {:is_null, :name}])
+  end
+
+  test "field" do
+    for {predicate, field} <- [
+          {{:cmp, :==, :a, 1}, :a},
+          {{:range, :b, {:>, 1}, {:<, 2}}, :b},
+          {{:like, :c, "x"}, :c},
+          {{:not_like, :d, "x"}, :d},
+          {{:in, :e, [1]}, :e},
+          {{:is_null, :f}, :f},
+          {{:not_null, :g}, :g}
+        ] do
+      assert Predicate.field(predicate) == field
+    end
+  end
+
+  test "pushdown says how the planner can serve a predicate" do
+    assert Predicate.pushdown({:cmp, :==, :_, "k"}) == :key
+    assert Predicate.pushdown({:range, :_, {:>=, "a"}, {:<, "b"}}) == :key
+    assert Predicate.pushdown({:in, :_, ["a", "b"]}) == :in
+    assert Predicate.pushdown({:in, :name, ["a"]}) == :in
+    assert Predicate.pushdown({:cmp, :>, :age, 1}) == :index
+    assert Predicate.pushdown({:range, :age, {:>, 1}, {:<, 2}}) == :index
+
+    for predicate <- [{:like, :a, "x"}, {:not_like, :a, "x"}, {:is_null, :a}, {:not_null, :a}] do
+      assert Predicate.pushdown(predicate) == :filter
+    end
+  end
+
+  test "equality and range" do
+    assert Predicate.equality?({:cmp, :==, :a, 1})
+    refute Predicate.equality?({:cmp, :>, :a, 1})
+    assert Predicate.range?({:cmp, :<=, :a, 1})
+    assert Predicate.range?({:range, :a, {:>, 1}, {:<, 2}})
+    refute Predicate.range?({:cmp, :==, :a, 1})
+    refute Predicate.range?({:like, :a, "x"})
+  end
+
+  test "to_ecto encodes values as the index stores them" do
+    ref = Predicate.field_ref(:at)
+    assert {:>=, [], [^ref, "20240301"]} = Predicate.to_ecto({:cmp, :>=, :at, ~D[2024-03-01]})
+
+    assert {{:>, [], [_, 1]}, {:<, [], [_, 2]}} =
+             Predicate.to_ecto({:range, :age, {:>, 1}, {:<, 2}})
+
+    assert_raise Efsql.Exception.Unsupported, fn -> Predicate.to_ecto({:like, :a, "x"}) end
+  end
+end
diff --git a/test/render_test.exs b/test/render_test.exs
index 30e8cde..9efe7fa 100644
--- a/test/render_test.exs
+++ b/test/render_test.exs
@@ -132,4 +132,20 @@ defmodule Efsql.RenderTest do
     assert String.length(Render.cell(uri, 30)) <= 30
     assert Render.full(uri) =~ "scheme:"
   end
+
+  describe "columns" do
+    test "follow the select list" do
+      plan = %Efsql.Physical.Plan{columns: [:name, :id]}
+      assert Render.columns(plan, [%{id: 1, name: "a"}]) == [:name, :id]
+    end
+
+    test "for select *, the rows' fields with id first" do
+      rows = [%{name: "a", id: 1}, %{id: 2, age: 3}]
+      assert Render.columns(%Efsql.Physical.Plan{}, rows) == [:id, :age, :name]
+    end
+
+    test "no rows, no columns" do
+      assert Render.columns(%Efsql.Physical.Plan{columns: [:a]}, []) == []
+    end
+  end
 end
diff --git a/test/rewrite_test.exs b/test/rewrite_test.exs
index 4f7913b..2c008a5 100644
--- a/test/rewrite_test.exs
+++ b/test/rewrite_test.exs
@@ -74,4 +74,14 @@ defmodule Efsql.RewriteTest do
       assert [{:in, :name, ["a", "b"]}] = normalize([{:in, :name, ["a", "b"]}])
     end
   end
+
+  test "Predicate.take_first takes the first match out, keeping the rest in order" do
+    preds = [{:is_null, :a}, {:cmp, :>, :b, 1}, {:cmp, :>, :c, 2}]
+    gt? = &match?({:cmp, :>, _, _}, &1)
+
+    assert Efsql.Predicate.take_first(preds, gt?) ==
+             {{:cmp, :>, :b, 1}, [{:is_null, :a}, {:cmp, :>, :c, 2}]}
+
+    assert Efsql.Predicate.take_first(preds, fn _ -> false end) == {nil, preds}
+  end
 end
diff --git a/test/session_test.exs b/test/session_test.exs
new file mode 100644
index 0000000..fd76949
--- /dev/null
+++ b/test/session_test.exs
@@ -0,0 +1,26 @@
+defmodule Efsql.SessionTest do
+  use ExUnit.Case, async: true
+
+  alias Efsql.Exception.Unsupported
+  alias Efsql.Physical.Plan
+  alias Efsql.Result
+  alias Efsql.Session
+
+  test "a table without a tenant needs an active one" do
+    assert_raise Unsupported, ~r/no active tenant/, fn ->
+      Session.run(%Session{}, "select id from users;")
+    end
+  end
+
+  test "a result knows its columns and how many transactions it took" do
+    rows = [%{id: 1, name: "a"}]
+
+    assert %Result{rows: ^rows, columns: [:name, :id], transactions: 1, elapsed_ms: 5} =
+             Result.new(%Plan{access: {:fan_out, []}, columns: [:name, :id]}, rows, 5)
+
+    batches = {:batches, [{:fan_out, []}, {:fan_out, []}, {:fan_out, []}], 2}
+
+    assert %Result{columns: [:id, :name], transactions: 3} =
+             Result.new(%Plan{access: batches}, rows)
+  end
+end
diff --git a/test/settings_test.exs b/test/settings_test.exs
new file mode 100644
index 0000000..c652f8e
--- /dev/null
+++ b/test/settings_test.exs
@@ -0,0 +1,47 @@
+defmodule Efsql.SettingsTest do
+  use ExUnit.Case, async: true
+
+  alias Efsql.Settings
+
+  test "defaults" do
+    assert %Settings{limit: 15, tenant_batch: nil} = %Settings{}
+    assert Settings.query_options(%Settings{}) == []
+  end
+
+  test "limit takes a positive integer" do
+    assert {:ok, %Settings{limit: 20}, "limit set to 20"} = Settings.set(%Settings{}, "limit 20")
+    assert {:ok, %Settings{limit: 3}, _} = Settings.set(%Settings{}, "  limit   3 ")
+
+    for bad <- ["limit 0", "limit -1", "limit x", "limit 2.5", "limit"] do
+      assert {:error, "usage: " <> _} = Settings.set(%Settings{}, bad), bad
+    end
+  end
+
+  test "tenant_batch takes a positive integer or off, and becomes a query option" do
+    assert {:ok, settings, "tenant_batch set to 25"} =
+             Settings.set(%Settings{}, "tenant_batch 25")
+
+    assert Settings.query_options(settings) == [tenant_batch: 25]
+
+    assert {:ok, settings, _} = Settings.set(settings, "tenant_batch off")
+    assert settings.tenant_batch == nil
+    assert Settings.query_options(settings) == []
+
+    assert {:error, "usage: \\set tenant_batch N|off"} =
+             Settings.set(%Settings{}, "tenant_batch 0")
+  end
+
+  test "an unknown setting says what there is" do
+    assert {:error, message} = Settings.set(%Settings{}, "colour blue")
+    assert message =~ "\\set limit N"
+    assert message =~ "\\set tenant_batch N|off"
+  end
+
+  test "describe shows the current values" do
+    assert [{"\\set limit N", limit}, {"\\set tenant_batch N|off", batch}] =
+             Settings.describe(%Settings{limit: 7, tenant_batch: 4})
+
+    assert limit =~ "(7)"
+    assert batch =~ "(4)"
+  end
+end
diff --git a/test/tui_app_test.exs b/test/tui_app_test.exs
index ccbd976..c5120e3 100644
--- a/test/tui_app_test.exs
+++ b/test/tui_app_test.exs
@@ -52,7 +52,13 @@ defmodule Efsql.Tui.AppTest do
         {:key, :enter},
         {:done, :nav_entries, {:ok, ["demo", "staging"]}},
         {:key, :enter},
-        {:done, :activate, {:ok, {"Ecto.Adapters.FoundationDB", "demo", :fake_tenant}}},
+        {:done, :activate,
+         {:ok,
+          %Efsql.Session{
+            storage_id: "Ecto.Adapters.FoundationDB",
+            tenant_name: "demo",
+            tenant: :fake_tenant
+          }}},
         {:done, :sources, {:ok, ["orders", "users"]}}
       ])
 
@@ -144,7 +150,9 @@ defmodule Efsql.Tui.AppTest do
     assert model.input == ""
 
     plan = %Efsql.Physical.Plan{access: {:pk_range, nil, nil, nil, []}, ops: []}
-    {model, _} = feed(model, [{:done, :query, {:ok, {plan, [], %{}, 1}}}])
+
+    {model, _} =
+      feed(model, [{:done, :query, {:ok, {Efsql.Result.new(plan, [], 1), model.session}}}])
 
     {model, _} = feed(model, [{:key, :up}])
     assert model.input == "select id from users where name;"
@@ -155,7 +163,9 @@ defmodule Efsql.Tui.AppTest do
     rows = for i <- 1..3, do: %{id: "000#{i}", name: "User #{i}", notes: nil}
 
     plan = %Efsql.Physical.Plan{access: {:pk_range, nil, nil, nil, []}, ops: []}
-    {model, _} = feed(model, [{:done, :query, {:ok, {plan, rows, %{}, 7}}}])
+
+    {model, _} =
+      feed(model, [{:done, :query, {:ok, {Efsql.Result.new(plan, rows, 7), model.session}}}])
 
     text = frame_text(model)
     assert text =~ "(3 rows, 7 ms)"
@@ -184,8 +194,10 @@ defmodule Efsql.Tui.AppTest do
       columns: [:name, :"count(*)", :id]
     }
 
-    {model, _} = feed(model, [{:done, :query, {:ok, {plan, rows, %{}, 1}}}])
-    assert model.columns == [:name, :"count(*)", :id]
+    {model, _} =
+      feed(model, [{:done, :query, {:ok, {Efsql.Result.new(plan, rows, 1), model.session}}}])
+
+    assert model.result.columns == [:name, :"count(*)", :id]
     assert frame_text(model) =~ ~r/name\s+count\(\*\)\s+id/
 
     # the inspector lists fields in the same order
@@ -198,22 +210,25 @@ defmodule Efsql.Tui.AppTest do
     rows = [%{name: "Alice", id: "0001", age: 30}]
 
     plan = %Efsql.Physical.Plan{access: {:pk_range, nil, nil, nil, []}, ops: []}
-    {model, _} = feed(model, [{:done, :query, {:ok, {plan, rows, %{}, 1}}}])
-    assert model.columns == [:id, :age, :name]
+
+    {model, _} =
+      feed(model, [{:done, :query, {:ok, {Efsql.Result.new(plan, rows, 1), model.session}}}])
+
+    assert model.result.columns == [:id, :age, :name]
   end
 
   test "\\set tenant_batch sets and clears the tenants per transaction" do
     model = %{activated() | mode: :query}
 
     {model, _} = feed(model, chars("\\set tenant_batch 25") ++ [{:key, :enter}])
-    assert model.tenant_batch == 25
+    assert model.session.settings.tenant_batch == 25
     assert model.flash == {:info, "tenant_batch set to 25"}
 
     {model, _} = feed(model, chars("\\set tenant_batch off") ++ [{:key, :enter}])
-    assert model.tenant_batch == nil
+    assert model.session.settings.tenant_batch == nil
 
     {model, _} = feed(model, chars("\\set tenant_batch 0") ++ [{:key, :enter}])
-    assert model.tenant_batch == nil
+    assert model.session.settings.tenant_batch == nil
     assert {:error, "usage: " <> _} = model.flash
   end
 
@@ -223,11 +238,17 @@ defmodule Efsql.Tui.AppTest do
     batches = {:batches, [{:fan_out, []}, {:fan_out, []}], 4}
 
     plan = %Efsql.Physical.Plan{access: batches, ops: [], columns: [:name, :_tenant]}
-    {batched, _} = feed(model, [{:done, :query, {:ok, {plan, rows, %{}, 1}}}])
+
+    {batched, _} =
+      feed(model, [{:done, :query, {:ok, {Efsql.Result.new(plan, rows, 1), model.session}}}])
+
     assert frame_text(batched) =~ "(2 rows, 1 ms) · 2 transactions"
 
     plan = %Efsql.Physical.Plan{plan | access: {:fan_out, []}}
-    {single, _} = feed(model, [{:done, :query, {:ok, {plan, rows, %{}, 1}}}])
+
+    {single, _} =
+      feed(model, [{:done, :query, {:ok, {Efsql.Result.new(plan, rows, 1), model.session}}}])
+
     refute frame_text(single) =~ "transactions"
   end
 
@@ -237,7 +258,9 @@ defmodule Efsql.Tui.AppTest do
     rows = [%{id: uuid, item: "widget"}]
 
     plan = %Efsql.Physical.Plan{access: {:pk_range, nil, nil, nil, []}, ops: []}
-    {model, _} = feed(model, [{:done, :query, {:ok, {plan, rows, %{}, 1}}}])
+
+    {model, _} =
+      feed(model, [{:done, :query, {:ok, {Efsql.Result.new(plan, rows, 1), model.session}}}])
 
     text = frame_text(model)
     assert text =~ uuid
@@ -250,7 +273,10 @@ defmodule Efsql.Tui.AppTest do
     rows = [%{id: "0001", blob: long}]
 
     plan = %Efsql.Physical.Plan{access: {:pk_range, nil, nil, nil, []}, ops: []}
-    {model, _} = feed(model, [{:done, :query, {:ok, {plan, rows, %{}, 1}}}])
+
+    {model, _} =
+      feed(model, [{:done, :query, {:ok, {Efsql.Result.new(plan, rows, 1), model.session}}}])
+
     {model, _} = feed(model, [{:key, :tab}, {:key, :enter}])
     assert model.mode == :inspector
 
@@ -273,7 +299,10 @@ defmodule Efsql.Tui.AppTest do
       plan = %Efsql.Physical.Plan{access: {:pk_range, nil, nil, nil, []}, ops: []}
 
       {model, _} =
-        feed(model, [{:done, :query, {:ok, {plan, [%{id: "0001", blob: long}], %{}, 1}}}])
+        feed(model, [
+          {:done, :query,
+           {:ok, {Efsql.Result.new(plan, [%{id: "0001", blob: long}], 1), model.session}}}
+        ])
 
       {model, _} = feed(model, [{:key, :tab}, {:key, :enter}])
       model
@@ -383,7 +412,12 @@ defmodule Efsql.Tui.AppTest do
     test "a value that fits shows no scroll indicator" do
       model = %{activated() | mode: :query}
       plan = %Efsql.Physical.Plan{access: {:pk_range, nil, nil, nil, []}, ops: []}
-      {model, _} = feed(model, [{:done, :query, {:ok, {plan, [%{id: "0001"}], %{}, 1}}}])
+
+      {model, _} =
+        feed(model, [
+          {:done, :query, {:ok, {Efsql.Result.new(plan, [%{id: "0001"}], 1), model.session}}}
+        ])
+
       {model, _} = feed(model, [{:key, :tab}, {:key, :enter}])
 
       # the position indicator only appears when the value overflows
diff --git a/test/tui_width_test.exs b/test/tui_width_test.exs
index 0242b0e..d7de904 100644
--- a/test/tui_width_test.exs
+++ b/test/tui_width_test.exs
@@ -7,6 +7,7 @@ defmodule Efsql.Tui.WidthTest do
   alias Efsql.Tui.View
 
   @plan %Efsql.Physical.Plan{access: {:pk_range, nil, nil, nil, []}, ops: []}
+  @session %Efsql.Session{tenant: :t, tenant_name: "demo", storage_id: "s"}
 
   defp results(rows, size) do
     {model, _} =
@@ -15,11 +16,9 @@ defmodule Efsql.Tui.WidthTest do
           size: size,
           mode: :query,
           busy: "q",
-          tenant: :t,
-          tenant_id: "demo",
-          storage_id: "s"
+          session: @session
         },
-        {:done, :query, {:ok, {@plan, rows, %{}, 1}}}
+        {:done, :query, {:ok, {Efsql.Result.new(@plan, rows, 1), @session}}}
       )
 
     model
diff --git a/test/types_test.exs b/test/types_test.exs
index 64b33b6..5c683e0 100644
--- a/test/types_test.exs
+++ b/test/types_test.exs
@@ -65,4 +65,21 @@ defmodule Efsql.TypesTest do
       assert Types.index_key(:active) == :active
     end
   end
+
+  test "equality_key is equal exactly when compare/2 says equal" do
+    pairs = [
+      {Decimal.new("1.0"), Decimal.new("1.00")},
+      {~N[2024-01-01 00:00:00], ~N[2024-01-01 00:00:00.000000]},
+      {~U[2024-01-01 00:00:00Z], ~U[2024-01-01 00:00:00.000Z]},
+      {~T[09:30:00], ~T[09:30:00.000000]}
+    ]
+
+    for {a, b} <- pairs do
+      assert Types.compare(a, b) == :eq
+      assert Types.equality_key(a) == Types.equality_key(b)
+    end
+
+    refute Types.equality_key(Decimal.new("1.0")) == Types.equality_key(Decimal.new("1.1"))
+    assert Types.equality_key("x") == "x"
+  end
 end