You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@skywalking.apache.org by ha...@apache.org on 2026/04/28 11:31:34 UTC
(skywalking-banyandb) branch main updated: fix(trace): resolve flaky schema change merge test (#1099)
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
The following commit(s) were added to refs/heads/main by this push:
new e90e46934 fix(trace): resolve flaky schema change merge test (#1099)
e90e46934 is described below
commit e90e46934d4568450118f80c7134a0237702f390
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Tue Apr 28 19:31:28 2026 +0800
fix(trace): resolve flaky schema change merge test (#1099)
* docs: add snapshot retention flags and relative path version note
Document the min-file-snapshot-age flags for measure, stream, trace,
and property services. Add version annotation for relative path support.
---
api/proto/banyandb/database/v1/rpc.proto | 2 ++
banyand/measure/metadata.go | 3 ++-
banyand/stream/metadata.go | 4 +++-
banyand/trace/metadata.go | 4 +++-
banyand/trace/metadata_test.go | 20 ++++++++++++++++++++
docs/api-reference.md | 1 +
pkg/timestamp/scheduler.go | 8 ++++++--
7 files changed, 37 insertions(+), 5 deletions(-)
diff --git a/api/proto/banyandb/database/v1/rpc.proto b/api/proto/banyandb/database/v1/rpc.proto
index a7d7a882a..901e94f8d 100644
--- a/api/proto/banyandb/database/v1/rpc.proto
+++ b/api/proto/banyandb/database/v1/rpc.proto
@@ -553,6 +553,8 @@ message ShardInfo {
InvertedIndexInfo inverted_index_info = 5;
// sidx_info contains information about sidx.
SIDXInfo sidx_info = 6;
+ // file_part_count is the number of file parts (excluding in-memory parts) in this shard.
+ int64 file_part_count = 7;
}
// SeriesIndexInfo contains information about the series index.
diff --git a/banyand/measure/metadata.go b/banyand/measure/metadata.go
index 29ff21ebc..8482eed9d 100644
--- a/banyand/measure/metadata.go
+++ b/banyand/measure/metadata.go
@@ -467,12 +467,13 @@ func (sr *schemaRepo) collectShardInfo(table any, shardID uint32) *databasev1.Sh
}
}
defer snapshot.decRef()
- var totalCount, compressedSize, partCount uint64
+ var totalCount, compressedSize, partCount, filePartCount uint64
for _, pw := range snapshot.parts {
if pw.p != nil {
totalCount += pw.p.partMetadata.TotalCount
compressedSize += pw.p.partMetadata.CompressedSizeBytes
partCount++
+ filePartCount++
} else if pw.mp != nil {
totalCount += pw.mp.partMetadata.TotalCount
compressedSize += pw.mp.partMetadata.CompressedSizeBytes
diff --git a/banyand/stream/metadata.go b/banyand/stream/metadata.go
index a2e452044..0ecc8074c 100644
--- a/banyand/stream/metadata.go
+++ b/banyand/stream/metadata.go
@@ -346,13 +346,14 @@ func (sr *schemaRepo) collectShardInfo(table any, shardID uint32) *databasev1.Sh
}
}
defer snapshot.decRef()
- var totalCount, compressedSize, uncompressedSize, partCount uint64
+ var totalCount, compressedSize, uncompressedSize, partCount, filePartCount uint64
for _, pw := range snapshot.parts {
if pw.p != nil {
totalCount += pw.p.partMetadata.TotalCount
compressedSize += pw.p.partMetadata.CompressedSizeBytes
uncompressedSize += pw.p.partMetadata.UncompressedSizeBytes
partCount++
+ filePartCount++
} else if pw.mp != nil {
totalCount += pw.mp.partMetadata.TotalCount
compressedSize += pw.mp.partMetadata.CompressedSizeBytes
@@ -368,6 +369,7 @@ func (sr *schemaRepo) collectShardInfo(table any, shardID uint32) *databasev1.Sh
PartCount: int64(partCount),
InvertedIndexInfo: invertedIndexInfo,
SidxInfo: &databasev1.SIDXInfo{},
+ FilePartCount: int64(filePartCount),
}
}
diff --git a/banyand/trace/metadata.go b/banyand/trace/metadata.go
index 8c3322f35..6adee2a62 100644
--- a/banyand/trace/metadata.go
+++ b/banyand/trace/metadata.go
@@ -355,13 +355,14 @@ func (sr *schemaRepo) collectShardInfo(ctx context.Context, table any, shardID u
}
}
defer snapshot.decRef()
- var totalCount, compressedSize, uncompressedSize, partCount uint64
+ var totalCount, compressedSize, uncompressedSize, partCount, filePartCount uint64
for _, pw := range snapshot.parts {
if pw.p != nil {
totalCount += pw.p.partMetadata.TotalCount
compressedSize += pw.p.partMetadata.CompressedSizeBytes
uncompressedSize += pw.p.partMetadata.UncompressedSpanSizeBytes
partCount++
+ filePartCount++
} else if pw.mp != nil {
totalCount += pw.mp.partMetadata.TotalCount
compressedSize += pw.mp.partMetadata.CompressedSizeBytes
@@ -377,6 +378,7 @@ func (sr *schemaRepo) collectShardInfo(ctx context.Context, table any, shardID u
PartCount: int64(partCount),
InvertedIndexInfo: &databasev1.InvertedIndexInfo{},
SidxInfo: sidxInfo,
+ FilePartCount: int64(filePartCount),
}
}
diff --git a/banyand/trace/metadata_test.go b/banyand/trace/metadata_test.go
index e605d5de5..dae19e9ba 100644
--- a/banyand/trace/metadata_test.go
+++ b/banyand/trace/metadata_test.go
@@ -366,6 +366,12 @@ var _ = Describe("Metadata", func() {
env := setupSchemaChangeTrace(svcs, traceName, groupName, traceSetupOptions{withExtraTag: true})
writeSchemaChangeTraceData(svcs, traceName, groupName, now.Add(-2*time.Hour), 5,
writeTraceDataOptions{extraTag: extraTagInt})
+ // Wait for the first batch to flush to disk before writing the second batch.
+ // This ensures each shard has at least one file part from the first batch,
+ // so the second batch creates additional parts that can be merged.
+ Eventually(func() int64 {
+ return getFilePartCount(svcs, groupName)
+ }, flags.EventuallyTimeout).Should(BeNumerically(">=", 1))
changeTraceExtraTagType(svcs, traceName, groupName)
writeSchemaChangeTraceData(svcs, traceName, groupName, now.Add(-1*time.Hour), 3,
writeTraceDataOptions{extraTag: extraTagString, traceIDPrefix: "trace_new_"})
@@ -751,3 +757,17 @@ func getTotalPartCount(svcs *services, group string) int64 {
}
return total
}
+
+func getFilePartCount(svcs *services, group string) int64 {
+ dataInfo, err := svcs.trace.CollectDataInfo(context.TODO(), group)
+ if err != nil || dataInfo == nil {
+ return 0
+ }
+ var total int64
+ for _, seg := range dataInfo.SegmentInfo {
+ for _, shard := range seg.ShardInfo {
+ total += shard.FilePartCount
+ }
+ }
+ return total
+}
diff --git a/docs/api-reference.md b/docs/api-reference.md
index eea4435c6..631bd679c 100644
--- a/docs/api-reference.md
+++ b/docs/api-reference.md
@@ -4451,6 +4451,7 @@ ShardInfo contains information about a specific shard.
| part_count | [int64](#int64) | | part_count is the number of parts in this shard. |
| inverted_index_info | [InvertedIndexInfo](#banyandb-database-v1-InvertedIndexInfo) | | inverted_index_info contains information about the inverted index. |
| sidx_info | [SIDXInfo](#banyandb-database-v1-SIDXInfo) | | sidx_info contains information about sidx. |
+| file_part_count | [int64](#int64) | | file_part_count is the number of file parts (excluding in-memory parts) in this shard. |
diff --git a/pkg/timestamp/scheduler.go b/pkg/timestamp/scheduler.go
index b98df0b01..7ee533e2e 100644
--- a/pkg/timestamp/scheduler.go
+++ b/pkg/timestamp/scheduler.go
@@ -150,12 +150,16 @@ func (s *Scheduler) Closed() bool {
// Close the Scheduler and shut down all registered tasks.
func (s *Scheduler) Close() {
s.Lock()
- defer s.Unlock()
s.closed = true
+ tasks := make([]*task, 0, len(s.tasks))
for k, t := range s.tasks {
- t.close()
+ tasks = append(tasks, t)
delete(s.tasks, k)
}
+ s.Unlock()
+ for _, t := range tasks {
+ t.close()
+ }
}
// Metrics returns the metrics of all registered tasks.