Compact a quiet table's fresh files every minute, between sweeps (T-627)

config/runtime.exs

@@ -578,6 +578,20 @@ if ms = System.get_env("SMOLQUERY_COMPACT_SPAN_BUDGET_MS") do
       Smolquery.RuntimeConfig.non_negative_integer!("SMOLQUERY_COMPACT_SPAN_BUDGET_MS", ms)
 end
 
+# T-627: how often the hour level runs again for a quiet table between
+# sweeps; 0 turns it off.
+if ms = System.get_env("SMOLQUERY_COMPACT_FRESH_INTERVAL_MS") do
+  config :smolquery, Smolquery.StorageService,
+    compact_fresh_interval_ms:
+      Smolquery.RuntimeConfig.non_negative_integer!("SMOLQUERY_COMPACT_FRESH_INTERVAL_MS", ms)
+end
+
+if bytes = System.get_env("SMOLQUERY_COMPACT_FRESH_BELOW_BYTES") do
+  config :smolquery, Smolquery.StorageService,
+    compact_fresh_below_bytes:
+      Smolquery.RuntimeConfig.positive_integer!("SMOLQUERY_COMPACT_FRESH_BELOW_BYTES", bytes)
+end
+
 if ms = System.get_env("SMOLQUERY_COMPACT_SPAN_MS") do
   config :smolquery, Smolquery.StorageService,
     compact_span_ms: Smolquery.RuntimeConfig.positive_integer!("SMOLQUERY_COMPACT_SPAN_MS", ms)

config/test.exs

@@ -42,7 +42,7 @@ config :smolquery, Smolquery.BufferService,
   encode_concurrency: 1,
   commit_siblings: 0
 
-config :smolquery, Smolquery.StorageService, engine_extensions: []
+config :smolquery, Smolquery.StorageService, engine_extensions: [], compact_fresh_interval_ms: 0
 
 config :smolquery, Smolquery.QueryService, engine_extensions: []
 

docs/configuration.md

@@ -92,6 +92,8 @@ received value, and the accepted shape.
 | `SMOLQUERY_COMPACT_SPILL_FLOOR_BYTES` | A sweep runs the hour level only while the spill filesystem (`SMOLQUERY_SPILL_DIR`) has less free space than this (T-601). The same happens while a recycled compaction engine's abandoned merge still holds a spill directory. Default `2147483648` (2 GiB); `0` turns the free-space check off. |
 | `SMOLQUERY_COMPACT_SPILL_SHARE` | Each compaction engine instance sets its `max_temp_directory_size` to 1/N of the spill filesystem's free space when it starts, at most `SMOLQUERY_MAX_TEMP_DIRECTORY_SIZE` (T-601). Default `4`. A rebuilt engine then takes a share of what an abandoned merge left, not another fixed cap. |
 | `SMOLQUERY_COMPACT_SPAN_MS` | The widest time span one compacted file covers (`86400000`, a day, T-592). Retention drops a file only once all of it has aged out, so this is how far past its TTL a row can live. Span-level work is owned per table and span, like the hour level's buckets. Must be larger than `SMOLQUERY_COMPACT_BUCKET_MS` |
+| `SMOLQUERY_COMPACT_FRESH_INTERVAL_MS` | How often the compactor runs the hour level again, between sweeps, for a quiet table (`60000`, T-627). A quiet table is one whose files in the current or previous hour include one under `SMOLQUERY_COMPACT_FRESH_BELOW_BYTES`: it seals on age, a file per seal, and without this a dozen tiny files pile up in the current hour between sweeps. Only those tables are listed and planned on the tick. `0` turns it off |
+| `SMOLQUERY_COMPACT_FRESH_BELOW_BYTES` | The file size under which a recent file makes its table quiet for the fresh tick (`4194304`, 4 MiB) |
 | `SMOLQUERY_S3_BUCKET` | Puts the sealed tier on an S3-compatible store. It points the `store:` of both the storage service and the query service at `Segments.Store.S3` |
 | `SMOLQUERY_S3_ACCESS_KEY_ID` / `SMOLQUERY_S3_SECRET_ACCESS_KEY` | Static S3 credentials. Set both, or neither. With both unset, the store uses the AWS (Amazon Web Services) [credential chain](#s3-credentials) instead. Startup rejects one without the other |
 | `SMOLQUERY_S3_ENDPOINT` | The S3-compatible endpoint. Unset targets AWS S3 |

docs/deployment.md

@@ -127,6 +127,15 @@ answers nothing at the TCP level.
 
 One note per release, newest first.
 
+### 0.22.0: quiet tables are compacted every minute (T-627)
+
+A table that seals on age, not size, leaves one sealed file per seal. On the sandbox `logs.cluster` (about 110 rows a minute, three write partitions) sealed two or three files a minute of 1-30 rows each. Compaction swept every 5 minutes, so a dozen of them sat in the current hour, and every query over recent data opened each one over S3: a 60-minute histogram ending now cost about 5x one ending 15 minutes ago, on the same row count.
+
+- **What changed:** between sweeps, every `SMOLQUERY_COMPACT_FRESH_INTERVAL_MS` (60 s), each storage node runs the hour level again for the quiet tables its last sweep found. A quiet table has a file in the current or previous hour under `SMOLQUERY_COMPACT_FRESH_BELOW_BYTES` (4 MiB). A busy table, whose seals fill by size, is not listed on the tick.
+- **Cost:** one catalog listing per quiet table per minute, and up to one merge and swap each. The tick merges only files under `SMOLQUERY_COMPACT_FRESH_BELOW_BYTES`, so it rewrites at most 4 MiB a table a minute; a file past that waits for the 5-minute sweep, as before.
+- **Watch:** each swap can conflict with a seal of the same table (T-573). `smolquery_compaction_conflicts_total` counts them; `smolquery_compactions_total{result}` and `smolquery_compaction_lane_microseconds_total{lane="fresh"}` show the tick's work.
+- **Off:** set `SMOLQUERY_COMPACT_FRESH_INTERVAL_MS=0`.
+
 ### 0.22.0: an optional shared read cache for query engines (T-626)
 
 Each query job runs on its own DuckDB engine, which stops when the job ends, and DuckDB's caches stop with it. So every job read again from S3 the files and Parquet footers the previous job had just read. A dashboard's second query was as cold as its first. On the sandbox, a HyperDX page spent 1,981 ms in DuckDB with a fresh engine per query and 548 ms with the caches kept.

lib/smolquery/storage_service/runtime.ex

@@ -173,6 +173,16 @@ defmodule Smolquery.StorageService.Runtime do
   `compact_target_bytes` of `nil` does, while keeping the settled files out
   of the hour level.
 
+  `compact_fresh_interval_ms` (60 s, T-627) runs the hour level again between
+  sweeps for a quiet table: one whose newest files, in the current or the
+  previous `compact_bucket_ms`, include one under `compact_fresh_below_bytes`
+  (4 MiB). A table that seals on age rather than size leaves a file per seal,
+  two or three a minute, and a sweep every `compact_interval_ms` let a dozen
+  pile up in the current hour, each one a file every recent query opens. The
+  fresh tick merges them a minute later instead of five. `0` turns it off.
+  It lists only the quiet tables the last sweep found, so a busy table pays
+  nothing; see `Smolquery.StorageService.Scheduler`.
+
   Compaction runs on its own engine, `compact_engine/1` (T-259).
   `compact_engine_memory_limit` sizes it the way `engine_memory_limit` sizes
   the merge engine, one rung down: explicit knob, else a quarter of the
@@ -322,6 +332,8 @@ defmodule Smolquery.StorageService.Runtime do
     compact_target_bytes: 1_073_741_824,
     compact_span_ms: 86_400_000,
     compact_span_budget_ms: 120_000,
+    compact_fresh_interval_ms: 60_000,
+    compact_fresh_below_bytes: 4_194_304,
     compact_engine_memory_limit: nil,
     compact_engine_mib_per_thread: 256,
     compact_span_decoded_bytes: 8_589_934_592,
@@ -369,6 +381,8 @@ defmodule Smolquery.StorageService.Runtime do
           compact_target_bytes: pos_integer() | nil,
           compact_span_ms: pos_integer(),
           compact_span_budget_ms: non_neg_integer(),
+          compact_fresh_interval_ms: non_neg_integer(),
+          compact_fresh_below_bytes: pos_integer(),
           compact_engine_memory_limit: String.t() | nil,
           compact_engine_mib_per_thread: pos_integer(),
           compact_span_decoded_bytes: pos_integer(),
@@ -411,6 +425,8 @@ defmodule Smolquery.StorageService.Runtime do
     :compact_target_bytes,
     :compact_span_ms,
     :compact_span_budget_ms,
+    :compact_fresh_interval_ms,
+    :compact_fresh_below_bytes,
     :compact_engine_memory_limit,
     :compact_engine_mib_per_thread,
     :compact_span_decoded_bytes,

lib/smolquery/storage_service/scheduler.ex

@@ -107,6 +107,36 @@ defmodule Smolquery.StorageService.Scheduler do
   (T-601), the span lane does not run at all, so its cooldowns keep their
   count instead of being cleared by plans that could not happen.
 
+  ## A fresh tick between sweeps for quiet tables (T-627)
+
+  A table that seals on age, not size, leaves one file per seal: about
+  110 rows a minute over three write partitions was two or three files a
+  minute of 1-30 rows each, and a sweep every `compact_interval_ms` let a
+  dozen pile up in the current hour. Every query over recent data opened
+  each one over S3: a 60-minute histogram ending now cost about 5x one
+  ending 15 minutes ago, on the same row count.
+
+  So every `compact_fresh_interval_ms` (60 s) this process runs the hour
+  lane again for the tables the last sweep found quiet
+  (`Smolquery.StorageService.Scheduler.Planner.fresh_tables/3`): a recent
+  file under `compact_fresh_below_bytes`. It lists and plans only those, so
+  a busy table costs nothing extra, and it runs in this process, so it can
+  never race a sweep over the same files. Its outcomes feed the row caps,
+  the quarantine and the backoff as a sweep's do; a table whose listing is
+  no longer quiet leaves the set until the next sweep adds it back.
+
+  The tick merges only files under `compact_fresh_below_bytes`, not
+  `compact_below_bytes`: a quiet table's current-hour file grows with every
+  merge, and without the lower floor the tick would rewrite it every minute
+  up to 32 MiB. Past 4 MiB it waits for the sweep, as before. The tick
+  plans with the un-gated runtime, so it takes recent files only and never
+  reports the span level paused; with no quiet table it does nothing.
+
+  What a node learns advances per tick as well as per sweep: a row cap's
+  patience and the quarantine's threshold count ticks too, and a lost
+  commit parks the table for `compact_interval_ms`, skipping its ticks
+  until then.
+
   ## The parts
 
     * `Smolquery.StorageService.Scheduler.Planner`: which files of a table to
@@ -143,7 +173,8 @@ defmodule Smolquery.StorageService.Scheduler do
     quarantined_groups: MapSet.new(),
     cooldowns: %{},
     span_cooldowns: %{},
-    span_widths: %{}
+    span_widths: %{},
+    fresh: MapSet.new()
   ]
 
   use Smolquery.StorageService.Sweeper, interval: :compact_interval_ms
@@ -277,11 +308,75 @@ defmodule Smolquery.StorageService.Scheduler do
            quarantined_groups: quarantined_groups,
            cooldowns: cooldowns,
            span_cooldowns: span_cooldowns,
-           span_widths: span_widths
+           span_widths: span_widths,
+           fresh: Planner.fresh_tables(listings, runtime)
        }}
     end
   end
 
+  @doc false
+  @spec on_start(%__MODULE__{}) :: %__MODULE__{}
+  def on_start(state), do: schedule_fresh(state)
+
+  @doc false
+  @spec handle_tick(term(), %__MODULE__{}) :: {:noreply, %__MODULE__{}}
+  def handle_tick(:fresh, state), do: {:noreply, state |> fresh_run() |> schedule_fresh()}
+  def handle_tick(_message, state), do: {:noreply, state}
+
+  defp schedule_fresh(%__MODULE__{runtime: %Runtime{compact_fresh_interval_ms: 0}} = state),
+    do: state
+
+  defp schedule_fresh(%__MODULE__{runtime: runtime} = state) do
+    Process.send_after(self(), :fresh, runtime.compact_fresh_interval_ms)
+
+    state
+  end
+
+  defp fresh_run(%__MODULE__{fresh: fresh} = state) do
+    case fresh |> Enum.sort() |> Enum.reject(&Backoff.cooling_down?(state.cooldowns, &1)) do
+      [] -> state
+      tables -> fresh_run(state, tables)
+    end
+  end
+
+  defp fresh_run(%__MODULE__{fresh: fresh} = state, tables) do
+    runtime = %{
+      state.runtime
+      | compact_below_bytes:
+          min(state.runtime.compact_below_bytes, state.runtime.compact_fresh_below_bytes)
+    }
+
+    {hour, listings, deferred, _stopped} =
+      timed(:fresh, fn -> hour_lane(runtime, state, tables) end)
+
+    swept = tables -- deferred
+
+    {quarantine, quarantined_groups} =
+      Quarantine.adjusted_quarantine(
+        state.quarantine,
+        state.quarantined_groups,
+        hour,
+        Quarantine.threshold()
+      )
+
+    still_fresh = Planner.fresh_tables(listings, runtime)
+
+    gone =
+      for table_ref <- Map.keys(listings),
+          not MapSet.member?(still_fresh, table_ref),
+          do: table_ref
+
+    %{
+      state
+      | row_caps: Caps.adjusted_row_caps(state.row_caps, hour, runtime.compact_max_rows),
+        quarantine: quarantine,
+        quarantined_groups: quarantined_groups,
+        cooldowns:
+          Backoff.adjusted_cooldowns(state.cooldowns, swept, hour, runtime, state.row_caps),
+        fresh: MapSet.difference(fresh, MapSet.new(gone))
+    }
+  end
+
   defp timed(lane, run) do
     started = System.monotonic_time(:microsecond)
     result = run.()
@@ -326,7 +421,7 @@ defmodule Smolquery.StorageService.Scheduler do
     if stops_sweep?(outcome) do
       Logger.warning(fn ->
         "compaction sweep stopped after a call exited on #{inspect(table_ref)}: " <>
-          "#{length(rest)} table(s) deferred to the next sweep"
+          "#{length(rest)} table(s) deferred to the next sweep or fresh tick"
       end)
 
       {[outcome], %{}, rest, true}

lib/smolquery/storage_service/scheduler/planner.ex

@@ -232,6 +232,35 @@ defmodule Smolquery.StorageService.Scheduler.Planner do
     Enum.reject(owned, &MapSet.member?(active, &1))
   end
 
+  @doc """
+  The tables of `listings` the fresh tick re-runs the hour level for (T-627):
+  each one with a file in the current or the previous `compact_bucket_ms`
+  under `compact_fresh_below_bytes`. A table whose recent files all fill by
+  size is not one; none is when `compact_fresh_interval_ms` is `0`.
+  """
+  @spec fresh_tables(%{Catalog.table_ref() => [map()]}, Runtime.t(), integer()) ::
+          MapSet.t(Catalog.table_ref())
+  def fresh_tables(listings, runtime, now_ms \\ System.os_time(:millisecond))
+
+  def fresh_tables(_listings, %Runtime{compact_fresh_interval_ms: 0}, _now_ms), do: MapSet.new()
+
+  def fresh_tables(listings, runtime, now_ms) do
+    since = div(now_ms, runtime.compact_bucket_ms) - 1
+
+    for {table_ref, files} <- listings,
+        Enum.any?(files, &small_since?(&1, runtime, since)),
+        into: MapSet.new(),
+        do: table_ref
+  end
+
+  defp small_since?(%{path: path, bytes: bytes}, runtime, since) do
+    bytes < runtime.compact_fresh_below_bytes and
+      case bucket(path, runtime.compact_bucket_ms) do
+        :error -> false
+        hour -> hour >= since
+      end
+  end
+
   @doc """
   `tables` in the order the span lane takes them: most span candidates first
   (settled files under half of `compact_target_bytes` in the table's

lib/smolquery/storage_service/sweeper.ex

@@ -15,8 +15,28 @@ defmodule Smolquery.StorageService.Sweeper do
   before the `use`. A sweeper that carries state across sweeps returns
   `{:ok, report, state}` or `{:error, reason, state}` instead; the two-tuple
   forms keep the state unchanged.
+
+  A sweeper with a second timer defines two optional callbacks:
+  `on_start/1` takes the initial state and returns it with the timer
+  scheduled, and `handle_tick/2` takes every message other than `:sweep`,
+  answering as `handle_info/2` would. The compactor's fresh tick (T-627) is
+  the one such timer.
   """
 
+  @doc false
+  @spec started(module(), state) :: state when state: var
+  def started(module, state) do
+    if function_exported?(module, :on_start, 1), do: module.on_start(state), else: state
+  end
+
+  @doc false
+  @spec tick(module(), term(), state) :: {:noreply, state} when state: var
+  def tick(module, message, state) do
+    if function_exported?(module, :handle_tick, 2),
+      do: module.handle_tick(message, state),
+      else: {:noreply, state}
+  end
+
   @doc false
   @spec split_result(term(), state) :: {term(), state} when state: var
   def split_result({:ok, report, state}, _state), do: {{:ok, report}, state}
@@ -35,7 +55,7 @@ defmodule Smolquery.StorageService.Sweeper do
 
       @impl GenServer
       def init(%Smolquery.StorageService.Runtime{} = runtime) do
-        {:ok, schedule(%__MODULE__{runtime: runtime})}
+        {:ok, schedule(Sweeper.started(__MODULE__, %__MODULE__{runtime: runtime}))}
       end
 
       @impl GenServer
@@ -60,7 +80,7 @@ defmodule Smolquery.StorageService.Sweeper do
       end
 
       @impl GenServer
-      def handle_info(_message, state), do: {:noreply, state}
+      def handle_info(message, state), do: Sweeper.tick(__MODULE__, message, state)
 
       defp schedule(state) do
         Process.send_after(self(), :sweep, Map.fetch!(state.runtime, unquote(interval)))

lib/smolquery/telemetry.ex

@@ -97,8 +97,9 @@ defmodule Smolquery.Telemetry do
                                           — a table whose swap lost its catalog commit to a
                                           concurrent write, left out for one sweep interval;
                                           conflicts counts them in a row (T-595)
-      [:smolquery, :compact, :lane]       %{duration_us}, meta %{lane: :hour | :span}
-                                          — one per lane per compaction sweep (T-603)
+      [:smolquery, :compact, :lane]       %{duration_us}, meta %{lane: :hour | :span | :fresh}
+                                          — one per lane per compaction sweep (T-603), and one
+                                          per fresh tick (T-627)
       [:smolquery, :compact, :span_paused] %{count}, meta %{reason: :abandoned_spill |
                                           :spill_floor}
                                           — a sweep that ran its hour level only, because
@@ -397,7 +398,7 @@ defmodule Smolquery.Telemetry do
         "abandoned_spill (a recycled engine's merge still spills) or spill_floor (the spill " <>
         "filesystem is below compact_spill_floor_bytes) (T-601).",
     "smolquery_compaction_lane_microseconds_total" =>
-      "Time each compaction lane of a sweep took, by lane (hour or span), listing, sizing " <>
+      "Time each compaction lane took, by lane (hour or span per sweep, fresh per fresh tick), listing, sizing " <>
         "and planning included; rate over wall time is each lane's share of a node (T-603).",
     "smolquery_compaction_conflicts_total" =>
       "Compactions of a table that lost the catalog commit to a concurrent write, after the " <>
@@ -1270,7 +1271,7 @@ defmodule Smolquery.Telemetry do
 
   defp catalog_kind(_meta), do: :unknown
 
-  defp compact_level(%{level: level}) when level in [:hour, :span], do: level
+  defp compact_level(%{level: level}) when level in [:hour, :span, :fresh], do: level
   defp compact_level(_meta), do: :hour
 
   defp span_paused_reason(%{reason: reason}) when reason in [:abandoned_spill, :spill_floor],

test/smolquery/storage_service/scheduler/planner_test.exs

@@ -18,6 +18,33 @@ defmodule Smolquery.StorageService.Scheduler.PlannerTest do
   defp files(count, bytes, at_ms),
     do: for(n <- 1..count//1, do: %{path: "#{Id.generate(at_ms + n)}.parquet", bytes: bytes})
 
+  describe "fresh_tables/3 (T-627)" do
+    test "is the tables with a small file in the current or the previous bucket" do
+      runtime = Runtime.new(name: __MODULE__.Fresh, compact_fresh_interval_ms: 60_000)
+      hour = runtime.compact_bucket_ms
+      quiet = {"logs", "cluster"}
+      last_hour = {"logs", "last_hour"}
+      busy = {"logs", "busy"}
+      old = {"logs", "old"}
+
+      listings = %{
+        quiet => files(3, 4_000, @now - 60_000),
+        last_hour => files(1, 4_000, @now - hour),
+        busy => files(3, 64_000_000, @now - 60_000),
+        old => files(3, 4_000, @now - 3 * hour)
+      }
+
+      assert Planner.fresh_tables(listings, runtime, @now) == MapSet.new([quiet, last_hour])
+    end
+
+    test "is none with the fresh tick off" do
+      runtime = Runtime.new(name: __MODULE__.FreshOff, compact_fresh_interval_ms: 0)
+
+      assert Planner.fresh_tables(%{{"a", "t"} => files(3, 10, @now)}, runtime, @now) ==
+               MapSet.new()
+    end
+  end
+
   describe "by_need/4 (T-603)" do
     test "puts the table with the most settled span candidates first, then orders by name" do
       runtime = Runtime.new(name: __MODULE__.Need, compact_target_bytes: 1_073_741_824)