perf(logs): collapse IPv4 at emit time instead of in a second pass (#54930)

### What does this PR do?

Targets `mwdd146980/tokenize-ipv4-hybrid-token`, not `main`. It keeps
the IPv4 hybrid token exactly as-is and only changes *when* the collapse
happens, to recover a tokenizer throughput regression.

`collapseHybridTokens` walked the whole token list after tokenization
and called `ipv4At` at every position. Two problems:

1. **`ipv4At` isn't inlinable.** Two slice headers plus an index, seven
comparisons — over the budget, confirmed by its absence from the `can
inline` list under `go build -gcflags=-m`. So it was a real function
call at *every* token position, ~250 of them on a 1 KB line.
2. **The pass rewrote both buffers unconditionally.**
`tsBuf[w]=tsBuf[r]` / `idxBuf[w]=idxBuf[r]` ran for every token even
when `w == r` and nothing collapsed — the common case, since most log
lines have no IP.

A dotted quad can only ever be closed by its **final octet**, so the
check doesn't need a second pass at all — it only needs to run when a
1-3 digit run is emitted:

```go
t.tsBuf = append(t.tsBuf, token+Token(r))
if token == D1 && runLen <= maxIPv4OctetDigits {
    t.collapseIPv4Tail()
}
```

`collapseIPv4Tail` reads the fields directly (no slice params, and
`isIPv4OctetToken` still inlines), tests separators first (`ts[5] !=
Period` rejects almost every call in one compare), skips re-checking the
trailing octet since the caller already established it, and doesn't
touch the buffers unless there's a real match. On a match it overwrites
one slot and truncates — capacity is preserved, so buffer reuse and the
allocation-free borrowed path are unaffected.

The second commit handles `tokenizeWithoutHybridCollapse`, which called
`emitRuns` to get *uncollapsed* tokens. That half of the IIS test is
really about the scorer — given a token sequence where the client IP is
still seven run tokens, Kadane averages 0.5 and the line stays aggregate
— so the sequence is a fixture, not something the tokenizer needs to
reproduce. It is now written out explicitly, split around the client IP
so the same fixture expresses both the pre- and post-collapse shapes,
with a `require.Equal` against the real `Tokenize` output pinning it so
the baseline cannot drift. No test-only state on `Tokenizer`. Both IIS
tests pass unchanged.

### Motivation

Measured against this PR's parent commit. Binaries built from each
commit and run **interleaved** (alternating each round, so thermal drift
on the M4 Max cancels rather than biasing one side), 10 rounds,
`benchstat` n=10:

| | parent | PR head (`93bfd45`) | this branch |
|---|---|---|---|
| Latency geomean (33 benchmarks) | 193.8 ns | **+38.79%** | **+1.82%**
|
| Throughput geomean (~1 KB lines) | 647 MiB/s | **−35.84%** |
**−2.02%** |

Worst cases on the PR head were `AppLog/Borrowed` 1.43 → 2.43 µs
(+69.9%), `TimestampHeavy` +70.5%, `RealisticApacheLog` +56.7%, all at
`p=0.000`. This is `tokenizeBorrowed`, the per-line hot path for auto
multi-line detection.

With this change, over half the benchmarks are statistically
indistinguishable from the parent (`p>0.05`); the worst remaining is
`TimestampHeavy` at +8.4%.

### Describe how you validated your changes

Collapsing during emission rather than after isn't *obviously*
equivalent on chained quads like `1.2.3.4.5.6.7.8`, so I verified it
instead of reasoning about it:

- **Differential test, 250,042 inputs.** Dumped tokens **and** start
indices from both implementations and diffed — byte-for-byte identical.
51k of those inputs contain at least one quad (16k×1, 13k×2, 12.7k×3,
8k×4, 764 with 5+), generated from a quad-biased grammar plus long
dotted chains. My first corpus only produced 26 quads, which would have
proven nothing, so I regenerated before trusting it.
- Full package suite passes, including both new IIS tests, under
`-count=2` and `-race`.
- `go vet` clean; `gofmt` clean.
- Added `1.2.3.4.5.6.7`, `1.2.3.4.5.6.7.8` and `12.34.56.78.90.12.34.56`
to `TestTokenizerIPv4`. That leftmost-greedy, non-overlapping behavior
is load-bearing for this refactor and was previously unguarded.
- Reworded the `HybridTokenPromotion` invariant in `tokenizer.allium`:
it described the collapse as a scan performed *after* promotion, which
was the two-pass structure. Now states the observable rule (leftmost,
non-overlapping) and documents the chained-quad case. Semantics
unchanged. I couldn't run `allium check` — the binary isn't installed
locally, so that wording is worth a look.

### Additional Notes

- No release note: this is a perf refinement to an unmerged feature that
already carries one.
- No `BUILD.bazel` change — no files or imports added.
- The benchmark numbers are **tokenizer-local**. I didn't measure the
full decoder path, so I can't say what share of real per-line pipeline
cost this represents.
- Feel free to squash this into your branch or cherry-pick it, whichever
is less disruptive.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: gh123man

pkg/logs/internal/decoder/preprocessor/timestamp_detector_test.go

@@ -159,15 +159,34 @@ func defaultTimestampDetectorSettings(t *testing.T) (threshold float64, labelerM
 	return threshold, labelerMaxBytes
 }
 
-func tokenizeWithoutHybridCollapse(tok *Tokenizer, input []byte) []Token {
-	maxBytes := len(input)
-	if tok.maxEvalBytes > 0 && tok.maxEvalBytes < maxBytes {
-		maxBytes = tok.maxEvalBytes
+// The token sequence for the first tokenizer_max_input_bytes of
+// iisW3CFailingShape, split around the client IP so the same fixture expresses
+// both the pre- and post-collapse shapes:
+//
+//	DDDD-DD-DD DD:DD:DD CDCCCD <ip> CCC /CCCCC/CCCCCCC/CDD
+//
+// Before the fix the IP was seven separate run tokens, which is what let Kadane
+// extend the timestamp match across it; now it is one IPv4 token.
+var (
+	iisW3CTokenPrefix = []Token{
+		D4, Dash, D2, Dash, D2, Space, // 2026-08-11
+		D2, Colon, D2, Colon, D2, Space, // 10:34:49
+		C1, D1, C3, D1, Space, // W3SVC1
+	}
+	iisW3CClientIPRuns = []Token{D2, Period, D1, Period, D2, Period, D2} // 10.1.48.10
+	iisW3CTokenSuffix  = []Token{
+		Space, C3, Space, // GET
+		Fslash, C5, Fslash, C7, Fslash, C1, D2, // /ZenIT/Service/v13
 	}
-	tok.emitRuns(input[:maxBytes])
-	out := make([]Token, len(tok.tsBuf))
-	copy(out, tok.tsBuf)
-	return out
+)
+
+// iisW3CTokens builds the fixture with the client IP rendered as the given
+// tokens: the seven run tokens for the pre-collapse shape, or a single IPv4.
+func iisW3CTokens(clientIP ...Token) []Token {
+	out := make([]Token, 0, len(iisW3CTokenPrefix)+len(clientIP)+len(iisW3CTokenSuffix))
+	out = append(out, iisW3CTokenPrefix...)
+	out = append(out, clientIP...)
+	return append(out, iisW3CTokenSuffix...)
 }
 
 // TestIISW3CDottedQuadDoesNotDiluteTimestampScore is the unit-level proof
@@ -184,7 +203,7 @@ func TestIISW3CDottedQuadDoesNotDiluteTimestampScore(t *testing.T) {
 	detector := NewTimestampDetector(threshold)
 	raw := []byte(iisW3CFailingShape)
 
-	without := tokenizeWithoutHybridCollapse(tok, raw)
+	without := iisW3CTokens(iisW3CClientIPRuns...)
 	matchWithout := staticTokenGraph.MatchProbability(without)
 	assert.Equal(t, 0.5, matchWithout.probability, "pre-fix Kadane average over timestamp+IP must be 0.5")
 
@@ -193,6 +212,11 @@ func TestIISW3CDottedQuadDoesNotDiluteTimestampScore(t *testing.T) {
 	assert.Equal(t, aggregate, ctxWithout.label, "without IPv4 collapse the IIS line must stay aggregate at threshold 0.5")
 
 	with, _ := tok.Tokenize(raw)
+	// Pins the fixture above to what the tokenizer actually emits, so the
+	// pre-collapse baseline cannot drift away from the real token sequence.
+	require.Equal(t, iisW3CTokens(IPv4), with,
+		"fixture must match the tokenizer output apart from the collapsed client IP")
+
 	matchWith := staticTokenGraph.MatchProbability(with)
 	assert.Equal(t, 1.0, matchWith.probability, "after collapse Kadane must stay on the timestamp-only run")
 

pkg/logs/internal/decoder/preprocessor/tokenizer.go

@@ -16,6 +16,9 @@ import (
 const (
 	maxRun         = 10
 	ipv4TokenWidth = 7 // D Period D Period D Period D
+	// maxIPv4OctetDigits is the longest digit run that can be an octet, so a
+	// run longer than this can never close a dotted quad.
+	maxIPv4OctetDigits = 3
 )
 
 // maxSpecialTokenLen and the special-token/debug-string tables are generated
@@ -150,9 +153,15 @@ func (t *Tokenizer) emitToken(input []byte, token Token, start, end int) {
 			r = maxRun - 1
 		}
 		t.tsBuf = append(t.tsBuf, token+Token(r))
-	} else {
-		t.tsBuf = append(t.tsBuf, token)
+		// A dotted quad can only ever be closed by its final octet, so this is
+		// the one emission that can complete the pattern. Checking here keeps
+		// the cost off every other token and avoids a second pass entirely.
+		if token == D1 && runLen <= maxIPv4OctetDigits {
+			t.collapseIPv4Tail()
+		}
+		return
 	}
+	t.tsBuf = append(t.tsBuf, token)
 }
 
 // tokenizeIntoBuffers scans input a single time and emits tokens into the
@@ -163,13 +172,11 @@ func (t *Tokenizer) tokenizeIntoBuffers(input []byte) ([]Token, []int) {
 		return nil, nil
 	}
 	t.emitRuns(input)
-	t.collapseHybridTokens()
 	return t.tsBuf, t.idxBuf
 }
 
-// emitRuns run-length-encodes input into tsBuf/idxBuf without hybrid collapse.
-// Tests use this to reconstruct the pre-IPv4-token tokenizer for the IIS
-// false-aggregate case.
+// emitRuns run-length-encodes input into tsBuf/idxBuf, collapsing hybrid tokens
+// via emitToken as each one completes.
 func (t *Tokenizer) emitRuns(input []byte) {
 	inputLen := len(input)
 	if inputLen == 0 {
@@ -213,49 +220,40 @@ func isIPv4OctetToken(tok Token) bool {
 	return tok >= D1 && tok <= D3
 }
 
-// ipv4At reports whether tokens[i:] starts with an IPv4 dotted quad.
-// Each octet is a 1-3 digit run and each separator is a single '.'.
-// A collapsed ".." / "..." Period token is rejected via the start indices.
-func ipv4At(tokens []Token, indices []int, i int) bool {
-	if i+ipv4TokenWidth > len(tokens) {
-		return false
-	}
-	if !isIPv4OctetToken(tokens[i]) || tokens[i+1] != Period ||
-		!isIPv4OctetToken(tokens[i+2]) || tokens[i+3] != Period ||
-		!isIPv4OctetToken(tokens[i+4]) || tokens[i+5] != Period ||
-		!isIPv4OctetToken(tokens[i+6]) {
-		return false
-	}
-	return indices[i+2] == indices[i+1]+1 &&
-		indices[i+4] == indices[i+3]+1 &&
-		indices[i+6] == indices[i+5]+1
-}
-
-// collapseHybridTokens rewrites multi-token patterns into a single token.
-// IPv4 dotted quads are the first of these: as separate digit/period tokens
-// they look like timestamp fragments to the detector, and addresses with
-// different octet widths would otherwise be different sampler patterns.
-func (t *Tokenizer) collapseHybridTokens() {
+// collapseIPv4Tail rewrites a dotted quad ending at the last emitted token into
+// a single IPv4 token. As separate digit/period tokens a quad looks like a
+// timestamp fragment to the detector, and addresses with different octet widths
+// would otherwise be different sampler patterns.
+//
+// Called from emitToken immediately after a 1-3 digit run is appended, which is
+// the only emission that can close the pattern, so the tokenizer never makes a
+// second pass over the token list. Each octet must be a 1-3 digit run and each
+// separator a single '.'; a collapsed ".." / "..." Period token is rejected via
+// the start indices.
+func (t *Tokenizer) collapseIPv4Tail() {
 	n := len(t.tsBuf)
 	if n < ipv4TokenWidth {
 		return
 	}
-	w := 0
-	for r := 0; r < n; {
-		if ipv4At(t.tsBuf, t.idxBuf, r) {
-			t.tsBuf[w] = IPv4
-			t.idxBuf[w] = t.idxBuf[r]
-			w++
-			r += ipv4TokenWidth
-			continue
-		}
-		t.tsBuf[w] = t.tsBuf[r]
-		t.idxBuf[w] = t.idxBuf[r]
-		w++
-		r++
+	ts := t.tsBuf[n-ipv4TokenWidth:]
+
+	// Separators first: they reject nearly every call in a single compare, and
+	// the trailing octet is already known from the caller.
+	if ts[5] != Period || ts[3] != Period || ts[1] != Period {
+		return
+	}
+	if !isIPv4OctetToken(ts[0]) || !isIPv4OctetToken(ts[2]) || !isIPv4OctetToken(ts[4]) {
+		return
 	}
-	t.tsBuf = t.tsBuf[:w]
-	t.idxBuf = t.idxBuf[:w]
+	idx := t.idxBuf[n-ipv4TokenWidth:]
+	if idx[2] != idx[1]+1 || idx[4] != idx[3]+1 || idx[6] != idx[5]+1 {
+		return
+	}
+
+	// Keep idx[0] (the address start) and drop the six tokens it absorbed.
+	ts[0] = IPv4
+	t.tsBuf = t.tsBuf[:n-ipv4TokenWidth+1]
+	t.idxBuf = t.idxBuf[:n-ipv4TokenWidth+1]
 }
 
 // tokensToString converts a list of tokens to a debug string.