diff --git a/README.md b/README.md index a602be5..d02d99b 100644 --- a/README.md +++ b/README.md @@ -233,6 +233,7 @@ Use `analytics.GetStatistics()` and host logging to monitor the engine. See [dep | [Product analytics][product] | [Secondary indexes][indexes] | | [Service telemetry][telemetry] | [HTTP API reference][http-api] | | [Sites and tenants][tenants] | [Extensibility][extensibility] | +| [Deployment and lifecycle][deployment] | [Storage format][storage-format] | ## Build and contribute @@ -291,6 +292,7 @@ Released under the [MIT license](https://github.com/ZoneTree/ZoneTree.Analytics/ [accuracy]: https://analytics.zonetree.dev/docs/concepts/time-and-accuracy/ [architecture]: https://analytics.zonetree.dev/docs/concepts/architecture/ [configuration]: https://analytics.zonetree.dev/docs/reference/configuration/ +[storage-format]: https://analytics.zonetree.dev/docs/reference/storage-format/ [web-reports]: https://analytics.zonetree.dev/docs/guides/web-reports/ [rollups]: https://analytics.zonetree.dev/docs/performance/rollups/ [product]: https://analytics.zonetree.dev/docs/guides/product-analytics/ diff --git a/apps/ZoneTree.Analytics.Dashboard/server/DashboardAnalyticsOptions.cs b/apps/ZoneTree.Analytics.Dashboard/server/DashboardAnalyticsOptions.cs index 11c1b56..2186517 100644 --- a/apps/ZoneTree.Analytics.Dashboard/server/DashboardAnalyticsOptions.cs +++ b/apps/ZoneTree.Analytics.Dashboard/server/DashboardAnalyticsOptions.cs @@ -34,7 +34,7 @@ public static AnalyticsOptions CreatePerformanceSamples(string directory, TimeZo int workers, ILogger logger = null) { var options = Create(directory, PerformanceSite, timeZone, logger); - options.Ingestion.WorkerCount = workers; + SetWorkers(options, workers); options.Rollups.ValueAggregates["load_time"] = ValueAggregates.Percentiles; return options; } @@ -43,7 +43,7 @@ public static AnalyticsOptions CreateBenchmark(string directory, TimeZoneInfo ti int workers, bool percentiles, ILogger logger = null) { var options = Create(directory, BenchmarkSite, timeZone, logger); - options.Ingestion.WorkerCount = workers; + SetWorkers(options, workers); options.Ingestion.QueueFullBehavior = QueueFullBehavior.Wait; if (percentiles) options.Rollups.ValueAggregates["load_time"] = ValueAggregates.Percentiles; return options; @@ -54,6 +54,14 @@ public static AnalyticsOptions CreateBenchmarkReport(string directory, string ti bool percentiles, ILogger logger = null) => CreateBenchmark(directory, TimeZoneInfo.FindSystemTimeZoneById(timeZoneId), 1, percentiles, logger); + // Generated ingestion runs as many rollup writers as workers, so the rows that every worker + // buffers are merged and written in parallel at the same scale. + static void SetWorkers(AnalyticsOptions options, int workers) + { + options.Ingestion.WorkerCount = workers; + options.Ingestion.RollupWriterCount = workers; + } + static AnalyticsOptions Create(string directory, string site, TimeZoneInfo timeZone, ILogger logger) { var options = new AnalyticsOptions(); diff --git a/docs-site/content/docs/performance/tuning/tuning.mdx b/docs-site/content/docs/performance/tuning/tuning.mdx index 24a7c1a..edf9b47 100644 --- a/docs-site/content/docs/performance/tuning/tuning.mdx +++ b/docs-site/content/docs/performance/tuning/tuning.mdx @@ -20,6 +20,7 @@ Measure a consistent Release build. Include representative filtered and unfilter | Setting | Tradeoff | | --- | --- | | WorkerCount | More processing concurrency, with larger combined per-worker budgets | +| RollupWriterCount | More parallel rollup writes, with one more thread each; rows of every worker are merged before they are written | | QueueCapacity | More burst capacity, with more queued memory | | MaxBufferedRollupRows | More aggregation state held per worker before writing | | Sessions.MaxCachedSessions | More active-session state held per worker | diff --git a/docs-site/content/docs/reference/configuration/configuration.mdx b/docs-site/content/docs/reference/configuration/configuration.mdx index 526bd1f..d3128fe 100644 --- a/docs-site/content/docs/reference/configuration/configuration.mdx +++ b/docs-site/content/docs/reference/configuration/configuration.mdx @@ -102,6 +102,7 @@ The dashboard extends raw retention and enables selected aggregates and indexes. | --- | --- | --- | | `QueueCapacity` | `int` | Maximum number of queued events per worker. Default is 100,000. | | `WorkerCount` | `int` | Number of background ingestion workers. Events are routed to workers by visitor, so each visitor's events are processed in order. Default is 1, which handles well over a hundred thousand events per second. | +| `RollupWriterCount` | `int` | Number of dedicated rollup writer threads, 1 to 64. Each writer owns a share of the rollup rows: it merges the rows that all workers buffered for them and writes each row once per flush, instead of once per worker. Default is 1, which suits single-processor machines; with several workers on several cores, more writers write rows in parallel. | | `FlushInterval` | `TimeSpan` | How often buffered aggregates and session states are written to storage. Rollup-based queries lag behind ingestion by at most this interval. Default is 2 seconds. Must not exceed approximately 49.7 days, the timer's maximum period. | | `MaxBufferedRollupRows` | `int` | Forces a flush when a worker buffers more distinct rollup rows than this. Default is 100,000. | | `QueueFullBehavior` | `QueueFullBehavior` | Behavior of `AnalyticsEngine.Track` when the queue is full. Default is `QueueFullBehavior.Drop`, so request threads never block. | diff --git a/docs-site/content/docs/reference/engine/engine.mdx b/docs-site/content/docs/reference/engine/engine.mdx index 4e9ad80..b5e4506 100644 --- a/docs-site/content/docs/reference/engine/engine.mdx +++ b/docs-site/content/docs/reference/engine/engine.mdx @@ -80,7 +80,7 @@ Partition record counts are approximate and may be null for partitions not opene ## Errors and cancellation -`AnalyticsException` is the base for engine-specific errors. `AnalyticsConfigurationException` reports incompatible options or dataset configuration. `AnalyticsQueryException` reports unanswerable or bounded queries; `AnalyticsQueryTimeoutException` is its timeout subclass. +`AnalyticsException` is the base for engine-specific errors. `AnalyticsConfigurationException` reports incompatible options or dataset configuration. `AnalyticsQueryException` reports unanswerable or bounded queries; `AnalyticsQueryTimeoutException` is its timeout subclass. `AnalyticsStorageException` reports a flush whose rollup rows could not be written; its inner exception is the storage failure, and the rows are retried by later flushes. Invalid arguments use standard argument exceptions. Caller cancellation uses the cancellation token; configured query timeout additionally protects expensive report execution. @@ -132,6 +132,8 @@ than ZoneTree and is not exposed. For shared indexes, restrict iterators using the returned logical `KeyPrefix`. Each accessor exposes a logical namespace within the shared physical store. +See the [storage format reference](../storage-format/storage-format.mdx) for directory layouts, exact key and payload encodings, metadata generations, and recovery behavior. + ## Related pages - [First events and reports](../../getting-started/first-events/first-events.mdx) diff --git a/docs-site/content/docs/reference/index.mdx b/docs-site/content/docs/reference/index.mdx index e3b4375..16fcc3e 100644 --- a/docs-site/content/docs/reference/index.mdx +++ b/docs-site/content/docs/reference/index.mdx @@ -11,7 +11,7 @@ Look up exact API contracts and defaults. These pages complement the worked exam ## Find the contract -Use the [engine API](./engine/engine.mdx) for entry points, [query model](./query-model/query-model.mdx) for calculation semantics, [configuration](./configuration/configuration.mdx) for settings, and [HTTP API](./http-api/http-api.mdx) for integration. +Use the [engine API](./engine/engine.mdx) for entry points, [query model](./query-model/query-model.mdx) for calculation semantics, [configuration](./configuration/configuration.mdx) for settings, [HTTP API](./http-api/http-api.mdx) for integration, and [storage format](./storage-format/storage-format.mdx) for persisted layouts and recovery contracts. ## In this section @@ -20,6 +20,7 @@ Use the [engine API](./engine/engine.mdx) for entry points, [query model](./quer + ## Continue exploring diff --git a/docs-site/content/docs/reference/storage-format/storage-format.mdx b/docs-site/content/docs/reference/storage-format/storage-format.mdx new file mode 100644 index 0000000..d36dd0c --- /dev/null +++ b/docs-site/content/docs/reference/storage-format/storage-format.mdx @@ -0,0 +1,488 @@ +--- +title: "Storage format" +description: "Directory layout, binary keys and values, metadata, partition routing, and recovery contracts." +order: 5 +template: doc +--- + +# Storage format + +ZoneTree.Analytics stores a dataset as several ordered ZoneTree stores plus small, checksummed metadata files. This reference describes the format implemented by the current source: dataset `format-version` **2**, dimension-index format **2**, and rollup-index format **2**. + +Use this page to inspect a dataset, understand storage costs, or review a codec change. For application code, prefer the public [storage accessors and codecs](../engine/engine.mdx#advanced-storage-access). Those APIs return typed records without requiring callers to reproduce byte offsets. + +## Format boundaries and compatibility + +There are two layers to the persisted format: + +| Layer | Defined by | Includes | +| --- | --- | --- | +| Analytics records | ZoneTree.Analytics | Directory routing, keys, event payloads, rollup and session serializers, metadata, index namespaces, and recovery checkpoints | +| Storage engine files | ZoneTree | Tree metadata, segments, write-ahead logs, compression, block layout, and merge machinery | + +The byte layouts below describe the serialized keys and values passed to ZoneTree. They are not offsets into segment or WAL files. Segment mode and compression can be configured through `StoreOptions`; opening trees through the matching ZoneTree version handles that layer. The current core project references ZoneTree **2.0.1**. + +On dataset creation, the engine persists these settings and requires exact matches on every reopen: + +| Metadata key | Stored value | Why it must match | +| --- | --- | --- | +| `format-version` | `"2"` | Selects the dataset format understood by the engine; it increases with every incompatible layout change | +| `time-zone` | `AnalyticsOptions.TimeZone.Id` | Daily rollup keys encode local calendar dates | +| `hll-precision` | Decimal `Rollups.HyperLogLogPrecision` | Sketches merged together must use compatible precision | +| `wal-directory` | Absolute configured WAL root, or `""` | Unflushed records can live outside the data directory | + +A mismatch raises `AnalyticsConfigurationException`. Index formats have their own versions, persisted as `dimension-index.format` and `rollup-index.format`, and the historical joint-rollup journal carries one in its body. They follow the same rule as `format-version`: each increases with an incompatible change of its layout, and a stored version other than the current one causes startup to fail with the same older, newer, or unknown explanation. + +`RollupValueSerializer` and `SessionStateSerializer` type names are recorded in ZoneTree metadata and must remain stable. Changes to field order, primitive encodings, positional session dimensions, hashing, and key comparison also need compatibility review. + +`format-version` is an increasing number: every incompatible change to a persisted layout increases it. A release refuses a directory written in another format, and the error says whether an older or a newer release wrote it. Within a format, readers reject what they do not recognize instead of skipping it: unknown rollup header bits, unknown raw-event flags, unknown session headers, and trailing bytes raise `FormatException`. An older reader therefore never silently drops a part that a newer layout added. + +## Dataset directory + +Paths in this table are relative to `AnalyticsOptions.DataDirectory`. A partition directory contains one ZoneTree. Optional stores appear according to the engine configuration and workload; a tier can exist before it has any partitions. + +| Path | Contents | +| --- | --- | +| `engine.lock` | Exclusive handle held for the engine's lifetime | +| `meta/meta-a.json` and `meta/meta-b.json` | Alternating generations of dataset metadata | +| `events/yyyy-MM-dd/` | Raw event partitions, when `StoreRawEvents` is enabled | +| `rollups/minute/yyyy-MM-dd/` | Minute aggregate partitions, when enabled | +| `rollups/hour/yyyy-MM-dd/` | Hour aggregate partitions, when enabled | +| `rollups/day/yyyy-MM-dd/` | Day aggregate partitions, when enabled | +| `sessions/` | One unpartitioned tree of active sessions, when sessions are enabled | +| `indexes/yyyy-MM-dd/` | Shared covering dimension-index partitions | +| `rollup-indexes/minute/yyyy-MM-dd/` | Shared covering rollup-index partitions for the minute tier | +| `rollup-indexes/hour/yyyy-MM-dd/` | Shared covering rollup-index partitions for the hour tier | +| `rollup-indexes/day/yyyy-MM-dd/` | Shared covering rollup-index partitions for the day tier | +| `high-water` in each partitioned store's root | The latest time written to each partition of that store, one `name ticks` line per partition | +| `backfill/joint-rollups/journal-a.bin` and `journal-b.bin` | Alternating checkpoints of historical joint-rollup reconstruction, present while one runs | + +Sites share these trees. A site name is part of each event and rollup key; it is not a subdirectory. Logical dimension indexes share `indexes/`, and logical rollup indexes share one physical store per enabled tier. + +When `Storage.WriteAheadLogDirectory` is set, every tree's WAL directory uses the same relative path under that separate root. For example, data in `events/2026-10-15/` uses logs under `events/2026-10-15/` beneath the WAL root. Metadata files remain under the data root. + +The existence of `engine.lock` does not indicate a stale owner. Ownership is the open exclusive handle, released when the engine closes or its process exits. + +## Partition names and time routing + +Partitions are sized by approximate record count. They are not daily tables, even though their names use invariant `yyyy-MM-dd` dates. + +For ordered starts S0, S1, and S2, routing works as follows: + +| Partition | Routed time range | +| --- | --- | +| S0 | Every timestamp before S1, including timestamps before S0 | +| S1 | S1 inclusive to S2 exclusive | +| S2 | S2 inclusive and every later timestamp | + +The first record creates a partition named for its day. The map is reconstructed from date-named subdirectories at startup. There is no partition-map file; the store's `high-water` file only records the latest time written to each partition, its high-water mark. + +When the partition receiving new records reaches the target, rollover creates a new partition at the latest of these day boundaries: + +- The first boundary at or after the housekeeping clock plus `Ingestion.MaxFutureSkew` plus one day. With the default one-hour skew, a new partition is usually named two days ahead. +- One day after the receiving partition's start. +- The day after the receiving partition's high-water mark. + +Every record already written therefore keeps its routing, including records written while a clock ran ahead of the housekeeping clock. Normally the receiving partition is the latest one. When a clock that ran ahead started an empty partition in the future, records still go to an earlier partition. Rollover then inserts the new partition before the future one, provided the earlier partition's high-water mark is known and the boundary fits before the future start. + +High-water marks are saved at housekeeping, about once a minute, and at shutdown. A crash loses only marks raised since the last save; the boundary's day margin covers them unless a clock ran ahead during that minute. The target is not a hard capacity limit: checks and day-boundary routing can let a partition exceed it. Approximate counts can include deleted records that merges have not removed yet. + +Each tier routes on the timestamp it stores: raw events and dimension indexes use event UTC ticks; minute and hour rollups use UTC bucket ticks; day rollups use local calendar-day ticks. Index stores roll over independently from their source stores, so a copied record need not have the same partition name as its source. + +The core defaults target **10,000,000 records per partition** for raw events, each rollup tier, and each shared index store. Index targets count all copied rows across logical namespaces. Changing a target affects future rollover; it does not repartition existing records. Explicit historical timestamps route back into the partition covering their time. + +Queries visit overlapping partitions using half-open ranges. Partitions open on demand and close after the idle timeout, which defaults to ten minutes. A storage lease pins a partition against idle closure and defers its physical deletion, while record expiration continues to apply. + +## Binary primitives + +Layouts below list fields in serialization order. Repeated groups occur exactly as many times as their preceding count specifies. There is no implicit alignment or padding in the Analytics payloads. + +| Notation | Encoding | +| --- | --- | +| `u8` | One byte | +| `u64be` | Eight-byte unsigned integer, most significant byte first | +| `u64le` | Eight-byte unsigned integer, least significant byte first | +| `f64le` | IEEE 754 binary64 bits in eight little-endian bytes | +| `vu` | Unsigned base-128 variable-length integer: low seven bits first, high bit set when another byte follows; at most ten bytes for a 64-bit value | +| `vi` | Signed 64-bit integer mapped with zig-zag, then encoded as `vu` | +| `str` | `vu(UTF-8 byte length + 1)` followed by UTF-8 bytes; `vu(0)` means null, `vu(1)` means empty | +| `segment` | UTF-8 bytes followed by `00`; no length prefix | +| `bytes` | Exact byte sequence, with length determined by the surrounding layout | + +Zig-zag maps 0 to 0, -1 to 1, 1 to 2, and -2 to 3. For example, unsigned 300 encodes as `AC 02`, while signed -1 encodes as `01`. `str("A")` is `02 41`; a segment containing `A` is `41 00`. String lengths count UTF-8 bytes, not .NET characters. + +Time values use .NET ticks: 100 nanoseconds since `0001-01-01`. Event timestamps and minute/hour buckets are nonnegative UTC ticks. A daily bucket uses the ticks of the reporting zone's local midnight date, without a UTC offset encoded in the key. Interpret it with the persisted reporting time zone, including daylight-saving transitions; it is not a UTC midnight instant. + +Binary-key trees use `ByteArrayLexicographicComparerAscending`. Bytes compare lexicographically, and a key sorts before any longer key sharing it as a prefix. Big-endian time fields preserve chronological ordering within a fixed preceding prefix. UTF-8 segments sort by their encoded bytes, without culture-aware collation. + +### Names, missing values, and empty values + +Ingestion sanitizes names by removing control characters, trimming, and truncating to 128 .NET characters without splitting a surrogate pair. Dimension values replace control characters with spaces, trim, and use the configured length limit. Both replace each unpaired surrogate with U+FFFD, the character its UTF-8 encoding stores, so hashes and in-memory keys agree with the stored bytes. Ingestion ignores dimensions with a null value or a sanitized name that is empty or contains a comma. Public codecs operate on stored names and values and do not repeat ingestion normalization. + +Keys reserve `00` as the segment terminator. Rollup dimension tuples additionally reserve `1F` as their value separator and `01` as the missing-dimension sentinel. Public rollup encoders reject control characters in real dimension values and reject commas in dimension names, keeping those representations unambiguous. The public raw encoder and decoder also reject control characters in dimension values, so a raw event can always be aggregated into a rollup key. + +The three empty or absent cases differ: + +| Context | Missing | Explicitly empty | +| --- | --- | --- | +| Length-prefixed string | `00` | `01` | +| One dimension value in a rollup key | Raw byte `01` | Zero bytes for that value part | +| Raw event dimension list | Dimension can be absent; the binary codec also supports a null string | Present pair with an empty string | + +## Raw events + +The primary raw store uses `Memory` keys and values. It contains enriched event data; it does not persist the original `AnalyticsEvent` object or its raw IP address and user-agent string as dedicated fields. + +### Raw key + +| Order | Field | Encoding | +| --- | --- | --- | +| 1 | Site | `segment` | +| 2 | Event timestamp | `u64be`, UTC ticks | +| 3 | Sequence | `u64be`, nonnegative sequence | + +For a site whose UTF-8 length is N, the key is N + 17 bytes. Events sort by site, then timestamp, then sequence. Event name is in the value, so scanning one site's time range can include several event types. + +The engine obtains sequences with an atomic increment. At startup its counter begins at the larger of current UTC ticks and the persisted `raw-sequence`; the counter is saved every minute and at shutdown. Sequence is a tie-breaker for equal timestamps, not an event count or the event timestamp itself. + +### Raw value + +| Order | Field | Encoding | +| --- | --- | --- | +| 1 | Event name | `str` | +| 2 | Visitor hash | `u64le` | +| 3 | Session ID | `u64le`; zero means no session | +| 4 | Session start offset | `vi(event ticks - session start ticks)`; zero when there is no session | +| 5 | Flags | `u8`; bit `01` means interactive; every reader rejects other bits | +| 6 | Dimension count | `vu` | +| 7 | Each dimension | Name `str`, then value `str` | +| 8 | Measured value count | `vu` | +| 9 | Each measured value | Name `str`, then number `f64le` | + +Site, event timestamp, and sequence are supplied by the key and do not repeat in the payload. Recover the session start by subtracting the signed offset from the event timestamp. With session ID zero, the decoded session start has no semantic meaning; use `RawEvent.SessionStart`, which returns null in that case. + +Dimension and measured-value pairs preserve encoded order. There is no requirement that raw pairs have the canonical order used by rollup keys. Visitor is the engine's stored hash, not the original visitor identifier. Derived query-time dimensions such as local hour and weekday are calculated when needed. + +An empty raw value is a tombstone. A live event value has no version or presence byte; its first bytes are the event name's length-prefixed string. `RawEventCodec.Encode(null)` returns an empty value. + +## Rollups + +Each enabled granularity has its own partitioned `Memory` / `RollupValue` store. Granularity is determined by that store's path, not by a field in the key or value. + +### Rollup key + +| Order | Field | Encoding | +| --- | --- | --- | +| 1 | Site | `segment` | +| 2 | Event name | `segment` | +| 3 | Dimension-set identifier | `segment`; empty for totals | +| 4 | Bucket | `u64be` | +| 5 | Combined dimension values | Raw UTF-8, without a length prefix or final terminator | + +Dimension names are distinct and ordered with `StringComparer.Ordinal`. Their comma-joined names form the dimension-set identifier. Values follow that same order, separated by byte `1F`. A missing dimension contributes sentinel `01`; an explicitly empty dimension contributes an empty part. Totals use an empty identifier and no combined-value bytes. + +For dimensions supplied as `source=GitHub` and `country=DE`, the identifier is `country,source` and the trailing tuple is `DE`, byte `1F`, then `GitHub`. For `country=null` and `source=""`, the tuple is exactly `01 1F`. A one-dimension empty value has no trailing value bytes; its nonempty dimension-set identifier distinguishes it from totals. + +Keys order by site, event, dimension set, bucket, then combined values. All values in a series and time range therefore form one contiguous range in each overlapping partition. To scan, seek to the three-segment series prefix plus the inclusive starting bucket; stop when the prefix changes or the bucket reaches the exclusive end. + +### Rollup value header + +`RollupValueSerializer` begins with one byte: + +| Value or bit | Meaning | +| --- | --- | +| `00` | Deleted row; the only byte | +| `01` | Present bit, set for every live row | +| `02` | Visitor sketch section | +| `04` | Value detail section | +| `08` | Distinct-dimension sketch section | + +Live flags combine with bitwise OR. Readers reject other bits, a live header without the present bit, and trailing bytes. A live row containing no sketches, measured values, or optional details still has a header, event count, and zero measured-value count. Count zero does not make a row a tombstone: session and page metrics can contribute sum-only updates without adding another event. + +### Rollup value body + +| Order | Field | Encoding and condition | +| --- | --- | --- | +| 1 | Header | `u8` | +| 2 | Event/contribution count | `vi` | +| 3 | Measured value count | `vu` | +| 4 | Each scalar measured value | Scalar layout below | +| 5 | Visitor section | `vu` byte length, then the HLL layout below; only with header bit `02` | +| 6 | Detail section | `vu` byte length, then `vu` detail count equal to the measured value count, then per measured value a `vu` entry length and its detail record; length `0` means the value has no details; only with header bit `04` | +| 7 | Distinct section | `vu` byte length, then `vu` dimension count, then per dimension its name `str`, a nonzero `vu` entry length, and its HLL sketch; only with header bit `08` | + +Sections are nonempty and appear in this order, after every scalar field. The reader decodes the count and the scalar values with the row, checks each section's length and the detail count, and keeps the sections as stored bytes. When the value arrives as a slice of a larger buffer, such as a decompressed disk block, the sections are first copied into a buffer of their own, so a cached or merged row never keeps the block alive. + +Decoding happens one part at a time, the first time a query or a merge reads it: the visitor section, then a single value detail or a single distinct sketch. Reading a section's entry lengths and dimension names does not decode any sketch. A query that reads only counts and sums therefore never decodes a sketch, and a query for one distinct dimension or one value's percentiles decodes only that entry. A merge or index copy writes every section and entry it leaves unchanged back from its stored bytes, without re-encoding it. + +Measured-value names and distinct-dimension names are stored in ordinal order by the engine. Empty visitor sketches are omitted. A distinct-dimension sketch represents distinct values of that named dimension, independently of the row's visitor sketch. + +Each scalar measured value has this layout: + +| Order | Field | Encoding and condition | +| --- | --- | --- | +| 1 | Name | `str` | +| 2 | Sum | `f64le` | +| 3 | Measured count | `vi` | +| 4 | Minimum | `f64le`, when measured count is nonzero | +| 5 | Maximum | `f64le`, when measured count is nonzero | +| 6 | Mean | `f64le`, when measured count exceeds one | +| 7 | Population standard deviation | `f64le`, when measured count exceeds one | + +Measured count zero denotes a sum-only delta, such as a session metric correction. Minimum, maximum, mean, and standard deviation are omitted in that case. With one measurement, the reader reconstructs mean from minimum and standard deviation as zero. The final field is standard deviation, not variance or an M2 accumulator; `ValueStats.Variance` squares it when requested. + +### Optional value details + +When header bit `04` is set, the detail section holds an entry for every scalar value, in the same order. A value without optional data has an entry of length zero. A nonempty entry holds this record, whose `u8` flags enable first/last with bit `01` and a quantile sketch with bit `02`; at least one of them is set. + +| Order | Field | Encoding and condition | +| --- | --- | --- | +| 1 | Detail flags | `u8` | +| 2 | First value | `f64le`, with bit `01` | +| 3 | First timestamp | `vi`, absolute UTC ticks, with bit `01` | +| 4 | Last value | `f64le`, with bit `01` | +| 5 | Last timestamp offset | `vi(last ticks - first ticks)`, with bit `01` | +| 6 | Quantile sketch | Layout below, with bit `02` | + +First/last timestamps identify measurements, not bucket boundaries. Equal-timestamp ties select the smaller first value and larger last value so merge order does not change the result. + +## Sketch encodings + +In a rollup value, every sketch is length-prefixed, so a reader can skip it without decoding it: the visitor sketch fills its section, each distinct sketch and each value detail has its own entry length, and a quantile sketch lies inside its detail. Within those bounds, a sketch's internal counts determine where its fields end. + +### HyperLogLog + +| Order | Field | Encoding | +| --- | --- | --- | +| 1 | Precision P | `u8`, supported range 8 through 18 | +| 2 | Representation | `u8`: `01` sparse, `02` dense | +| 3, sparse | Entry count | `vu` | +| 4, sparse | Each sorted entry's difference from the previous entry | `vu`; previous begins at zero | +| 3, dense | Register array | Exactly 2^P bytes, one byte per register | + +Sparse entries are the top 32 bits of the internally mixed 64-bit hashes, sorted and deduplicated. Reconstruct each entry by adding its delta to the preceding entry. They are hashes, not visitor IDs. Sparse counts avoid estimation error except for collisions in those 32-bit values. The implementation switches larger sets to dense storage, with an exact-count limit of 2^(P - 2). + +Default precision 14 gives 16,384 dense registers, an embedded dense payload of 16,386 bytes including precision and representation, and approximately 0.81% standard error. A rollup with several visitor or distinct sketches can therefore be much larger than its scalar fields suggest, before ZoneTree compression. + +Dimension distinct counts begin with the stable FNV-1a hash implemented by `Text.StableHash`: offset basis 14695981039346656037, XOR each UTF-16 code unit, then multiply by 1099511628211 with 64-bit wrapping. This is a hash over UTF-16 code units, rather than UTF-8 bytes. HLL then applies its own SplitMix64 finalizer. Changing either hash changes the meaning of persisted sketches. + +### Quantile sketch + +The quantile payload stores an unsigned zero count, then a positive bucket store, then a negative bucket store. Each sign store uses: + +| Order | Field | Encoding and condition | +| --- | --- | --- | +| 1 | Contiguous bucket span length | `vu`; zero ends an empty store | +| 2 | Minimum bucket index | `vi`, only for a nonempty store | +| 3 | Count for every index in the span | `vu` per bucket, including zero counts | + +Indices use the DDSketch mapping `ceil(log(magnitude) / log(gamma))`, with `gamma = (1 + 0.01) / (1 - 0.01)`. Magnitudes strictly below `1e-9` use the zero count. Each sign store retains at most 2,048 contiguous buckets; when a wider span is needed, the smallest magnitudes collapse into the lowest retained bucket. The usual 1% relative-error bound is subject to that collapse and the near-zero bucket. + +The payload does not store an accuracy parameter, gamma, or minimum magnitude. Those constants are part of the implementation's format assumptions. + +## Active sessions + +Sessions use an unpartitioned `IZoneTree` keyed by visitor hash. The key is a typed `ulong` handled by ZoneTree's default comparer and serializer, rather than an Analytics binary key with a site prefix. Visitor hashing includes site scope. The session ID is a separate field in the value. + +`SessionStateSerializer` writes: + +| Order | Field | Encoding | +| --- | --- | --- | +| 1 | Presence | `u8`: `00` deleted, then no further bytes; `01` live | +| 2 | Session ID | `vu` | +| 3 | Start ticks | `vi`, absolute UTC ticks | +| 4 | Last-event offset | `vi(last ticks - start ticks)` | +| 5 | Expires-at ticks | `vi`, absolute wall-clock UTC ticks | +| 6 | Page views | `vi` | +| 7 | Interactions | `vi` | +| 8 | Events | `vi` | +| 9 | Entry page | `str` | +| 10 | Entry hostname | `str` | +| 11 | Last page | `str` | +| 12 | Last hostname | `str` | +| 13 | Last-page offset | `vi(last page ticks - start ticks)` | +| 14 | Standard dimension values | Sixteen `str` fields in the fixed order below; no count or names | +| 15 | Custom dimension count | `vu` | +| 16 | Each custom dimension | Name `str`, then value `str` | + +The sixteen positional names are `source`, `referrer`, `channel`, `utm_source`, `utm_medium`, `utm_campaign`, `utm_term`, `utm_content`, `country`, `region`, `city`, `browser`, `browser_version`, `os`, `os_version`, and `device`, in that order. + +Null means a session dimension was not collected; empty means collected but unknown. Positions missing from a shorter `Dimensions` array, such as the default empty one, are written as null. Only nonnull custom dimensions are written. On reopen, the reader maps custom names into the current session schema and ignores names no longer configured. This permits custom session dimensions to change between runs without changing the sixteen standard positions. The reader rejects any other presence byte and trailing bytes. + +The worker's cached site, pending session-start state, and pending metric deltas are transient and are not serialized here. A stored session is hidden when `ExpiresAtUtcTicks` is at or before current wall-clock UTC ticks; normal merges reclaim its bytes. This TTL is independent of the historical timestamps of events in a backfill. + +## Covering dimension indexes + +All configured dimension indexes share a partitioned `Memory` / `Memory` store. Each event with a nonnull value for an indexed dimension contributes a row; missing or null values contribute none, while empty strings can be indexed. + +| Order | Key field | Encoding | +| --- | --- | --- | +| 1 | Indexed dimension name | `segment` | +| 2 | Site | `segment` | +| 3 | Equality value | `str` | +| 4 | Event ticks | `u64be`, UTC | +| 5 | Original raw sequence | `u64be` | + +The value is the **complete original raw payload**, unchanged. It is not a raw-key pointer. Event name remains in that payload, and time and sequence remain in the index key's final sixteen bytes. + +The logical namespace returned by `GetDimensionIndex(dimension).KeyPrefix` is only the first segment. To read one equality range, append the site segment, encoded equality value, and starting timestamp. Stop when the equality prefix changes or the timestamp reaches the exclusive end. + +The equality value is length-prefixed, so global ordering across different values is not ordinary string ordering. The useful contiguous range is for a fixed dimension, site, and exact value. Do not pass a full index key to `StorageKeyCodec.DecodeRawEvent`: it has an additional leading namespace and value. Reconstruct the primary raw key from site and the final time/sequence fields before using the public raw key/value decoder. + +One event carrying four indexed dimensions can store four full payload copies in addition to its primary raw record. The shared partition target counts those four rows. + +## Covering rollup indexes + +Each enabled tier has one shared `Memory` / `RollupValue` index store. A logical index identifies an equality-leading dimension and a joint rollup definition. + +| Order | Key field | Encoding | +| --- | --- | --- | +| 1 | Indexed dimension | `segment` | +| 2 | Joint dimension-set identifier | `segment`, canonical comma-joined names | +| 3 | Site | `segment` | +| 4 | Event name | `segment` | +| 5 | Equality value | `str` | +| 6 | Bucket | `u64be` | +| 7 | Original primary rollup key | Complete key bytes through its combined dimension values | + +The value is the **complete current primary `RollupValue`**, serialized exactly like a normal rollup row. Index updates copy absolute replacements after primary merges. They do not apply another additive delta to the copied value. + +`GetRollupIndex(...).KeyPrefix` contains the first two segments. The remaining fields give a contiguous scan for one site, event, exact value, and bucket range. Appending the original key distinguishes rows that share the equality value and bucket but have different remaining group values. Bucket, site, and event consequently occur both in the index prefix and in the appended original key. + +A missing rollup dimension is converted to the `01` sentinel string before encoding the equality value as `str`; this differs from a null length-prefixed string. An empty dimension is an empty `str`. To use `StorageKeyCodec.DecodeRollup`, first locate the appended original key; passing the entire index key does not decode a primary rollup key. + +## Metadata file format + +Each `meta-a.json` or `meta-b.json` is a UTF-8 header line, a line-feed byte, and an indented JSON object. Despite the extension, the complete file is not a standalone JSON document. + +The header consists of three space-separated tokens: `zonetree-analytics-meta`, a positive decimal generation, and the uppercase hexadecimal SHA-256 digest of **all bytes after the first LF**. The digest covers the JSON bytes exactly as written, including whitespace. It does not cover the header. + +The JSON body maps string keys to string values, with keys emitted in ordinal order. Numbers, booleans, manifests, and base64 are represented as strings at this layer. The writer escapes some characters, for example `+` in base64 as `+`, the `1F` separator as `\u001F`, and non-ASCII text, so read the file with a JSON parser rather than by text search. + +| Key or prefix | Meaning | +| --- | --- | +| `format-version`, `time-zone`, `hll-precision`, `wal-directory` | Reopen compatibility checks | +| `raw-sequence` | Last saved sequence counter | +| `visitor-secret` | Base64 generated 32-byte identity secret; an explicitly configured secret can be supplied externally instead | +| `salt:yyyy-MM-dd` | Base64 daily 32-byte salt for anonymous identity hashing, by UTC date. Kept for the current and previous day; the salt of an older day, created for events with historical timestamps, is kept until a day after its last use | +| `salt-used:yyyy-MM-dd` | UTC ticks of the last use of an older day's salt, saved at most once an hour | +| `rollup:` | Rollup definition activation ticks; event-restricted definitions also encode their event list in the key | +| `aggregate:first-last:` and `aggregate:percentiles:` | Measured-value aggregate activation ticks | +| `distinct:` | Distinct-dimension activation ticks | +| `granularity:` | Activation ticks for an enabled time tier | +| `dimension-index.format` | Dimension-index format, currently `"2"` | +| `dimension-index.dimensions` | Ordinally sorted, newline-separated indexed names | +| `dimension-index.clean` | Whether the completed index checkpoint is reusable with durable source and destination stores | +| `dimension-index.partitions.shared` | Comma-separated shared partition names | +| `dimension-index.dirty.` | Day-tick suffixes for durable pending reconciliation intents | +| `dimension-index.backfill.` | Copy-backfill configuration, phase, cursor, and count | +| `rollup-index.format` | Rollup-index format, currently `"2"` | +| `rollup-index.config` | Sorted, newline-separated registrations; each joins dimension, definition ID, and granularity using `1F` | +| `rollup-index.clean` | Completed reusable rollup-index checkpoint | +| `rollup-index.partitions.` | Per-tier shared partition manifests | +| `rollup-index.dirty.` | Per-tier day intents, with tier and day ticks in the key suffix | +| `rollup-index.backfill.` | Rollup copy-backfill configuration, phase, cursor, and count | +| `rollup-index.backfill-pending` | Newline-separated pending historical joint-rollup definitions | + +### Generations and writes + +A metadata update creates the next generation by replacing the older file in place, then flushing the stream to disk. Generation N goes to `meta-a.json` when N is even and `meta-b.json` when N is odd. Startup validates available headers and checksums and selects the valid file with the highest generation. A missing or checksum-invalid generation can therefore leave the other one usable; existing metadata files with no valid generation cause startup to fail rather than create a new identity secret or reset format settings. + +When keys are removed, the store writes two successive generations so the removed entries disappear from both files. This applies to expired salts as well as completed recovery checkpoints. It is removal from the live metadata files, not secure erasure of historical backups or underlying storage media. + +Activation metadata records historical coverage. Definitions and aggregates present at dataset creation start at tick zero. A newly enabled definition, aggregate, or tier on an existing dataset starts at its activation time; removing and later restoring it does not claim continuous coverage of the gap. Eligible historical joint-rollup reconstruction can publish earlier coverage after its durable backfill completes. Existing row bytes alone are insufficient to infer complete coverage. + +## Writes, durability, and recovery + +An accepted event first enters the ingestion queue. Raw records, session snapshots, rollups, and index copies are then maintained through different write paths. There is no dataset-wide transaction committing all of them together. + +`StoreOptions.WriteAheadLogMode` defaults to `AsyncCompressed`. Persistent tree writes follow the configured ZoneTree WAL mode. A successful tracking call reports queue acceptance, and a crash can lose queued events and buffered aggregates. `Flush` drains prior accepted ingestion work and writes buffered aggregates and session state; it does not make every configured WAL mode synchronous. Clean disposal drains work, flushes, closes stores, and releases the directory lock. `Compact` additionally moves open trees toward merged disk storage. + +Ingestion workers do not write rollup rows themselves. On flush, a worker hands its buffered rows to dedicated rollup writers (`Ingestion.RollupWriterCount`), each of which owns a share of the rows by key. A writer merges the rows of all workers that flushed since its last write, writes each row once, and only then lets those workers persist their session states. Rollup writes serialize the updates of each row. A row's key selects one of 64 lock stripes of its partition, so writers update different rows of a partition in parallel and the same row one at a time; its covering index copies are written under the same lock, in order. The writer merges a delta with the existing value once, retains the absolute replacement for retries, and finishes a pending replacement before applying a newer delta in its stripe. This prevents retrying a failed upsert from applying the same contribution twice within that write path. It does not provide crash-atomic coordination between raw events and every aggregate tier. + +### Index reconciliation + +The primary raw store is authoritative for dimension copies; primary joint rollups are authoritative for rollup copies. Before changing a source/copy pair, the engine persists a dirty-day intent in metadata. The day is obtained by flooring that record's time coordinate to a whole day; minute/hour/raw coordinates are UTC, while day-rollup coordinates represent local calendar dates. + +Startup completes index checks and required repairs before ingestion workers begin: + +| State | Startup behavior | +| --- | --- | +| Matching configuration, format, retained partitions, and clean checkpoint | Reuse completed indexes | +| Dirty-day intents with an otherwise reusable checkpoint | Remove affected copies and reconstruct those days from authoritative records, including interrupted deletions | +| Configuration change, incomplete checkpoint, or missing retained index partitions | Rebuild the affected shared index stores | +| Source or destination WAL disabled | Rebuild instead of trusting a durable completed checkpoint | +| Explicit incompatible index format | Fail the format check | + +Copy backfills use metadata keys ending in `.config`, `.phase`, `.partition`, `.key`, and `.count`. Phase is `clear` or `copy`; the cursor key is base64, and the count is decimal text. Checkpoints advance after destination data is persisted, every 100,000 input records and at partition boundaries. Absolute upserts make replay of an unfinished chunk idempotent. + +Historical reconstruction of joint rollups has a separate binary journal, `backfill/joint-rollups/journal-a.bin` and `journal-b.bin`, written alternately like the metadata files: + +| Order | Field | Encoding | +| --- | --- | --- | +| 1 | Magic | The 8 ASCII bytes `ZTAJRNL1` | +| 2 | Generation | `u64le`, positive | +| 3 | Body length | `u64le` | +| 4 | Body checksum | 32-byte SHA-256 of the body | +| 5 | Body | The checkpoint | + +The header is written last, so a torn write never validates; startup reads the valid file with the highest generation. The body holds the configuration signature, coverage, site/partition/key cursor, progress counters, active reconstructed sessions, and a redo list of absolute rollup replacements. Redo records include granularity, partition, primary key bytes, and serialized rollup value bytes, stored as bytes rather than text. Replaying replacements avoids adding the same contributions twice after a crash; completed coverage is published only after the durable reconstruction checkpoint, and the journal files are deleted when reconstruction completes. + +Recovery of derived indexes cannot recover raw events lost before persistence or history already removed by retention. Ordinary startup index reconciliation also does not imply that all ordinary raw history is replayed into every primary rollup after a crash. + +## Deletion, expiration, and storage reclamation + +| Store | Explicit deletion representation | Expiration rule | +| --- | --- | --- | +| Raw events and dimension indexes | Empty `Memory` value | Event ticks older than the raw retention cutoff | +| Rollups and rollup indexes | Null `RollupValue`, serialized as byte `00` | Bucket older than its tier's cutoff; day cutoffs use local-day ticks | +| Active sessions | Null `SessionState`, serialized as byte `00` | Expires-at ticks at or before wall-clock UTC ticks | + +The store deletion predicates hide expired records as retention cutoffs advance. A lease does not restore expired records. Logical absence can therefore precede physical removal of the serialized bytes. Iterators configured to include deleted records are a lower-level diagnostic view. + +Maintenance drops fully expired partitions and merges partially expired trees. Whole segment ranges can be discarded only when the key range proves that every record has expired; series-leading and equality-leading layouts require that check within the appropriate prefix. Bottom-segment reclamation follows ZoneTree's merge schedule. A directory size is consequently not an exact measure of currently queryable data. + +Defaults retain raw history for 90 days, minute rollups for two days when enabled, hourly rollups for 400 days, and daily rollups indefinitely. The default enabled tiers are Hour and Day. Null retention means keep indefinitely. Retention is rounded in whole UTC days, or whole hours for durations under one day; daily aggregate comparisons use the corresponding reporting-zone date. + +Visitor erasure removes retained raw events, their dimension-index copies, and the active session for an explicit visitor. It does not subtract contributions from existing aggregate rollups or their covering indexes. See [retention and erasure](../../operations/retention-and-erasure/retention-and-erasure.mdx) for the public operation. + +## Inspecting and transferring a dataset + +Use `analytics.Storage.RawEvents`, `GetRollups`, `GetDimensionIndex`, and `GetRollupIndex` to list partitions and acquire disposable leases. `Sessions` exposes its typed tree directly. Metadata files are not exposed as a ZoneTree. Accessors return the engine's actual trees and existing maintainers, and remain valid only while the engine is open. + +For raw records, decode the iterator's complete primary key/value pair with `RawEventCodec.Decode`. For rollups, use `StorageKeyCodec.DecodeRollup` and interpret the bucket with `GetBucketStart(granularity, analytics.TimeZone)`. Public key encoders restore canonical layouts. The public raw decoder validates UTF-8, lengths, names, control characters in values, flags, timestamp ranges, and trailing bytes and raises `FormatException` for malformed records. The rollup serializer checks the header, section framing, and trailing bytes when a row is read, and a section's contents when it is decoded; malformed bytes raise `FormatException`. Neither it nor the session serializer validates every field, so do not treat them as general untrusted-file validators. + +Index accessors expose a logical namespace inside a shared tree. Restrict iterators to `KeyPrefix`, seek into that prefix, and stop as soon as it no longer matches. Their keys use the index layouts above, not the public primary-key decoder layouts. + +Treat stored instances, arrays, and sketches as immutable. Direct tree writes bypass ingestion, aggregation, index coordination, session caches, and coverage metadata. Do not dispose or drop engine-owned trees or maintainers, or create an additional maintainer. Release leases before engine disposal. + +For a coordinated file copy, close the engine cleanly and copy the entire data root plus the separate WAL root, if configured. Preserve metadata, sessions, all tiers, and any active recovery journal together. If the identity secret is supplied in configuration, retain that configuration securely as well. Reopening requires the persisted format settings, including the same configured absolute WAL root; relocation of that root is not supported by simply changing the option. Test a restored copy in isolation before depending on it. [Deployment and lifecycle](../../operations/deployment/deployment.mdx) covers operational ownership and backups. + +## Implementation map + +These source files define the contracts described here: + +| Contract | Source | +| --- | --- | +| Every format version and its check | [StorageFormats.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/StorageFormats.cs) | +| Dataset settings checks and directory ownership | [AnalyticsEngine.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/AnalyticsEngine.cs) | +| Store types, comparers, serializers, deletion predicates, and WAL routing | [StoreFactory.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/StoreFactory.cs) | +| Partition discovery, time routing, and high-water marks | [PartitionedStore.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/PartitionedStore.cs) | +| Visitor hashing and salt lifetime | [VisitorHasher.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Privacy/VisitorHasher.cs) | +| Primitive binary encoding | [ByteWriter.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Internal/ByteWriter.cs) | +| Primary key layouts and public key codecs | [KeyCodec.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/KeyCodec.cs), [StorageKeyCodec.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/StorageKeyCodec.cs) | +| Raw payloads | [RawEventCodec.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/RawEventCodec.cs) | +| Aggregate payloads and optional value details | [RollupValue.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/RollupValue.cs) | +| Session payloads | [SessionState.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/SessionState.cs), [StandardNames.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/StandardNames.cs) | +| Embedded sketches | [HyperLogLog.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Sketches/HyperLogLog.cs), [QuantileSketch.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Sketches/QuantileSketch.cs) | +| Index key layouts | [DimensionIndexKeyCodec.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/DimensionIndexKeyCodec.cs), [RollupIndexKeyCodec.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/RollupIndexKeyCodec.cs) | +| Metadata generations and historical coverage | [MetaStore.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/MetaStore.cs), [RollupActivations.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/RollupActivations.cs) | +| Copy recovery and historical reconstruction journal | [IndexRecovery.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/IndexRecovery.cs), [JointBackfillCheckpoint.cs](https://github.com/ZoneTree/ZoneTree.Analytics/blob/main/src/ZoneTree.Analytics/Storage/JointBackfillCheckpoint.cs) | + +## Related pages + +- [Engine API and advanced storage access](../engine/engine.mdx) +- [Secondary indexes and backfilling](../../performance/secondary-indexes/secondary-indexes.mdx) +- [Configuration reference](../configuration/configuration.mdx) +- [Time and accuracy](../../concepts/time-and-accuracy/time-and-accuracy.mdx) +- [Deployment and lifecycle](../../operations/deployment/deployment.mdx) diff --git a/src/ZoneTree.Analytics.UnitTests/DashboardAnalyticsOptionsTests.cs b/src/ZoneTree.Analytics.UnitTests/DashboardAnalyticsOptionsTests.cs index 1249b83..f50ae9a 100644 --- a/src/ZoneTree.Analytics.UnitTests/DashboardAnalyticsOptionsTests.cs +++ b/src/ZoneTree.Analytics.UnitTests/DashboardAnalyticsOptionsTests.cs @@ -37,6 +37,10 @@ public void ProfilesShareRetentionAndIndexCoverageWithoutSharingMutableOptions() Assert.That(samples.Ingestion.WorkerCount, Is.EqualTo(4)); Assert.That(benchmark.Ingestion.WorkerCount, Is.EqualTo(20)); Assert.That(reports.Ingestion.WorkerCount, Is.EqualTo(1)); + // Generated ingestion runs one rollup writer per worker. + Assert.That(samples.Ingestion.RollupWriterCount, Is.EqualTo(4)); + Assert.That(benchmark.Ingestion.RollupWriterCount, Is.EqualTo(20)); + Assert.That(reports.Ingestion.RollupWriterCount, Is.EqualTo(1)); Assert.That(benchmark.Ingestion.QueueFullBehavior, Is.EqualTo(QueueFullBehavior.Wait)); Assert.That(primary.Rollups.ValueAggregates["load_time"], Is.EqualTo(ValueAggregates.Percentiles)); Assert.That(samples.Rollups.ValueAggregates["load_time"], Is.EqualTo(ValueAggregates.Percentiles)); diff --git a/src/ZoneTree.Analytics.UnitTests/FlushRecoveryTests.cs b/src/ZoneTree.Analytics.UnitTests/FlushRecoveryTests.cs index 71f7fe3..0e131e5 100644 --- a/src/ZoneTree.Analytics.UnitTests/FlushRecoveryTests.cs +++ b/src/ZoneTree.Analytics.UnitTests/FlushRecoveryTests.cs @@ -9,6 +9,10 @@ public sealed class FlushRecoveryTests : DirectoryTest { static readonly DateTime Day0 = DateTime.UtcNow.Date.AddDays(-2); + /// Asserts that writing rollups fails with a storage failure of the given type. + static void AssertRollupWriteFails(Func code) where TFailure : Exception => + Assert.That(Assert.ThrowsAsync(code).InnerException, Is.TypeOf()); + static RollupValue Value(ulong visitor, double amount) { var accumulator = new RollupAccumulator { Count = 1 }; @@ -148,7 +152,7 @@ void Configure(AnalyticsOptions options) using (var engine = TestEngine.Create(directory, Configure)) { Track(engine, "/entry", 0, 10); - Assert.ThrowsAsync(() => engine.FlushAsync().WaitAsync(TimeSpan.FromSeconds(5))); + AssertRollupWriteFails(() => engine.FlushAsync().WaitAsync(TimeSpan.FromSeconds(5))); Assert.That(engine.GetStatistics().Flushes, Is.Zero); Track(engine, "/exit", 1, 20); await engine.FlushAsync().WaitAsync(TimeSpan.FromSeconds(5)); @@ -261,8 +265,8 @@ public async Task RepeatedFailuresKeepThePartitionLeasedUntilRecovery() try { Track(engine, "/exit", 1, 20); - Assert.ThrowsAsync(() => engine.FlushAsync()); - Assert.ThrowsAsync(() => engine.FlushAsync()); + AssertRollupWriteFails(() => engine.FlushAsync()); + AssertRollupWriteFails(() => engine.FlushAsync()); store.CloseIdle(TimeSpan.Zero, keepLatestOpen: false); Assert.Multiple(() => { @@ -338,7 +342,7 @@ public void FinalFlushFailureIsReportedAfterReleasingStorage() }; }); Track(engine, "/entry", 0, 0); - Assert.ThrowsAsync(() => engine.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + AssertRollupWriteFails(() => engine.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5))); unavailable = false; using var reopened = TestEngine.Create(directory); Assert.That(reopened.Query(new AnalyticsQuery @@ -359,6 +363,8 @@ public async Task PersistentFailureUsesBackpressureAndShutdownWakesTheRetryWait( options.Sessions.Enabled = false; options.Ingestion.MaxBufferedRollupRows = 1; options.Ingestion.QueueCapacity = 8; + // One writer opens one partition per flush attempt, so two opens are two attempts. + options.Ingestion.RollupWriterCount = 1; // A long delay makes it clear that shutdown wakes the retry wait immediately. options.Storage.Rollups.ConfigureMaintainer = _ => { @@ -382,7 +388,7 @@ public async Task PersistentFailureUsesBackpressureAndShutdownWakesTheRetryWait( }); if (shutdown) { - Assert.ThrowsAsync(() => harness.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5))); + AssertRollupWriteFails(() => harness.DisposeAsync().AsTask().WaitAsync(TimeSpan.FromSeconds(5))); } else { @@ -414,6 +420,7 @@ public async Task PersistentFailureUsesBackpressureAndShutdownWakesTheRetryWait( sealed class WorkerHarness : IAsyncDisposable { readonly RollupStore rollups; + readonly RollupWriterPool rollupWriters; readonly IZoneTree sessions; readonly IMaintainer maintainer; bool disposed; @@ -444,10 +451,11 @@ public WorkerHarness(string directory, Action configure = null var definitions = options.Rollups.Definitions.ToArray(); rollups = new RollupStore(Path.Combine(directory, "rollups"), options.Rollups.Granularities, options.Storage.Partitioning, definitions, factory, logger); + rollupWriters = new RollupWriterPool(rollups, options.Ingestion.RollupWriterCount, logger); Worker = new IngestionWorker(new IngestionServices { Options = options, Calendar = calendar, Enricher = enricher, Sessions = SessionFault, - Rollups = rollups, Definitions = definitions, Logger = logger, Counters = Counters + Rollups = rollups, RollupWriters = rollupWriters, Definitions = definitions, Logger = logger, Counters = Counters }); Worker.Start(); } @@ -476,6 +484,7 @@ public async ValueTask DisposeAsync() } finally { + rollupWriters.Dispose(); rollups.Dispose(); maintainer?.Dispose(); sessions?.Dispose(); diff --git a/src/ZoneTree.Analytics.UnitTests/ReleaseRegressionTests.cs b/src/ZoneTree.Analytics.UnitTests/ReleaseRegressionTests.cs index b1f79a0..9e441e0 100644 --- a/src/ZoneTree.Analytics.UnitTests/ReleaseRegressionTests.cs +++ b/src/ZoneTree.Analytics.UnitTests/ReleaseRegressionTests.cs @@ -116,7 +116,7 @@ public void FlushFailureIsLoggedAndIngestionContinues(int failOnOpen) }; }); engine.Track(new AnalyticsEvent("before") { VisitorId = "alice", Timestamp = Utc(Day0.AddHours(1)) }); - Assert.Throws(() => engine.Flush()); + Assert.That(Assert.Throws(() => engine.Flush()).InnerException, Is.TypeOf()); Assert.That(engine.Track(new AnalyticsEvent("after") { VisitorId = "bob", Timestamp = Utc(Day0.AddHours(2)) }), Is.True); engine.Flush(); Assert.Multiple(() => diff --git a/src/ZoneTree.Analytics.UnitTests/RetentionTests.cs b/src/ZoneTree.Analytics.UnitTests/RetentionTests.cs index 23cb5dd..c3eab69 100644 --- a/src/ZoneTree.Analytics.UnitTests/RetentionTests.cs +++ b/src/ZoneTree.Analytics.UnitTests/RetentionTests.cs @@ -100,8 +100,8 @@ public void RangesExpireOnlyWithinOneHead() const long cutoff = 1_000; ExpiredKeyPredicate rawExpired = key => KeyCodec.ReadRawTicks(key) < cutoff; ExpiredKeyPredicate rollupExpired = key => KeyCodec.ReadRollupBucket(key) < cutoff; - bool Raw(Memory first, Memory last) => ExpiredRecords.IsRangeExpired(first.Span, last.Span, 1, rawExpired); - bool Rollups(Memory first, Memory last) => ExpiredRecords.IsRangeExpired(first.Span, last.Span, 3, rollupExpired); + bool Raw(Memory first, Memory last) => ExpiredRecords.IsRangeExpired(first.Span, last.Span, KeyCodec.RawEventHead, rawExpired); + bool Rollups(Memory first, Memory last) => ExpiredRecords.IsRangeExpired(first.Span, last.Span, KeyCodec.RollupHead, rollupExpired); Assert.Multiple(() => { // One site, older than the cutoff up to the last key: every event between is older too. @@ -176,7 +176,7 @@ public void RetentionRemovesExpiredRecordsFromDisk() foreach (var name in raw.ListPartitions()) { using var lease = raw.Acquire(name); - Assert.That(ExpiredRecords.DeleteFirstOfExpiredHeads(lease.Tree, 1, clock.IsRawEventExpired), Is.Zero); + Assert.That(ExpiredRecords.DeleteFirstOfExpiredHeads(lease.Tree, KeyCodec.RawEventHead, clock.IsRawEventExpired), Is.Zero); } } diff --git a/src/ZoneTree.Analytics.UnitTests/RollupWriterPoolTests.cs b/src/ZoneTree.Analytics.UnitTests/RollupWriterPoolTests.cs new file mode 100644 index 0000000..b8028b6 --- /dev/null +++ b/src/ZoneTree.Analytics.UnitTests/RollupWriterPoolTests.cs @@ -0,0 +1,172 @@ +using ZoneTree.Analytics.Ingestion; +using ZoneTree.Analytics.Internal; +using ZoneTree.Analytics.Storage; + +namespace ZoneTree.Analytics.UnitTests; + +public sealed class RollupWriterPoolTests : DirectoryTest +{ + static readonly DateTime Day0 = DateTime.UtcNow.Date.AddDays(-2); + + static Dictionary Rows(RollupBufferKey key, ulong visitor) + { + var accumulator = new RollupAccumulator { Count = 1 }; + accumulator.AddVisitor(visitor, 14); + return new() { [key] = accumulator }; + } + + static StoreFactory Factory(AnalyticsOptions options) => new(options.Storage, + Path.GetFullPath(options.DataDirectory), new RecordingLogger(), + new RetentionClock(options.Retention, new ReportingCalendar(options.TimeZone, options.FirstDayOfWeek))); + + [Test] + public async Task AWriterWritesTheRowOfWorkersThatFlushedTogetherOnce() + { + using var opening = new ManualResetEventSlim(); + using var release = new ManualResetEventSlim(); + // The first write opens the partition and waits there, while two more workers flush. + var options = TestEngine.Options(directory, o => o.Storage.Rollups.ConfigureMaintainer = _ => + { + opening.Set(); + release.Wait(TimeSpan.FromSeconds(30)); + }); + using var rollups = new RollupStore(Path.Combine(directory, "rollups"), Granularity.Hour, + options.Storage.Partitioning, [new RollupDefinition()], Factory(options), new RecordingLogger()); + using var pool = new RollupWriterPool(rollups, 1, new RecordingLogger()); + var bucket = Day0.AddHours(3).Ticks; + var key = new RollupBufferKey(Granularity.Hour, bucket, "default", "purchase", 0, ""); + Task Flush(ulong visitor) => Task.Run(() => pool.Write(Rows(key, visitor), new Dictionary[1], new RollupHandoff(1))); + + var first = Flush(1); + Assert.That(opening.Wait(TimeSpan.FromSeconds(10)), Is.True); + Task[] others = [Flush(2), Flush(3)]; + Assert.That(SpinWait.SpinUntil(() => pool.QueuedPieces == 2, TimeSpan.FromSeconds(10)), Is.True); + release.Set(); + await Task.WhenAll([first, .. others]); + + var store = rollups.GetStore(Granularity.Hour); + using var lease = store.Acquire(store.ListPartitions().Single()); + Assert.That(lease.Tree.TryGet(KeyCodec.RollupKey("default", "purchase", "", bucket, ""), out var stored), Is.True); + Assert.Multiple(() => + { + // The first flush alone, then the two queued flushes as one merged row. + Assert.That(pool.RowsWritten, Is.EqualTo(2)); + Assert.That(stored.Count, Is.EqualTo(3)); + Assert.That(stored.Visitors.Estimate(), Is.EqualTo(3)); + }); + } + + [Test] + public void WorkersOfAFailedBatchEachGetAnExceptionOfTheirOwn() + { + using var opening = new ManualResetEventSlim(); + using var release = new ManualResetEventSlim(); + var opens = 0; + // Every partition open fails; the first one waits there, while two more workers flush. + var options = TestEngine.Options(directory, o => o.Storage.Rollups.ConfigureMaintainer = _ => + { + if (Interlocked.Increment(ref opens) == 1) + { + opening.Set(); + release.Wait(TimeSpan.FromSeconds(30)); + } + throw new IOException("Rollup storage unavailable"); + }); + using var rollups = new RollupStore(Path.Combine(directory, "rollups"), Granularity.Hour, + options.Storage.Partitioning, [new RollupDefinition()], Factory(options), new RecordingLogger()); + using var pool = new RollupWriterPool(rollups, 1, new RecordingLogger()); + var key = new RollupBufferKey(Granularity.Hour, Day0.AddHours(3).Ticks, "default", "purchase", 0, ""); + Task Flush(ulong visitor) => Task.Run(() => pool.Write(Rows(key, visitor), new Dictionary[1], new RollupHandoff(1))); + + var first = Flush(1); + Assert.That(opening.Wait(TimeSpan.FromSeconds(10)), Is.True); + Task[] others = [Flush(2), Flush(3)]; + Assert.That(SpinWait.SpinUntil(() => pool.QueuedPieces == 2, TimeSpan.FromSeconds(10)), Is.True); + release.Set(); + Assert.ThrowsAsync(() => first); + // The retry of the first batch fails once, for both queued flushes. + var failures = others.Select(task => Assert.ThrowsAsync(() => task)).ToArray(); + Assert.Multiple(() => + { + // Before: both workers rethrew the one failure, each rewriting its stack trace. + Assert.That(failures[0], Is.Not.SameAs(failures[1])); + Assert.That(failures[0].InnerException, Is.SameAs(failures[1].InnerException)); + Assert.That(failures[0].InnerException, Is.TypeOf()); + Assert.That(failures[0].Message, Does.Contain("Rollup storage unavailable")); + }); + } + + [Test] + public void ManyWorkersAndWritersCountEveryEventOnce() + { + using var engine = TestEngine.Create(directory, o => + { + o.Ingestion.WorkerCount = 8; + o.Ingestion.RollupWriterCount = 3; + // Small buffers make workers flush on their own, not only together. + o.Ingestion.MaxBufferedRollupRows = 50; + }); + for (var i = 0; i < 5_000; ++i) + { + Assert.That(engine.Track(new AnalyticsEvent(StandardEvents.PageView) + { + VisitorId = "v" + (i % 300), Url = "/p" + (i % 7), Timestamp = Utc(Day0.AddMinutes(i % 1_440)) + }), Is.True); + } + engine.Flush(); + AnalyticsQuery Query(QuerySource source) => new() + { + Range = Day(Day0), Source = source, GroupBy = { StandardDimensions.Page }, Metrics = { Metric.PageViews, Metric.Visitors } + }; + var rollups = engine.Query(Query(QuerySource.Rollups)); + var raw = engine.Query(Query(QuerySource.RawEvents)); + Assert.Multiple(() => + { + Assert.That(rollups.Totals[Metric.PageViews], Is.EqualTo(5_000)); + Assert.That(rollups.Totals[Metric.Visitors], Is.EqualTo(300)); + Assert.That(rollups.Rows.Select(row => (row[StandardDimensions.Page], row[Metric.PageViews], row[Metric.Visitors])), + Is.EquivalentTo(raw.Rows.Select(row => (row[StandardDimensions.Page], row[Metric.PageViews], row[Metric.Visitors])))); + }); + } + + [Test] + public void FailuresOfSeveralWritersAreAllLogged() + { + var logger = new RecordingLogger(); + var storageUp = false; + var engine = TestEngine.Create(directory, o => + { + o.Ingestion.RollupWriterCount = 8; + o.Storage.Logger = logger; + // Every rollup partition open fails, so every writer with rows fails on its own. + o.Storage.Rollups.ConfigureMaintainer = _ => + { + if (!storageUp) + throw new IOException("Rollup storage unavailable"); + }; + }); + try + { + for (var i = 0; i < 200; ++i) + engine.Track(new AnalyticsEvent("purchase") { Site = "s" + i, Timestamp = Utc(Day0.AddMinutes(i)) }); + Assert.That(Assert.Throws(() => engine.Flush()).InnerException, Is.TypeOf()); + var failures = logger.Problems.Count(problem => problem.Contains("Rollup storage unavailable", StringComparison.Ordinal)); + // Before: the worker reported the first writer's failure and dropped the others. + Assert.That(failures, Is.GreaterThan(1)); + } + finally + { + storageUp = true; + engine.Dispose(); + } + } + + [TestCase(0)] + [TestCase(65)] + public void RollupWriterCountIsValidated(int count) + { + var options = TestEngine.Options(directory, o => o.Ingestion.RollupWriterCount = count); + var error = Assert.Throws(() => new AnalyticsEngine(options).Dispose()); + Assert.That(error.Message, Does.Contain("RollupWriterCount")); + } +} diff --git a/src/ZoneTree.Analytics.UnitTests/SessionRollupConsistencyTests.cs b/src/ZoneTree.Analytics.UnitTests/SessionRollupConsistencyTests.cs new file mode 100644 index 0000000..c796155 --- /dev/null +++ b/src/ZoneTree.Analytics.UnitTests/SessionRollupConsistencyTests.cs @@ -0,0 +1,83 @@ +using System.Globalization; + +namespace ZoneTree.Analytics.UnitTests; + +/// +/// Rollups rebuilt from raw events and queries of raw events derive session metrics from the +/// same reconstructed sessions, so they must agree with each other and with live ingestion. +/// +public sealed class SessionRollupConsistencyTests : DirectoryTest +{ + static readonly DateTime Day0 = DateTime.UtcNow.Date.AddDays(-3); + + // Session metrics group by session-level dimensions; page metrics by page. + static readonly (string[] GroupBy, Metric[] Metrics)[] Queries = + [ + ([StandardDimensions.Source, StandardDimensions.EntryPage], + [Metric.Sessions, Metric.Bounces, Metric.BounceRate, Metric.SessionDuration, Metric.PageViewsPerSession]), + ([StandardDimensions.Source, StandardDimensions.Page], [Metric.Entrances, Metric.Exits, Metric.PageViews]) + ]; + + static void Ingest(AnalyticsEngine engine) + { + for (var visitor = 0; visitor < 30; ++visitor) + { + // Every third visitor bounces; the others view three or four pages, minutes apart. + var pages = visitor % 3 == 0 ? 1 : 2 + visitor % 3; + var start = Day0.AddHours(1 + visitor % 20).AddMinutes(visitor); + for (var page = 0; page < pages; ++page) + { + Assert.That(engine.Track(new AnalyticsEvent(StandardEvents.PageView) + { + VisitorId = "v" + visitor, Url = "/p" + (visitor + page) % 4, Timestamp = Utc(start.AddMinutes(page * 5)), + Dimensions = { [StandardDimensions.Source] = visitor % 2 == 0 ? "GitHub" : "Google" } + }), Is.True); + } + } + engine.Flush(); + } + + static (AnalyticsResult Result, string[] Rows) Run(AnalyticsEngine engine, QuerySource source, string[] groupBy, Metric[] metrics) + { + var query = new AnalyticsQuery { Range = Day(Day0), Source = source }; + query.GroupBy.AddRange(groupBy); + query.Metrics.AddRange(metrics); + var result = engine.Query(query); + return (result, [.. result.Rows.Select(row => + string.Join("|", groupBy.Select(dimension => row[dimension])) + "|" + + string.Join(",", metrics.Select(metric => row[metric].ToString("R", CultureInfo.InvariantCulture))))]); + } + + [Test] + public void RebuiltRollupsRawQueriesAndLiveIngestionAgreeOnSessionMetrics() + { + using (var engine = TestEngine.Create(directory)) + Ingest(engine); + // Rollup indexes added later rebuild their joint rollups from the retained raw events. + using var rebuilt = TestEngine.Create(directory, o => + { + o.Storage.RollupIndexes.Definitions.Add(new(StandardDimensions.Source, StandardDimensions.EntryPage)); + o.Storage.RollupIndexes.Definitions.Add(new(StandardDimensions.Source, StandardDimensions.Page)); + }); + using var live = TestEngine.Create(Path.Combine(directory, "live"), o => + { + o.Rollups.Definitions.Add(new RollupDefinition(StandardDimensions.EntryPage, StandardDimensions.Source)); + o.Rollups.Definitions.Add(new RollupDefinition(StandardDimensions.Page, StandardDimensions.Source)); + }); + Ingest(live); + foreach (var (groupBy, metrics) in Queries) + { + var fromRollups = Run(rebuilt, QuerySource.Rollups, groupBy, metrics); + var fromRaw = Run(rebuilt, QuerySource.RawEvents, groupBy, metrics); + var fromLive = Run(live, QuerySource.Rollups, groupBy, metrics); + var name = string.Join(",", groupBy); + Assert.Multiple(() => + { + Assert.That(fromRollups.Result.Plan.Steps.All(step => step.Source != PlanSource.RawEvents), Is.True, name + ": answered by the rebuilt rollup"); + Assert.That(fromRollups.Rows, Is.Not.Empty, name); + Assert.That(fromRollups.Rows, Is.EquivalentTo(fromRaw.Rows), name + ": rebuilt rollups and raw queries"); + Assert.That(fromLive.Rows, Is.EquivalentTo(fromRaw.Rows), name + ": live ingestion and raw queries"); + }); + } + } +} diff --git a/src/ZoneTree.Analytics/AnalyticsEngine.cs b/src/ZoneTree.Analytics/AnalyticsEngine.cs index b16713b..8d03e56 100644 --- a/src/ZoneTree.Analytics/AnalyticsEngine.cs +++ b/src/ZoneTree.Analytics/AnalyticsEngine.cs @@ -176,12 +176,16 @@ public AnalyticsEngine(AnalyticsOptions options) ownedStores.Add(rollupIndexes); rollupIndexes.Initialize(rawEvents, activations, calendar, enricher.Schema); rollups.Indexes = rollupIndexes; + // Registered after the stores it writes to, so it stops before they close. + var rollupWriters = new RollupWriterPool(rollups, options.Ingestion.RollupWriterCount, logger); + ownedStores.Add(rollupWriters); var services = new IngestionServices { Options = options, Enricher = enricher, Rollups = rollups, + RollupWriters = rollupWriters, RawEvents = rawEvents, DimensionIndexes = dimensionIndexes.Indexes.Count == 0 ? null : dimensionIndexes, Sessions = sessions, @@ -481,8 +485,10 @@ async ValueTask TrackCoreAsync(AnalyticsEvent analyticsEvent, b /// aggregates are written to storage. /// /// - /// A storage failure is reported to the caller. Pending aggregates are retained and retried - /// by subsequent flushes, including automatic flushes, without applying successful writes twice. + /// A storage failure is reported to the caller; a failure to write rollups as an + /// whose inner exception is the storage failure. + /// Pending aggregates are retained and retried by subsequent flushes, including automatic + /// flushes, without applying successful writes twice. /// public async Task FlushAsync(CancellationToken cancellationToken = default) { @@ -795,7 +801,7 @@ int ApplyRetentionCore() if (!rollups.IsEnabled(granularity) || cutoffs.Of(granularity) == long.MinValue) continue; // Daily partitions are named after local days of the reporting time zone. - var cutoff = granularity == Granularity.Day ? cutoffs.DayRollupsLocalDay : cutoffs.Of(granularity); + var cutoff = cutoffs.RollupCutoff(granularity); dropped += rollups.GetStore(granularity).DropPartitionsEndingBefore(cutoff); } RemoveExpiredRecords(cutoffs); @@ -816,20 +822,20 @@ void RemoveExpiredRecords(RetentionCutoffs cutoffs) if (rawEvents != null && cutoffs.RawEvents != long.MinValue && cutoffs.RawEvents != removedUpTo.RawEvents) { rawEvents.Store.RemoveExpired( - cutoffs.RawEvents, tree => ExpiredRecords.DeleteFirstOfExpiredHeads(tree, 1, retentionClock.IsRawEventExpired)); + cutoffs.RawEvents, tree => ExpiredRecords.DeleteFirstOfExpiredHeads(tree, KeyCodec.RawEventHead, retentionClock.IsRawEventExpired)); foreach (var store in dimensionIndexes.AllStores) store.RemoveExpired(cutoffs.RawEvents, - tree => DimensionIndexKeyCodec.DeleteFirstExpired(tree, retentionClock.IsRawEventExpired)); + tree => ExpiredRecords.DeleteFirstOfExpiredHeads(tree, DimensionIndexKeyCodec.HeadLength, retentionClock.IsRawEventExpired)); removedUpTo = removedUpTo with { RawEvents = cutoffs.RawEvents }; } foreach (var granularity in RollupStore.Granularities) { - var cutoff = granularity == Granularity.Day ? cutoffs.DayRollupsLocalDay : cutoffs.Of(granularity); - var removed = granularity == Granularity.Day ? removedUpTo.DayRollupsLocalDay : removedUpTo.Of(granularity); + var cutoff = cutoffs.RollupCutoff(granularity); + var removed = removedUpTo.RollupCutoff(granularity); if (!rollups.IsEnabled(granularity) || cutoff == long.MinValue || cutoff == removed) continue; rollups.GetStore(granularity).RemoveExpired( - cutoff, tree => ExpiredRecords.DeleteFirstOfExpiredHeads(tree, 3, key => retentionClock.IsRollupExpired(granularity, key))); + cutoff, tree => ExpiredRecords.DeleteFirstOfExpiredHeads(tree, KeyCodec.RollupHead, key => retentionClock.IsRollupExpired(granularity, key))); removedUpTo = granularity switch { Granularity.Minute => removedUpTo with { MinuteRollups = cutoff }, diff --git a/src/ZoneTree.Analytics/AnalyticsExceptions.cs b/src/ZoneTree.Analytics/AnalyticsExceptions.cs index 846795a..77fd962 100644 --- a/src/ZoneTree.Analytics/AnalyticsExceptions.cs +++ b/src/ZoneTree.Analytics/AnalyticsExceptions.cs @@ -44,6 +44,33 @@ public AnalyticsConfigurationException(string message, Exception innerException) } } +/// +/// Raised by a flush when its rollup rows could not be written to storage. The rows are retained +/// and retried by later flushes. is the storage failure. +/// +/// +/// Rollup writer threads write the rows of every worker that flushed together, so one storage +/// failure can fail several flushes at once. Each of them raises an exception of its own, with +/// that failure as its inner exception. +/// +public sealed class AnalyticsStorageException : AnalyticsException +{ + /// Creates the exception. + public AnalyticsStorageException() + { + } + + /// Creates the exception with a message. + public AnalyticsStorageException(string message) : base(message) + { + } + + /// Creates the exception with a message and the exception that caused it. + public AnalyticsStorageException(string message, Exception innerException) : base(message, innerException) + { + } +} + /// /// Raised when a query is invalid or cannot be answered with the configured storage. /// diff --git a/src/ZoneTree.Analytics/Ingestion/IngestionWorker.cs b/src/ZoneTree.Analytics/Ingestion/IngestionWorker.cs index 58e39a9..8092647 100644 --- a/src/ZoneTree.Analytics/Ingestion/IngestionWorker.cs +++ b/src/ZoneTree.Analytics/Ingestion/IngestionWorker.cs @@ -56,6 +56,9 @@ sealed class IngestionServices public RollupStore Rollups; + /// The threads that write every worker's rollup rows. + public RollupWriterPool RollupWriters; + public RawEventStore RawEvents; public DimensionIndexStore DimensionIndexes; @@ -88,7 +91,8 @@ sealed class IngestionServices /// /// /// Raw events are written to ZoneTree immediately. Rollup rows are pre-aggregated in -/// memory and merged into ZoneTree on flush, together with the changed session states. +/// memory and handed to the on flush, which merges the rows of +/// all workers; the changed session states are written once the rollups are. /// /// sealed class IngestionWorker : IAsyncDisposable @@ -138,7 +142,11 @@ sealed class IngestionWorker : IAsyncDisposable PartitionLease, Memory> rawLease; - RollupWriteBatch pendingRollups; + // Rows a rollup writer did not accept because storage was failing, by writer, retried by the + // next flush. Rows a writer accepted are its own to retry. + readonly Dictionary[] keptRollups; + + readonly RollupHandoff handoff; Exception lastFlushFailure; @@ -173,6 +181,8 @@ public IngestionWorker(IngestionServices services) storeRawEvents = options.Storage.StoreRawEvents && services.RawEvents != null; indexWriter = services.DimensionIndexes?.CreateWriter(); maxBufferedRows = options.Ingestion.MaxBufferedRollupRows; + keptRollups = new Dictionary[services.RollupWriters.Count]; + handoff = new RollupHandoff(services.RollupWriters.Count); lookupHorizon = services.StoredSessionHorizon; RefreshRetentionCutoffs(); } @@ -256,7 +266,20 @@ async Task RunAsync() } finally { - pendingRollups?.Dispose(); + // Rows still kept at shutdown are lost with the failure that kept them. + Array.Clear(keptRollups); + } + } + + /// The rollup rows waiting for a flush: buffered, or kept from failed flushes. + int RetainedRollupRows + { + get + { + var count = buffer.Count; + foreach (var kept in keptRollups) + count += kept?.Count ?? 0; + return count; } } @@ -268,7 +291,7 @@ async Task RunAsync() async Task WaitForFlushRecoveryAsync() { while (lastFlushFailure != null && !stopping.Task.IsCompleted && - (buffer.Count >= maxBufferedRows || sessions.Count >= services.Options.Sessions.MaxCachedSessions)) + (RetainedRollupRows >= maxBufferedRows || sessions.Count >= services.Options.Sessions.MaxCachedSessions)) { await Task.WhenAny(Task.Delay(services.Options.Ingestion.FlushInterval), stopping.Task).ConfigureAwait(false); if (!stopping.Task.IsCompleted) @@ -568,29 +591,19 @@ Exception TryFlush() return lastFlushFailure; } - void WritePendingRollups() - { - if (pendingRollups == null) - return; - services.Rollups.Write(pendingRollups); - pendingRollups.Dispose(); - pendingRollups = null; - } - void Flush() { lastFlushTimestamp = Stopwatch.GetTimestamp(); RefreshRetentionCutoffs(); try { - WritePendingRollups(); foreach (var visitor in dirtySessions) { if (sessions.TryGetValue(visitor, out var dirty)) EmitSessionDeltas(dirty); } - pendingRollups = new RollupWriteBatch(buffer.Swap()); - WritePendingRollups(); + // Waits until every writer has written these rows and the rows it accepted before. + services.RollupWriters.Write(buffer.Swap(), keptRollups, handoff); if (dirtySessions.Count > 0) { diff --git a/src/ZoneTree.Analytics/Ingestion/RollupWriterPool.cs b/src/ZoneTree.Analytics/Ingestion/RollupWriterPool.cs new file mode 100644 index 0000000..06440fc --- /dev/null +++ b/src/ZoneTree.Analytics/Ingestion/RollupWriterPool.cs @@ -0,0 +1,327 @@ +using ZoneTree.Analytics.Internal; +using ZoneTree.Analytics.Storage; +using ZoneTree.Logger; + +namespace ZoneTree.Analytics.Ingestion; + +/// +/// One worker's flush through every rollup writer: the outcome of its piece for each writer. +/// Owned and reused by its worker; a worker has at most one flush in flight. +/// +sealed class RollupHandoff(int writers) +{ + readonly object sync = new(); + + int remaining; + + public readonly bool[] Accepted = new bool[writers]; + + public readonly Exception[] Failures = new Exception[writers]; + + public void Begin() + { + Array.Clear(Accepted); + Array.Clear(Failures); + remaining = Accepted.Length; + } + + public void Complete(int writer, bool accepted, Exception failure) + { + lock (sync) + { + Accepted[writer] = accepted; + Failures[writer] = failure; + if (--remaining == 0) + Monitor.PulseAll(sync); + } + } + + public void Wait() + { + lock (sync) + { + while (remaining > 0) + Monitor.Wait(sync); + } + } +} + +/// A worker's rows for one rollup writer. +sealed record RollupPiece(RollupHandoff Handoff, int Writer, Dictionary Rows); + +/// +/// Dedicated threads that write the rollup rows of every worker. Each writer owns the rows +/// whose key hashes to it, so a row is written by one writer only. +/// +/// +/// +/// Workers are routed by visitor, so the busy rows of a bucket, such as its totals, are buffered +/// by every worker. A writer merges the pieces of all workers that flushed since its last write, +/// and writes each row once instead of once per worker. Flushes of all workers start together +/// on the flush timer, so their pieces meet. +/// +/// +/// A worker waits for every writer before it persists its session states, as when it wrote its +/// rollups itself. A writer that accepts a piece owns its rows: when their write fails, it keeps +/// them and retries them before any later piece. While that retry fails, it rejects new pieces, +/// so their workers keep their rows, and the buffers they bound, until storage recovers. +/// +/// +sealed class RollupWriterPool : IDisposable +{ + /// The rows of a piece without rows. Never modified: merges skip empty pieces. + static readonly Dictionary NoRows = []; + + readonly RollupStore rollups; + + readonly ILogger logger; + + readonly Writer[] writers; + + public RollupWriterPool(RollupStore rollups, int count, ILogger logger) + { + this.rollups = rollups; + this.logger = logger; + writers = new Writer[count]; + for (var i = 0; i < count; ++i) + writers[i] = new Writer(this, i); + } + + public int Count => writers.Length; + + long rowsWritten; + + /// The rows the writers have written, each merged row once. + internal long RowsWritten => Interlocked.Read(ref rowsWritten); + + /// The pieces waiting for a writer. + internal int QueuedPieces => writers.Sum(writer => writer.Queued); + + /// The writer that owns a row. + int WriterOf(in RollupBufferKey key) => (int)((uint)key.GetHashCode() % (uint)writers.Length); + + /// + /// Writes a worker's flushed rows with its rows kept from failed flushes, and waits. Rows a + /// writer did not accept stay in , by writer, for the next flush. + /// + /// + /// A writer failed; its rows are retained for a retry. Failures of further writers are logged. + /// + public void Write(Dictionary rows, + Dictionary[] kept, RollupHandoff handoff) + { + if (writers.Length == 1 && kept[0] == null) + { + // The only writer takes the swapped buffer as it is. + kept[0] = rows; + } + else + { + foreach (var (key, accumulator) in rows) + RollupAccumulator.MergeInto(kept[WriterOf(key)] ??= [], key, accumulator); + } + handoff.Begin(); + // Every writer answers, even without rows: its answer also tells whether the rows it + // accepted from an earlier flush have been written since. + for (var i = 0; i < writers.Length; ++i) + writers[i].Post(new RollupPiece(handoff, i, kept[i] ?? NoRows)); + handoff.Wait(); + Exception failure = null; + for (var i = 0; i < writers.Length; ++i) + { + if (handoff.Accepted[i]) + kept[i] = null; + var writerFailure = handoff.Failures[i]; + if (writerFailure == null || ReferenceEquals(writerFailure, failure)) + continue; + if (failure == null) + failure = writerFailure; + else + logger.LogError(writerFailure); + } + // A writer completes every piece of a failed batch with the same failure. Each worker throws + // an exception of its own: throwing one instance on several threads mixes their stack traces. + if (failure != null) + throw new AnalyticsStorageException("Writing rollups failed: " + failure.Message, failure); + } + + public void Dispose() + { + foreach (var writer in writers) + writer.Stop(); + var failures = new List(); + foreach (var writer in writers) + { + writer.Join(); + if (writer.ShutdownFailure != null) + failures.Add(writer.ShutdownFailure); + } + Disposal.ThrowFailures(failures); + } + + sealed class Writer + { + readonly RollupWriterPool pool; + + readonly object sync = new(); + + readonly Thread thread; + + List queue = []; + + bool stopping; + + // Accepted rows whose write failed, retried before any later piece. Changed by the writer + // thread only; hasFailed publishes whether it is set to workers posting empty pieces. + RollupWriteBatch failed; + + volatile bool hasFailed; + + /// A failure to release the rows still failing at shutdown, reported by Dispose. + public Exception ShutdownFailure { get; private set; } + + public Writer(RollupWriterPool pool, int index) + { + this.pool = pool; + thread = new Thread(Run) { IsBackground = true, Name = "ZoneTree.Analytics rollup writer " + index }; + thread.Start(); + } + + public void Post(RollupPiece piece) + { + lock (sync) + { + if (stopping) + { + piece.Handoff.Complete(piece.Writer, false, new ObjectDisposedException(nameof(RollupWriterPool))); + return; + } + // A worker's earlier rows are written before it flushes again, or kept as the failed + // batch: without rows and without a failed batch, there is nothing to wait for. + if (piece.Rows.Count == 0 && !hasFailed) + { + piece.Handoff.Complete(piece.Writer, true, null); + return; + } + queue.Add(piece); + Monitor.Pulse(sync); + } + } + + public void Stop() + { + lock (sync) + { + stopping = true; + Monitor.Pulse(sync); + } + } + + public void Join() => thread.Join(); + + public int Queued + { + get + { + lock (sync) + return queue.Count; + } + } + + void Run() + { + while (true) + { + List pieces; + lock (sync) + { + while (queue.Count == 0 && !stopping) + Monitor.Wait(sync); + if (queue.Count == 0) + break; + pieces = queue; + queue = []; + } + Process(pieces); + } + // Rows still failing at shutdown are lost, as a worker's retained rows are. Nothing may + // escape this thread: an unhandled exception would end the process. + try + { + failed?.Dispose(); + } + catch (Exception exception) + { + ShutdownFailure = exception; + } + failed = null; + hasFailed = false; + } + + /// Writes the pieces; every piece is completed, whatever fails. + void Process(List pieces) + { + // Until their rows are in a batch, rejected pieces stay with their workers. + var accepted = false; + Exception failure = null; + try + { + if (failed != null) + { + // While storage is still failing, the exception keeps the new rows with the workers. + pool.rollups.Write(failed); + failed.Dispose(); + failed = null; + hasFailed = false; + } + var batch = Merge(pieces, out var rows); + accepted = true; + if (batch != null) + { + try + { + pool.rollups.Write(batch); + batch.Dispose(); + Interlocked.Add(ref pool.rowsWritten, rows); + } + catch + { + // Published before the pieces complete, so their workers' next empty pieces wait. + failed = batch; + hasFailed = true; + throw; + } + } + } + catch (Exception exception) + { + failure = exception; + } + foreach (var piece in pieces) + piece.Handoff.Complete(piece.Writer, accepted, failure); + } + + /// One batch of the rows of all pieces, each row once; null when they have none. + static RollupWriteBatch Merge(List pieces, out int rows) + { + rows = 0; + Dictionary merged = null; + foreach (var piece in pieces) + { + if (piece.Rows.Count == 0) + continue; + if (merged == null) + { + merged = piece.Rows; + continue; + } + foreach (var (key, accumulator) in piece.Rows) + RollupAccumulator.MergeInto(merged, key, accumulator); + } + if (merged == null) + return null; + rows = merged.Count; + return new RollupWriteBatch(merged); + } + } +} diff --git a/src/ZoneTree.Analytics/Internal/RetentionCutoffs.cs b/src/ZoneTree.Analytics/Internal/RetentionCutoffs.cs index 8d8eeaa..f90d879 100644 --- a/src/ZoneTree.Analytics/Internal/RetentionCutoffs.cs +++ b/src/ZoneTree.Analytics/Internal/RetentionCutoffs.cs @@ -65,12 +65,19 @@ static long Cutoff(TimeSpan? retention, long nowUtcTicks) _ => DayRollups }; + /// + /// The cutoff in the time coordinate rollup keys of a granularity hold: UTC for minute and + /// hour buckets, the local day for day buckets. + /// + public long RollupCutoff(Granularity granularity) => + granularity == Granularity.Day ? DayRollupsLocalDay : Of(granularity); + /// /// True when a rollup row is within the retention of its granularity. Minute and hour rows /// are compared by event time, day rows by their local day bucket. /// public bool KeepsRollup(Granularity granularity, long utcTicks, long bucket) => - granularity == Granularity.Day ? bucket >= DayRollupsLocalDay : utcTicks >= Of(granularity); + (granularity == Granularity.Day ? bucket : utcTicks) >= RollupCutoff(granularity); /// /// A query plan warning when a raw event read starting at diff --git a/src/ZoneTree.Analytics/Options/IngestionOptions.cs b/src/ZoneTree.Analytics/Options/IngestionOptions.cs index 0a80c21..7c9765d 100644 --- a/src/ZoneTree.Analytics/Options/IngestionOptions.cs +++ b/src/ZoneTree.Analytics/Options/IngestionOptions.cs @@ -17,6 +17,14 @@ public sealed class IngestionOptions /// public int WorkerCount { get; set; } = 1; + /// + /// Number of dedicated rollup writer threads. Each writer owns a share of the rollup rows: it + /// merges the rows that all workers buffered for them and writes each row once per flush, + /// instead of once per worker. Default is 1, which suits single-processor machines; with + /// several workers on several cores, more writers write rows in parallel. + /// + public int RollupWriterCount { get; set; } = 1; + /// /// How often buffered aggregates and session states are written to storage. /// Rollup-based queries lag behind ingestion by at most this interval. @@ -51,6 +59,8 @@ internal void Validate() throw new ArgumentException("Ingestion.QueueCapacity must be positive."); if (WorkerCount is < 1 or > 64) throw new ArgumentException("Ingestion.WorkerCount must be between 1 and 64."); + if (RollupWriterCount is < 1 or > 64) + throw new ArgumentException("Ingestion.RollupWriterCount must be between 1 and 64."); if (FlushInterval <= TimeSpan.Zero) throw new ArgumentException("Ingestion.FlushInterval must be positive."); if (FlushInterval > TimeSpan.FromMilliseconds(uint.MaxValue - 1)) diff --git a/src/ZoneTree.Analytics/Query/QueryEngine.cs b/src/ZoneTree.Analytics/Query/QueryEngine.cs index 7e3d84a..3d17301 100644 --- a/src/ZoneTree.Analytics/Query/QueryEngine.cs +++ b/src/ZoneTree.Analytics/Query/QueryEngine.cs @@ -323,15 +323,8 @@ static RollupAccumulator GetGroup(QueryContext context, Dictionary target, Dictionary source) { foreach (var (key, accumulator) in source) - { - if (target.TryGetValue(key, out var existing)) - { - existing.Merge(accumulator); + if (RollupAccumulator.MergeInto(target, key, accumulator)) context.GroupBudget.Release(); - } - else - target.Add(key, accumulator); - } source.Clear(); } @@ -515,12 +508,10 @@ static void AddRawSession( ++accumulator.Count; if (source.ReadsVisitors) accumulator.AddVisitor(session.Visitor, context.Precision); - accumulator.AddValueDelta(StandardValues.Bounces, session.Interactions <= 1 ? 1 : 0); - var duration = session.Last - session.Start; - if (duration > 0) - accumulator.AddValueDelta(StandardValues.Duration, (double)duration / TimeSpan.TicksPerSecond); - if (session.PageViews > 0) - accumulator.AddValueDelta(StandardValues.SessionPageViews, session.PageViews); + var values = sessionValues ??= []; + RawSessionContributions.SessionValues(session, values); + foreach (var value in values) + accumulator.AddValueDelta(value.Name, value.Value); foreach (var dimension in source.DistinctDimensions) { var value = new FilterValues(session.Dimensions, session.Start, context.Calendar).Get(dimension); @@ -529,9 +520,13 @@ static void AddRawSession( } } + // The values of one session row, reused by the scan threads (see RawSessionContributions). + [ThreadStatic] + static List sessionValues; + /// - /// Entrances count at the first event of a session when it is a page view; exits count at - /// the last page view of a session. This matches how ingestion maintains them in rollups. + /// The entrance and the exit of a session (see ), when + /// they lie in the segment. /// static void AddRawEntranceAndExit( QueryContext context, @@ -542,24 +537,29 @@ static void AddRawEntranceAndExit( ReadSegment segment, PageDimensions pageDimensions) { - if (session.FirstIsPageView && session.Start >= segment.From && session.Start < segment.To) - { - var dimensions = pageDimensions.Of(session.Dimensions, session.FirstPage, session.FirstHostname); - if (MatchAndGroup(context, source, dimensions, session.Start, groupValues)) - { - GetGroup(context, groups, context.Calendar.Bucket(session.Start, context.Interval), groupValues) - .AddValueDelta(StandardValues.Entrances, 1); - } - } - if (session.LastPage != null && session.LastPageTicks >= segment.From && session.LastPageTicks < segment.To) - { - var dimensions = pageDimensions.Of(session.Dimensions, session.LastPage, session.LastHostname); - if (MatchAndGroup(context, source, dimensions, session.LastPageTicks, groupValues)) - { - GetGroup(context, groups, context.Calendar.Bucket(session.LastPageTicks, context.Interval), groupValues) - .AddValueDelta(StandardValues.Exits, 1); - } - } + if (RawSessionContributions.Entrance(session, out var ticks, out var page, out var hostname)) + AddPageMark(context, source, groups, session, groupValues, segment, pageDimensions, ticks, page, hostname, StandardValues.Entrances); + if (RawSessionContributions.Exit(session, out ticks, out page, out hostname)) + AddPageMark(context, source, groups, session, groupValues, segment, pageDimensions, ticks, page, hostname, StandardValues.Exits); + } + + static void AddPageMark( + QueryContext context, + SourceContext source, + Dictionary groups, + RawSession session, + string[] groupValues, + ReadSegment segment, + PageDimensions pageDimensions, + long ticks, + string page, + string hostname, + string value) + { + if (ticks < segment.From || ticks >= segment.To) + return; + if (MatchAndGroup(context, source, pageDimensions.Of(session.Dimensions, page, hostname), ticks, groupValues)) + GetGroup(context, groups, context.Calendar.Bucket(ticks, context.Interval), groupValues).AddValueDelta(value, 1); } static bool MatchAndGroup( diff --git a/src/ZoneTree.Analytics/Query/RawSessionContributions.cs b/src/ZoneTree.Analytics/Query/RawSessionContributions.cs new file mode 100644 index 0000000..a66a030 --- /dev/null +++ b/src/ZoneTree.Analytics/Query/RawSessionContributions.cs @@ -0,0 +1,48 @@ +using ZoneTree.Analytics.Ingestion; + +namespace ZoneTree.Analytics.Query; + +/// +/// What a reconstructed session contributes to rollups: its session row, and the entrance and +/// exit of its page views. Rollups rebuilt from raw events and queries of raw events both take +/// them from here; ingestion maintains the same values incrementally, as deltas. +/// +static class RawSessionContributions +{ + /// + /// The values of the session row, in place of the list's contents: a bounce when the session + /// has at most one interaction, its duration and its page views. Zero deltas are omitted, as + /// ingestion omits them. + /// + public static void SessionValues(RawSession session, List values) + { + values.Clear(); + if (session.Interactions <= 1) + values.Add(new ValueChange(StandardValues.Bounces, 1, IsDelta: true)); + var duration = session.Last - session.Start; + if (duration > 0) + values.Add(new ValueChange(StandardValues.Duration, (double)duration / TimeSpan.TicksPerSecond, IsDelta: true)); + if (session.PageViews > 0) + values.Add(new ValueChange(StandardValues.SessionPageViews, session.PageViews, IsDelta: true)); + } + + /// The entrance: the first event of the session, when it is a page view. + public static bool Entrance(RawSession session, out long ticks, out string page, out string hostname) + { + var entered = session.FirstIsPageView && session.FirstPage != null; + ticks = entered ? session.Start : 0; + page = entered ? session.FirstPage : null; + hostname = entered ? session.FirstHostname : null; + return entered; + } + + /// The exit: the last page view of the session. + public static bool Exit(RawSession session, out long ticks, out string page, out string hostname) + { + var exited = session.LastPage != null; + ticks = exited ? session.LastPageTicks : 0; + page = exited ? session.LastPage : null; + hostname = exited ? session.LastHostname : null; + return exited; + } +} diff --git a/src/ZoneTree.Analytics/Storage/DimensionIndexKeyCodec.cs b/src/ZoneTree.Analytics/Storage/DimensionIndexKeyCodec.cs index 06eb113..cf9f301 100644 --- a/src/ZoneTree.Analytics/Storage/DimensionIndexKeyCodec.cs +++ b/src/ZoneTree.Analytics/Storage/DimensionIndexKeyCodec.cs @@ -48,35 +48,6 @@ public static Memory Key(ReadOnlySpan indexNamespace, string site, s public static long Sequence(ReadOnlySpan key) => (long)BinaryPrimitives.ReadUInt64BigEndian(key[^8..]); - public static bool IsRangeExpired(ReadOnlySpan first, ReadOnlySpan last, ExpiredKeyPredicate expired) => - first.Length == last.Length && first[..^16].SequenceEqual(last[..^16]) && expired(last); - - public static Memory AfterPrefix(ReadOnlySpan prefix) - { - for (var i = prefix.Length - 1; i >= 0; --i) - { - if (prefix[i] == byte.MaxValue) continue; - var next = prefix[..(i + 1)].ToArray(); - ++next[i]; - return next; - } - throw new InvalidOperationException("An index prefix must have a successor."); - } - - public static int DeleteFirstExpired(IZoneTree, Memory> tree, ExpiredKeyPredicate expired) - { - var deleted = 0; - using var iterator = tree.CreateIterator(new IteratorOptions - { - IteratorType = IteratorType.NoRefresh, IncludeDeletedRecords = true - }); - iterator.SeekFirst(); - while (iterator.Next()) - { - var key = iterator.CurrentKey; - if (expired(key.Span)) { tree.ForceDelete(key); ++deleted; } - iterator.Seek(AfterPrefix(key.Span[..^16])); - } - return deleted; - } + /// The head of an index key: everything before its timestamp and sequence. + public static int HeadLength(ReadOnlySpan key) => key.Length - 16; } diff --git a/src/ZoneTree.Analytics/Storage/DimensionIndexStore.cs b/src/ZoneTree.Analytics/Storage/DimensionIndexStore.cs index 128ac7a..8bbbbc4 100644 --- a/src/ZoneTree.Analytics/Storage/DimensionIndexStore.cs +++ b/src/ZoneTree.Analytics/Storage/DimensionIndexStore.cs @@ -128,7 +128,7 @@ void Backfill(RawEventStore rawEvents, long[] days) { using var lease = Store.Acquire(partition); if (lease == null) continue; - IndexRecoveryScan.Visit(lease.Tree, days, key => key.Length - 16, (key, _) => lease.Tree.ForceDelete(key)); + IndexRecoveryScan.Visit(lease.Tree, days, DimensionIndexKeyCodec.HeadLength, (key, _) => lease.Tree.ForceDelete(key)); } IndexRecoveryScan.Persist(AllStores); checkpoint.Cleared(); @@ -151,7 +151,7 @@ void Backfill(RawEventStore rawEvents, long[] days) if (lease == null) continue; partitions++; Memory lastKey = default; - IndexRecoveryScan.Visit(lease.Tree, days, key => KeyCodec.HeadLength(key, 1), (key, value) => + IndexRecoveryScan.Visit(lease.Tree, days, KeyCodec.RawEventHead, (key, value) => { lastKey = key; var rawKey = StorageKeyCodec.DecodeRawEvent(key.Span); diff --git a/src/ZoneTree.Analytics/Storage/ExpiredRecords.cs b/src/ZoneTree.Analytics/Storage/ExpiredRecords.cs index fcafa2c..a94483e 100644 --- a/src/ZoneTree.Analytics/Storage/ExpiredRecords.cs +++ b/src/ZoneTree.Analytics/Storage/ExpiredRecords.cs @@ -14,8 +14,8 @@ namespace ZoneTree.Analytics.Storage; /// A ZoneTree merge drops the deleted records it reads. It reads every disk segment part whose /// first record is deleted, and every part that a newer record falls into; it copies the other /// parts as they are. Keys are ordered by head (the site of a raw event; the site, event and -/// dimension set of a rollup row), then by time, so the expired records of a head are its first -/// ones. +/// dimension set of a rollup row; everything before the time of an index row), then by time, so +/// the expired records of a head are its first ones. /// /// /// Deleting the first record of every head whose first record has expired makes the next merge @@ -42,12 +42,12 @@ static class ExpiredRecords /// /// The first key of the range. /// The last key of the range. - /// The number of key segments that make up a head. + /// The head length of the tree's keys. /// The retention part of the tree's deletion predicate. public static bool IsRangeExpired( - ReadOnlySpan first, ReadOnlySpan last, int headSegments, ExpiredKeyPredicate isExpired) + ReadOnlySpan first, ReadOnlySpan last, KeyHeadLength headLength, ExpiredKeyPredicate isExpired) { - var head = KeyCodec.HeadLength(first, headSegments); + var head = headLength(first); return last.Length > head && last[..head].SequenceEqual(first[..head]) && isExpired(last); } @@ -56,10 +56,10 @@ public static bool IsRangeExpired( /// Returns the number of records deleted. /// /// The tree. - /// The number of key segments that make up a head. + /// The head length of the tree's keys. /// The retention part of the tree's deletion predicate. public static int DeleteFirstOfExpiredHeads( - IZoneTree, TValue> tree, int headSegments, ExpiredKeyPredicate isExpired) + IZoneTree, TValue> tree, KeyHeadLength headLength, ExpiredKeyPredicate isExpired) { var deleted = 0; // Deleted records are included: hidden expired records are what it looks for. Each delete @@ -79,8 +79,7 @@ public static int DeleteFirstOfExpiredHeads( ++deleted; } // Skip to the next head: past every key that starts with this head. - var head = key.Span[..(KeyCodec.HeadLength(key.Span, headSegments) - 1)]; - iterator.Seek(KeyCodec.AfterSegment(head)); + iterator.Seek(KeyCodec.AfterPrefix(key.Span[..headLength(key.Span)])); } return deleted; } diff --git a/src/ZoneTree.Analytics/Storage/IndexRecovery.cs b/src/ZoneTree.Analytics/Storage/IndexRecovery.cs index c390c95..371abd4 100644 --- a/src/ZoneTree.Analytics/Storage/IndexRecovery.cs +++ b/src/ZoneTree.Analytics/Storage/IndexRecovery.cs @@ -38,8 +38,6 @@ public void Clear(Dictionary entries) /// Seeks past untouched time ranges rather than scanning old event history. static class IndexRecoveryScan { - public delegate int HeadLength(ReadOnlySpan key); - public static bool MissingRetainedPartitions(string[] expected, IReadOnlyList actual, long cutoff) { var all = expected.Union(actual).Order(StringComparer.Ordinal).ToArray(); @@ -54,7 +52,7 @@ public static bool MissingRetainedPartitions(string[] expected, IReadOnlyList(IZoneTree, TValue> tree, long[] days, - HeadLength headLength, Action, TValue> visit, Memory after = default) + KeyHeadLength headLength, Action, TValue> visit, Memory after = default) { using var iterator = tree.CreateIterator(new IteratorOptions { @@ -73,7 +71,7 @@ public static void Visit(IZoneTree, TValue> tree, long[] da if (index >= 0) { visit(key, iterator.CurrentValue); continue; } index = ~index; iterator.Seek(index < days.Length - ? KeyCodec.SeekKey(head, days[index]) : DimensionIndexKeyCodec.AfterPrefix(head)); + ? KeyCodec.SeekKey(head, days[index]) : KeyCodec.AfterPrefix(head)); } } diff --git a/src/ZoneTree.Analytics/Storage/KeyCodec.cs b/src/ZoneTree.Analytics/Storage/KeyCodec.cs index 41ec3fa..d038d65 100644 --- a/src/ZoneTree.Analytics/Storage/KeyCodec.cs +++ b/src/ZoneTree.Analytics/Storage/KeyCodec.cs @@ -4,6 +4,12 @@ namespace ZoneTree.Analytics.Storage; +/// +/// The length of the head of a key: its leading bytes, after which the keys of one head are +/// ordered by time. +/// +delegate int KeyHeadLength(ReadOnlySpan key); + /// /// Order-preserving binary key layouts. /// @@ -149,6 +155,13 @@ public static long ReadRawTicks(ReadOnlySpan key) => public static long ReadRollupBucket(ReadOnlySpan key) => (long)BinaryPrimitives.ReadUInt64BigEndian(key.Slice(HeadLength(key, 3), 8)); + /// + /// The dimension-set identifier of a rollup key: its third segment, read without decoding + /// the rest of the key. + /// + public static string ReadDimensionSetId(ReadOnlySpan key) => + ReadSegment(key, HeadLength(key, 2), out _); + /// /// The length of the leading segments of a key up to and including its /// -th terminator: the site of a raw key (1), or the site, event and @@ -162,6 +175,12 @@ public static int HeadLength(ReadOnlySpan key, int segments) return length; } + /// The head of a raw key: its site. + public static int RawEventHead(ReadOnlySpan key) => HeadLength(key, 1); + + /// The head of a rollup key: its site, event and dimension set. + public static int RollupHead(ReadOnlySpan key) => HeadLength(key, 3); + /// /// Reads the segment that starts at and ends at the next terminator. /// @@ -188,4 +207,19 @@ public static Memory AfterSegment(ReadOnlySpan prefixWithSegment) result[^1] = 1; return result; } + + /// + /// Returns the smallest key greater than every key that starts with the given prefix. + /// + public static Memory AfterPrefix(ReadOnlySpan prefix) + { + for (var i = prefix.Length - 1; i >= 0; --i) + { + if (prefix[i] == byte.MaxValue) continue; + var next = prefix[..(i + 1)].ToArray(); + ++next[i]; + return next; + } + throw new InvalidOperationException("A key prefix must have a successor."); + } } diff --git a/src/ZoneTree.Analytics/Storage/RetentionClock.cs b/src/ZoneTree.Analytics/Storage/RetentionClock.cs index 6a43474..cfc38bc 100644 --- a/src/ZoneTree.Analytics/Storage/RetentionClock.cs +++ b/src/ZoneTree.Analytics/Storage/RetentionClock.cs @@ -50,7 +50,7 @@ public bool IsRawEventExpired(ReadOnlySpan key) => public bool IsRollupExpired(Granularity granularity, ReadOnlySpan key) { var cutoffs = current.Cutoffs; - var cutoff = granularity == Granularity.Day ? cutoffs.DayRollupsLocalDay : cutoffs.Of(granularity); + var cutoff = cutoffs.RollupCutoff(granularity); return cutoff != long.MinValue && KeyCodec.ReadRollupBucket(key) < cutoff; } } diff --git a/src/ZoneTree.Analytics/Storage/RollupIndexBackfill.cs b/src/ZoneTree.Analytics/Storage/RollupIndexBackfill.cs index 174aaae..72d73a1 100644 --- a/src/ZoneTree.Analytics/Storage/RollupIndexBackfill.cs +++ b/src/ZoneTree.Analytics/Storage/RollupIndexBackfill.cs @@ -89,24 +89,22 @@ void Finish(string site, RawSession session) state.Sessions++; if (session.Start >= coverage) { - changes.Clear(); - changes.Add(new(StandardValues.Bounces, session.Interactions <= 1 ? 1 : 0, true)); - changes.Add(new(StandardValues.Duration, Math.Max(0, session.Last - session.Start) / (double)TimeSpan.TicksPerSecond, true)); - changes.Add(new(StandardValues.SessionPageViews, session.PageViews, true)); + RawSessionContributions.SessionValues(session, changes); buffer.Contribute(site, StandardEvents.Session, session.Start, session.Dimensions, 1, session.Visitor, changes, DimensionScope.Session); } - if (session.FirstIsPageView && session.FirstPage != null && session.Start >= coverage) - { - changes.Clear(); changes.Add(new(StandardValues.Entrances, 1, true)); - buffer.Contribute(site, StandardEvents.PageView, session.Start, - pageDimensions.Of(session.Dimensions, session.FirstPage, session.FirstHostname), 0, 0, changes, DimensionScope.PageView); - } - if (session.LastPage != null && session.LastPageTicks >= coverage) - { - changes.Clear(); changes.Add(new(StandardValues.Exits, 1, true)); - buffer.Contribute(site, StandardEvents.PageView, session.LastPageTicks, - pageDimensions.Of(session.Dimensions, session.LastPage, session.LastHostname), 0, 0, changes, DimensionScope.PageView); - } + if (RawSessionContributions.Entrance(session, out var ticks, out var page, out var hostname)) + ContributePageMark(site, session, ticks, page, hostname, StandardValues.Entrances); + if (RawSessionContributions.Exit(session, out ticks, out page, out hostname)) + ContributePageMark(site, session, ticks, page, hostname, StandardValues.Exits); + } + void ContributePageMark(string site, RawSession session, long ticks, string page, string hostname, string value) + { + if (ticks < coverage) + return; + changes.Clear(); + changes.Add(new(value, 1, true)); + buffer.Contribute(site, StandardEvents.PageView, ticks, + pageDimensions.Of(session.Dimensions, page, hostname), 0, 0, changes, DimensionScope.PageView); } // One chronological stream per site keeps only sessions within an inactivity window. // Original session IDs/starts preserve attribution; events are never re-ingested or re-enriched. @@ -216,7 +214,7 @@ static void DeleteDefinitions(RollupStore rollups, HashSet definitions) using var iterator = lease.Tree.CreateIterator(new IteratorOptions { IteratorType = IteratorType.NoRefresh }); iterator.SeekFirst(); while (iterator.Next()) - if (definitions.Contains(StorageKeyCodec.DecodeRollup(iterator.CurrentKey.Span).DimensionSetId)) + if (definitions.Contains(KeyCodec.ReadDimensionSetId(iterator.CurrentKey.Span))) lease.Tree.ForceDelete(iterator.CurrentKey); } } diff --git a/src/ZoneTree.Analytics/Storage/RollupIndexKeyCodec.cs b/src/ZoneTree.Analytics/Storage/RollupIndexKeyCodec.cs index 6cde25b..4f6eaf9 100644 --- a/src/ZoneTree.Analytics/Storage/RollupIndexKeyCodec.cs +++ b/src/ZoneTree.Analytics/Storage/RollupIndexKeyCodec.cs @@ -52,24 +52,4 @@ public static int HeadLength(ReadOnlySpan key) } public static ReadOnlySpan OriginalKey(ReadOnlySpan key) => key[(HeadLength(key) + 8)..]; - - public static bool IsRangeExpired(ReadOnlySpan first, ReadOnlySpan last, ExpiredKeyPredicate expired) - { - var head = HeadLength(first); - return last.Length > head && last[..head].SequenceEqual(first[..head]) && expired(last); - } - - public static int DeleteFirstExpired(IZoneTree, RollupValue> tree, ExpiredKeyPredicate expired) - { - var count = 0; - using var iterator = tree.CreateIterator(new IteratorOptions { IteratorType = IteratorType.NoRefresh, IncludeDeletedRecords = true }); - iterator.SeekFirst(); - while (iterator.Next()) - { - var key = iterator.CurrentKey; - if (expired(key.Span)) { tree.ForceDelete(key); ++count; } - iterator.Seek(DimensionIndexKeyCodec.AfterPrefix(key.Span[..HeadLength(key.Span)])); - } - return count; - } } diff --git a/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs b/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs index 806bab2..1061235 100644 --- a/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs +++ b/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs @@ -150,7 +150,7 @@ void Backfill(Dictionary dirtyDays, bool historicalBackfill if (lease == null) continue; sourcePartitions++; Memory lastKey = default; - IndexRecoveryScan.Visit(lease.Tree, dirtyDays?.GetValueOrDefault(source.Granularity), key => KeyCodec.HeadLength(key, 3), (key, value) => + IndexRecoveryScan.Visit(lease.Tree, dirtyDays?.GetValueOrDefault(source.Granularity), KeyCodec.RollupHead, (key, value) => { lastKey = key; var written = CopyCore(source.Granularity, key, value); @@ -204,7 +204,7 @@ public void Copy(Granularity granularity, Memory key, RollupValue value) public void BeforeWrite(Granularity granularity, Memory key) { if (!recovery.TryGetValue(granularity, out var journal)) return; - var definition = KeyCodec.ReadSegment(key.Span, KeyCodec.HeadLength(key.Span, 2), out _); + var definition = KeyCodec.ReadDimensionSetId(key.Span); if (copies.ContainsKey((granularity, definition))) journal.Track(KeyCodec.ReadRollupBucket(key.Span)); } @@ -212,7 +212,7 @@ int CopyCore(Granularity granularity, Memory originalKey, RollupValue valu { // Ordinary rollup rows do not need dimension arrays/dictionaries merely to discover // that their combination is not indexed. - var definitionId = KeyCodec.ReadSegment(originalKey.Span, KeyCodec.HeadLength(originalKey.Span, 2), out _); + var definitionId = KeyCodec.ReadDimensionSetId(originalKey.Span); if (!copies.TryGetValue((granularity, definitionId), out var destinations)) return 0; var key = StorageKeyCodec.DecodeRollup(originalKey.Span); var store = destinations[0].Store; @@ -281,14 +281,14 @@ public int ApplyRetention(RetentionCutoffs cutoffs, RetentionClock clock) var dropped = 0; foreach (var granularity in RollupStore.Granularities) { - var cutoff = granularity == Granularity.Day ? cutoffs.DayRollupsLocalDay : cutoffs.Of(granularity); + var cutoff = cutoffs.RollupCutoff(granularity); if (cutoff == long.MinValue) continue; if (stores.TryGetValue(granularity, out var store)) { dropped += store.DropPartitionsEndingBefore(cutoff); if (removed.GetValueOrDefault(granularity, long.MinValue) == cutoff) continue; ExpiredKeyPredicate expired = key => clock.IsRollupExpired(granularity, RollupIndexKeyCodec.OriginalKey(key)); - store.RemoveExpired(cutoff, tree => RollupIndexKeyCodec.DeleteFirstExpired(tree, expired)); + store.RemoveExpired(cutoff, tree => ExpiredRecords.DeleteFirstOfExpiredHeads(tree, RollupIndexKeyCodec.HeadLength, expired)); } removed[granularity] = cutoff; } diff --git a/src/ZoneTree.Analytics/Storage/RollupStore.cs b/src/ZoneTree.Analytics/Storage/RollupStore.cs index 53db637..e034c69 100644 --- a/src/ZoneTree.Analytics/Storage/RollupStore.cs +++ b/src/ZoneTree.Analytics/Storage/RollupStore.cs @@ -11,8 +11,10 @@ namespace ZoneTree.Analytics.Storage; /// /// Each granularity has its own partitioned store, which starts a new partition after the /// number of rows configured in . -/// A writer per partition serializes merges and resumes failed writes, which keeps concurrent -/// flushes from several ingestion workers consistent even when an upsert throws after updating memory. +/// A per partition serializes the merges of each row by lock stripe +/// and resumes failed writes, which keeps concurrent writes consistent even when an upsert throws +/// after updating memory. During ingestion, the threads of the +/// call . /// sealed class RollupStore : IDisposable { diff --git a/src/ZoneTree.Analytics/Storage/RollupValue.cs b/src/ZoneTree.Analytics/Storage/RollupValue.cs index d98a3ec..0dd7aab 100644 --- a/src/ZoneTree.Analytics/Storage/RollupValue.cs +++ b/src/ZoneTree.Analytics/Storage/RollupValue.cs @@ -863,6 +863,22 @@ void MergeValues(RollupValue value) } } + /// + /// Merges into the row of , or adds it as + /// that row. Returns true when it merged into an existing row. + /// + public static bool MergeInto(Dictionary rows, TKey key, RollupAccumulator accumulator) + { + ref var existing = ref CollectionsMarshal.GetValueRefOrAddDefault(rows, key, out var exists); + if (!exists) + { + existing = accumulator; + return false; + } + existing.Merge(accumulator); + return true; + } + public void Merge(RollupAccumulator other) { Count += other.Count; @@ -1049,7 +1065,7 @@ static DetailEntries ReadDetails(ReadOnlyMemory bytes) var entries = new RollupSection[CheckedCount(ref reader, 1)]; for (var i = 0; i < entries.Length; ++i) { - var entry = ReadEntry(ref reader, bytes, allowEmpty: true); + var entry = ReadLengthPrefixed(ref reader, bytes, allowEmpty: true); if (!entry.IsEmpty) entries[i] = new(entry, DecodeDetail); } @@ -1077,7 +1093,7 @@ static DistinctEntries ReadDistinct(ReadOnlyMemory bytes) for (var i = 0; i < count; ++i) { names[i] = reader.ReadString() ?? throw new FormatException("A distinct dimension has no name."); - sketches[i] = new(ReadEntry(ref reader, bytes, allowEmpty: false), DecodeVisitors); + sketches[i] = new(ReadLengthPrefixed(ref reader, bytes, allowEmpty: false), DecodeVisitors); } RequireEnd(reader); return new DistinctEntries(names, sketches); @@ -1088,12 +1104,15 @@ static DistinctEntries ReadDistinct(ReadOnlyMemory bytes) } } - /// Reads the length of an entry and returns its bytes without decoding them. - static ReadOnlyMemory ReadEntry(ref ByteReader reader, ReadOnlyMemory bytes, bool allowEmpty) + /// + /// Reads the length of a section or entry and returns its bytes without decoding them. + /// Only a detail entry may be empty: sections are written only when they hold something. + /// + static ReadOnlyMemory ReadLengthPrefixed(ref ByteReader reader, ReadOnlyMemory bytes, bool allowEmpty) { var length = reader.ReadVarUInt64(); if ((length == 0 && !allowEmpty) || length > (ulong)reader.Remaining) - throw new FormatException("A rollup value entry is empty or truncated."); + throw new FormatException("A rollup value section or entry is empty or truncated."); var start = bytes.Length - reader.Remaining; reader.ReadBytes((int)length); return bytes.Slice(start, (int)length); @@ -1191,10 +1210,10 @@ static RollupValue Read(ReadOnlyMemory bytes) RollupSection details = null; RollupSection distinct = null; if ((header & HasVisitors) != 0) - visitors = new(ReadSection(ref sectionReader, sections), DecodeVisitors); + visitors = new(ReadLengthPrefixed(ref sectionReader, sections, allowEmpty: false), DecodeVisitors); if ((header & HasDetails) != 0) { - var section = ReadSection(ref sectionReader, sections); + var section = ReadLengthPrefixed(ref sectionReader, sections, allowEmpty: false); // Details align with the values: their count is checked now, the details decoded when read. var detailReader = new ByteReader(section.Span); if (detailReader.ReadVarUInt64() != (ulong)nameCount) @@ -1202,7 +1221,7 @@ static RollupValue Read(ReadOnlyMemory bytes) details = new(section, DecodeDetails); } if ((header & HasDistinct) != 0) - distinct = new(ReadSection(ref sectionReader, sections), DecodeDistinct); + distinct = new(ReadLengthPrefixed(ref sectionReader, sections, allowEmpty: false), DecodeDistinct); if (sectionReader.Remaining != 0) throw new FormatException("The rollup value has trailing bytes."); return new RollupValue(count, visitors, names, values, details, distinct); @@ -1224,17 +1243,6 @@ static int CheckedCount(ref ByteReader reader, int minimumEntryBytes) static bool OwnsBuffer(ReadOnlyMemory bytes) => MemoryMarshal.TryGetArray(bytes, out var segment) && segment.Offset == 0 && segment.Count == segment.Array.Length; - /// Reads the length of a section and returns its bytes without decoding them. - static ReadOnlyMemory ReadSection(ref ByteReader reader, ReadOnlyMemory bytes) - { - var length = reader.ReadVarUInt64(); - if (length == 0 || length > (ulong)reader.Remaining) - throw new FormatException("A rollup value section is empty or truncated."); - var start = bytes.Length - reader.Remaining; - reader.ReadBytes((int)length); - return bytes.Slice(start, (int)length); - } - public Memory Serialize(in RollupValue entry) { if (entry == null) diff --git a/src/ZoneTree.Analytics/Storage/StoreFactory.cs b/src/ZoneTree.Analytics/Storage/StoreFactory.cs index 35b2a1d..db0652d 100644 --- a/src/ZoneTree.Analytics/Storage/StoreFactory.cs +++ b/src/ZoneTree.Analytics/Storage/StoreFactory.cs @@ -40,8 +40,7 @@ sealed class StoreFactory public long RawRetentionCutoff => retention.Current.RawEvents; - public long RollupRetentionCutoff(Granularity granularity) => granularity == Granularity.Day - ? retention.Current.DayRollupsLocalDay : retention.Current.Of(granularity); + public long RollupRetentionCutoff(Granularity granularity) => retention.Current.RollupCutoff(granularity); public StoreFactory(StorageOptions storage, string dataRoot, ILogger logger, RetentionClock retention) { @@ -163,7 +162,7 @@ ZoneTreeFactory, Memory> DimensionIndexFactory(string directo .SetComparer(new ByteArrayLexicographicComparerAscending()) .SetIsDeletedDelegate((in Memory key, in Memory value) => value.Length == 0 || expired(key.Span)) .SetIsRangeDeletedDelegate((in Memory first, in Memory _, in Memory last, in Memory _) => - DimensionIndexKeyCodec.IsRangeExpired(first.Span, last.Span, expired)) + ExpiredRecords.IsRangeExpired(first.Span, last.Span, DimensionIndexKeyCodec.HeadLength, expired)) .SetMarkValueDeletedDelegate((ref Memory value) => value = Memory.Empty); } @@ -184,7 +183,7 @@ ZoneTreeFactory, RollupValue> RollupIndexFactory(string directory, .SetValueSerializer(new RollupValueSerializer()) .SetIsDeletedDelegate((in Memory key, in RollupValue value) => value == null || expired(key.Span)) .SetIsRangeDeletedDelegate((in Memory first, in RollupValue _, in Memory last, in RollupValue _) => - RollupIndexKeyCodec.IsRangeExpired(first.Span, last.Span, expired)) + ExpiredRecords.IsRangeExpired(first.Span, last.Span, RollupIndexKeyCodec.HeadLength, expired)) .SetMarkValueDeletedDelegate((ref RollupValue value) => value = null); } @@ -206,7 +205,7 @@ ZoneTreeFactory, Memory> RawEventsFactory(string directory) .SetIsDeletedDelegate((in Memory key, in Memory value) => value.Length == 0 || isExpired(key.Span)) .SetIsRangeDeletedDelegate((in Memory first, in Memory _, in Memory last, in Memory _) => - ExpiredRecords.IsRangeExpired(first.Span, last.Span, 1, isExpired)) + ExpiredRecords.IsRangeExpired(first.Span, last.Span, KeyCodec.RawEventHead, isExpired)) .SetMarkValueDeletedDelegate((ref Memory value) => value = Memory.Empty); } @@ -225,7 +224,7 @@ ZoneTreeFactory, RollupValue> RollupsFactory(string directory, Gran .SetIsDeletedDelegate((in Memory key, in RollupValue value) => value == null || isExpired(key.Span)) .SetIsRangeDeletedDelegate((in Memory first, in RollupValue _, in Memory last, in RollupValue _) => - ExpiredRecords.IsRangeExpired(first.Span, last.Span, 3, isExpired)) + ExpiredRecords.IsRangeExpired(first.Span, last.Span, KeyCodec.RollupHead, isExpired)) .SetMarkValueDeletedDelegate((ref RollupValue value) => value = null); }