diff --git a/apps/ZoneTree.Analytics.Dashboard/ui/package.json b/apps/ZoneTree.Analytics.Dashboard/ui/package.json index 1750a63..ffa409f 100644 --- a/apps/ZoneTree.Analytics.Dashboard/ui/package.json +++ b/apps/ZoneTree.Analytics.Dashboard/ui/package.json @@ -1,6 +1,6 @@ { "name": "zonetree-analytics-dashboard", - "version": "0.1.0", + "version": "1.0.1", "private": true, "type": "module", "scripts": { diff --git a/docs-site/package.json b/docs-site/package.json index a18c4cb..59c8872 100644 --- a/docs-site/package.json +++ b/docs-site/package.json @@ -2,6 +2,7 @@ "private": true, "type": "module", "name": "analytics.zonetree.dev", + "version": "1.0.1", "scripts": { "check": "tsc --noEmit", "build": "purestack build --content ./content --clean", diff --git a/package.json b/package.json index 6cdc359..140f62f 100644 --- a/package.json +++ b/package.json @@ -11,6 +11,7 @@ "node": ">=22.17.0" }, "scripts": { + "bump-version": "tsx scripts/bump-version.ts", "check-dependencies": "tsx scripts/check-dependencies.ts && yarn dedupe --check", "check:scripts": "tsc --project scripts/tsconfig.json", "check:workspaces": "yarn workspaces foreach --all --exclude zonetree-analytics-repo run check", diff --git a/scripts/bump-version.test.ts b/scripts/bump-version.test.ts new file mode 100644 index 0000000..86e7e64 --- /dev/null +++ b/scripts/bump-version.test.ts @@ -0,0 +1,118 @@ +import assert from 'node:assert/strict' +import { spawnSync } from 'node:child_process' +import { + mkdirSync, + mkdtempSync, + readFileSync, + rmSync, + writeFileSync, +} from 'node:fs' +import path from 'node:path' +import { fileURLToPath } from 'node:url' +import { test } from 'vitest' + +const root = fileURLToPath(new URL('../', import.meta.url)) +const script = fileURLToPath(new URL('./bump-version.ts', import.meta.url)) +const loader = import.meta.resolve('tsx') + +const props = (versions: string) => + `\r\n \r\n${versions} false\r\n \r\n\r\n` + +const prefix = (version: string) => + ` ${version}\r\n` + +const manifest = (name: string, version: string) => + `{\n "name": "${name}",\n "version": "${version}",\n "private": true\n}\n` + +// The files the script edits, by their path relative to the repository root. +type Files = Record + +const versioned = (version: string, props: string = propsOf(version)) => ({ + 'src/Directory.Build.props': props, + 'apps/ZoneTree.Analytics.Dashboard/ui/package.json': manifest( + 'zonetree-analytics-dashboard', + version, + ), + 'docs-site/package.json': manifest('analytics.zonetree.dev', version), +}) + +function propsOf(version: string) { + return props(prefix(version)) +} + +function fixture( + sources: Files, + run: ( + read: () => Files, + bump: (...args: string[]) => ReturnType, + ) => void, +) { + const temporaryRoot = path.join(root, '.tmp') + mkdirSync(temporaryRoot, { recursive: true }) + const directory = mkdtempSync(path.join(temporaryRoot, 'bump-version-')) + try { + for (const [relativePath, source] of Object.entries(sources)) { + const file = path.join(directory, relativePath) + mkdirSync(path.dirname(file), { recursive: true }) + writeFileSync(file, source) + } + run( + () => + Object.fromEntries( + Object.keys(sources).map((relativePath) => [ + relativePath, + readFileSync(path.join(directory, relativePath), 'utf8'), + ]), + ), + (...args) => + spawnSync(process.execPath, ['--import', loader, script, ...args], { + cwd: directory, + encoding: 'utf8', + }), + ) + } finally { + assert.equal(path.dirname(directory), temporaryRoot) + assert(path.basename(directory).startsWith('bump-version-')) + rmSync(directory, { recursive: true, force: true }) + } +} + +test('bumps every version declaration and keeps the rest of each file', () => { + fixture(versioned('1.0.0'), (read, bump) => { + const first = bump() + assert.equal(first.status, 0, String(first.stderr)) + assert.deepEqual(read(), versioned('1.0.1')) + assert.match(String(first.stdout), /1\.0\.0 -> 1\.0\.1: 3 file\(s\)/) + + // A patch past 99 carries into the minor version. + const carried = bump('100') + assert.equal(carried.status, 0, String(carried.stderr)) + assert.deepEqual(read(), versioned('1.1.1')) + }) +}) + +test('changes nothing for an ambiguous or mismatched version or an invalid delta', () => { + const mismatched = { + ...versioned('1.0.0'), + 'docs-site/package.json': manifest('analytics.zonetree.dev', '0.1.0'), + } + for (const sources of [ + versioned('1.0.0', props(prefix('1.0.0') + prefix('1.0.0'))), + mismatched, + ]) { + fixture(sources, (read, bump) => { + const result = bump() + assert.notEqual(result.status, 0) + assert.match(String(result.stderr), /No files have been changed/) + assert.deepEqual(read(), sources) + }) + } + const single = versioned('1.0.0') + fixture(single, (read, bump) => { + for (const delta of ['0', '-1', 'two']) { + const result = bump(delta) + assert.notEqual(result.status, 0, delta) + assert.deepEqual(read(), single, delta) + } + }) +}) diff --git a/scripts/bump-version.ts b/scripts/bump-version.ts new file mode 100644 index 0000000..060df27 --- /dev/null +++ b/scripts/bump-version.ts @@ -0,0 +1,85 @@ +import { readFileSync, writeFileSync } from 'node:fs' +import path from 'node:path' +import { bumpVersion, versionPattern } from './version' + +// Run from the repository root: yarn bump-version [delta] +const projectRoot = process.cwd() +if (process.argv.length > 3) { + throw new Error('Usage: yarn bump-version [delta] (default: 1)') +} +const delta = process.argv[2] ?? '1' + +// Both NuGet packages share VersionPrefix; package and assembly versions derive from it. +const versionSource = 'src/Directory.Build.props' +const versionDeclaration = /()([^<\r\n]*)(<\/VersionPrefix>)/g + +// The private workspaces are not published, but carry the product version too. +const manifests = [ + 'apps/ZoneTree.Analytics.Dashboard/ui/package.json', + 'docs-site/package.json', +] +const manifestVersion = /^(\s*"version"\s*:\s*")([^"\r\n]+)(")/gm + +// Every file that declares the product version, with the pattern of its declaration. +const targets: { relativePath: string; pattern: RegExp }[] = [ + { relativePath: versionSource, pattern: versionDeclaration }, + ...manifests.map((relativePath) => ({ + relativePath, + pattern: manifestVersion, + })), +] + +function read(relativePath: string) { + return readFileSync(path.join(projectRoot, relativePath), 'utf8') +} + +function declarations(relativePath: string, source: string, pattern: RegExp) { + const matches = [...source.matchAll(pattern)] + if (matches.length !== 1) { + throw new Error( + `Expected exactly one version declaration in ${relativePath}. No files have been changed.`, + ) + } + return matches[0] +} + +const currentVersion = declarations( + versionSource, + read(versionSource), + versionDeclaration, +)[2] +if (!versionPattern.test(currentVersion)) { + throw new Error( + `Invalid version ${currentVersion} in ${versionSource}. No files have been changed.`, + ) +} +const nextVersion = bumpVersion(currentVersion, delta) + +type Update = { relativePath: string; source: string } +const updates: Update[] = [] +for (const { relativePath, pattern } of targets) { + const source = read(relativePath) + const [, , version] = declarations(relativePath, source, pattern) + if (version !== currentVersion) { + throw new Error( + `${relativePath} declares ${version}, but ${versionSource} declares ${currentVersion}. No files have been changed.`, + ) + } + const updated = source.replace( + pattern, + (_match, prefix: string, _version: string, suffix: string) => + `${prefix}${nextVersion}${suffix}`, + ) + if (updated !== source) updates.push({ relativePath, source: updated }) +} + +// Read and validate every target before writing; keep formatting and line endings. +for (const update of updates) { + writeFileSync( + path.join(projectRoot, update.relativePath), + update.source, + 'utf8', + ) + console.log(`Updated ${update.relativePath}`) +} +console.log(`${currentVersion} -> ${nextVersion}: ${updates.length} file(s)`) diff --git a/scripts/version.ts b/scripts/version.ts new file mode 100644 index 0000000..070b410 --- /dev/null +++ b/scripts/version.ts @@ -0,0 +1,23 @@ +export const versionPattern = /^(0|[1-9]\d*)\.(0|[1-9]\d*)\.(0|[1-9]\d?)$/ + +export function bumpVersion(version: string, delta = '1') { + const increment = Number(delta) + if (!/^\+?[1-9]\d*$/.test(delta) || !Number.isSafeInteger(increment)) { + throw new Error(`Expected a positive integer delta, received ${delta}.`) + } + + const current = versionPattern.exec(version) + if (!current) { + throw new Error( + `Expected a major.minor.patch version with patch 0–99, received ${version}.`, + ) + } + + const [, major, minor, patch] = current + const totalPatch = Number(patch) + increment + const nextMinor = Number(minor) + Math.floor(totalPatch / 100) + if (!Number.isSafeInteger(totalPatch) || !Number.isSafeInteger(nextMinor)) { + throw new Error('The resulting version exceeds the safe integer range.') + } + return `${major}.${nextMinor}.${totalPatch % 100}` +} diff --git a/src/Directory.Build.props b/src/Directory.Build.props index 1beab31..614ed86 100644 --- a/src/Directory.Build.props +++ b/src/Directory.Build.props @@ -12,7 +12,7 @@ matching ZoneTree.Analytics. Pass VersionSuffix (for example preview.1) to pack a pre-release. --> - 1.0.0 + 1.0.1 false Ahmed Yasin Koculu ZoneTree diff --git a/src/ZoneTree.Analytics.UnitTests/RollupValueLayoutTests.cs b/src/ZoneTree.Analytics.UnitTests/RollupValueLayoutTests.cs new file mode 100644 index 0000000..4766d33 --- /dev/null +++ b/src/ZoneTree.Analytics.UnitTests/RollupValueLayoutTests.cs @@ -0,0 +1,336 @@ +using System.Runtime.InteropServices; +using ZoneTree.Analytics.Ingestion; +using ZoneTree.Analytics.Internal; +using ZoneTree.Analytics.Sketches; +using ZoneTree.Analytics.Storage; + +namespace ZoneTree.Analytics.UnitTests; + +/// +/// The rollup value layout: scalars first, then length-prefixed sketch sections that are decoded +/// only when read, and readers that reject what they do not understand. +/// +public sealed class RollupValueLayoutTests +{ + static readonly RollupValueSerializer Serializer = new(); + + /// A row with a dense visitor sketch, percentiles and a distinct-value sketch. + static RollupValue HeavyRow(int visitors = 5_000) + { + var accumulator = new RollupAccumulator { Count = visitors }; + for (var i = 0; i < visitors; ++i) + { + accumulator.AddVisitor((ulong)i + 1, HyperLogLog.DefaultPrecision); + accumulator.AddValue("load_time", i); + accumulator.AddValueDetail("load_time", i, 1000 + i, ValueAggregates.All, QuantileSketch.BucketOf(i)); + accumulator.AddDistinct("page", Text.StableHash("/p" + i), HyperLogLog.DefaultPrecision); + } + return accumulator.ToValue(); + } + + static long AllocatedPerCall(Action action, int iterations = 200) + { + action(); + var before = GC.GetAllocatedBytesForCurrentThread(); + for (var i = 0; i < iterations; ++i) + action(); + return (GC.GetAllocatedBytesForCurrentThread() - before) / iterations; + } + + [Test] + public void QueriesDecodeOnlyTheSketchesTheyRead() + { + var bytes = Serializer.Serialize(HeavyRow()); + var scalarsOnly = new RollupReadRequirements(false, null, []); + // Before: every read decoded the 16 KiB visitor sketch, the percentiles and the distinct sketch. + var allocated = AllocatedPerCall(() => new RollupAccumulator().MergeForQuery(Serializer.Deserialize(bytes), scalarsOnly)); + Assert.That(allocated, Is.LessThan(1_024), $"a scalar read allocated {allocated:N0} bytes"); + + var group = new RollupAccumulator(); + group.MergeForQuery(Serializer.Deserialize(bytes), new RollupReadRequirements( + true, new() { ["load_time"] = ValueAggregates.All }, ["page"])); + Assert.Multiple(() => + { + Assert.That(group.Count, Is.EqualTo(5_000)); + Assert.That(group.HasData, Is.True); + Assert.That(group.Visitors.Estimate(), Is.EqualTo(5_000).Within(5_000 * 0.03)); + Assert.That(group.DistinctOf("page").Estimate(), Is.EqualTo(5_000).Within(5_000 * 0.03)); + Assert.That(group.DetailOf("load_time").Quantiles.Count, Is.EqualTo(5_000)); + Assert.That(group.DetailOf("load_time").First, Is.Zero); + Assert.That(group.DetailOf("load_time").Last, Is.EqualTo(4_999)); + }); + } + + static readonly string[] DistinctDimensions = ["browser", "country", "page", "source"]; + + static readonly string[] AggregatedValues = ["a", "b", "c"]; + + /// A row with dense distinct sketches of four dimensions and percentiles of three values. + static RollupValue WideRow() + { + var accumulator = new RollupAccumulator { Count = 5_000 }; + for (var i = 0; i < 5_000; ++i) + { + foreach (var value in AggregatedValues) + { + accumulator.AddValue(value, i); + accumulator.AddValueDetail(value, i, 1000 + i, ValueAggregates.Percentiles, QuantileSketch.BucketOf(i)); + } + foreach (var dimension in DistinctDimensions) + accumulator.AddDistinct(dimension, Text.StableHash(dimension + i), HyperLogLog.DefaultPrecision); + } + return accumulator.ToValue(); + } + + [Test] + public void QueriesDecodeOnlyTheDistinctSketchAndValueDetailTheyRead() + { + var bytes = Serializer.Serialize(WideRow()); + long Read(RollupReadRequirements requirements) => + AllocatedPerCall(() => new RollupAccumulator().MergeForQuery(Serializer.Deserialize(bytes), requirements), 50); + var oneDistinct = Read(new(false, null, ["page"])); + var allDistinct = Read(new(false, null, DistinctDimensions)); + var oneDetail = Read(new(false, new() { ["b"] = ValueAggregates.Percentiles }, [])); + var allDetails = Read(new(false, AggregatedValues.ToDictionary(x => x, _ => ValueAggregates.Percentiles), [])); + TestContext.Out.WriteLine($"distinct: one {oneDistinct:N0} B, all {allDistinct:N0} B; details: one {oneDetail:N0} B, all {allDetails:N0} B"); + Assert.Multiple(() => + { + // Before: reading one entry decoded every sketch of its section. + Assert.That(oneDistinct, Is.LessThan(allDistinct / 3), $"one distinct sketch {oneDistinct:N0} B, all {allDistinct:N0} B"); + Assert.That(oneDetail, Is.LessThan(allDetails / 2), $"one value detail {oneDetail:N0} B, all {allDetails:N0} B"); + }); + + var group = new RollupAccumulator(); + group.MergeForQuery(Serializer.Deserialize(bytes), new RollupReadRequirements( + false, new() { ["b"] = ValueAggregates.Percentiles }, ["page"])); + Assert.Multiple(() => + { + Assert.That(group.DistinctOf("page").Estimate(), Is.EqualTo(5_000).Within(5_000 * 0.03)); + Assert.That(group.DistinctOf("country"), Is.Null); + Assert.That(group.DetailOf("b").Quantiles.Count, Is.EqualTo(5_000)); + Assert.That(group.DetailOf("a"), Is.Null); + }); + } + + [Test] + public void MergesKeepUntouchedDetailsAndDistinctSketchesAsStoredBytes() + { + var bytes = Serializer.Serialize(WideRow()).ToArray(); + var stored = Serializer.Deserialize(bytes); + // An ingestion delta that updates the percentiles of "b" and the distinct pages only. + var delta = new RollupAccumulator { Count = 1 }; + delta.AddValue("b", 7); + delta.AddValueDetail("b", 7, 99, ValueAggregates.Percentiles, QuantileSketch.BucketOf(7)); + delta.AddDistinct("page", Text.StableHash("/new"), HyperLogLog.DefaultPrecision); + var merged = RollupValue.Merge(stored, delta.ToValue()); + Assert.Multiple(() => + { + for (var i = 0; i < AggregatedValues.Length; ++i) + { + var updated = AggregatedValues[i] == "b"; + Assert.That(ReferenceEquals(merged.DetailEntryAt(i), stored.DetailEntryAt(i)), Is.EqualTo(!updated), AggregatedValues[i]); + Assert.That(merged.DetailEntryAt(i).IsStored, Is.EqualTo(!updated), AggregatedValues[i]); + } + for (var i = 0; i < DistinctDimensions.Length; ++i) + { + var updated = DistinctDimensions[i] == "page"; + Assert.That(ReferenceEquals(merged.DistinctEntryAt(i), stored.DistinctEntryAt(i)), Is.EqualTo(!updated), DistinctDimensions[i]); + } + }); + + var restored = Serializer.Deserialize(Serializer.Serialize(merged)); + var original = Serializer.Deserialize(bytes); + Assert.Multiple(() => + { + Assert.That(restored.DetailAt(1).Quantiles.Count, Is.EqualTo(5_001)); + Assert.That(restored.DetailAt(0).Quantiles.Count, Is.EqualTo(5_000)); + Assert.That(restored.DistinctAt(2).Estimate(), Is.EqualTo(original.DistinctAt(2).Estimate() + 1).Within(50)); + Assert.That(restored.DistinctAt(0).Estimate(), Is.EqualTo(original.DistinctAt(0).Estimate())); + Assert.That(restored.Distinct, Has.Length.EqualTo(4)); + Assert.That(restored.Details, Has.Length.EqualTo(3)); + }); + } + + [Test] + public void EntryFramingIsValidated() + { + const byte live = 1, details = 4, distinct = 8; + // One value "x" without moments: name, sum and a count of zero. + byte[] value = [2, (byte)'x', .. new byte[8], 0]; + Assert.Multiple(() => + { + // A detail entry must hold data: a value without details has length zero. + Assert.Throws(() => _ = Serializer.Deserialize( + (byte[])[live | details, 0, 1, .. value, .. Section([1, .. Section(0)])]).DetailAt(0), "empty detail"); + // A distinct entry cannot be empty, and its length must fit the section. + Assert.Throws(() => _ = Serializer.Deserialize( + (byte[])[live | distinct, 0, 0, .. Section(1, 2, (byte)'d', 0, 0, 0, 0)]).DistinctNames, "empty distinct sketch"); + Assert.Throws(() => _ = Serializer.Deserialize( + (byte[])[live | distinct, 0, 0, .. Section(1, 2, (byte)'d', 9, 14, 1, 0)]).DistinctNames, "truncated distinct sketch"); + }); + // A value without details decodes as null. + Assert.That(Serializer.Deserialize((byte[])[live | details, 0, 1, .. value, .. Section(1, 0)]).DetailAt(0), Is.Null); + } + + [Test] + public void AScalarQueryStillSeesVisitorOnlyRows() + { + var accumulator = new RollupAccumulator(); + accumulator.AddVisitor(7, HyperLogLog.DefaultPrecision); + var group = new RollupAccumulator(); + group.MergeForQuery(Serializer.Deserialize(Serializer.Serialize(accumulator.ToValue())), new RollupReadRequirements(false, null, [])); + Assert.That(group.HasData, Is.True); + } + + [Test] + public void UndecodedSectionsArePassedThroughByteForByte() + { + var bytes = Serializer.Serialize(HeavyRow()).ToArray(); + // Merges and index copies re-serialize stored rows: an untouched section is copied, not re-encoded. + var reserialized = AllocatedPerCall(() => Serializer.Serialize(Serializer.Deserialize(bytes))); + Assert.That(Serializer.Serialize(Serializer.Deserialize(bytes)).ToArray(), Is.EqualTo(bytes)); + Assert.That(reserialized, Is.LessThan(bytes.Length + 2_048), $"re-serializing allocated {reserialized:N0} bytes"); + + // A delta without sketches keeps the stored sections. + var delta = new RollupValue(1, null, ["load_time"], [ValueStats.Of(7)]); + var merged = Serializer.Deserialize(Serializer.Serialize(RollupValue.Merge(Serializer.Deserialize(bytes), delta))); + var original = Serializer.Deserialize(bytes); + Assert.Multiple(() => + { + Assert.That(merged.Count, Is.EqualTo(5_001)); + Assert.That(merged.Values[0].Count, Is.EqualTo(5_001)); + Assert.That(merged.Visitors.Estimate(), Is.EqualTo(original.Visitors.Estimate())); + Assert.That(merged.Distinct[0].Estimate(), Is.EqualTo(original.Distinct[0].Estimate())); + Assert.That(merged.Details[0].Quantiles.Count, Is.EqualTo(5_000), "the delta added no detail"); + }); + } + + [Test] + public void RowsReadFromALargeBlockDoNotKeepTheBlockAlive() + { + var bytes = Serializer.Serialize(HeavyRow()).ToArray(); + // ZoneTree reads values from decompressed disk blocks of 4 MiB by default. + var block = new byte[4 * 1024 * 1024]; + bytes.CopyTo(block, 1_000); + var row = Serializer.Deserialize(block.AsMemory(1_000, bytes.Length)); + // A merged row keeps the sections of the stored row it shares. + var merged = RollupValue.Merge(row, new RollupValue(1, null, ["load_time"], [ValueStats.Of(1)])); + static int BackingLength(ReadOnlyMemory section) => + MemoryMarshal.TryGetArray(section, out var segment) ? segment.Array.Length : -1; + foreach (var value in new[] { row, merged }) + { + Assert.Multiple(() => + { + Assert.That(BackingLength(value.VisitorSection.Bytes), Is.InRange(1, bytes.Length)); + Assert.That(BackingLength(value.DetailSection.Bytes), Is.InRange(1, bytes.Length)); + Assert.That(BackingLength(value.DistinctSection.Bytes), Is.InRange(1, bytes.Length)); + }); + } + Assert.Multiple(() => + { + Assert.That(merged.Visitors.Estimate(), Is.EqualTo(5_000).Within(5_000 * 0.03)); + Assert.That(merged.Details[0].Quantiles.Count, Is.EqualTo(5_000)); + Assert.That(Serializer.Serialize(row).ToArray(), Is.EqualTo(bytes)); + }); + + // A row read from a buffer of its own exact size is used without a copy. + var exact = Serializer.Deserialize(bytes); + Assert.That(MemoryMarshal.TryGetArray(exact.VisitorSection.Bytes, out var own) && ReferenceEquals(own.Array, bytes), Is.True); + } + + [Test] + public void ConcurrentReadersShareOneDecodedSketch() + { + var row = Serializer.Deserialize(Serializer.Serialize(HeavyRow(1_000))); + var sketches = new HyperLogLog[64]; + Parallel.For(0, sketches.Length, i => sketches[i] = row.Visitors); + Assert.That(sketches, Is.All.SameAs(sketches[0])); + Assert.That(sketches[0].Estimate(), Is.EqualTo(1_000)); + } + + [TestCase(1)] + [TestCase(100)] + [TestCase(4_000)] + [TestCase(4_097)] + public void SketchesOfEverySizeRoundTrip(int visitors) + { + var row = HeavyRow(visitors); + var restored = Serializer.Deserialize(Serializer.Serialize(row)); + Assert.Multiple(() => + { + Assert.That(restored.Visitors.Estimate(), Is.EqualTo(row.Visitors.Estimate())); + Assert.That(restored.Visitors.IsSparse, Is.EqualTo(row.Visitors.IsSparse)); + Assert.That(restored.Distinct[0].Estimate(), Is.EqualTo(row.Distinct[0].Estimate())); + Assert.That(Serializer.Serialize(restored).ToArray(), Is.EqualTo(Serializer.Serialize(row).ToArray())); + }); + } + + [Test] + public void RollupReaderRejectsWhatItDoesNotUnderstand() + { + var bytes = Serializer.Serialize(HeavyRow(10)).ToArray(); + var unknownBit = bytes.ToArray(); + unknownBit[0] |= 0x10; + var notPresent = bytes.ToArray(); + notPresent[0] &= 0xFE; + Assert.Multiple(() => + { + Assert.Throws(() => Serializer.Deserialize(unknownBit), "unknown section"); + Assert.Throws(() => Serializer.Deserialize(notPresent), "sections of a row without the present bit"); + Assert.Throws(() => Serializer.Deserialize(bytes.Append((byte)0).ToArray()), "trailing bytes"); + Assert.Throws(() => Serializer.Deserialize(bytes.AsMemory(0, bytes.Length - 1)), "a truncated section"); + }); + } + + // int.MaxValue as a varint: an array of that many elements exceeds the supported size. + static readonly byte[] HugeCount = [0xFF, 0xFF, 0xFF, 0xFF, 0x07]; + + static byte[] Section(params byte[] payload) => [(byte)payload.Length, .. payload]; + + [Test] + public void CorruptCountsAreFormatErrorsNotAllocations() + { + const byte live = 1, visitors = 2, distinct = 8; + // count 0, then a value count that the remaining bytes cannot hold. + byte[] values = [live, 0, .. HugeCount]; + // A sparse visitor sketch (precision 14) and a distinct section with an impossible count. + byte[] sparse = [live | visitors, 0, 0, .. Section([14, 1, .. HugeCount])]; + byte[] dimensions = [live | distinct, 0, 0, .. Section(HugeCount)]; + Assert.Multiple(() => + { + Assert.Throws(() => Serializer.Deserialize(values), "value count"); + Assert.Throws(() => _ = Serializer.Deserialize(sparse).Visitors, "sparse entry count"); + Assert.Throws(() => _ = Serializer.Deserialize(dimensions).Distinct, "distinct count"); + }); + } + + [Test] + public void ASparseDeltaWithOverflowingBitsIsRejected() + { + // One entry whose five-byte delta sets bits above 32 in its last byte. + byte[] row = [1 | 2, 0, 0, .. Section([14, 1, 1, 0xFF, 0xFF, 0xFF, 0xFF, 0x7F])]; + Assert.Throws(() => _ = Serializer.Deserialize(row).Visitors); + // The largest valid delta still decodes. + byte[] largest = [1 | 2, 0, 0, .. Section([14, 1, 1, 0xFF, 0xFF, 0xFF, 0xFF, 0x0F])]; + Assert.That(Serializer.Deserialize(largest).Visitors.Estimate(), Is.EqualTo(1)); + } + + [Test] + public void RawAndSessionReadersRejectWhatTheyDoNotUnderstand() + { + var raw = RawEventCodec.Encode("e", 1000, 1, 0, 0, false, [], []).ToArray(); + var flagsAt = 1 + 1 + 8 + 8 + 1; + raw[flagsAt] = 0x02; + Assert.Throws(() => RawEventCodec.TryDecode(raw, 1000, null, default, new RawEvent()), "unknown raw flag"); + + var sessions = new SessionStateSerializer(SessionSchema.Standard); + var session = sessions.Serialize(new SessionState { SessionId = 1 }).ToArray(); + var unknownPresence = session.ToArray(); + unknownPresence[0] = 2; + Assert.Multiple(() => + { + Assert.Throws(() => sessions.Deserialize(unknownPresence), "unknown session header"); + Assert.Throws(() => sessions.Deserialize(session.Append((byte)0).ToArray()), "trailing session bytes"); + }); + } +} diff --git a/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs b/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs index 60d0d7e..4ca15fd 100644 --- a/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs +++ b/src/ZoneTree.Analytics.UnitTests/StorageFormatRegressionTests.cs @@ -35,6 +35,23 @@ static double RawPageViews(AnalyticsEngine engine, DateTime from, DateTime to) = Range = Range(from, to), Source = QuerySource.RawEvents, Metrics = { Metric.PageViews } }).Totals[Metric.PageViews]; + [TestCase(StorageFormats.DatasetKey, "1", "older release")] + [TestCase(StorageFormats.DatasetKey, "3", "newer release")] + [TestCase(StorageFormats.DatasetKey, "two", "unknown storage format")] + [TestCase(StorageFormats.DimensionIndexKey, "1", "older release")] + [TestCase(StorageFormats.DimensionIndexKey, "3", "newer release")] + [TestCase(StorageFormats.RollupIndexKey, "1", "older release")] + [TestCase(StorageFormats.RollupIndexKey, "two", "unknown storage format")] + public void DirectoriesOfOtherStorageFormatsAreRefused(string key, string stored, string explanation) + { + using (TestEngine.Create(directory)) { } + var meta = new MetaStore(Path.Combine(directory, "meta")); + Assert.That(meta.Get(key), Is.Not.Null, "every format is persisted, even for unused indexes"); + meta.Set(key, stored); + var error = Assert.Throws(() => TestEngine.Create(directory).Dispose()); + Assert.That(error.Message, Does.Contain(explanation)); + } + [Test] public void RolloverNeverHidesRecordsWrittenAfterItsClock() { @@ -265,6 +282,23 @@ public void JointBackfillJournalHoldsAFlushOfDenseRows() Assert.That(redo[^1].Value, Is.EqualTo(value)); } + [Test] + public void AJointBackfillJournalOfAnOlderReleaseIsRefused() + { + // A valid journal file whose body starts with an older version. + var body = BitConverter.GetBytes(1); + var header = new byte[56]; + "ZTAJRNL1"u8.CopyTo(header); + BitConverter.GetBytes(1L).CopyTo(header, 8); + BitConverter.GetBytes((long)body.Length).CopyTo(header, 16); + System.Security.Cryptography.SHA256.HashData(body).CopyTo(header, 24); + var journal = Path.Combine(directory, "backfill", "joint-rollups"); + Directory.CreateDirectory(journal); + File.WriteAllBytes(Path.Combine(journal, "journal-b.bin"), [.. header, .. body]); + var error = Assert.Throws(() => _ = new JointBackfillCheckpoint(directory, "configuration", 0)); + Assert.That(error.Message, Does.Contain("older release")); + } + [Test] public void ATornJointBackfillJournalFallsBackToThePreviousCheckpoint() { diff --git a/src/ZoneTree.Analytics/AnalyticsEngine.cs b/src/ZoneTree.Analytics/AnalyticsEngine.cs index 41f8bd7..b16713b 100644 --- a/src/ZoneTree.Analytics/AnalyticsEngine.cs +++ b/src/ZoneTree.Analytics/AnalyticsEngine.cs @@ -38,8 +38,6 @@ namespace ZoneTree.Analytics; /// public sealed class AnalyticsEngine : IAnalyticsTracker, IDisposable, IAsyncDisposable { - const string FormatVersion = "1"; - readonly AnalyticsOptions options; readonly ILogger logger; @@ -316,17 +314,17 @@ bool ValidateConfiguration() ("hll-precision", options.Rollups.HyperLogLogPrecision.ToString(CultureInfo.InvariantCulture), "HyperLogLog precision"), ("wal-directory", walDirectory, "write-ahead-log directory"), ]; - if (meta.Get("format-version") == null) + if (meta.Get(StorageFormats.DatasetKey) is not { } format) { meta.Update(entries => { foreach (var (key, value, _) in settings) entries[key] = value; - entries["format-version"] = FormatVersion; + entries[StorageFormats.DatasetKey] = StorageFormats.Text(StorageFormats.Dataset); }); return true; } - Require("format-version", FormatVersion, "format version"); + StorageFormats.Require(format, StorageFormats.Dataset, "The analytics data directory"); foreach (var (key, value, description) in settings) Require(key, value, description); return false; diff --git a/src/ZoneTree.Analytics/Internal/ByteWriter.cs b/src/ZoneTree.Analytics/Internal/ByteWriter.cs index 0207029..6d5618d 100644 --- a/src/ZoneTree.Analytics/Internal/ByteWriter.cs +++ b/src/ZoneTree.Analytics/Internal/ByteWriter.cs @@ -33,6 +33,9 @@ public static ByteWriter Reuse(ref ByteWriter cached, int capacity) public byte[] ToArray() => buffer.AsSpan(0, length).ToArray(); + /// The bytes written since the last reset. + public ReadOnlySpan Written => buffer.AsSpan(0, length); + void Ensure(int additional) { var required = length + additional; @@ -140,6 +143,9 @@ public ByteReader(ReadOnlySpan span, bool strict = false) public readonly int Remaining => span.Length - position; + /// The bytes not read yet. + public readonly ReadOnlySpan Rest => span[position..]; + public byte ReadByte() => span[position++]; public ReadOnlySpan ReadBytes(int count) diff --git a/src/ZoneTree.Analytics/Sketches/HyperLogLog.cs b/src/ZoneTree.Analytics/Sketches/HyperLogLog.cs index b2dee2f..5d93faa 100644 --- a/src/ZoneTree.Analytics/Sketches/HyperLogLog.cs +++ b/src/ZoneTree.Analytics/Sketches/HyperLogLog.cs @@ -375,14 +375,34 @@ internal static HyperLogLog Read(ref ByteReader reader) } if (format != SparseFormat) throw new FormatException("Unknown HyperLogLog format."); - var count = (int)reader.ReadVarUInt64(); + var count = checked((int)reader.ReadVarUInt64()); + // Every entry takes at least one byte: a larger count is corrupt, not an allocation to try. + if (count > reader.Remaining) + throw new FormatException("The HyperLogLog entry count exceeds its bytes."); var entries = new uint[count]; + // Deltas are at most five bytes: decoding them in a local loop keeps the position in a register. + var bytes = reader.Rest; + var position = 0; uint previous = 0; for (var i = 0; i < count; ++i) { - previous += (uint)reader.ReadVarUInt64(); + uint delta = 0; + var shift = 0; + byte b; + do + { + b = bytes[position++]; + // A fifth byte carries the top four bits of a 32-bit delta and ends it. + if (shift == 28 && b > 0x0F) + throw new FormatException("Invalid HyperLogLog entry."); + delta |= (uint)(b & 0x7F) << shift; + shift += 7; + } + while ((b & 0x80) != 0); + previous += delta; entries[i] = previous; } + reader.ReadBytes(position); return new HyperLogLog(precision, entries, count, null); } diff --git a/src/ZoneTree.Analytics/Storage/DimensionIndexStore.cs b/src/ZoneTree.Analytics/Storage/DimensionIndexStore.cs index 1afc5cc..128ac7a 100644 --- a/src/ZoneTree.Analytics/Storage/DimensionIndexStore.cs +++ b/src/ZoneTree.Analytics/Storage/DimensionIndexStore.cs @@ -21,7 +21,6 @@ sealed class DimensionIndexStore : IDisposable { const string DimensionsKey = "dimension-index.dimensions"; const string CleanKey = "dimension-index.clean"; - const string Format = "2"; const string PartitionsKey = "dimension-index.partitions.shared"; const int ProbeLimit = 1024; readonly MetaStore meta; @@ -62,12 +61,17 @@ public void Initialize() { using var startup = new StartupLog(logger, "Dimension index initialization", directory); using var check = new StartupLog(logger, "Dimension index backfill check", directory); - if (meta.Get("dimension-index.format") is { } format && format != Format) - throw new AnalyticsConfigurationException("Dimension indexes use a new shared storage format. Start with a fresh analytics data directory."); + var format = meta.Get(StorageFormats.DimensionIndexKey); + if (format != null) + StorageFormats.Require(format, StorageFormats.DimensionIndex, "The dimension indexes of the analytics data directory"); var reasons = new List(); - if (meta.Get(CleanKey) != "true") reasons.Add("missing or unclean checkpoint"); - if (meta.Get(DimensionsKey) != manifest) reasons.Add("configuration changed"); - if (meta.Get("dimension-index.format") != Format) reasons.Add("format changed"); + // Indexes that were never initialized have no checkpoint, manifest or format to compare. + if (format == null) reasons.Add("new indexes"); + else + { + if (meta.Get(CleanKey) != "true") reasons.Add("missing or unclean checkpoint"); + if (meta.Get(DimensionsKey) != manifest) reasons.Add("configuration changed"); + } if (!factory.RawEventsDurable || !factory.DimensionIndexesDurable) reasons.Add("write-ahead logging disabled"); var dirtyDays = recovery.Days; var partitions = 0; @@ -279,7 +283,7 @@ public void MarkClean() meta.Update(entries => { entries[DimensionsKey] = manifest; - entries["dimension-index.format"] = Format; + entries[StorageFormats.DimensionIndexKey] = StorageFormats.Text(StorageFormats.DimensionIndex); entries[PartitionsKey] = Store == null ? "" : string.Join(',', Store.ListPartitions()); entries[CleanKey] = factory.RawEventsDurable && factory.DimensionIndexesDurable ? "true" : "false"; recovery.Clear(entries); diff --git a/src/ZoneTree.Analytics/Storage/JointBackfillCheckpoint.cs b/src/ZoneTree.Analytics/Storage/JointBackfillCheckpoint.cs index 40671ac..fa1c842 100644 --- a/src/ZoneTree.Analytics/Storage/JointBackfillCheckpoint.cs +++ b/src/ZoneTree.Analytics/Storage/JointBackfillCheckpoint.cs @@ -46,8 +46,6 @@ sealed class JointBackfillReplacement /// sealed class JointBackfillCheckpoint { - const int Version = 1; - const int HeaderLength = 56; static readonly byte[] Magic = "ZTAJRNL1"u8.ToArray(); @@ -157,7 +155,7 @@ static bool TryReadHeader(FileStream stream, out long fileGeneration) static void Write(BinaryWriter writer, JointBackfillState state) { - writer.Write(Version); + writer.Write(StorageFormats.JointBackfillJournal); writer.Write(state.Configuration); writer.Write(state.Copying); writer.Write(state.Coverage); @@ -209,8 +207,7 @@ static void Write(BinaryWriter writer, JointBackfillState state) static JointBackfillState Read(BinaryReader reader) { - if (reader.ReadInt32() != Version) - throw new AnalyticsConfigurationException("The joint rollup backfill journal has an unknown version."); + StorageFormats.Require(reader.ReadInt32(), StorageFormats.JointBackfillJournal, "The joint rollup backfill journal"); var state = new JointBackfillState { Configuration = reader.ReadString(), diff --git a/src/ZoneTree.Analytics/Storage/RawEventCodec.cs b/src/ZoneTree.Analytics/Storage/RawEventCodec.cs index b0c3ebe..823fa7a 100644 --- a/src/ZoneTree.Analytics/Storage/RawEventCodec.cs +++ b/src/ZoneTree.Analytics/Storage/RawEventCodec.cs @@ -296,11 +296,14 @@ internal static bool TryDecode( var sessionOffset = reader.ReadVarInt64(); target.SessionStartTicks = checked(ticks - sessionOffset); var flags = reader.ReadByte(); + // A flag this reader does not know is a layout it cannot read. + if ((flags & ~InteractiveFlag) != 0) + throw new FormatException("The raw event has an unknown layout."); target.Interactive = (flags & InteractiveFlag) != 0; - if (requireEnd && (string.IsNullOrWhiteSpace(target.Name) || target.Name.Any(char.IsControl) || (flags & ~InteractiveFlag) != 0 || + if (requireEnd && (string.IsNullOrWhiteSpace(target.Name) || target.Name.Any(char.IsControl) || (target.SessionId == 0 && sessionOffset != 0) || target.SessionStartTicks < 0 || target.SessionStartTicks > DateTime.MaxValue.Ticks)) - throw new FormatException("The raw event header contains an invalid name, session start or flags."); + throw new FormatException("The raw event header contains an invalid name or session start."); var dimensionCount = checked((int)reader.ReadVarUInt64()); for (var i = 0; i < dimensionCount; ++i) { diff --git a/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs b/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs index 913d671..806bab2 100644 --- a/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs +++ b/src/ZoneTree.Analytics/Storage/RollupIndexStore.cs @@ -13,7 +13,6 @@ sealed record RollupIndex(string Dimension, string DefinitionId, Granularity Gra sealed class RollupIndexStore : IDisposable { const string CleanKey = "rollup-index.clean"; - const string Format = "2"; readonly string root; readonly AnalyticsOptions options; readonly RollupStore rollups; @@ -51,17 +50,22 @@ public void Initialize(RawEventStore rawEvents, RollupActivations activations, R { using var startup = new StartupLog(logger, "Rollup index initialization", root); using var check = new StartupLog(logger, "Rollup index backfill check", root); - if (meta.Get("rollup-index.format") is { } format && format != Format) - throw new AnalyticsConfigurationException("Rollup indexes use a new shared storage format. Start with a fresh analytics data directory."); + var format = meta.Get(StorageFormats.RollupIndexKey); + if (format != null) + StorageFormats.Require(format, StorageFormats.RollupIndex, "The rollup indexes of the analytics data directory"); var configured = options.Storage.RollupIndexes.Definitions.SelectMany(definition => RollupStore.Granularities .Where(granularity => rollups.IsEnabled(granularity)) .Select(granularity => Registration(definition.Dimension, definition.Definition.Id, granularity))) .Order(StringComparer.Ordinal).ToArray(); manifest = string.Join('\n', configured); var reasons = new List(); - if (meta.Get(CleanKey) != "true") reasons.Add("missing or unclean checkpoint"); - if (meta.Get("rollup-index.config") != manifest) reasons.Add("configuration changed"); - if (meta.Get("rollup-index.format") != Format) reasons.Add("format changed"); + // Indexes that were never initialized have no checkpoint, manifest or format to compare. + if (format == null) reasons.Add("new indexes"); + else + { + if (meta.Get(CleanKey) != "true") reasons.Add("missing or unclean checkpoint"); + if (meta.Get("rollup-index.config") != manifest) reasons.Add("configuration changed"); + } if (!factory.RollupsDurable || !factory.RollupIndexesDurable) reasons.Add("write-ahead logging disabled"); var partitions = 0; foreach (var granularity in RollupStore.Granularities.Where(granularity => @@ -297,7 +301,7 @@ public void MarkClean() meta.Update(entries => { entries[CleanKey] = factory.RollupsDurable && factory.RollupIndexesDurable ? "true" : "false"; - entries["rollup-index.format"] = Format; + entries[StorageFormats.RollupIndexKey] = StorageFormats.Text(StorageFormats.RollupIndex); entries["rollup-index.config"] = manifest; foreach (var (granularity, store) in stores) entries[PartitionsKey(granularity)] = string.Join(',', store.ListPartitions()); foreach (var journal in recovery.Values) journal.Clear(entries); diff --git a/src/ZoneTree.Analytics/Storage/RollupValue.cs b/src/ZoneTree.Analytics/Storage/RollupValue.cs index d08e2c0..d98a3ec 100644 --- a/src/ZoneTree.Analytics/Storage/RollupValue.cs +++ b/src/ZoneTree.Analytics/Storage/RollupValue.cs @@ -1,3 +1,4 @@ +using System.Runtime.InteropServices; using ZoneTree.Analytics.Internal; using ZoneTree.Analytics.Sketches; using ZoneTree.Serializers; @@ -213,29 +214,134 @@ internal static ValueDetail Read(ref ByteReader reader) } } +/// +/// A part of a stored row that is kept as its stored bytes until it is read. Concurrent +/// readers share one decoded instance. +/// +sealed class RollupSection where T : class +{ + readonly Func, T> decode; + + T value; + + /// A section created from a value. + public RollupSection(T value) => this.value = value; + + /// A stored section, decoded on first read. + public RollupSection(ReadOnlyMemory bytes, Func, T> decode) + { + Bytes = bytes; + this.decode = decode; + } + + /// The stored encoding; empty for a section created from a value. + public ReadOnlyMemory Bytes { get; } + + public bool IsStored => !Bytes.IsEmpty; + + public T Value + { + get + { + var current = Volatile.Read(ref value); + if (current != null) + return current; + var decoded = decode(Bytes); + return Interlocked.CompareExchange(ref value, decoded, null) ?? decoded; + } + } +} + +/// +/// The value details of a row, aligned with its value names: one entry per value, null for a +/// value without details. Each entry stays in its stored bytes until it is read. +/// +sealed class DetailEntries(RollupSection[] entries) +{ + public readonly RollupSection[] Entries = entries; + + ValueDetail[] all; + + public bool HasAny => Array.Exists(Entries, x => x != null); + + /// The detail of a value, decoding only that detail. + public ValueDetail At(int index) => Entries[index]?.Value; + + /// Every detail, decoded once. + public ValueDetail[] All + { + get + { + var current = Volatile.Read(ref all); + if (current != null) + return current; + var decoded = Array.ConvertAll(Entries, x => x?.Value); + return Interlocked.CompareExchange(ref all, decoded, null) ?? decoded; + } + } + + /// The entries of details created in memory; a detail without data has none. + public static DetailEntries Of(ValueDetail[] details) => new(Array.ConvertAll(details, + x => x != null && (x.HasFirstLast || x.Quantiles != null) ? new RollupSection(x) : null)); +} + +/// +/// The distinct-value sketches of a row, by dimension in ordinal order. Each sketch stays in its +/// stored bytes until it is read. +/// +sealed class DistinctEntries(string[] names, RollupSection[] sketches) +{ + public readonly string[] Names = names; + + public readonly RollupSection[] Sketches = sketches; + + HyperLogLog[] all; + + /// Every sketch, decoded once. + public HyperLogLog[] All + { + get + { + var current = Volatile.Read(ref all); + if (current != null) + return current; + var decoded = Array.ConvertAll(Sketches, x => x.Value); + return Interlocked.CompareExchange(ref all, decoded, null) ?? decoded; + } + } + + /// The entries of sketches created in memory. + public static DistinctEntries Of(string[] names, HyperLogLog[] sketches) => + new(names, Array.ConvertAll(sketches, x => new RollupSection(x))); +} + /// /// A stored rollup row. Treat instances, their arrays and sketches as immutable: /// merges create new instances, so values held by ZoneTree segments are never modified. /// +/// +/// The count and the scalar statistics are decoded with the row. The visitor sketch, each value +/// detail and each distinct sketch stay in their stored bytes until they are read: most queries +/// need few of them, and merges or index copies that leave them unchanged write the stored bytes +/// back without decoding them. +/// public sealed class RollupValue { /// The number of events or session deltas represented by this row. public readonly long Count; - /// The visitor sketch, or null when the row has no visitors. - public readonly HyperLogLog Visitors; - /// The measured value names, in ordinal order. public readonly string[] ValueNames; /// The measured statistics, aligned with . public readonly ValueStats[] Values; - // The opt-in parts, null for the rows of values without them: queries read many rows, so the - // row itself stays small. - readonly OptInParts optIn; + // The sections, null for rows without them: queries read many rows, so the row stays small. + readonly RollupSection visitors; + + readonly RollupSection details; - sealed record OptInParts(ValueDetail[] Details, string[] DistinctNames, HyperLogLog[] Distinct); + readonly RollupSection distinct; /// Creates a stored row. Arrays and sketches are retained without copying. public RollupValue( @@ -246,79 +352,112 @@ public RollupValue( ValueDetail[] details = null, string[] distinctNames = null, HyperLogLog[] distinct = null) + : this( + count, + visitors == null ? null : new(visitors), + valueNames, + values, + details == null ? null : new(DetailEntries.Of(details)), + distinctNames is { Length: > 0 } ? new(DistinctEntries.Of(distinctNames, distinct ?? [])) : null) + { + } + + internal RollupValue( + long count, + RollupSection visitors, + string[] valueNames, + ValueStats[] values, + RollupSection details, + RollupSection distinct) { Count = count; - Visitors = visitors; ValueNames = valueNames ?? []; Values = values ?? []; - if (details != null || distinctNames is { Length: > 0 }) - optIn = new OptInParts(details, distinctNames ?? [], distinct ?? []); + this.visitors = visitors; + this.details = details; + this.distinct = distinct; } + /// The visitor sketch, or null when the row has no visitors. + public HyperLogLog Visitors => visitors?.Value; + + /// Whether the row has a visitor sketch with visitors, without decoding it. + public bool HasVisitors => visitors != null && (visitors.IsStored || !visitors.Value.IsEmpty); + /// /// Opt-in aggregates per value, aligned with ; null when no value has any. + /// Decodes every detail; decodes one. /// - public ValueDetail[] Details => optIn?.Details; + public ValueDetail[] Details => details?.Value.All; /// /// The dimensions whose distinct values are counted, in ordinal order. /// - public string[] DistinctNames => optIn?.DistinctNames ?? []; + public string[] DistinctNames => distinct?.Value.Names ?? []; /// - /// Distinct value sketches, aligned with . + /// Distinct value sketches, aligned with . Decodes every sketch; + /// decodes one. /// - public HyperLogLog[] Distinct => optIn?.Distinct ?? []; + public HyperLogLog[] Distinct => distinct?.Value.All ?? []; /// The opt-in detail at a value index, or null when no details were collected. - public ValueDetail DetailAt(int index) => Details?[index]; + public ValueDetail DetailAt(int index) => details?.Value.At(index); + + /// The distinct value sketch at an index of . + public HyperLogLog DistinctAt(int index) => distinct.Value.Sketches[index].Value; + + internal RollupSection VisitorSection => HasVisitors ? visitors : null; + + internal RollupSection DetailSection => + details != null && (details.IsStored || details.Value.HasAny) ? details : null; + + internal RollupSection DistinctSection => + distinct != null && (distinct.IsStored || distinct.Value.Names.Length > 0) ? distinct : null; + + internal RollupSection DetailEntryAt(int index) => details?.Value.Entries[index]; + + internal RollupSection DistinctEntryAt(int index) => distinct.Value.Sketches[index]; /// Merges two rows without modifying either row, its arrays or sketches. + /// + /// A section, detail or distinct sketch only one row has is shared, so its stored bytes are + /// written back as they are. + /// public static RollupValue Merge(RollupValue existing, RollupValue delta) { if (existing == null) return delta; if (delta == null) return existing; - var visitors = Union(existing.Visitors, delta.Visitors); + var visitors = !delta.HasVisitors ? existing.visitors : + !existing.HasVisitors ? delta.visitors : + new RollupSection(Union(existing.Visitors, delta.Visitors)); string[] names; ValueStats[] values; - ValueDetail[] details; + RollupSection details; if (delta.ValueNames.Length == 0) { names = existing.ValueNames; values = existing.Values; - details = existing.Details; + details = existing.details; } else if (existing.ValueNames.Length == 0) { names = delta.ValueNames; values = delta.Values; - details = delta.Details; + details = delta.details; } else { MergeValues(existing, delta, out names, out values, out details); } - string[] distinctNames; - HyperLogLog[] distinct; - if (delta.DistinctNames.Length == 0) - { - distinctNames = existing.DistinctNames; - distinct = existing.Distinct; - } - else if (existing.DistinctNames.Length == 0) - { - distinctNames = delta.DistinctNames; - distinct = delta.Distinct; - } - else - { - MergeDistinct(existing, delta, out distinctNames, out distinct); - } - return new RollupValue(existing.Count + delta.Count, visitors, names, values, details, distinctNames, distinct); + var distinct = delta.DistinctSection == null ? existing.distinct : + existing.DistinctSection == null ? delta.distinct : + new RollupSection(MergeDistinct(existing, delta)); + return new RollupValue(existing.Count + delta.Count, visitors, names, values, details, distinct); } /// @@ -335,13 +474,27 @@ static HyperLogLog Union(HyperLogLog a, HyperLogLog b) return union; } + /// The entry of two merged details: one row's entry as it is, or the merged details. + static RollupSection MergeEntries(RollupSection a, RollupSection b) => + a == null ? b : b == null ? a : new(ValueDetail.Combine(a.Value, b.Value)); + static void MergeValues( - RollupValue a, RollupValue b, out string[] names, out ValueStats[] values, out ValueDetail[] details) + RollupValue a, RollupValue b, out string[] names, out ValueStats[] values, out RollupSection details) { + // Rows of the same values keep the details of the only row that has them. + if (a.ValueNames.AsSpan().SequenceEqual(b.ValueNames) && (a.details == null || b.details == null)) + { + names = a.ValueNames; + values = new ValueStats[names.Length]; + for (var k = 0; k < values.Length; ++k) + values[k] = ValueStats.Combine(a.Values[k], b.Values[k]); + details = a.details ?? b.details; + return; + } var capacity = a.ValueNames.Length + b.ValueNames.Length; var resultNames = new List(capacity); var resultValues = new List(capacity); - var resultDetails = a.Details != null || b.Details != null ? new List(capacity) : null; + var resultDetails = a.details != null || b.details != null ? new List>(capacity) : null; int i = 0, j = 0; while (i < a.ValueNames.Length || j < b.ValueNames.Length) { @@ -352,58 +505,62 @@ static void MergeValues( { resultNames.Add(a.ValueNames[i]); resultValues.Add(a.Values[i]); - resultDetails?.Add(a.DetailAt(i)); + resultDetails?.Add(a.DetailEntryAt(i)); ++i; } else if (cmp > 0) { resultNames.Add(b.ValueNames[j]); resultValues.Add(b.Values[j]); - resultDetails?.Add(b.DetailAt(j)); + resultDetails?.Add(b.DetailEntryAt(j)); ++j; } else { resultNames.Add(a.ValueNames[i]); resultValues.Add(ValueStats.Combine(a.Values[i], b.Values[j])); - resultDetails?.Add(ValueDetail.Combine(a.DetailAt(i), b.DetailAt(j))); + resultDetails?.Add(MergeEntries(a.DetailEntryAt(i), b.DetailEntryAt(j))); ++i; ++j; } } names = [.. resultNames]; values = [.. resultValues]; - details = resultDetails?.ToArray(); + details = resultDetails == null ? null : new(new DetailEntries([.. resultDetails])); } - static void MergeDistinct(RollupValue a, RollupValue b, out string[] names, out HyperLogLog[] sketches) + static DistinctEntries MergeDistinct(RollupValue a, RollupValue b) { - var resultNames = new List(a.DistinctNames.Length + b.DistinctNames.Length); - var resultSketches = new List(resultNames.Capacity); + var aNames = a.DistinctNames; + var bNames = b.DistinctNames; + var resultNames = new List(aNames.Length + bNames.Length); + var resultSketches = new List>(resultNames.Capacity); int i = 0, j = 0; - while (i < a.DistinctNames.Length || j < b.DistinctNames.Length) + while (i < aNames.Length || j < bNames.Length) { - var cmp = i == a.DistinctNames.Length ? 1 : - j == b.DistinctNames.Length ? -1 : - string.CompareOrdinal(a.DistinctNames[i], b.DistinctNames[j]); + var cmp = i == aNames.Length ? 1 : + j == bNames.Length ? -1 : + string.CompareOrdinal(aNames[i], bNames[j]); if (cmp < 0) { - resultNames.Add(a.DistinctNames[i]); - resultSketches.Add(a.Distinct[i++]); + resultNames.Add(aNames[i]); + resultSketches.Add(a.DistinctEntryAt(i++)); } else if (cmp > 0) { - resultNames.Add(b.DistinctNames[j]); - resultSketches.Add(b.Distinct[j++]); + resultNames.Add(bNames[j]); + resultSketches.Add(b.DistinctEntryAt(j++)); } else { - resultNames.Add(a.DistinctNames[i]); - resultSketches.Add(Union(a.Distinct[i++], b.Distinct[j++])); + resultNames.Add(aNames[i]); + var stored = a.DistinctEntryAt(i++); + var union = Union(stored.Value, b.DistinctAt(j++)); + // A union that adds nothing keeps the stored entry and its bytes. + resultSketches.Add(ReferenceEquals(union, stored.Value) ? stored : new(union)); } } - names = [.. resultNames]; - sketches = [.. resultSketches]; + return new DistinctEntries([.. resultNames], [.. resultSketches]); } } @@ -641,7 +798,7 @@ public void Merge(RollupValue value) } var distinctNames = value.DistinctNames; for (var i = 0; i < distinctNames.Length; ++i) - MergeDistinct(distinctNames[i], value.Distinct[i]); + MergeDistinct(distinctNames[i], value.DistinctAt(i)); } /// Merges only the sketches and details a query reads, preserving scalar row presence. @@ -653,7 +810,7 @@ public void MergeForQuery(RollupValue value, RollupReadRequirements requirements if (requirements.Visitors) MergeVisitors(value.Visitors); else - hasSkippedVisitors |= value.Visitors != null && !value.Visitors.IsEmpty; + hasSkippedVisitors |= value.HasVisitors; MergeValues(value); if (requirements.ValueDetails != null) { @@ -668,7 +825,7 @@ public void MergeForQuery(RollupValue value, RollupReadRequirements requirements { var index = Array.IndexOf(value.DistinctNames, dimension); if (index >= 0) - MergeDistinct(dimension, value.Distinct[index]); + MergeDistinct(dimension, value.DistinctAt(index)); } } @@ -829,12 +986,15 @@ public RollupValue ToValue() /// /// /// header(1: 0 = deleted, otherwise Present | HasVisitors | HasDetails | HasDistinct) -/// count(varint) [visitor sketch] +/// count(varint) /// nameCount(varint) [name(string) sum(double) count(varint) [min(double) max(double) [mean(double) stddev(double)]]]* -/// [detail per value] when HasDetails, see ValueDetail.Write; flags 0 for none -/// [distinctCount(varint) [dimension(string) sketch]*] when HasDistinct +/// [length(varint) visitor sketch] when HasVisitors +/// [length(varint) detailCount(varint) [length(varint) detail]*] when HasDetails; length 0 for none +/// [length(varint) distinctCount(varint) [dimension(string) length(varint) sketch]*] when HasDistinct /// -/// Min and max are written when count is not zero, moments when count exceeds one. +/// Min and max are written when count is not zero, moments when count exceeds one. Sections, +/// details and distinct sketches are length-prefixed and decoded only when read, one at a time. +/// A reader rejects header bits it does not know and trailing bytes. /// sealed class RollupValueSerializer : ISerializer { @@ -848,18 +1008,151 @@ sealed class RollupValueSerializer : ISerializer const byte HasDistinct = 8; + const byte KnownBits = Present | HasVisitors | HasDetails | HasDistinct; + + delegate T SectionReader(ref ByteReader reader); + + static readonly Func, HyperLogLog> DecodeVisitors = + bytes => Decode(bytes, static (ref ByteReader reader) => HyperLogLog.Read(ref reader)); + + static readonly Func, ValueDetail> DecodeDetail = bytes => Decode(bytes, static (ref ByteReader reader) => + { + var detail = ValueDetail.Read(ref reader); + // A value without details has no entry: a stored detail always holds data. + if (!detail.HasFirstLast && detail.Quantiles == null) + throw new FormatException("A stored value detail is empty."); + return detail; + }); + + static readonly Func, DetailEntries> DecodeDetails = ReadDetails; + + static readonly Func, DistinctEntries> DecodeDistinct = ReadDistinct; + [ThreadStatic] static ByteWriter threadWriter; + [ThreadStatic] + static ByteWriter threadSectionWriter; + + [ThreadStatic] + static ByteWriter threadEntryWriter; + + /// + /// Reads the entries of the detail section: their lengths only, each detail is decoded when read. + /// + static DetailEntries ReadDetails(ReadOnlyMemory bytes) + { + try + { + var reader = new ByteReader(bytes.Span); + // A value without details takes one byte, its zero length. + var entries = new RollupSection[CheckedCount(ref reader, 1)]; + for (var i = 0; i < entries.Length; ++i) + { + var entry = ReadEntry(ref reader, bytes, allowEmpty: true); + if (!entry.IsEmpty) + entries[i] = new(entry, DecodeDetail); + } + RequireEnd(reader); + return new DetailEntries(entries); + } + catch (Exception exception) when (IsMalformed(exception)) + { + throw new FormatException("The rollup value details are malformed.", exception); + } + } + + /// + /// Reads the names and the entries of the distinct section; each sketch is decoded when read. + /// + static DistinctEntries ReadDistinct(ReadOnlyMemory bytes) + { + try + { + var reader = new ByteReader(bytes.Span); + // Every entry takes at least a name length, its length and a sparse sketch's precision, format and count. + var count = CheckedCount(ref reader, 5); + var names = new string[count]; + var sketches = new RollupSection[count]; + 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); + } + RequireEnd(reader); + return new DistinctEntries(names, sketches); + } + catch (Exception exception) when (IsMalformed(exception)) + { + throw new FormatException("The rollup value distinct sketches are malformed.", exception); + } + } + + /// Reads the length of an entry and returns its bytes without decoding them. + static ReadOnlyMemory ReadEntry(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."); + var start = bytes.Length - reader.Remaining; + reader.ReadBytes((int)length); + return bytes.Slice(start, (int)length); + } + + static void RequireEnd(ByteReader reader) + { + if (reader.Remaining != 0) + throw new FormatException("A rollup value section has trailing bytes."); + } + + /// The exceptions that malformed bytes raise while they are read. + static bool IsMalformed(Exception exception) => + exception is IndexOutOfRangeException or ArgumentOutOfRangeException or OverflowException; + + /// Decodes a whole section; malformed bytes raise . + static T Decode(ReadOnlyMemory bytes, SectionReader read) + { + try + { + var reader = new ByteReader(bytes.Span); + var result = read(ref reader); + if (reader.Remaining != 0) + throw new FormatException("A rollup value section has trailing bytes."); + return result; + } + catch (Exception exception) when (exception is IndexOutOfRangeException or ArgumentOutOfRangeException or OverflowException) + { + throw new FormatException("A rollup value section is malformed.", exception); + } + } + public RollupValue Deserialize(Memory bytes) + { + try + { + return Read(bytes); + } + catch (Exception exception) when (exception is IndexOutOfRangeException or ArgumentOutOfRangeException or OverflowException) + { + throw new FormatException("The rollup value is malformed or truncated.", exception); + } + } + + static RollupValue Read(ReadOnlyMemory bytes) { var reader = new ByteReader(bytes.Span); var header = reader.ReadByte(); if (header == Deleted) + { + if (reader.Remaining != 0) + throw new FormatException("A deleted rollup value has trailing bytes."); return null; + } + if ((header & ~KnownBits) != 0 || (header & Present) == 0) + throw new FormatException("The rollup value has an unknown layout."); var count = reader.ReadVarInt64(); - var visitors = (header & HasVisitors) != 0 ? HyperLogLog.Read(ref reader) : null; - var nameCount = (int)reader.ReadVarUInt64(); + // Every value takes at least a name length, its sum and its count. + var nameCount = CheckedCount(ref reader, 10); var names = new string[nameCount]; var values = new ValueStats[nameCount]; for (var i = 0; i < nameCount; ++i) @@ -881,30 +1174,65 @@ public RollupValue Deserialize(Memory bytes) } values[i] = new ValueStats(sum, valueCount, min, max, mean, deviation); } - ValueDetail[] details = null; - if ((header & HasDetails) != 0) + if ((header & (HasVisitors | HasDetails | HasDistinct)) == 0) { - details = new ValueDetail[nameCount]; - for (var i = 0; i < nameCount; ++i) - { - var detail = ValueDetail.Read(ref reader); - details[i] = detail.HasFirstLast || detail.Quantiles != null ? detail : null; - } + if (reader.Remaining != 0) + throw new FormatException("The rollup value has trailing bytes."); + return new RollupValue(count, null, names, values, null, null); } - string[] distinctNames = null; - HyperLogLog[] distinct = null; - if ((header & HasDistinct) != 0) + // The sections follow the scalar values. They keep their stored bytes for as long as the row + // lives, and ZoneTree reads values from decompressed disk blocks of megabytes: a slice of a + // block would keep the whole block alive in caches and in merged rows, so the sections of a + // value that does not own its buffer are copied into one buffer of their own. + var start = bytes.Length - reader.Remaining; + var sections = OwnsBuffer(bytes) ? bytes[start..] : bytes[start..].ToArray(); + var sectionReader = new ByteReader(sections.Span); + RollupSection visitors = null; + RollupSection details = null; + RollupSection distinct = null; + if ((header & HasVisitors) != 0) + visitors = new(ReadSection(ref sectionReader, sections), DecodeVisitors); + if ((header & HasDetails) != 0) { - var distinctCount = (int)reader.ReadVarUInt64(); - distinctNames = new string[distinctCount]; - distinct = new HyperLogLog[distinctCount]; - for (var i = 0; i < distinctCount; ++i) - { - distinctNames[i] = reader.ReadString(); - distinct[i] = HyperLogLog.Read(ref reader); - } + var section = ReadSection(ref sectionReader, sections); + // 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) + throw new FormatException("The rollup value details do not match its values."); + details = new(section, DecodeDetails); } - return new RollupValue(count, visitors, names, values, details, distinctNames, distinct); + if ((header & HasDistinct) != 0) + distinct = new(ReadSection(ref sectionReader, sections), DecodeDistinct); + if (sectionReader.Remaining != 0) + throw new FormatException("The rollup value has trailing bytes."); + return new RollupValue(count, visitors, names, values, details, distinct); + } + + /// + /// Reads a count of entries that take at least each. A + /// count the remaining bytes cannot hold is corrupt, not an allocation to try. + /// + static int CheckedCount(ref ByteReader reader, int minimumEntryBytes) + { + var count = reader.ReadVarUInt64(); + if (count > (ulong)(reader.Remaining / minimumEntryBytes)) + throw new FormatException("A rollup value count exceeds its bytes."); + return (int)count; + } + + /// Whether the bytes are a whole array, which holds nothing but this value. + 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) @@ -912,20 +1240,18 @@ public Memory Serialize(in RollupValue entry) if (entry == null) return new byte[] { Deleted }; var writer = ByteWriter.Reuse(ref threadWriter, 256); - var hasVisitors = entry.Visitors != null && !entry.Visitors.IsEmpty; - var hasDetails = entry.Details != null && Array.Exists(entry.Details, x => x != null); - var hasDistinct = entry.DistinctNames.Length > 0; + var visitors = entry.VisitorSection; + var details = entry.DetailSection; + var distinct = entry.DistinctSection; var header = Present; - if (hasVisitors) + if (visitors != null) header |= HasVisitors; - if (hasDetails) + if (details != null) header |= HasDetails; - if (hasDistinct) + if (distinct != null) header |= HasDistinct; writer.WriteByte(header); writer.WriteVarInt64(entry.Count); - if (hasVisitors) - entry.Visitors.Write(writer); writer.WriteVarUInt64((ulong)entry.ValueNames.Length); for (var i = 0; i < entry.ValueNames.Length; ++i) { @@ -944,25 +1270,54 @@ public Memory Serialize(in RollupValue entry) } } } - if (hasDetails) + if (visitors != null) + WriteSection(writer, visitors, ref threadSectionWriter, static (section, sketch) => sketch.Write(section)); + if (details != null) { - foreach (var detail in entry.Details) + WriteSection(writer, details, ref threadSectionWriter, static (section, entries) => { - if (detail == null) - writer.WriteByte(0); - else - detail.Write(writer); - } + section.WriteVarUInt64((ulong)entries.Entries.Length); + foreach (var detail in entries.Entries) + { + if (detail == null) + section.WriteVarUInt64(0); + else + WriteSection(section, detail, ref threadEntryWriter, static (entry, value) => value.Write(entry)); + } + }); } - if (hasDistinct) + if (distinct != null) { - writer.WriteVarUInt64((ulong)entry.DistinctNames.Length); - for (var i = 0; i < entry.DistinctNames.Length; ++i) + WriteSection(writer, distinct, ref threadSectionWriter, static (section, entries) => { - writer.WriteString(entry.DistinctNames[i]); - entry.Distinct[i].Write(writer); - } + section.WriteVarUInt64((ulong)entries.Names.Length); + for (var i = 0; i < entries.Names.Length; ++i) + { + section.WriteString(entries.Names[i]); + WriteSection(section, entries.Sketches[i], ref threadEntryWriter, static (entry, sketch) => sketch.Write(entry)); + } + }); } return writer.ToArray(); } + + /// + /// Writes a length-prefixed section or entry: its stored bytes as they are, or its value + /// encoded into . + /// + static void WriteSection(ByteWriter writer, RollupSection section, ref ByteWriter scratch, Action encode) + where T : class + { + if (section.IsStored) + { + writer.WriteVarUInt64((ulong)section.Bytes.Length); + writer.WriteBytes(section.Bytes.Span); + return; + } + var encoded = ByteWriter.Reuse(ref scratch, 256); + encode(encoded, section.Value); + writer.WriteVarUInt64((ulong)encoded.Written.Length); + writer.WriteBytes(encoded.Written); + } } + diff --git a/src/ZoneTree.Analytics/Storage/SessionState.cs b/src/ZoneTree.Analytics/Storage/SessionState.cs index 8b5e5d6..3a8789b 100644 --- a/src/ZoneTree.Analytics/Storage/SessionState.cs +++ b/src/ZoneTree.Analytics/Storage/SessionState.cs @@ -123,8 +123,12 @@ public SessionStateSerializer(SessionSchema schema) public SessionState Deserialize(Memory bytes) { var reader = new ByteReader(bytes.Span); - if (reader.ReadByte() == Deleted) + var header = reader.ReadByte(); + // A header or trailing bytes this reader does not know are a layout it cannot read. + if (header == Deleted && reader.Remaining == 0) return null; + if (header != Present) + throw new FormatException("The session state has an unknown layout."); var state = new SessionState { SessionId = reader.ReadVarUInt64(), @@ -153,6 +157,8 @@ public SessionState Deserialize(Memory bytes) if (index >= SessionSchema.StandardCount) dimensions[index] = value; } + if (reader.Remaining != 0) + throw new FormatException("The session state has trailing bytes."); state.Dimensions = dimensions; return state; } diff --git a/src/ZoneTree.Analytics/Storage/StorageFormats.cs b/src/ZoneTree.Analytics/Storage/StorageFormats.cs new file mode 100644 index 0000000..cfe639b --- /dev/null +++ b/src/ZoneTree.Analytics/Storage/StorageFormats.cs @@ -0,0 +1,66 @@ +using System.Globalization; + +namespace ZoneTree.Analytics.Storage; + +/// +/// The versions of every persisted layout, in one place. +/// +/// +/// A version increases with every incompatible change of its layout. +/// refuses data written in any other version. Within a version, readers reject header bits, +/// flags and trailing bytes they do not know. +/// +static class StorageFormats +{ + /// The data directory: raw event, rollup and session payloads, keys and metadata. + public const int Dataset = 2; + + /// The metadata key of . + public const string DatasetKey = "format-version"; + + /// The shared covering dimension-index store: its key layout and metadata. + public const int DimensionIndex = 2; + + /// The metadata key of . + public const string DimensionIndexKey = "dimension-index.format"; + + /// The covering rollup-index stores: their key layout and metadata. + public const int RollupIndex = 2; + + /// The metadata key of . + public const string RollupIndexKey = "rollup-index.format"; + + /// + /// The body of the historical joint-rollup backfill journal, whose redo rows hold serialized + /// rollup values. + /// + public const int JointBackfillJournal = 2; + + /// The text a format is persisted as in metadata. + public static string Text(int version) => version.ToString(CultureInfo.InvariantCulture); + + /// + /// Refuses a stored format other than the current one. names the + /// data, such as "The analytics data directory". + /// + /// The stored format is not the current one. + public static void Require(string stored, int current, string subject) + { + if (!int.TryParse(stored, NumberStyles.None, CultureInfo.InvariantCulture, out var version)) + throw new AnalyticsConfigurationException( + $"{subject} has an unknown storage format '{stored}'; this release reads format {current}."); + Require(version, current, subject); + } + + /// + public static void Require(int stored, int current, string subject) + { + if (stored > current) + throw new AnalyticsConfigurationException( + $"{subject} uses storage format {stored}, written by a newer release; this release reads format {current}."); + if (stored < current) + throw new AnalyticsConfigurationException( + $"{subject} uses storage format {stored}, written by an older release; this release reads format {current}. " + + "Start with a fresh analytics data directory."); + } +}