You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by "mrproliu (via GitHub)" <gi...@apache.org> on 2026/05/29 13:17:50 UTC
[PR] Fix lifecycle migration of non-multiple segment intervals via row-replay [skywalking-banyandb]
mrproliu opened a new pull request, #1145:
URL: https://github.com/apache/skywalking-banyandb/pull/1145
## Why
The lifecycle migration moves expired data from one stage to the next (hot → warm → cold). Each stage may define a **different `segment_interval`**. The original migration copied whole part files into
a single target segment chosen from the part's timestamp (the "chunk-sync" path). That is only correct when every source part falls entirely inside one target segment.
When the source and target segment intervals are **not multiples of each other** (their day-grids don't align — e.g. a 5-day source interval migrating into 7-day target segments), a single source
part's rows can **straddle a target segment boundary**. Chunk-copying such a part dumps **all** of its rows into one target segment, so the rows that belong to the adjacent segment end up in the wrong
place: the migrated part's `[minTimestamp, maxTimestamp]` escapes its target segment's `[start, start+interval)` bounds. This produces mis-segmented data and breaks per-segment assumptions for later
queries and retention.
## What
Detect source parts that cross a target segment boundary and migrate them with **row-level replay** instead of a file copy:
- **Part fully inside one target segment** → unchanged fast **chunk-sync** (copies the part file).
- **Part spanning ≥2 target segments** → **row-replay**: re-publish each row so the receiving node routes it to the target segment that matches the row's own timestamp. No row lands in the wrong
segment.
### Supporting changes
- **Per-part flush-and-confirm**: each migrated part is flushed and its server-side acks confirmed before it is marked complete (the batch publisher is closed/reopened per part), so a part is only
recorded as migrated once it is durably delivered; on failure it is recorded as an error instead of silently lost.
- **Source-segment completion key**: migration progress is keyed on the *source* segment instead of the target, fixing a collision where multiple source segments mapping to one target segment could be
wrongly treated as already-migrated.
- **`raw_element_id` consistency (stream)**: a new `raw_element_id` is carried through row-replay so an element migrated via both chunk-sync and row-replay keeps one storage element id; the warm-stage
query dedup then converges to a single row instead of duplicating it.
- **Migration report** now breaks down `chunk_sync` vs `row_replay` parts/rows per group so the path taken is observable.
## Tests
- New `row_replay_*` unit tests for the per-type replayers and progress tracking.
- Distributed e2e (`TestPropertyLifecycle`): added measure / stream / trace cross-segment specs that seed boundary-straddling data, run the migration, and verify (a) the **physical** layout — every
migrated part stays within its own target segment, segments split correctly, exact segment count + contiguous boundaries — and (b) the **gRPC query** results from the warm stage (per-timestamp,
same-segment, and cross-segment aggregate). Stream specs also assert `raw_element_id` dedup; trace adds a `cross_segment_timestamp` index rule so warm-stage queries are verifiable.
- `make lint`, lifecycle unit tests, and the distributed e2e (5/5) all pass.
- [ ] If this pull request closes/resolves/fixes an existing issue, replace the issue number. Fixes apache/skywalking#<issue number>.
- [x] Update the [`CHANGES` log](https://github.com/apache/skywalking-banyandb/blob/main/CHANGES.md).
--
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: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix lifecycle migration of non-multiple segment intervals via row-replay [skywalking-banyandb]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
Copilot commented on code in PR #1145:
URL: https://github.com/apache/skywalking-banyandb/pull/1145#discussion_r3324611032
##########
banyand/backup/lifecycle/segment_boundary_utils.go:
##########
@@ -24,17 +24,16 @@ import (
"github.com/apache/skywalking-banyandb/pkg/timestamp"
)
-func calculateTargetSegments(partMinTS, partMaxTS int64, targetInterval storage.IntervalRule) []time.Time {
- // Use time.Local so sender's bucket math stays on the same per-timezone
- // grid as the receiver, which sees ctx.MinTimestamp as time.Local too.
- minTime := time.Unix(0, partMinTS)
- maxTime := time.Unix(0, partMaxTS)
-
+// calculateTargetSegments returns every target bucket overlapping the half-open
+// source range [start, end). Every migration caller passes the source segment's
+// own boundaries (segment/shard time range), where end is exclusive, so the
+// strict current.Before(end) guard naturally excludes the empty next bucket
+// when the source ends exactly on a target boundary. Using time.Local keeps the
+// sender's bucket math on the same per-timezone grid as the receiver, which
+// sees ctx.MinTimestamp as time.Local too.
+func calculateTargetSegments(start, end time.Time, targetInterval storage.IntervalRule) []time.Time {
var targetSegments []time.Time
- // current starts at the bucket containing minTime, so segmentEnd is
- // always > minTime and the loop guard already excludes anything past
- // maxTime - every iteration produces a real overlap.
- for current := targetInterval.Standard(minTime); !current.After(maxTime); current = targetInterval.NextTime(current) {
+ for current := targetInterval.Standard(start); current.Before(end); current = targetInterval.NextTime(current) {
targetSegments = append(targetSegments, current)
}
Review Comment:
calculateTargetSegments' doc says the bucket math should be on the time.Local grid, but the implementation currently uses start/end's Location implicitly (storage.IntervalRule.Standard anchors at 1970-01-01 in t.Location()). Converting start/end to time.Local inside the helper makes the behavior match the comment and avoids timezone-dependent behavior if a caller passes a UTC time.Time.
##########
test/cases/lifecycle/lifecycle.go:
##########
@@ -398,3 +487,687 @@ func asInt(v interface{}) int {
}
return 0
}
+
+// crossSegmentTimestamps returns three probe timestamps shared by the
+// cross-segment suites: single sits in a source segment fully inside one target
+// segment, while left/right sit in a source segment that straddles a target
+// boundary (left and right of the crossing). The picked source day N satisfies
+// N%6==2 (LCM of the 2-day source and 3-day target grids), shifted >=8 local
+// days back so it clears the 5-day source TTL.
+func crossSegmentTimestamps() (single, left, right time.Time) {
+ nowLocal := time.Now().Local().Truncate(24 * time.Hour)
+ nowDay := nowLocal.Unix() / (24 * 3600)
+ crossSrcDayStart := nowDay - 8
+ for crossSrcDayStart%6 != 2 {
+ crossSrcDayStart--
+ }
+ crossSrcStart := nowLocal.AddDate(0, 0, int(crossSrcDayStart-nowDay))
+ return crossSrcStart.Add(-12 * time.Hour), crossSrcStart.Add(12 * time.Hour), crossSrcStart.Add(36 * time.Hour)
+}
+
+// runLifecycleMigration runs a single hot->warm lifecycle migration, pointing
+// every root path at the shared source dir and writing its report to reportDir.
+func runLifecycleMigration(progressFile, reportDir string) {
+ lifecycleCmd := lifecycle.NewCommand()
+ args := []string{
+ "--grpc-addr", SharedContext.DataAddr,
+ "--stream-root-path", SharedContext.SrcDir,
+ "--measure-root-path", SharedContext.SrcDir,
+ "--trace-root-path", SharedContext.SrcDir,
+ "--progress-file", progressFile,
+ "--report-dir", reportDir,
+ }
+ args = append(args, SharedContext.MetadataFlags...)
+ lifecycleCmd.SetArgs(args)
+ gomega.Expect(lifecycleCmd.Execute()).To(gomega.Succeed())
+}
+
+// drainWriteAcks closes a client-streaming write and fails the spec if any
+// server-side ack reports a non-success status, surfacing per-row rejections
+// that a bare CloseSend would swallow.
+func drainWriteAcks[R interface{ GetStatus() string }](recv func() (R, error), closeSend func() error) {
+ ackErrs := make(chan error, 8)
+ go func() {
+ defer close(ackErrs)
+ for {
+ resp, recvErr := recv()
+ if recvErr != nil {
+ return
+ }
+ if status := resp.GetStatus(); status != "" && status != "STATUS_SUCCEED" {
+ ackErrs <- fmt.Errorf("write rejected: %s", status)
+ }
+ }
+ }()
+ gomega.Expect(closeSend()).To(gomega.Succeed())
+ for ackErr := range ackErrs {
+ gomega.Expect(ackErr).NotTo(gomega.HaveOccurred(), "write ack must succeed")
+ }
+}
Review Comment:
drainWriteAcks currently (1) ignores Recv() errors (it just returns), so non-EOF stream failures can be missed, and (2) can deadlock if more than 8 non-success acks are produced before CloseSend returns (bounded channel fills and blocks the recv goroutine). A simple synchronous receive loop after CloseSend avoids both issues and fails the spec on any non-success status or non-EOF error.
##########
test/cases/lifecycle/lifecycle.go:
##########
@@ -19,17 +19,27 @@
package lifecycle_test
import (
+ "context"
"encoding/json"
+ "fmt"
"io/fs"
"os"
Review Comment:
The new drainWriteAcks helper should verify the Recv() terminal error (io.EOF) and surface non-EOF stream errors. That requires importing io; otherwise Recv errors can be silently swallowed and the test can pass even if the write stream fails server-side.
--
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: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix lifecycle migration of non-multiple segment intervals via row-replay [skywalking-banyandb]
Posted by "mrproliu (via GitHub)" <gi...@apache.org>.
mrproliu commented on code in PR #1145:
URL: https://github.com/apache/skywalking-banyandb/pull/1145#discussion_r3329553954
##########
banyand/backup/lifecycle/row_replay_measure.go:
##########
@@ -0,0 +1,336 @@
+// 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 lifecycle
+
+import (
+ "context"
+ "fmt"
+ "path/filepath"
+ "strconv"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/dump"
+ dumpmeasure "github.com/apache/skywalking-banyandb/banyand/internal/dump/measure"
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+)
+
+const (
+ measureReplayBatchSize = 2000
+ measureReplayBatchTimeout = 30 * time.Second
+)
+
+// cachedMeasureSchema bundles a measure's schema with its derived entity and
+// sharding-key routers so each row only requires one schema lookup per
+// measure. shardingKey is non-nil only when the measure declares a non-empty
+// ShardingKey; otherwise the entity-based shardID wins.
+type cachedMeasureSchema struct {
+ schema *databasev1.Measure
+ shardingKey partition.Router
+ entityLocator partition.Locator
+}
+
+// measureRowReplayer replays a group's measure parts row-by-row through the
+// real Write API, resolving each row's schema on demand via EntityValues.
+type measureRowReplayer struct {
+ selector node.Selector
+ client queue.Client
+ publisher queue.BatchPublisher
+ fs fs.FileSystem
+ logger *logger.Logger
+ measureSchemas map[string]*databasev1.Measure
+ schemaCache map[string]*cachedMeasureSchema
+ irCache map[string]*dump.IndexResolver
+ mergedRuleToTag map[uint32]dump.IndexedTagSpec
+ counter *uint64
+ group string
+ batch []bus.Message
+ schemaCacheMu sync.Mutex
+ irCacheMu sync.Mutex
+ batchMu sync.Mutex
+ targetShardNum uint32
+}
+
+// newMeasureRowReplayer constructs a measure replayer, eagerly enumerating the
+// group's measures and index rules for IndexResolver decoding.
+func newMeasureRowReplayer(
+ ctx context.Context,
+ group string, targetShardNum uint32,
+ selector node.Selector, client queue.Client,
+ md metadata.Repo, fileSystem fs.FileSystem,
+ l *logger.Logger, counter *uint64,
+) (*measureRowReplayer, error) {
+ measures, err := md.MeasureRegistry().ListMeasure(ctx, schema.ListOpt{Group: group})
+ if err != nil {
+ return nil, fmt.Errorf("list measures of group %s: %w", group, err)
+ }
+ rules, err := md.IndexRuleRegistry().ListIndexRule(ctx, schema.ListOpt{Group: group})
+ if err != nil {
+ return nil, fmt.Errorf("list index rules of group %s: %w", group, err)
+ }
+ bindings, err := md.IndexRuleBindingRegistry().ListIndexRuleBinding(ctx, schema.ListOpt{Group: group})
+ if err != nil {
+ return nil, fmt.Errorf("list index rule bindings of group %s: %w", group, err)
+ }
+ // Index the already-listed measures by name so loadSchema resolves each
+ // subject from this snapshot instead of issuing a redundant GetMeasure
+ // per measure. Using one consistent snapshot also avoids mid-migration
+ // schema drift.
+ measureSchemas := make(map[string]*databasev1.Measure, len(measures))
+ for _, m := range measures {
+ measureSchemas[m.GetMetadata().GetName()] = m
+ }
+ return &measureRowReplayer{
+ group: group,
+ targetShardNum: targetShardNum,
+ selector: selector,
+ client: client,
+ publisher: client.NewBatchPublisher(measureReplayBatchTimeout),
+ fs: fileSystem,
+ logger: l,
+ measureSchemas: measureSchemas,
+ schemaCache: make(map[string]*cachedMeasureSchema),
+ irCache: make(map[string]*dump.IndexResolver),
+ mergedRuleToTag: deriveMergedRuleToTag(measures, rules, bindings),
+ counter: counter,
+ batch: make([]bus.Message, 0, measureReplayBatchSize),
+ }, nil
+}
+
+// Close flushes pending messages, closes the publisher and releases cached
+// IndexResolver handles.
+func (r *measureRowReplayer) Close() (map[string]*common.Error, error) {
+ flushCtx, cancel := context.WithTimeout(context.Background(), measureReplayBatchTimeout)
+ defer cancel()
+ flushErr := r.flushBatch(flushCtx)
+ r.irCacheMu.Lock()
+ for _, ir := range r.irCache {
+ _ = ir.Close()
+ }
+ r.irCache = nil
+ r.irCacheMu.Unlock()
+ cee, closeErr := r.publisher.Close()
+ if flushErr != nil {
+ return cee, flushErr
+ }
+ return cee, closeErr
+}
+
+// flushAndConfirm flushes the buffered batch, closes the publisher to obtain the
+// per-node delivery result for everything sent since the last call, then opens a
+// fresh publisher for subsequent parts. The batch publisher is client-streaming,
+// so per-node errors are only observable once its stream closes; this lets a
+// row-replayed part be confirmed durable before it is marked completed.
+func (r *measureRowReplayer) flushAndConfirm(ctx context.Context) (map[string]*common.Error, error) {
+ flushErr := r.flushBatch(ctx)
+ cee, closeErr := r.publisher.Close()
+ r.publisher = r.client.NewBatchPublisher(measureReplayBatchTimeout)
+ if flushErr != nil {
+ return cee, flushErr
+ }
+ return cee, closeErr
+}
+
+// loadSchema resolves a measure schema by name from the snapshot listed at
+// construction and lazily caches its derived routers.
+func (r *measureRowReplayer) loadSchema(measureName string) (*cachedMeasureSchema, error) {
+ r.schemaCacheMu.Lock()
+ defer r.schemaCacheMu.Unlock()
+ if c, ok := r.schemaCache[measureName]; ok {
+ return c, nil
+ }
+ m, ok := r.measureSchemas[measureName]
+ if !ok {
+ return nil, fmt.Errorf("measure schema %s/%s not found in group snapshot", r.group, measureName)
+ }
+ c := &cachedMeasureSchema{
+ schema: m,
+ entityLocator: partition.NewEntityLocator(m.TagFamilies, m.Entity, m.GetMetadata().GetModRevision()),
+ }
+ if sk := m.GetShardingKey(); sk != nil && len(sk.GetTagNames()) > 0 {
+ c.shardingKey = partition.NewShardingKeyLocator(m.TagFamilies, sk)
+ }
+ r.schemaCache[measureName] = c
+ return c, nil
+}
+
+// loadIndexResolver lazily opens an IndexResolver per source segment.
+func (r *measureRowReplayer) loadIndexResolver(segmentPath string) (*dump.IndexResolver, error) {
+ r.irCacheMu.Lock()
+ defer r.irCacheMu.Unlock()
+ if ir, ok := r.irCache[segmentPath]; ok {
+ return ir, nil
+ }
+ ir, err := dump.NewIndexResolver(segmentPath, dump.DefaultIndexCacheSize, r.mergedRuleToTag)
+ if err != nil {
+ return nil, fmt.Errorf("open index resolver for %s: %w", segmentPath, err)
+ }
+ r.irCache[segmentPath] = ir
Review Comment:
Since all shards should walk from the same segment, I delete the map reference to a single reference to reduce the memory usage.
--
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: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix lifecycle migration of non-multiple segment intervals via row-replay [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily commented on code in PR #1145:
URL: https://github.com/apache/skywalking-banyandb/pull/1145#discussion_r3328474707
##########
banyand/backup/lifecycle/row_replay_measure.go:
##########
@@ -0,0 +1,336 @@
+// 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 lifecycle
+
+import (
+ "context"
+ "fmt"
+ "path/filepath"
+ "strconv"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "google.golang.org/protobuf/types/known/timestamppb"
+
+ "github.com/apache/skywalking-banyandb/api/common"
+ "github.com/apache/skywalking-banyandb/api/data"
+ commonv1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/common/v1"
+ databasev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/database/v1"
+ measurev1 "github.com/apache/skywalking-banyandb/api/proto/banyandb/measure/v1"
+ "github.com/apache/skywalking-banyandb/banyand/internal/dump"
+ dumpmeasure "github.com/apache/skywalking-banyandb/banyand/internal/dump/measure"
+ "github.com/apache/skywalking-banyandb/banyand/metadata"
+ "github.com/apache/skywalking-banyandb/banyand/metadata/schema"
+ "github.com/apache/skywalking-banyandb/banyand/queue"
+ "github.com/apache/skywalking-banyandb/pkg/bus"
+ "github.com/apache/skywalking-banyandb/pkg/fs"
+ "github.com/apache/skywalking-banyandb/pkg/logger"
+ "github.com/apache/skywalking-banyandb/pkg/node"
+ "github.com/apache/skywalking-banyandb/pkg/partition"
+)
+
+const (
+ measureReplayBatchSize = 2000
+ measureReplayBatchTimeout = 30 * time.Second
+)
+
+// cachedMeasureSchema bundles a measure's schema with its derived entity and
+// sharding-key routers so each row only requires one schema lookup per
+// measure. shardingKey is non-nil only when the measure declares a non-empty
+// ShardingKey; otherwise the entity-based shardID wins.
+type cachedMeasureSchema struct {
+ schema *databasev1.Measure
+ shardingKey partition.Router
+ entityLocator partition.Locator
+}
+
+// measureRowReplayer replays a group's measure parts row-by-row through the
+// real Write API, resolving each row's schema on demand via EntityValues.
+type measureRowReplayer struct {
+ selector node.Selector
+ client queue.Client
+ publisher queue.BatchPublisher
+ fs fs.FileSystem
+ logger *logger.Logger
+ measureSchemas map[string]*databasev1.Measure
+ schemaCache map[string]*cachedMeasureSchema
+ irCache map[string]*dump.IndexResolver
+ mergedRuleToTag map[uint32]dump.IndexedTagSpec
+ counter *uint64
+ group string
+ batch []bus.Message
+ schemaCacheMu sync.Mutex
+ irCacheMu sync.Mutex
+ batchMu sync.Mutex
+ targetShardNum uint32
+}
+
+// newMeasureRowReplayer constructs a measure replayer, eagerly enumerating the
+// group's measures and index rules for IndexResolver decoding.
+func newMeasureRowReplayer(
+ ctx context.Context,
+ group string, targetShardNum uint32,
+ selector node.Selector, client queue.Client,
+ md metadata.Repo, fileSystem fs.FileSystem,
+ l *logger.Logger, counter *uint64,
+) (*measureRowReplayer, error) {
+ measures, err := md.MeasureRegistry().ListMeasure(ctx, schema.ListOpt{Group: group})
+ if err != nil {
+ return nil, fmt.Errorf("list measures of group %s: %w", group, err)
+ }
+ rules, err := md.IndexRuleRegistry().ListIndexRule(ctx, schema.ListOpt{Group: group})
+ if err != nil {
+ return nil, fmt.Errorf("list index rules of group %s: %w", group, err)
+ }
+ bindings, err := md.IndexRuleBindingRegistry().ListIndexRuleBinding(ctx, schema.ListOpt{Group: group})
+ if err != nil {
+ return nil, fmt.Errorf("list index rule bindings of group %s: %w", group, err)
+ }
+ // Index the already-listed measures by name so loadSchema resolves each
+ // subject from this snapshot instead of issuing a redundant GetMeasure
+ // per measure. Using one consistent snapshot also avoids mid-migration
+ // schema drift.
+ measureSchemas := make(map[string]*databasev1.Measure, len(measures))
+ for _, m := range measures {
+ measureSchemas[m.GetMetadata().GetName()] = m
+ }
+ return &measureRowReplayer{
+ group: group,
+ targetShardNum: targetShardNum,
+ selector: selector,
+ client: client,
+ publisher: client.NewBatchPublisher(measureReplayBatchTimeout),
+ fs: fileSystem,
+ logger: l,
+ measureSchemas: measureSchemas,
+ schemaCache: make(map[string]*cachedMeasureSchema),
+ irCache: make(map[string]*dump.IndexResolver),
+ mergedRuleToTag: deriveMergedRuleToTag(measures, rules, bindings),
+ counter: counter,
+ batch: make([]bus.Message, 0, measureReplayBatchSize),
+ }, nil
+}
+
+// Close flushes pending messages, closes the publisher and releases cached
+// IndexResolver handles.
+func (r *measureRowReplayer) Close() (map[string]*common.Error, error) {
+ flushCtx, cancel := context.WithTimeout(context.Background(), measureReplayBatchTimeout)
+ defer cancel()
+ flushErr := r.flushBatch(flushCtx)
+ r.irCacheMu.Lock()
+ for _, ir := range r.irCache {
+ _ = ir.Close()
+ }
+ r.irCache = nil
+ r.irCacheMu.Unlock()
+ cee, closeErr := r.publisher.Close()
+ if flushErr != nil {
+ return cee, flushErr
+ }
+ return cee, closeErr
+}
+
+// flushAndConfirm flushes the buffered batch, closes the publisher to obtain the
+// per-node delivery result for everything sent since the last call, then opens a
+// fresh publisher for subsequent parts. The batch publisher is client-streaming,
+// so per-node errors are only observable once its stream closes; this lets a
+// row-replayed part be confirmed durable before it is marked completed.
+func (r *measureRowReplayer) flushAndConfirm(ctx context.Context) (map[string]*common.Error, error) {
+ flushErr := r.flushBatch(ctx)
+ cee, closeErr := r.publisher.Close()
+ r.publisher = r.client.NewBatchPublisher(measureReplayBatchTimeout)
+ if flushErr != nil {
+ return cee, flushErr
+ }
+ return cee, closeErr
+}
+
+// loadSchema resolves a measure schema by name from the snapshot listed at
+// construction and lazily caches its derived routers.
+func (r *measureRowReplayer) loadSchema(measureName string) (*cachedMeasureSchema, error) {
+ r.schemaCacheMu.Lock()
+ defer r.schemaCacheMu.Unlock()
+ if c, ok := r.schemaCache[measureName]; ok {
+ return c, nil
+ }
+ m, ok := r.measureSchemas[measureName]
+ if !ok {
+ return nil, fmt.Errorf("measure schema %s/%s not found in group snapshot", r.group, measureName)
+ }
+ c := &cachedMeasureSchema{
+ schema: m,
+ entityLocator: partition.NewEntityLocator(m.TagFamilies, m.Entity, m.GetMetadata().GetModRevision()),
+ }
+ if sk := m.GetShardingKey(); sk != nil && len(sk.GetTagNames()) > 0 {
+ c.shardingKey = partition.NewShardingKeyLocator(m.TagFamilies, sk)
+ }
+ r.schemaCache[measureName] = c
+ return c, nil
+}
+
+// loadIndexResolver lazily opens an IndexResolver per source segment.
+func (r *measureRowReplayer) loadIndexResolver(segmentPath string) (*dump.IndexResolver, error) {
+ r.irCacheMu.Lock()
+ defer r.irCacheMu.Unlock()
+ if ir, ok := r.irCache[segmentPath]; ok {
+ return ir, nil
+ }
+ ir, err := dump.NewIndexResolver(segmentPath, dump.DefaultIndexCacheSize, r.mergedRuleToTag)
+ if err != nil {
+ return nil, fmt.Errorf("open index resolver for %s: %w", segmentPath, err)
+ }
+ r.irCache[segmentPath] = ir
Review Comment:
`irCache` accumulates one `dump.IndexResolver` (bluge index reader + its cache) per source segment and only releases them in `Close()`, which for the measure/stream replayers is the whole group migration. A migration that spans many source segments therefore keeps every segment's index reader open simultaneously, so peak memory grows with the number of cross-segment source segments rather than staying bounded.
Since segments are visited in order and an `IndexResolver` is only needed while that segment's parts are being replayed, could the resolver for a segment be closed/evicted once its last part has been replayed (or capped with a small LRU)? That would keep resident memory bounded regardless of migration span. The same applies to `streamRowReplayer.irCache`.
(Not a correctness issue — the replayers run against snapshot dirs, so there's no live-serving conflict — purely a resource-footprint concern on large-group migrations.)
--
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: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix lifecycle migration of non-multiple segment intervals via row-replay [skywalking-banyandb]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #1145:
URL: https://github.com/apache/skywalking-banyandb/pull/1145#issuecomment-4575772525
## [Codecov](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1145?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
:white_check_mark: All modified and coverable lines are covered by tests.
:white_check_mark: Project coverage is 41.05%. Comparing base ([`3530dd9`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/3530dd9b5cefaa4fef60eec91ecbfbdfcaf243c5?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)) to head ([`192d8cb`](https://app.codecov.io/gh/apache/skywalking-banyandb/commit/192d8cb4642539287cff250211b0c708f8bb69d8?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)).
:warning: Report is 266 commits behind head on main.
<details><summary>Additional details and impacted files</summary>
```diff
@@ Coverage Diff @@
## main #1145 +/- ##
==========================================
- Coverage 45.97% 41.05% -4.93%
==========================================
Files 328 196 -132
Lines 55505 20348 -35157
==========================================
- Hits 25520 8353 -17167
+ Misses 27909 11145 -16764
+ Partials 2076 850 -1226
```
| [Flag](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1145/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [pkg](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1145/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `41.05% <100.00%> (?)` | |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
</details>
[:umbrella: View full report in Codecov by Sentry](https://app.codecov.io/gh/apache/skywalking-banyandb/pull/1145?dropdown=coverage&src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
:loudspeaker: Have feedback on the report? [Share it here](https://about.codecov.io/codecov-pr-comment-feedback/?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
<details><summary> :rocket: New features to boost your workflow: </summary>
- :snowflake: [Test Analytics](https://docs.codecov.com/docs/test-analytics): Detect flaky tests, report on failures, and find test suite problems.
- :package: [JS Bundle Analysis](https://docs.codecov.com/docs/javascript-bundle-analysis): Save yourself from yourself by tracking and limiting bundle sizes in JS merges.
</details>
--
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: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix lifecycle migration of non-multiple segment intervals via row-replay [skywalking-banyandb]
Posted by "hanahmily (via GitHub)" <gi...@apache.org>.
hanahmily merged PR #1145:
URL: https://github.com/apache/skywalking-banyandb/pull/1145
--
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: notifications-unsubscribe@skywalking.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org