Copilot commented on code in PR #1210:
URL:
https://github.com/apache/skywalking-banyandb/pull/1210#discussion_r3541040584
##########
banyand/trace/merger.go:
##########
@@ -315,32 +338,63 @@ func (tst *tsTable) releaseDispatchRequest(req
*mergeDispatchRequest) {
}
}
+// mergeOverrides lets the finalize round drive mergePartsThenSendIntroduction
with a
+// pre-built sampler filter (bypassing the hot-path isMergeHot/merge_grace
build) and a
+// finalize-generation stamp for the output part. A nil *mergeOverrides means
the hot /
+// flusher path: build the filter from merge-time rules and min-propagate
finalizeGen.
+type mergeOverrides struct {
+ filter *mergeFilter
+ finalizeGen *uint64
+}
+
+// buildHotMergeFilter builds the in-merge retention filter for the hot /
flusher merge
+// path from the merge-time rules (registered samplers + merge_grace
maturity). It
+// returns nil when the native pipeline is off, no samplers are registered, or
the merge
+// is still hot (parts younger than merge_grace), in which case the merge runs
unfiltered.
+func (tst *tsTable) buildHotMergeFilter(parts []*partWrapper) *mergeFilter {
+ if !tst.option.nativePipelineEnabled {
+ return nil
+ }
+ samplers := lookupSamplers(tst.group)
+ if len(samplers) == 0 {
+ return nil
+ }
+ graceNs := lookupMergeGrace(tst.group)
+ if graceNs <= 0 {
+ graceNs = int64(tst.option.mergeGraceDefault)
+ }
+ if isMergeHot(parts, graceNs, time.Now().UnixNano()) {
Review Comment:
`enabled_events` no longer reliably disables the MERGE-time retention
filter. With a FINALIZE-only config, `reconcilePipeline` still registers
samplers, and `buildHotMergeFilter` applies them to hot merges whenever
`nativePipelineEnabled` is on (it only checks `lookupSamplers(tst.group)`).
That contradicts the comment in `reconcilePipeline` that filtering can be
disabled via `enabled_events`, and makes it impossible to enable FINALIZE
without also enabling MERGE filtering.
Consider tracking per-group merge-event enablement separately (e.g., a small
registry boolean set by `reconcilePipeline`) and have `buildHotMergeFilter`
return nil when MERGE is disabled for the group, even if samplers are
registered for FINALIZE.
##########
banyand/trace/finalize_state.go:
##########
@@ -0,0 +1,91 @@
+// Licensed to Apache Software Foundation (ASF) under one or more contributor
+// license agreements. See the NOTICE file distributed with
+// this work for additional information regarding copyright
+// ownership. Apache Software Foundation (ASF) licenses this file to you under
+// the Apache License, Version 2.0 (the "License"); you may
+// not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package trace
+
+import (
+ "encoding/json"
+ "path/filepath"
+
+ "github.com/pkg/errors"
+
+ "github.com/apache/skywalking-banyandb/banyand/internal/storage"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+)
+
+// finalizeStateFilename is the per-shard finalization-sampling state document,
+// stored in the shard directory alongside snapshots.
+const finalizeStateFilename = "finalize.json"
+
+// finalizeState is the per-shard finalization-sampling state, persisted
atomically
+// in the shard directory (Rev 8: per-shard, NOT in the per-segment metadata).
It is
+// the sole authority for a shard's finalize generation, cooldown clock, round
count,
+// and arrival-based unsampled-bytes counter. The finalize worker is the only
writer
+// (serialized node-wide); readers seed the cached tsTable fields at open. A
missing
+// or corrupt file decodes to the zero value ("never finalized"), so
pre-finalize
+// shards and partial writes both fail open.
+type finalizeState struct {
+ // LastFinalizedAt is the RFC3339Nano wall-clock of the last successful
round;
+ // empty means never finalized. It is the cooldown clock.
+ LastFinalizedAt string `json:"lastFinalizedAt,omitempty"`
+ // FinalizeGeneration is the shard's monotonic current generation; 0 =
never
+ // finalized. A part is selectable when its FinalizeGen < this value.
+ FinalizeGeneration uint64 `json:"finalizeGeneration,omitempty"`
Review Comment:
The `FinalizeGeneration` comment doesn’t match the implemented selection
predicate. `runFinalizeRound` selects parts with `FinalizeGen < gNext`, where
`gNext = FinalizeGeneration + 1`, which effectively means `FinalizeGen <=
FinalizeGeneration` (full rewrite per round). The current comment says
“selectable when FinalizeGen < this value”, which would imply survivors stamped
at `FinalizeGeneration` are *not* selected, contradicting the current semantics.
##########
banyand/trace/svc_standalone.go:
##########
@@ -218,6 +221,16 @@ func (s *standalone) PreRun(ctx context.Context) error {
if err != nil {
return err
}
+ // Start the background finalization-sampling scanner when the native
pipeline is
+ // enabled. It is a single node-wide goroutine (concurrency-1) that
sweeps cooled
+ // segments of finalize-enabled groups; it is inert until a group
enables the
+ // FINALIZE event, so starting it unconditionally-under-the-flag is
safe.
+ if s.option.nativePipelineEnabled {
+ s.finalizeCloser = run.NewCloser(1)
+ //nolint:contextcheck // the finalize scanner uses the
run.Closer for cancellation, not a context (mirrors the tsTable loops).
+ go s.schemaRepo.finalizeScanLoop(s.finalizeCloser,
defaultFinalizeScanInterval)
+ }
Review Comment:
This starts a long-running background goroutine with a raw `go` statement.
Elsewhere in the codebase, service-lifetime goroutines are typically launched
via `run.Go(...)` to get centralized panic recovery/diagnostics. Using a bare
`go` here means any unexpected panic inside `finalizeScanLoop` would crash the
whole process.
Consider switching this launch to `run.Go` (similar to
`banyand/internal/storage/rotation.go:56`) so panics are recovered and reported
consistently.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
