diff --git a/docs/docs.logflare.com/docs/concepts/endpoints.md b/docs/docs.logflare.com/docs/concepts/endpoints.md
index 9d499871ab..b15df9e086 100644
--- a/docs/docs.logflare.com/docs/concepts/endpoints.md
+++ b/docs/docs.logflare.com/docs/concepts/endpoints.md
@@ -106,6 +106,8 @@ Caching is performed on a query parameter basis. As such, if there are three API
 
 When querying a specific endpoint version with `LF-ENDPOINT-VERSION`, the version number also partitions the cache. A request for the current endpoint and a request for version `1` use separate caches even when the query parameters match.
 
+ClickHouse scheduling classes do not partition successful result caches. Operators can apply [query-class policy](../self-hosting/clickhouse-query-policy.md) on cache misses and refreshes without changing query results. Current operator-enforced endpoint limits also apply to historical versions.
+
 ### Proactive Requerying
 
 Logflare endpoints can be proactively requeried to ensure that the cache does not become stale throughout the cache lifetime.
diff --git a/docs/docs.logflare.com/docs/self-hosting/clickhouse-query-policy.md b/docs/docs.logflare.com/docs/self-hosting/clickhouse-query-policy.md
new file mode 100644
index 0000000000..96374c67a8
--- /dev/null
+++ b/docs/docs.logflare.com/docs/self-hosting/clickhouse-query-policy.md
@@ -0,0 +1,94 @@
+---
+title: ClickHouse query-class policy
+---
+
+# ClickHouse query-class policy
+
+Operators can configure scheduling and resource limits independently of ClickHouse read-cluster routing. The backend configuration key is `query_class_settings`. It is disabled by default; an empty map removes the class policy without removing endpoint-owned limits.
+
+## Operator configuration
+
+Use the internal operator function with a persisted admin user. The admin flag is provisioned out of band and rechecked on every save. Customer backend changesets, the backend API, and customer forms cannot introduce, change, or clear this configuration. Public backend JSON omits it. An operator must clear this policy before changing backend type. Ordinary saves merge under a row lock so stale customer snapshots cannot erase newer operator policy. No new customer-supplied ClickHouse settings are enabled.
+
+```elixir
+admin = Logflare.Users.get(admin_id)
+backend = Logflare.Backends.get_backend(backend_id)
+
+settings = %{
+  "default" => %{"priority" => 5},
+  "dashboard_logs_free" => %{"priority" => 1},
+  "dashboard_logs_paid" => %{"priority" => 1},
+  "dashboard_reports_free" => %{"priority" => 1},
+  "dashboard_reports_paid" => %{"priority" => 1},
+  "dashboard_observability" => %{"priority" => 1},
+  "api_free" => %{"priority" => 10, "max_threads" => 4},
+  "api_paid" => %{"priority" => 10, "max_threads" => 4},
+  "mcp" => %{"priority" => 10, "max_threads" => 4}
+}
+
+{:ok, backend} =
+  Logflare.Backends.configure_query_class_settings(admin, backend, settings)
+```
+
+These values are an example, not production configuration. Choose them after checking the deployed query-user profile and load-testing in staging.
+
+A nonempty configuration must include a `default` with a positive integer `priority`. A class can inherit its priority from this default. Missing, unknown, or unconfigured labels receive the default policy. Only the classes listed above are accepted at save time.
+
+The class is selected from the original `LF-ENDPOINT-CLICKHOUSE-READ-CLUSTER-LABEL` request header, even when routing resolves to another cluster or retries on the default cluster. Known query-class labels do not generate unconfigured-routing warnings; unknown routing labels still do.
+
+The header is a scheduling hint, not an authorization boundary. Preserve the existing endpoint authentication, and ensure the trusted Management API classifier supplies the header rather than forwarding an untrusted client's value. Selecting a label does not grant access to another tenant's data.
+
+## Allowed settings and precedence
+
+| Setting | Accepted value |
+| --- | --- |
+| `priority` | Positive integer; 1 is highest priority, larger values are lower priority |
+| `max_threads` | Positive integer |
+| `max_memory_usage` | Positive integer bytes |
+| `max_bytes_to_read` | Positive integer bytes |
+| `max_rows_to_read` | Positive integer rows |
+| `max_execution_time` | Positive integer or fractional seconds |
+| `read_overflow_mode` | Only `throw`; inserted automatically with byte or row limits |
+| `timeout_overflow_mode` | Only `throw`; inserted automatically with an execution-time limit |
+
+Zero (unlimited or no priority), negative values, wrong types, unknown keys, and overflow modes returning partial results are rejected. String values are serialized only from the fixed enum allowlist. Output-format, join semantics, sandbox permissions, and external table/URL settings are not allowed.
+
+ClickHouse profile defaults supply settings not specified by Logflare. Logflare combines class default, selected class, and endpoint-owned settings. Resource limits use the smallest configured value across all three policies; non-limit settings such as priority use the later policy. Hard profile ceilings must be ClickHouse constraints, not merely default values. The server still enforces those constraints after merging.
+
+Endpoint-owned policy is passed separately from consumer SQL. The adaptor checks for consumer overrides anywhere in the final AST, then injects the combined policy once. Supported `EXPLAIN SELECT` wrappers are preserved. Previews use the same policy builder:
+
+```elixir
+Logflare.Endpoints.get_transformed_query(endpoint, params, read_cluster: "api_paid")
+```
+
+Historical endpoint versions retain current operator ceilings. Cached snapshot reloads also overlay current endpoint policy. Saving endpoint-owned limits invalidates caches for all endpoint versions and stops their active refresh tasks.
+
+Successful result caches remain class-agnostic: changing the class does not create a different result-cache key or run a query on a cache hit. Misses and background refreshes enforce policy at execution time. A refresh retains the class from the request that originally created the cache, but reads current backend policy. Backend class-policy changes invalidate backend metadata across the cluster, but do not evict reusable successful results, restart ingesters, or tear down active read connections.
+
+## Deployment preflight
+
+Before enabling this policy on a backend, connect using its actual query credentials, including any dedicated query user. Inspect the deployed settings and constraints:
+
+```sql
+SELECT name, type, value, min, max, readonly
+FROM system.settings
+WHERE name IN (
+  'readonly', 'priority', 'max_threads', 'max_execution_time',
+  'max_memory_usage', 'max_bytes_to_read', 'max_rows_to_read',
+  'read_overflow_mode', 'timeout_overflow_mode'
+);
+```
+
+For every newly used key, issue a small SELECT with the intended setting using those credentials. Verify both the accepted values and rejection of values outside profile constraints. `readonly = 1` can prohibit changes unless the setting is specifically changeable in readonly mode; `readonly = 2` allows settings changes subject to constraints. Do not weaken `readonly`, grant DDL, or make `readonly` itself changeable to make these probes pass. See [ClickHouse setting constraints](https://clickhouse.com/docs/operations/settings/constraints-on-settings).
+
+Local regression tests exercise every allowlisted key with a temporary `readonly = 2` user and profile ceilings. They do not establish that the production `logflare_prod` profile permits these values. Production verification is an operator rollout requirement.
+
+Deploy Logflare support on every instance and drain older versions before enabling the policy. Apply the backend class configuration and a matching `priority = 5` baseline in the ClickHouse `logflare` profile in the same coordinated rollout window, **not the profile change before Logflare class policy**. Consider `priority MIN 1` to prevent bypass paths from opting out with zero. Ensure configured limits fit the profile's constraints; use server-side MAX constraints where cluster-wide ceilings are required.
+
+## Verification and rollout monitoring
+
+Ensure query-setting logging is enabled for the verification queries. For actual API/MCP, dashboard, missing-label, and unknown-label requests, inspect `system.query_log.Settings` on Logflare Reads and verify priority and limits, including requests routed to the default cluster. Use the query ID to identify the request; a small local regression test verifies the recorded API values.
+
+In staging, compare dashboard p95 latency and timeout rates under competing API load. During rollout, monitor query errors, delayed concurrency slots, paused queries, memory, and connections. Lower-priority queries retain memory and connections while paused and may see more execution-time failures. Priority is not tenant isolation or aggregate admission control.
+
+Aggregate Management API throttling (O11Y-2662) and moving API/MCP classes to separately sized compute remain separate mitigations. Roll back class policy through the admin function with `%{}`; coordinate any profile rollback after accounting for requests and instances still using class policy.
diff --git a/lib/logflare/backends.ex b/lib/logflare/backends.ex
index 1dff44a8f4..ddc6888348 100644
--- a/lib/logflare/backends.ex
+++ b/lib/logflare/backends.ex
@@ -8,6 +8,7 @@ defmodule Logflare.Backends do
 
   alias Ecto.Changeset
   alias Logflare.Backends.Adaptor
+  alias Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryClassSettings
   alias Logflare.Backends.Backend
   alias Logflare.Backends.BackendRegistry
   alias Logflare.Backends.ConsolidatedSup
@@ -318,7 +319,27 @@ defmodule Logflare.Backends do
   Updates the config of a Backend.
   """
   @spec update_backend(Backend.t(), map()) :: {:ok, Backend.t()} | {:error, Changeset.t()}
-  def update_backend(%Backend{} = backend, attrs) do
+  def update_backend(%Backend{} = backend, attrs), do: update_backend_config(backend, attrs, [])
+
+  @spec configure_query_class_settings(User.t(), Backend.t(), map()) ::
+          {:ok, Backend.t()} | {:error, term()}
+  def configure_query_class_settings(%User{id: user_id}, %Backend{id: id}, settings) do
+    with %User{admin: true} <- Repo.get(User, user_id),
+         %Backend{type: :clickhouse} = backend <- get_backend(id),
+         {:ok, settings} <- QueryClassSettings.normalize(settings) do
+      update_backend_config(backend, %{config: %{query_class_settings: settings}},
+        allow_query_class_settings: true,
+        query_policy_only: true
+      )
+    else
+      nil -> {:error, :not_found}
+      %User{} -> {:error, :forbidden}
+      %Backend{} -> {:error, :not_clickhouse_backend}
+      error -> error
+    end
+  end
+
+  defp update_backend_config(%Backend{} = backend, attrs, opts) do
     alerts_modified = Map.has_key?(attrs, :alert_queries)
 
     default_ingest_modified? =
@@ -328,20 +349,29 @@ defmodule Logflare.Backends do
 
     source_id = Map.get(attrs, "source_id") || Map.get(attrs, :source_id)
 
-    changeset =
-      backend
-      |> Backend.changeset(attrs)
-      |> validate_default_ingest_source(source_id)
-      |> then(fn changeset ->
-        if alerts_modified do
-          Changeset.put_assoc(changeset, :alert_queries, Map.get(attrs, :alert_queries))
-        else
-          changeset
+    result =
+      Repo.transact(fn ->
+        current =
+          from(b in Backend, where: b.id == ^backend.id, lock: "FOR UPDATE")
+          |> Repo.one!()
+          |> typecast_config_string_map_to_atom_map()
+
+        current = if alerts_modified, do: Repo.preload(current, :alert_queries), else: current
+
+        changeset =
+          current
+          |> Backend.changeset(attrs, opts)
+          |> validate_default_ingest_source(source_id)
+          |> maybe_put_backend_alerts(attrs, alerts_modified)
+
+        with :ok <- validate_query_policy_backend(current, opts),
+             {:ok, updated} <- Repo.update(changeset) do
+          {:ok, {current, updated}}
         end
       end)
 
-    case Repo.update(changeset) do
-      {:ok, updated} ->
+    case result do
+      {:ok, {backend, updated}} ->
         updated = preload_sources(updated)
 
         updated =
@@ -357,11 +387,15 @@ defmodule Logflare.Backends do
             updated
           end
 
-        Enum.each(updated.sources, &restart_source_sup(&1))
-
-        apply_consolidated_pipeline_change(backend, updated, default_ingest_modified?)
-
-        if config_modified?, do: Adaptor.on_backend_config_changed(updated)
+        if Keyword.get(opts, :query_policy_only, false) do
+          keys = [{__MODULE__, updated.id}]
+          Logflare.ContextCache.bust_keys(keys)
+          Cluster.Utils.rpc_multicast(Logflare.ContextCache, :bust_keys, [keys])
+        else
+          Enum.each(updated.sources, &restart_source_sup(&1))
+          apply_consolidated_pipeline_change(backend, updated, default_ingest_modified?)
+          notify_backend_config_change(updated, config_modified?)
+        end
 
         {:ok, typecast_config_string_map_to_atom_map(updated)}
 
@@ -370,11 +404,33 @@ defmodule Logflare.Backends do
     end
   end
 
+  @spec maybe_put_backend_alerts(Changeset.t(), map(), boolean()) :: Changeset.t()
+  defp maybe_put_backend_alerts(changeset, attrs, true),
+    do: Changeset.put_assoc(changeset, :alert_queries, Map.get(attrs, :alert_queries))
+
+  defp maybe_put_backend_alerts(changeset, _attrs, false), do: changeset
+
+  @spec notify_backend_config_change(Backend.t(), boolean()) :: term()
+  defp notify_backend_config_change(_backend, false), do: :ok
+
+  defp notify_backend_config_change(backend, true),
+    do: Adaptor.on_backend_config_changed(backend)
+
+  @spec validate_query_policy_backend(Backend.t(), Keyword.t()) ::
+          :ok | {:error, :not_clickhouse_backend}
+  defp validate_query_policy_backend(%Backend{type: :clickhouse}, _opts), do: :ok
+
+  defp validate_query_policy_backend(_backend, opts) do
+    if Keyword.get(opts, :query_policy_only, false),
+      do: {:error, :not_clickhouse_backend},
+      else: :ok
+  end
+
   @spec validate_default_ingest_source(Changeset.t(), String.t() | integer() | nil) ::
           Changeset.t()
   defp validate_default_ingest_source(%{changes: %{default_ingest?: true}} = changeset, source_id)
        when is_non_empty_binary(source_id) or is_integer(source_id) do
-    case Sources.get(source_id) do
+    case Repo.get(Source, source_id) do
       %Source{default_ingest_backend_enabled?: true} ->
         changeset
 
diff --git a/lib/logflare/backends/adaptor/clickhouse_adaptor.ex b/lib/logflare/backends/adaptor/clickhouse_adaptor.ex
index 44a51449cc..ddddac9dc0 100644
--- a/lib/logflare/backends/adaptor/clickhouse_adaptor.ex
+++ b/lib/logflare/backends/adaptor/clickhouse_adaptor.ex
@@ -21,6 +21,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
   alias __MODULE__.Ingester
   alias __MODULE__.Pipeline
   alias __MODULE__.Provisioner
+  alias __MODULE__.QueryClassSettings
   alias __MODULE__.QueryConnectionSup
   alias __MODULE__.QueryErrorNormalizer
   alias __MODULE__.QueryTemplates
@@ -34,6 +35,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
   alias Logflare.Backends.IngestEventQueue
   alias Logflare.Backends.Adaptor.QueryResult
   alias Logflare.Backends.QueryError
+  alias Logflare.Endpoints.ClickHouseSettings
   alias Logflare.LogEvent
   alias Logflare.LogEvent.TypeDetection
   alias Logflare.Sources.Source
@@ -57,6 +59,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
     :labeled_read_pool_size,
     :read_only_urls,
     :default_read_cluster,
+    :query_class_settings,
     :query_user,
     :query_password,
     :replica_routing_param
@@ -144,6 +147,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
   def redact_config(config) do
     config
     |> Map.put(:password, "REDACTED")
+    |> Map.delete(:query_class_settings)
     |> redact_query_password()
   end
 
@@ -198,6 +202,13 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
       )
       when is_non_empty_binary(query_string) and is_list(declared_params) and is_map(input_params) and
              is_list(opts) do
+    endpoint_settings =
+      if is_map(endpoint_query),
+        do: Map.get(endpoint_query, :enforced_clickhouse_settings, %{}),
+        else: %{}
+
+    opts = Keyword.put(opts, :enforced_clickhouse_settings, endpoint_settings)
+
     with {:ok, {limited_query, max_rows}} <- limit_endpoint_query(query_string, endpoint_query) do
       execute_query_with_params(
         backend,
@@ -314,6 +325,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
        labeled_read_pool_size: :integer,
        read_only_urls: {:map, :string},
        default_read_cluster: :string,
+       query_class_settings: :map,
        use_async_inserts_for_small_batches: :boolean,
        use_async_inserts_only: :boolean,
        async_insert_cluster_url: :string,
@@ -335,6 +347,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
       :labeled_read_pool_size,
       :read_only_urls,
       :default_read_cluster,
+      :query_class_settings,
       :use_async_inserts_for_small_batches,
       :use_async_inserts_only,
       :async_insert_cluster_url,
@@ -413,6 +426,7 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
     )
     |> validate_read_only_urls()
     |> validate_default_read_cluster()
+    |> validate_query_class_settings()
     |> validate_user_pass()
     |> validate_query_user_pass()
     |> validate_number(:read_pool_size,
@@ -745,27 +759,62 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
 
   def execute_ch_query(%Backend{} = backend, statement, params, opts)
       when is_list_or_map(params) and is_list(opts) do
-    requested = Keyword.get(opts, :read_cluster)
-    label = resolve_read_cluster_label(backend, requested)
+    with {:ok, statement} <- prepare_query(backend, statement, opts) do
+      requested = Keyword.get(opts, :read_cluster)
+      label = resolve_read_cluster_label(backend, requested)
 
-    warn_on_unconfigured_read_cluster(backend, requested, label)
-    headers = Keyword.get(opts, :headers, [])
+      warn_on_unconfigured_read_cluster(backend, requested, label)
+      headers = Keyword.get(opts, :headers, [])
 
-    {result, queried_label} =
-      case do_ch_query_on_label(backend, statement, params, label, headers) do
-        {:error, %QueryError{kind: :connection_error}} = error ->
-          maybe_retry_on_default_cluster(backend, statement, params, label, headers, error)
+      {result, queried_label} =
+        case do_ch_query_on_label(backend, statement, params, label, headers) do
+          {:error, %QueryError{kind: :connection_error}} = error ->
+            maybe_retry_on_default_cluster(backend, statement, params, label, headers, error)
 
-        result ->
-          {result, label}
-      end
+          result ->
+            {result, label}
+        end
 
-    emit_query_error_telemetry(result, %{
-      backend_id: backend.id,
-      read_cluster: read_cluster_tag(queried_label)
-    })
+      emit_query_error_telemetry(result, %{
+        backend_id: backend.id,
+        read_cluster: read_cluster_tag(queried_label)
+      })
 
-    result
+      result
+    else
+      {:error, reason} ->
+        {:error, %{query_error(:invalid_query, reason) | description: reason}}
+    end
+  end
+
+  @spec prepare_query(Backend.t(), iodata(), Keyword.t()) ::
+          {:ok, String.t()} | {:error, String.t()}
+  def prepare_query(%Backend{config: config}, statement, opts \\ []) do
+    classes = Map.get(config, :query_class_settings) || %{}
+    endpoint_settings = Keyword.get(opts, :enforced_clickhouse_settings, %{})
+
+    with {:ok, settings} <-
+           QueryClassSettings.resolve(
+             classes,
+             Keyword.get(opts, :read_cluster),
+             endpoint_settings
+           ) do
+      ClickHouseSettings.enforce(IO.iodata_to_binary(statement), settings)
+    end
+  end
+
+  @spec validate_query_class_settings(Changeset.t()) :: Changeset.t()
+  defp validate_query_class_settings(changeset) do
+    case Changeset.get_field(changeset, :query_class_settings) do
+      nil ->
+        changeset
+
+      settings ->
+        case QueryClassSettings.normalize(settings) do
+          {:ok, settings} -> Changeset.put_change(changeset, :query_class_settings, settings)
+          {:error, reason} -> Changeset.add_error(changeset, :query_class_settings, reason)
+        end
+    end
   end
 
   @spec do_ch_query_on_label(
@@ -816,7 +865,10 @@ defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor do
   @spec warn_on_unconfigured_read_cluster(Backend.t(), String.t() | nil, String.t() | nil) :: :ok
   defp warn_on_unconfigured_read_cluster(%Backend{} = backend, requested, label)
        when is_non_empty_binary(requested) and requested != label do
-    Logger.warning(
+    level = if QueryClassSettings.class_label?(requested), do: :debug, else: :warning
+
+    Logger.log(
+      level,
       "ClickHouse read cluster not configured, falling back to resolved read cluster",
       user_id: backend.user_id,
       backend_id: backend.id,
diff --git a/lib/logflare/backends/adaptor/clickhouse_adaptor/query_class_settings.ex b/lib/logflare/backends/adaptor/clickhouse_adaptor/query_class_settings.ex
new file mode 100644
index 0000000000..f9b6cf082c
--- /dev/null
+++ b/lib/logflare/backends/adaptor/clickhouse_adaptor/query_class_settings.ex
@@ -0,0 +1,53 @@
+defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryClassSettings do
+  @moduledoc """
+  Admin-owned ClickHouse scheduling policy, independent of read-cluster routing.
+  An empty map disables class policy. Configured policy requires a default priority.
+  """
+
+  alias Logflare.Endpoints.ClickHouseSettings
+
+  @classes ~w(dashboard_logs_free dashboard_logs_paid dashboard_reports_free dashboard_reports_paid dashboard_observability mcp api_free api_paid)
+
+  @spec class_label?(term()) :: boolean()
+  def class_label?(label), do: label in @classes
+
+  @spec normalize(term()) :: {:ok, map()} | {:error, String.t()}
+  def normalize(settings) when settings == %{}, do: {:ok, settings}
+
+  def normalize(settings) when is_map(settings) do
+    with {:ok, normalized} <- normalize_classes(settings),
+         %{"priority" => priority} when is_integer(priority) <- normalized["default"] do
+      {:ok, normalized}
+    else
+      {:error, _} = error -> error
+      _ -> {:error, "Query class settings require a default with a positive priority"}
+    end
+  end
+
+  def normalize(_), do: {:error, "Query class settings must be a map"}
+
+  @spec resolve(term(), term(), map()) :: {:ok, map()} | {:error, String.t()}
+  def resolve(classes, requested, endpoint_settings) do
+    with {:ok, classes} <- normalize(classes) do
+      ClickHouseSettings.merge([
+        Map.get(classes, "default", %{}),
+        Map.get(classes, requested, %{}),
+        endpoint_settings
+      ])
+    end
+  end
+
+  @spec normalize_classes(map()) :: {:ok, map()} | {:error, String.t()}
+  defp normalize_classes(settings) do
+    Enum.reduce_while(settings, {:ok, %{}}, fn
+      {label, policy}, {:ok, normalized} when label in @classes or label == "default" ->
+        case ClickHouseSettings.normalize(policy) do
+          {:ok, policy} -> {:cont, {:ok, Map.put(normalized, label, policy)}}
+          {:error, reason} -> {:halt, {:error, "Query class #{label}: #{reason}"}}
+        end
+
+      {label, _policy}, _acc ->
+        {:halt, {:error, "Unknown query class #{inspect(label)}"}}
+    end)
+  end
+end
diff --git a/lib/logflare/backends/backend.ex b/lib/logflare/backends/backend.ex
index 6736e39a15..738fa49e8d 100644
--- a/lib/logflare/backends/backend.ex
+++ b/lib/logflare/backends/backend.ex
@@ -105,16 +105,44 @@ defmodule Logflare.Backends.Backend do
     |> Enum.sort()
   end
 
-  def changeset(backend, attrs) do
+  def changeset(backend, attrs, opts \\ []) do
     backend
     |> cast(attrs, [:type, :config, :name, :description, :metadata, :default_ingest?])
     |> validate_required([:user_id, :type, :config, :name])
     |> validate_inclusion(:type, Map.keys(@adaptor_mapping))
     |> validate_config()
+    |> protect_query_class_settings(opts)
     |> validate_default_ingest()
     |> do_config_change()
   end
 
+  @spec protect_query_class_settings(Changeset.t(), Keyword.t()) :: Changeset.t()
+  defp protect_query_class_settings(changeset, opts) do
+    current = query_class_policy(changeset.data.config_encrypted)
+
+    proposed =
+      query_class_policy(get_field(changeset, :config) || changeset.data.config_encrypted)
+
+    changing_type? = get_field(changeset, :type) != changeset.data.type
+    changing_policy? = current != proposed or (current != %{} and changing_type?)
+
+    if changing_policy? and not Keyword.get(opts, :allow_query_class_settings, false) do
+      add_error(
+        changeset,
+        :"config.query_class_settings",
+        "can only be configured by an administrator"
+      )
+    else
+      changeset
+    end
+  end
+
+  @spec query_class_policy(term()) :: term()
+  defp query_class_policy(config) when is_map(config),
+    do: Map.get(config, :query_class_settings) || Map.get(config, "query_class_settings") || %{}
+
+  defp query_class_policy(_config), do: %{}
+
   # temp function
   defp do_config_change(%Changeset{changes: %{config: config}} = changeset) do
     changeset
diff --git a/lib/logflare/endpoints.ex b/lib/logflare/endpoints.ex
index d23be6e135..2b32784367 100644
--- a/lib/logflare/endpoints.ex
+++ b/lib/logflare/endpoints.ex
@@ -9,6 +9,7 @@ defmodule Logflare.Endpoints do
   alias Logflare.Alerting
   alias Logflare.Alerting.AlertQuery
   alias Logflare.Backends
+  alias Logflare.Backends.Adaptor.ClickHouseAdaptor
   alias Logflare.Backends.Adaptor.QueryResult
   alias Logflare.Backends.Backend
   alias Logflare.Backends.QueryError
@@ -672,8 +673,7 @@ defmodule Logflare.Endpoints do
              else: expanded_query
            ),
          {:ok, transformed_query} <-
-           Sql.transform(query_language, transform_input, user_id),
-         {:ok, transformed_query} <- apply_enforced_settings(endpoint_query, transformed_query) do
+           Sql.transform(query_language, transform_input, user_id) do
       :telemetry.span(
         [:logflare, :endpoints, :run_query, :exec_query_on_backend],
         %{endpoint_id: endpoint_query.id, language: query_language},
@@ -822,10 +822,10 @@ defmodule Logflare.Endpoints do
 
   Useful for debugging and UI display of merged sandbox queries.
   """
-  @spec get_transformed_query(EndpointQuery.t(), params :: map()) ::
+  @spec get_transformed_query(EndpointQuery.t(), map(), Keyword.t()) ::
           {:ok, String.t()} | {:error, String.t()}
-  def get_transformed_query(%EndpointQuery{} = endpoint_query, params \\ %{})
-      when is_map(params) do
+  def get_transformed_query(%EndpointQuery{} = endpoint_query, params \\ %{}, opts \\ [])
+      when is_map(params) and is_list(opts) do
     %EndpointQuery{
       query: query_string,
       user_id: user_id,
@@ -863,19 +863,24 @@ defmodule Logflare.Endpoints do
              else: expanded_query
            ),
          {:ok, transformed_query} <- Sql.transform(query_language, transform_input, user_id),
-         {:ok, transformed_query} <- apply_enforced_settings(endpoint_query, transformed_query) do
+         {:ok, transformed_query} <- preview_query_policy(endpoint_query, transformed_query, opts) do
       {:ok, transformed_query}
     end
   end
 
-  defp apply_enforced_settings(
-         %EndpointQuery{language: :ch_sql, enforced_clickhouse_settings: settings},
-         transformed_query
-       )
-       when is_map(settings),
-       do: ClickHouseSettings.enforce(transformed_query, settings)
+  @spec preview_query_policy(EndpointQuery.t(), String.t(), Keyword.t()) ::
+          {:ok, String.t()} | {:error, term()}
+  defp preview_query_policy(%EndpointQuery{language: :ch_sql} = endpoint, query, opts) do
+    with {:ok, backend} <- get_backend_for_query(endpoint) do
+      ClickHouseAdaptor.prepare_query(
+        backend,
+        query,
+        Keyword.put(opts, :enforced_clickhouse_settings, endpoint.enforced_clickhouse_settings)
+      )
+    end
+  end
 
-  defp apply_enforced_settings(_endpoint_query, transformed_query), do: {:ok, transformed_query}
+  defp preview_query_policy(_endpoint_query, query, _opts), do: {:ok, query}
 
   @spec maybe_convert_lql_to_sql(
           lql_param :: String.t() | nil,
@@ -975,7 +980,11 @@ defmodule Logflare.Endpoints do
         end
 
       redact_pii = Keyword.get(opts, :redact_pii, endpoint_query.redact_pii)
-      query_opts = Keyword.put(opts, :query_type, :endpoint)
+
+      query_opts =
+        opts
+        |> Keyword.put(:query_type, :endpoint)
+        |> Keyword.put(:enforced_clickhouse_settings, endpoint_query.enforced_clickhouse_settings)
 
       case adaptor.execute_query(backend, query_args, query_opts) do
         {:ok, %QueryResult{rows: rows} = result} ->
diff --git a/lib/logflare/endpoints/clickhouse_settings.ex b/lib/logflare/endpoints/clickhouse_settings.ex
index e902a27156..990e5f6fb8 100644
--- a/lib/logflare/endpoints/clickhouse_settings.ex
+++ b/lib/logflare/endpoints/clickhouse_settings.ex
@@ -1,7 +1,7 @@
 defmodule Logflare.Endpoints.ClickHouseSettings do
   @moduledoc """
-  Endpoint-owned ClickHouse resource limits. Consumer SQL is validated separately;
-  these settings are applied only after the final sandboxed query is assembled.
+  Typed ClickHouse resource and scheduling policy. Consumer SQL is validated separately;
+  trusted settings are combined before being applied to the final assembled query.
 
   Only internal administrators can configure these limits; the public endpoint
   editor and API do not accept them. The final outer SETTINGS clause is built
@@ -12,7 +12,17 @@ defmodule Logflare.Endpoints.ClickHouseSettings do
 
   alias Logflare.Sql.Parser
 
-  @resource_limits ~w(max_bytes_to_read max_rows_to_read max_memory_usage max_execution_time)
+  @resource_limits ~w(max_bytes_to_read max_rows_to_read max_memory_usage max_execution_time max_threads)
+  @setting_types %{
+    "priority" => :integer,
+    "max_threads" => :integer,
+    "max_bytes_to_read" => :integer,
+    "max_rows_to_read" => :integer,
+    "max_memory_usage" => :integer,
+    "max_execution_time" => :float,
+    "read_overflow_mode" => {:enum, ["throw"]},
+    "timeout_overflow_mode" => {:enum, ["throw"]}
+  }
   @max_limit 9_223_372_036_854_775_807
 
   @spec normalize(map()) :: {:ok, map()} | {:error, String.t()}
@@ -25,12 +35,36 @@ defmodule Logflare.Endpoints.ClickHouseSettings do
           settings
         end
 
+      settings =
+        if Map.has_key?(settings, "max_execution_time"),
+          do: Map.put(settings, "timeout_overflow_mode", "throw"),
+          else: settings
+
       {:ok, settings}
     end
   end
 
   def normalize(_), do: {:error, "Enforced ClickHouse settings must be a map"}
 
+  @spec merge([map()]) :: {:ok, map()} | {:error, String.t()}
+  def merge(policies) do
+    Enum.reduce_while(policies, {:ok, %{}}, fn policy, {:ok, combined} ->
+      case normalize(policy) do
+        {:ok, policy} ->
+          {:cont, {:ok, Map.merge(combined, policy, &merge_value/3)}}
+
+        error ->
+          {:halt, error}
+      end
+    end)
+  end
+
+  @spec merge_value(String.t(), term(), term()) :: term()
+  defp merge_value(key, existing, incoming) when key in @resource_limits,
+    do: min(existing, incoming)
+
+  defp merge_value(_key, _existing, incoming), do: incoming
+
   @spec enforce(String.t(), map()) :: {:ok, String.t()} | {:error, String.t()}
   def enforce(query, settings) when settings == %{}, do: {:ok, query}
 
@@ -64,26 +98,31 @@ defmodule Logflare.Endpoints.ClickHouseSettings do
     do: {:error, "Expected one ClickHouse SELECT or EXPLAIN SELECT query"}
 
   defp validate_entries(settings) do
-    Enum.reduce_while(settings, :ok, fn
-      {key, value}, :ok
-      when key in @resource_limits and is_integer(value) and value > 0 and
-             value <= @max_limit ->
-        {:cont, :ok}
-
-      {"read_overflow_mode", "throw"}, :ok ->
+    Enum.reduce_while(settings, :ok, fn {key, value}, :ok ->
+      if valid_value?(Map.get(@setting_types, key), value) do
         {:cont, :ok}
-
-      {key, _value}, :ok ->
+      else
         {:halt, {:error, "Invalid enforced ClickHouse setting #{inspect(key)} or value"}}
+      end
     end)
   end
 
+  @spec valid_value?(term(), term()) :: boolean()
+  defp valid_value?(:integer, value),
+    do: is_integer(value) and value > 0 and value <= @max_limit
+
+  defp valid_value?(:float, value),
+    do: is_number(value) and value > 0 and value <= @max_limit
+
+  defp valid_value?({:enum, values}, value), do: value in values
+  defp valid_value?(_type, _value), do: false
+
   defp settings_sql(settings) do
     clauses =
       settings
       |> Enum.sort()
       |> Enum.map_join(", ", fn
-        {"read_overflow_mode", "throw"} -> "read_overflow_mode = 'throw'"
+        {key, value} when is_binary(value) -> "#{key} = '#{value}'"
         {key, value} -> "#{key} = #{value}"
       end)
 
@@ -93,7 +132,7 @@ defmodule Logflare.Endpoints.ClickHouseSettings do
   defp reject_conflicts(ast, keys) do
     case conflicting_setting(ast, keys) do
       nil -> :ok
-      key -> {:error, "ClickHouse setting #{key} is enforced by this endpoint"}
+      key -> {:error, "ClickHouse setting #{key} is enforced by query policy"}
     end
   end
 
diff --git a/test/logflare/backends/adaptor/clickhouse_adaptor/query_class_execution_test.exs b/test/logflare/backends/adaptor/clickhouse_adaptor/query_class_execution_test.exs
new file mode 100644
index 0000000000..3b50899008
--- /dev/null
+++ b/test/logflare/backends/adaptor/clickhouse_adaptor/query_class_execution_test.exs
@@ -0,0 +1,306 @@
+defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryClassExecutionTest do
+  use Logflare.DataCase
+
+  alias Logflare.Backends
+  alias Logflare.Backends.Adaptor.ClickHouseAdaptor
+  alias Logflare.Backends.QueryError
+  alias Logflare.Endpoints
+
+  @policy %{
+    "default" => %{"priority" => 5},
+    "api_free" => %{"priority" => 10, "max_threads" => 4},
+    "api_paid" => %{"priority" => 10, "max_threads" => 4},
+    "mcp" => %{"priority" => 10, "max_threads" => 4},
+    "dashboard_logs_free" => %{"priority" => 1},
+    "dashboard_logs_paid" => %{"priority" => 1},
+    "dashboard_reports_free" => %{"priority" => 1},
+    "dashboard_reports_paid" => %{"priority" => 1},
+    "dashboard_observability" => %{"priority" => 1}
+  }
+
+  setup do
+    insert(:plan)
+    admin = insert(:user, admin: true)
+
+    {_source, original} =
+      setup_clickhouse_test(
+        config: %{
+          read_only_urls: %{"dashboard_logs_paid" => "http://localhost:8123"},
+          default_read_cluster: "dashboard_logs_paid"
+        }
+      )
+
+    assert {:ok, backend} = Backends.configure_query_class_settings(admin, original, @policy)
+    start_supervised!({ClickHouseAdaptor, backend})
+    %{backend: backend, original: original, admin: admin}
+  end
+
+  test "effective priorities use requested classes despite routing fallback", %{backend: backend} do
+    for {label, priority} <-
+          Enum.map(@policy, fn {label, settings} -> {label, settings["priority"]} end) ++
+            [{nil, 5}, {"unknown", 5}] do
+      log =
+        ExUnit.CaptureLog.capture_log(fn ->
+          assert {:ok, {[%{"priority" => ^priority}], _}} =
+                   ClickHouseAdaptor.execute_ch_query(
+                     backend,
+                     "SELECT toInt64(getSetting('priority')) AS priority",
+                     [],
+                     read_cluster: label
+                   )
+        end)
+
+      if label in ["api_free", "api_paid", "mcp"] do
+        refute log =~ "read cluster not configured"
+      end
+    end
+  end
+
+  test "connection fallback retains the requested API policy", context do
+    assert {:ok, backend} =
+             Backends.update_backend(context.backend, %{
+               config: %{
+                 read_only_urls: %{
+                   "api_paid" => "http://localhost:8123",
+                   "dashboard_logs_paid" => "http://localhost:8123"
+                 }
+               }
+             })
+
+    api_pool = ClickHouseAdaptor.connection_pool_via(backend, "api_paid")
+
+    stub(Ch, :query, fn
+      ^api_pool, _sql, _params, _opts ->
+        {:error, %DBConnection.ConnectionError{message: "unreachable"}}
+
+      pool, sql, params, opts ->
+        Mimic.call_original(Ch, :query, [pool, sql, params, opts])
+    end)
+
+    log =
+      ExUnit.CaptureLog.capture_log(fn ->
+        assert {:ok, {[%{"priority" => 10}], _}} =
+                 ClickHouseAdaptor.execute_ch_query(
+                   backend,
+                   "SELECT toInt64(getSetting('priority')) AS priority",
+                   [],
+                   read_cluster: "api_paid"
+                 )
+      end)
+
+    assert log =~ "read cluster unhealthy"
+  end
+
+  test "results stay class-agnostic while refreshes use live policy and the original class",
+       context do
+    endpoint =
+      insert(:endpoint,
+        user: Logflare.Users.get(context.backend.user_id),
+        backend: context.backend,
+        language: :ch_sql,
+        query: "SELECT 'cached-policy' AS value",
+        cache_duration_seconds: 60,
+        proactive_requerying_seconds: 60
+      )
+
+    test_pid = self()
+
+    stub(Ch, :query, fn conn, sql, params, opts ->
+      sql = IO.iodata_to_binary(sql)
+      if sql =~ "cached-policy", do: send(test_pid, {:executed, sql})
+      Mimic.call_original(Ch, :query, [conn, sql, params, opts])
+    end)
+
+    pid = start_supervised!({Endpoints.ResultsCache, {endpoint, %{}, [read_cluster: "api_paid"]}})
+
+    assert {:ok, %{rows: [%{"value" => "cached-policy"}]}} =
+             Endpoints.run_cached_query(endpoint, %{}, read_cluster: "api_paid")
+
+    assert_received {:executed, first_sql}
+    assert first_sql =~ "priority = 10"
+
+    assert {:ok, %{rows: [%{"value" => "cached-policy"}]}} =
+             Endpoints.run_cached_query(endpoint, %{}, read_cluster: "dashboard_logs_paid")
+
+    refute_received {:executed, _}
+
+    policy = Map.put(@policy, "api_paid", %{"priority" => 12, "max_threads" => 2})
+
+    pool_via = ClickHouseAdaptor.connection_pool_via(context.backend, "dashboard_logs_paid")
+    pool_pid = GenServer.whereis(pool_via)
+    assert is_pid(pool_pid)
+    assert Backends.Cache.get_backend(context.backend.id).config.query_class_settings == @policy
+
+    assert {:ok, _} =
+             Backends.configure_query_class_settings(context.admin, context.backend, policy)
+
+    assert Backends.Cache.get_backend(context.backend.id).config.query_class_settings == policy
+    assert GenServer.whereis(pool_via) == pool_pid
+    send(pid, :refresh)
+    assert_receive {:executed, refreshed_sql}, 5_000
+    assert refreshed_sql =~ "priority = 12"
+    assert refreshed_sql =~ "max_threads = 2"
+    TestUtils.retry_assert(fn -> assert :sys.get_state(pid).query_tasks == [] end)
+  end
+
+  test "endpoint policy merges once, handles converted parameters and matches previews",
+       context do
+    backend = context.backend
+
+    endpoint =
+      insert(:endpoint,
+        user: Logflare.Users.get(backend.user_id),
+        backend: backend,
+        language: :ch_sql,
+        query: "SELECT @value AS value",
+        enforced_clickhouse_settings: %{"max_threads" => 2, "max_execution_time" => 0.5}
+      )
+
+    assert {:ok, preview} =
+             Endpoints.get_transformed_query(endpoint, %{"value" => "hello"},
+               read_cluster: "api_paid"
+             )
+
+    assert preview =~ "priority = 10"
+    assert preview =~ "max_threads = 2"
+    assert preview =~ "max_execution_time = 0.5"
+
+    assert {:ok, %{rows: [%{"value" => "hello"}]}} =
+             Endpoints.run_query(endpoint, %{"value" => "hello"},
+               read_cluster: "api_paid",
+               enforced_clickhouse_settings: %{"priority" => 1}
+             )
+
+    assert {:error, %QueryError{kind: :invalid_query, description: reason}} =
+             ClickHouseAdaptor.execute_ch_query(
+               backend,
+               "SELECT 1 AS n SETTINGS priority = 0",
+               [],
+               read_cluster: "api_paid"
+             )
+
+    assert reason =~ "priority is enforced"
+  end
+
+  test "effective API settings are recorded in system.query_log.Settings", context do
+    id = Ecto.UUID.generate()
+
+    assert {:ok, _} =
+             ClickHouseAdaptor.execute_ch_query(
+               context.backend,
+               "SELECT 1 AS n SETTINGS log_queries = 1, log_query_settings = 1",
+               [],
+               read_cluster: "api_paid",
+               headers: [{"x-clickhouse-query-id", id}]
+             )
+
+    assert {:ok, _} = ClickHouseAdaptor.execute_ch_query(context.original, "SYSTEM FLUSH LOGS")
+
+    TestUtils.retry_assert(fn ->
+      assert {:ok, {[%{"priority" => "10", "threads" => "4"}], _}} =
+               ClickHouseAdaptor.execute_ch_query(
+                 context.original,
+                 "SELECT Settings['priority'] AS priority, Settings['max_threads'] AS threads " <>
+                   "FROM system.query_log WHERE query_id = '#{id}' AND type = 'QueryFinish'"
+               )
+    end)
+  end
+
+  test "all allowed keys work under readonly=2 and profile ceilings remain enforced", context do
+    name = "query_class_reader_#{Ecto.UUID.generate() |> String.replace("-", "")}"
+    admin_backend = context.original
+
+    assert {:ok, _} =
+             ClickHouseAdaptor.execute_ch_query(
+               admin_backend,
+               "CREATE USER #{name} IDENTIFIED BY 'test-secret' SETTINGS " <>
+                 "readonly = 2 READONLY, priority = 5 MIN 1, max_threads = 8 MAX 8, " <>
+                 "max_execution_time = 2 MAX 2, max_memory_usage = 10000000 MAX 10000000, " <>
+                 "max_bytes_to_read = 1000000 MAX 1000000, max_rows_to_read = 1000 MAX 1000"
+             )
+
+    uri = URI.parse(admin_backend.config.url)
+
+    admin_opts =
+      [
+        scheme: uri.scheme,
+        hostname: uri.host,
+        port: uri.port,
+        database: admin_backend.config.database,
+        username: Map.get(admin_backend.config, :username),
+        password: Map.get(admin_backend.config, :password)
+      ]
+      |> Enum.reject(fn {_key, value} -> is_nil(value) end)
+
+    on_exit(fn ->
+      {:ok, pool} = Ch.start_link(admin_opts)
+
+      try do
+        assert {:ok, _} = Ch.query(pool, "DROP USER IF EXISTS #{name}")
+      after
+        GenServer.stop(pool)
+      end
+    end)
+
+    assert {:ok, _} =
+             ClickHouseAdaptor.execute_ch_query(admin_backend, "GRANT SELECT ON *.* TO #{name}")
+
+    policy = %{
+      "default" => %{"priority" => 5},
+      "api_paid" => %{
+        "priority" => 10,
+        "max_threads" => 4,
+        "max_execution_time" => 1.5,
+        "max_memory_usage" => 1_000_000,
+        "max_bytes_to_read" => 100_000,
+        "max_rows_to_read" => 100
+      }
+    }
+
+    assert {:ok, backend} =
+             Backends.configure_query_class_settings(context.admin, context.backend, policy)
+
+    assert {:ok, backend} =
+             Backends.update_backend(backend, %{
+               config: %{query_user: name, query_password: "test-secret"}
+             })
+
+    keys =
+      ~w(priority max_threads max_execution_time max_memory_usage max_bytes_to_read max_rows_to_read read_overflow_mode timeout_overflow_mode readonly)
+
+    sql = "SELECT " <> Enum.map_join(keys, ", ", &"toString(getSetting('#{&1}')) AS #{&1}")
+
+    TestUtils.retry_assert(fn ->
+      assert {:ok, {[settings], _}} =
+               ClickHouseAdaptor.execute_ch_query(backend, sql, [], read_cluster: "api_paid")
+
+      assert settings == %{
+               "priority" => "10",
+               "max_threads" => "4",
+               "max_execution_time" => "1.5",
+               "max_memory_usage" => "1000000",
+               "max_bytes_to_read" => "100000",
+               "max_rows_to_read" => "100",
+               "read_overflow_mode" => "throw",
+               "timeout_overflow_mode" => "throw",
+               "readonly" => "2"
+             }
+    end)
+
+    assert {:error, %QueryError{}} =
+             ClickHouseAdaptor.execute_ch_query(backend, "SELECT 1", [],
+               enforced_clickhouse_settings: %{"max_memory_usage" => 20_000_000}
+             )
+
+    assert {:ok, backend} =
+             Backends.update_backend(backend, %{config: %{query_user: nil, query_password: nil}})
+
+    TestUtils.retry_assert(fn ->
+      assert {:ok, {[%{"readonly" => 0}], _}} =
+               ClickHouseAdaptor.execute_ch_query(
+                 backend,
+                 "SELECT toInt64(getSetting('readonly')) AS readonly"
+               )
+    end)
+  end
+end
diff --git a/test/logflare/backends/adaptor/clickhouse_adaptor/query_class_settings_test.exs b/test/logflare/backends/adaptor/clickhouse_adaptor/query_class_settings_test.exs
new file mode 100644
index 0000000000..2c498c7d95
--- /dev/null
+++ b/test/logflare/backends/adaptor/clickhouse_adaptor/query_class_settings_test.exs
@@ -0,0 +1,98 @@
+defmodule Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryClassSettingsTest do
+  use ExUnit.Case, async: true
+
+  alias Logflare.Backends.Adaptor.ClickHouseAdaptor.QueryClassSettings
+  alias Logflare.Endpoints.ClickHouseSettings
+
+  @classes %{
+    "default" => %{"priority" => 5, "max_threads" => 16},
+    "api_paid" => %{"priority" => 10, "max_threads" => 4},
+    "mcp" => %{"priority" => 10},
+    "dashboard_logs_paid" => %{"priority" => 1}
+  }
+
+  test "selects the requested class and defaults missing or unknown labels" do
+    for label <- [nil, "unknown", "api_free"] do
+      assert {:ok, %{"priority" => 5, "max_threads" => 16}} =
+               QueryClassSettings.resolve(@classes, label, %{})
+    end
+
+    assert {:ok, %{"priority" => 10, "max_threads" => 4}} =
+             QueryClassSettings.resolve(@classes, "api_paid", %{})
+
+    assert {:ok, %{"priority" => 1, "max_threads" => 16}} =
+             QueryClassSettings.resolve(@classes, "dashboard_logs_paid", %{})
+
+    assert {:ok, %{}} = QueryClassSettings.resolve(%{}, "api_paid", %{})
+  end
+
+  test "limits use stricter-wins across defaults, class and endpoint policy" do
+    classes = %{
+      "default" => %{"priority" => 5, "max_threads" => 4, "max_execution_time" => 65},
+      "api_paid" => %{"priority" => 10, "max_threads" => 16, "max_execution_time" => 30}
+    }
+
+    assert {:ok, settings} =
+             QueryClassSettings.resolve(classes, "api_paid", %{
+               "max_threads" => 8,
+               "max_execution_time" => 0.5,
+               "max_memory_usage" => 1024
+             })
+
+    assert settings["priority"] == 10
+    assert settings["max_threads"] == 4
+    assert settings["max_execution_time"] == 0.5
+    assert settings["max_memory_usage"] == 1024
+    assert settings["timeout_overflow_mode"] == "throw"
+  end
+
+  test "rejects unknown classes, missing defaults, wrong types and unsafe settings" do
+    for config <- [
+          nil,
+          %{"api_paid" => %{"priority" => 10}},
+          %{"default" => %{}},
+          %{"default" => %{"priority" => 0}},
+          %{"default" => %{"priority" => 1.5}},
+          %{"default" => %{"priority" => 5}, "api_unknown" => %{}},
+          %{"default" => %{"priority" => 5}, "api_paid" => %{"max_threads" => "4"}},
+          %{"default" => %{"priority" => 5}, "api_paid" => %{"max_threads" => 1.5}},
+          %{"default" => %{"priority" => 5, "readonly" => 0}},
+          %{"default" => %{"priority" => 5, "join_use_nulls" => 1}},
+          %{"default" => %{"priority" => 5, "read_overflow_mode" => "break"}},
+          %{"default" => %{"priority" => 5, "timeout_overflow_mode" => "break"}}
+        ] do
+      assert {:error, _} = QueryClassSettings.normalize(config)
+    end
+  end
+
+  test "serializes fractional limits and fixed enums safely" do
+    assert {:ok, settings} =
+             ClickHouseSettings.normalize(%{
+               "max_execution_time" => 0.5,
+               "max_rows_to_read" => 100
+             })
+
+    assert settings["read_overflow_mode"] == "throw"
+    assert settings["timeout_overflow_mode"] == "throw"
+    assert {:ok, sql} = ClickHouseSettings.enforce("SELECT a FROM t", settings)
+    assert sql =~ "max_execution_time = 0.5"
+    assert sql =~ "read_overflow_mode = 'throw'"
+    assert sql =~ "timeout_overflow_mode = 'throw'"
+
+    assert {:error, _} =
+             ClickHouseSettings.normalize(%{"read_overflow_mode" => "throw'; SELECT 1"})
+  end
+
+  test "rejects class overrides inside SELECT and EXPLAIN nested queries" do
+    assert {:ok, policy} = QueryClassSettings.resolve(@classes, "api_paid", %{})
+
+    for sql <- [
+          "SELECT a FROM t SETTINGS priority = 0",
+          "WITH t AS (SELECT a FROM s SETTINGS max_threads = 32) SELECT a FROM t",
+          "EXPLAIN SELECT a FROM t SETTINGS priority = 1"
+        ] do
+      assert {:error, reason} = ClickHouseSettings.enforce(sql, policy)
+      assert reason =~ "is enforced"
+    end
+  end
+end
diff --git a/test/logflare/backends/query_class_settings_test.exs b/test/logflare/backends/query_class_settings_test.exs
new file mode 100644
index 0000000000..3559c41371
--- /dev/null
+++ b/test/logflare/backends/query_class_settings_test.exs
@@ -0,0 +1,100 @@
+defmodule Logflare.Backends.QueryClassSettingsTest do
+  use Logflare.DataCase
+
+  alias Logflare.Backends
+  alias Logflare.Backends.Backend
+  alias Logflare.Repo
+
+  @policy %{
+    "default" => %{"priority" => 5},
+    "api_paid" => %{"priority" => 10, "max_threads" => 4}
+  }
+
+  setup do
+    insert(:plan)
+    owner = insert(:user)
+    admin = insert(:user, admin: true)
+
+    backend =
+      insert(:backend,
+        user: owner,
+        type: :clickhouse,
+        config: %{url: "http://localhost:8123", port: 8123, database: "default"}
+      )
+      |> then(&Backends.get_backend(&1.id))
+
+    %{owner: owner, admin: admin, backend: backend}
+  end
+
+  test "operator policy is validated, persisted and protected from customer changes", context do
+    %{owner: owner, admin: admin, backend: backend} = context
+
+    assert {:error, :forbidden} = Backends.configure_query_class_settings(owner, backend, @policy)
+
+    assert {:error, :forbidden} =
+             Backends.configure_query_class_settings(%{owner | admin: true}, backend, @policy)
+
+    assert {:error, changeset} =
+             Backends.update_backend(backend, %{config: %{query_class_settings: @policy}})
+
+    assert Keyword.has_key?(changeset.errors, :"config.query_class_settings")
+
+    assert {:ok, configured} = Backends.configure_query_class_settings(admin, backend, @policy)
+    assert configured.config.query_class_settings == @policy
+    assert Backends.get_backend(backend.id).config.query_class_settings == @policy
+
+    refute Map.has_key?(
+             Jason.decode!(Jason.encode!(configured))["config"],
+             "query_class_settings"
+           )
+
+    for policy <- [%{}, nil, %{"default" => %{"priority" => 1}}] do
+      assert {:error, _} =
+               Backends.update_backend(configured, %{
+                 "config" => %{"query_class_settings" => policy}
+               })
+    end
+
+    changeset = Backend.changeset(configured, %{type: :postgres})
+    assert Keyword.has_key?(changeset.errors, :"config.query_class_settings")
+
+    assert {:ok, updated} = Backends.update_backend(backend, %{config: %{read_pool_size: 4}})
+    assert Map.get(updated.config, :query_class_settings) == @policy
+    assert {:ok, _} = Backends.update_backend(updated, %{name: "renamed"})
+
+    assert {:error, _} =
+             Backends.configure_query_class_settings(admin, backend, %{
+               "default" => %{"priority" => 0}
+             })
+
+    assert Backends.get_backend(backend.id).config.query_class_settings == @policy
+    assert {:ok, cleared} = Backends.configure_query_class_settings(admin, backend, %{})
+    assert cleared.config.query_class_settings == %{}
+  end
+
+  test "create and schema changesets cannot introduce operator policy", %{
+    owner: owner,
+    backend: backend
+  } do
+    attrs = %{
+      name: "customer",
+      type: :clickhouse,
+      config: Map.put(backend.config, :query_class_settings, @policy)
+    }
+
+    assert {:error, changeset} = Backends.create_backend(owner, attrs)
+    assert Keyword.has_key?(changeset.errors, :"config.query_class_settings")
+    refute Backend.changeset(backend, %{config: %{query_class_settings: @policy}}).valid?
+  end
+
+  test "revoked administrators and non-ClickHouse backends cannot configure policy", context do
+    %{admin: admin, backend: backend} = context
+    other = insert(:backend, user: context.owner, type: :bigquery)
+
+    assert {:error, :not_clickhouse_backend} =
+             Backends.configure_query_class_settings(admin, other, @policy)
+
+    admin |> Ecto.Changeset.change(admin: false) |> Repo.update!()
+    assert {:error, :forbidden} = Backends.configure_query_class_settings(admin, backend, @policy)
+  end
+end
diff --git a/test/logflare/endpoints/cache_test.exs b/test/logflare/endpoints/cache_test.exs
index 2b6af31529..d3c7c0db4e 100644
--- a/test/logflare/endpoints/cache_test.exs
+++ b/test/logflare/endpoints/cache_test.exs
@@ -258,8 +258,9 @@ defmodule Logflare.Endpoints.CacheTest do
         {:cached, {:ok, snapshot}}
       )
 
-      expect(ClickHouseAdaptor, :execute_query, fn _backend, query_args, _opts ->
-        sql = elem(query_args, 0)
+      expect(ClickHouseAdaptor, :execute_query, fn backend, query_args, opts ->
+        assert opts[:enforced_clickhouse_settings]["max_execution_time"] == 5
+        assert {:ok, sql} = ClickHouseAdaptor.prepare_query(backend, elem(query_args, 0), opts)
         assert sql =~ "'historical'"
         assert sql =~ "max_execution_time = 5"
         {:ok, QueryResult.new([%{"testing" => "historical"}], %{total_bytes_processed: 0})}
diff --git a/test/logflare/endpoints/clickhouse_settings_test.exs b/test/logflare/endpoints/clickhouse_settings_test.exs
index 2dfd8dfed6..848b7add5e 100644
--- a/test/logflare/endpoints/clickhouse_settings_test.exs
+++ b/test/logflare/endpoints/clickhouse_settings_test.exs
@@ -72,7 +72,7 @@ defmodule Logflare.Endpoints.ClickHouseSettingsTest do
           "EXPLAIN SELECT a FROM t SETTINGS max_execution_time = 10",
           "EXPLAIN WITH src AS (SELECT a FROM t SETTINGS max_execution_time = 10) SELECT a FROM src"
         ] do
-      assert {:error, "ClickHouse setting max_execution_time is enforced by this endpoint"} =
+      assert {:error, "ClickHouse setting max_execution_time is enforced by query policy"} =
                ClickHouseSettings.enforce(query, settings)
     end
   end
diff --git a/test/logflare_web/controllers/endpoints_controller_test.exs b/test/logflare_web/controllers/endpoints_controller_test.exs
index 01a72000e1..b3094c7f42 100644
--- a/test/logflare_web/controllers/endpoints_controller_test.exs
+++ b/test/logflare_web/controllers/endpoints_controller_test.exs
@@ -437,8 +437,9 @@ defmodule LogflareWeb.EndpointsControllerTest do
 
         Logflare.ContextCache.bust_keys([{Endpoints, endpoint.id}])
 
-        expect(ClickHouseAdaptor, :execute_query, fn _backend, query_args, _opts ->
-          sql = elem(query_args, 0)
+        expect(ClickHouseAdaptor, :execute_query, fn backend, query_args, opts ->
+          assert opts[:enforced_clickhouse_settings]["max_execution_time"] == 5
+          assert {:ok, sql} = ClickHouseAdaptor.prepare_query(backend, elem(query_args, 0), opts)
           assert sql =~ "'historical'"
           assert sql =~ "max_execution_time = 5"
 
diff --git a/test/logflare_web/live_views/endpoints_live_test.exs b/test/logflare_web/live_views/endpoints_live_test.exs
index 2bd31e8a8d..44940a050d 100644
--- a/test/logflare_web/live_views/endpoints_live_test.exs
+++ b/test/logflare_web/live_views/endpoints_live_test.exs
@@ -114,7 +114,8 @@ defmodule LogflareWeb.EndpointsLiveTest do
       assert Logflare.Endpoints.get_endpoint_query(endpoint.id).enforced_clickhouse_settings == %{
                "max_rows_to_read" => 100,
                "max_execution_time" => 5,
-               "read_overflow_mode" => "throw"
+               "read_overflow_mode" => "throw",
+               "timeout_overflow_mode" => "throw"
              }
 
       html =