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/09/27 10:06:56 UTC
(skywalking-banyandb) branch main updated: fix(validate): enforce tag-name uniqueness across a resource's tag families (#1370)
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 24150bf01 fix(validate): enforce tag-name uniqueness across a resource's tag families (#1370)
24150bf01 is described below
commit 24150bf01ef8d446b66d6695e810c3ec88c64df3
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Sun Sep 27 18:06:51 2026 +0800
fix(validate): enforce tag-name uniqueness across a resource's tag families (#1370)
---
CHANGES.md | 1 +
api/validate/validate.go | 34 ++++++++++++
api/validate/validate_test.go | 70 +++++++++++++++++++++++++
banyand/metadata/schema/property/client.go | 12 +++++
banyand/metadata/schema/property/client_test.go | 64 ++++++++++++++++++++++
pkg/query/logical/schema.go | 7 ++-
pkg/query/vectorized/measure/plan.go | 6 ++-
pkg/query/vectorized/measure/plan/analyzer.go | 11 ++--
8 files changed, 196 insertions(+), 9 deletions(-)
diff --git a/CHANGES.md b/CHANGES.md
index 225fd1515..89ad16dd1 100644
--- a/CHANGES.md
+++ b/CHANGES.md
@@ -40,6 +40,7 @@ Release Notes.
- Keep DNS discovery `ListNode` successful when some SRV addresses refuse connections but at least one node was discovered, so non-meta nodes no longer deadlock in PreRun on their own not-yet-listening gRPC address.
- Fix un-interruptible sleep on shutdown during snapshot sync retry and add jittered backoff for stream, measure, and trace.
- Report the real shard ID in `CollectDataInfo` shard info for stream, measure, and trace instead of the live table's slice index, so a node that owns only higher-numbered shards is no longer attributed to shard 0 in cross-node shard-load analysis.
+- Reject a measure or stream whose tag name appears in more than one tag family, or twice within one family, when it is created or updated. The query layer resolves a bare tag name against a flat map keyed by name alone, so such a schema previously registered successfully and then resolved that name differently depending on which query path read it. Already-persisted schemas keep loading unchanged; only the registry's create and update paths enforce the rule.
- Fix a silently wrong result from the vectorized measure liaison's aggregation reduce when a single per-node partial exceeds 65,536 rows: past that boundary the consuming row index wrapped, folding early rows into a group a second time and never visiting the rest, with no error. It is reachable today on `Top`-bearing queries, whose per-node limit is unbounded. The reduce now consumes an oversized partial in bounded chunks, and `BatchSize` is capped at `MaxUint16` so it cannot reopen the [...]
- Reject a `--trace-vectorized-batch-size` above 65,536 at startup instead of accepting it. The trace query path indexes a batch's active rows through a `uint16` selection, so a batch size above that bound silently wrapped -- reading early rows a second time and never reading the rest. The stream path already enforced this bound; measure now does too. A node configured above the bound will fail to start rather than return wrong results, and must lower the flag.
diff --git a/api/validate/validate.go b/api/validate/validate.go
index a458ca033..e01bc5a5a 100644
--- a/api/validate/validate.go
+++ b/api/validate/validate.go
@@ -450,6 +450,40 @@ func tagFamily(tagFamilies []*databasev1.TagFamilySpec) error {
return nil
}
+// UniqueTagNames rejects a tag name that appears more than once across a
+// resource's tag families.
+//
+// A tag name identifies a tag across the whole resource, not merely within its
+// own family: the query layer resolves a bare name against a flat map keyed by
+// name alone -- logical.CommonSchema.CreateRef states the invariant outright,
+// and BydbQL's transformer relies on it for every identifier it resolves. Two
+// families sharing a name make those lookups depend on iteration order. A
+// single resource-wide set also catches a name duplicated within one family.
+//
+// This is deliberately separate from Measure and Stream, which the data nodes
+// also run when loading an already-persisted schema. A schema registered
+// before this check existed must keep loading rather than be dropped, so only
+// the registry's create and update paths call this.
+func UniqueTagNames(tagFamilies []*databasev1.TagFamilySpec) error {
+ familyOfTag := make(map[string]string)
+ for i := range tagFamilies {
+ for j := range tagFamilies[i].Tags {
+ tagName := tagFamilies[i].Tags[j].Name
+ previousFamily, duplicated := familyOfTag[tagName]
+ if !duplicated {
+ familyOfTag[tagName] = tagFamilies[i].Name
+ continue
+ }
+ if previousFamily == tagFamilies[i].Name {
+ return fmt.Errorf("tag name %q is duplicated in tag family %q", tagName, previousFamily)
+ }
+ return fmt.Errorf("tag name %q is duplicated in tag families %q and %q: tag names must be unique across all tag families",
+ tagName, previousFamily, tagFamilies[i].Name)
+ }
+ }
+ return nil
+}
+
// Property validates the provided Property schema object.
func Property(property *databasev1.Property) error {
if property == nil {
diff --git a/api/validate/validate_test.go b/api/validate/validate_test.go
index c1c5ae815..61e125932 100644
--- a/api/validate/validate_test.go
+++ b/api/validate/validate_test.go
@@ -299,3 +299,73 @@ func TestCheckShardingKeySubset(t *testing.T) {
})
}
}
+
+// TestUniqueTagNamesRejectsDuplicateAcrossFamilies pins the resource-wide tag
+// name invariant. The query layer resolves a bare tag name against a flat map
+// (logical.CommonSchema.CreateRef, and BydbQL's allTags), so two families
+// sharing a name would make resolution depend on iteration order.
+func TestUniqueTagNamesRejectsDuplicateAcrossFamilies(t *testing.T) {
+ tagFamilies := []*databasev1.TagFamilySpec{
+ {Name: "default", Tags: []*databasev1.TagSpec{
+ {Name: "id", Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "svc", Type: databasev1.TagType_TAG_TYPE_STRING},
+ }},
+ {Name: "other", Tags: []*databasev1.TagSpec{
+ {Name: "svc", Type: databasev1.TagType_TAG_TYPE_INT},
+ }},
+ }
+ err := UniqueTagNames(tagFamilies)
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), `tag name "svc" is duplicated in tag families "default" and "other"`)
+}
+
+// TestUniqueTagNamesRejectsDuplicateWithinOneFamily falls out of the same flat
+// map and was likewise unchecked; it names the single family rather than
+// repeating it.
+func TestUniqueTagNamesRejectsDuplicateWithinOneFamily(t *testing.T) {
+ tagFamilies := []*databasev1.TagFamilySpec{
+ {Name: "default", Tags: []*databasev1.TagSpec{
+ {Name: "id", Type: databasev1.TagType_TAG_TYPE_STRING},
+ {Name: "id", Type: databasev1.TagType_TAG_TYPE_INT},
+ }},
+ }
+ err := UniqueTagNames(tagFamilies)
+ assert.Error(t, err)
+ assert.Contains(t, err.Error(), `tag name "id" is duplicated in tag family "default"`)
+}
+
+// TestUniqueTagNamesAcceptsDistinctNames guards against over-reach: distinct
+// names across several families are the normal case.
+func TestUniqueTagNamesAcceptsDistinctNames(t *testing.T) {
+ tagFamilies := []*databasev1.TagFamilySpec{
+ {Name: "default", Tags: []*databasev1.TagSpec{{Name: "id", Type: databasev1.TagType_TAG_TYPE_STRING}}},
+ {Name: "searchable", Tags: []*databasev1.TagSpec{{Name: "svc", Type: databasev1.TagType_TAG_TYPE_STRING}}},
+ }
+ assert.NoError(t, UniqueTagNames(tagFamilies))
+}
+
+// TestMeasureAndStreamStillLoadDuplicateTagNames pins the split deliberately:
+// the shared validators are what the data nodes run when loading an
+// already-persisted schema (banyand/{measure,stream}/metadata.go's
+// OnAddOrUpdate, which logs and drops the resource on failure). A schema
+// registered before UniqueTagNames existed must keep loading, so only the
+// registry's create and update paths reject it.
+func TestMeasureAndStreamStillLoadDuplicateTagNames(t *testing.T) {
+ duplicated := []*databasev1.TagFamilySpec{
+ {Name: "default", Tags: []*databasev1.TagSpec{{Name: "id", Type: databasev1.TagType_TAG_TYPE_STRING}}},
+ {Name: "searchable", Tags: []*databasev1.TagSpec{{Name: "id", Type: databasev1.TagType_TAG_TYPE_STRING}}},
+ }
+ measure := &databasev1.Measure{
+ Metadata: &commonv1.Metadata{Name: "legacy_measure", Group: "group1"},
+ Entity: &databasev1.Entity{TagNames: []string{"id"}},
+ TagFamilies: duplicated,
+ }
+ assert.NoError(t, Measure(measure))
+ stream := &databasev1.Stream{
+ Metadata: &commonv1.Metadata{Name: "legacy_stream", Group: "group1"},
+ Entity: &databasev1.Entity{TagNames: []string{"id"}},
+ TagFamilies: duplicated,
+ }
+ assert.NoError(t, Stream(stream))
+ assert.Error(t, UniqueTagNames(duplicated))
+}
diff --git a/banyand/metadata/schema/property/client.go b/banyand/metadata/schema/property/client.go
index 0d82752e7..2f16ccf7f 100644
--- a/banyand/metadata/schema/property/client.go
+++ b/banyand/metadata/schema/property/client.go
@@ -736,6 +736,9 @@ func (r *SchemaRegistry) CreateStream(ctx context.Context, stream *databasev1.St
if validateErr := validate.Stream(stream); validateErr != nil {
return 0, validateErr
}
+ if uniqueErr := validate.UniqueTagNames(stream.GetTagFamilies()); uniqueErr != nil {
+ return 0, uniqueErr
+ }
now := time.Now().UnixNano()
stream.Metadata.ModRevision = now
stream.UpdatedAt = timestamppb.Now()
@@ -747,6 +750,9 @@ func (r *SchemaRegistry) UpdateStream(ctx context.Context, stream *databasev1.St
if validateErr := validate.Stream(stream); validateErr != nil {
return 0, validateErr
}
+ if uniqueErr := validate.UniqueTagNames(stream.GetTagFamilies()); uniqueErr != nil {
+ return 0, uniqueErr
+ }
now := time.Now().UnixNano()
stream.Metadata.ModRevision = now
stream.UpdatedAt = timestamppb.Now()
@@ -783,6 +789,9 @@ func (r *SchemaRegistry) CreateMeasure(ctx context.Context, measure *databasev1.
if validateErr := validate.Measure(measure); validateErr != nil {
return 0, validateErr
}
+ if uniqueErr := validate.UniqueTagNames(measure.GetTagFamilies()); uniqueErr != nil {
+ return 0, uniqueErr
+ }
if subsetWarn := validate.CheckShardingKeySubset(measure); subsetWarn != nil {
r.l.Warn().Err(subsetWarn).Str("measure", measure.GetMetadata().GetName()).Msg("sharding key is not a subset of entity tags")
}
@@ -802,6 +811,9 @@ func (r *SchemaRegistry) UpdateMeasure(ctx context.Context, measure *databasev1.
if validateErr := validate.Measure(measure); validateErr != nil {
return 0, validateErr
}
+ if uniqueErr := validate.UniqueTagNames(measure.GetTagFamilies()); uniqueErr != nil {
+ return 0, uniqueErr
+ }
if subsetWarn := validate.CheckShardingKeySubset(measure); subsetWarn != nil {
r.l.Warn().Err(subsetWarn).Str("measure", measure.GetMetadata().GetName()).Msg("sharding key is not a subset of entity tags")
}
diff --git a/banyand/metadata/schema/property/client_test.go b/banyand/metadata/schema/property/client_test.go
index e0d1fc360..69fa5eac8 100644
--- a/banyand/metadata/schema/property/client_test.go
+++ b/banyand/metadata/schema/property/client_test.go
@@ -1948,3 +1948,67 @@ func TestWatchPush(t *testing.T) {
assert.Equal(t, "test-group", restartEvt.Group)
})
}
+
+// duplicateTagFamily returns a second tag family reusing a tag name already
+// present in the helper schemas' "default" family.
+func duplicateTagFamily() *databasev1.TagFamilySpec {
+ return &databasev1.TagFamilySpec{
+ Name: "searchable",
+ Tags: []*databasev1.TagSpec{
+ {Name: "service_name", Type: databasev1.TagType_TAG_TYPE_STRING},
+ },
+ }
+}
+
+// TestCreateRejectsDuplicateTagNameAcrossFamilies proves the registry actually
+// calls validate.UniqueTagNames. The unit tests in api/validate cover the rule
+// itself; this covers the wiring, which a mis-wired call site would otherwise
+// pass silently. The rule lives here rather than in validate.Measure/Stream
+// because the data nodes run those when loading an already-persisted schema.
+func TestCreateRejectsDuplicateTagNameAcrossFamilies(t *testing.T) {
+ addr := startTestSchemaServer(t)
+ reg := newTestRegistry(t, addr)
+ ctx := context.Background()
+ createTestGroup(t, reg)
+
+ m := testMeasure("test-group")
+ m.TagFamilies = append(m.TagFamilies, duplicateTagFamily())
+ _, measureErr := reg.CreateMeasure(ctx, m)
+ require.Error(t, measureErr)
+ assert.Contains(t, measureErr.Error(), `tag name "service_name" is duplicated`)
+
+ s := testStream()
+ s.TagFamilies = append(s.TagFamilies, duplicateTagFamily())
+ _, streamErr := reg.CreateStream(ctx, s)
+ require.Error(t, streamErr)
+ assert.Contains(t, streamErr.Error(), `tag name "service_name" is duplicated`)
+}
+
+// TestUpdateRejectsDuplicateTagNameAcrossFamilies covers the update path, which
+// is wired separately from create.
+func TestUpdateRejectsDuplicateTagNameAcrossFamilies(t *testing.T) {
+ addr := startTestSchemaServer(t)
+ reg := newTestRegistry(t, addr)
+ ctx := context.Background()
+ createTestGroup(t, reg)
+
+ m := testMeasure("test-group")
+ _, createErr := reg.CreateMeasure(ctx, m)
+ require.NoError(t, createErr)
+ stored, getErr := reg.GetMeasure(ctx, &commonv1.Metadata{Group: "test-group", Name: "test-measure"})
+ require.NoError(t, getErr)
+ stored.TagFamilies = append(stored.TagFamilies, duplicateTagFamily())
+ _, updateErr := reg.UpdateMeasure(ctx, stored)
+ require.Error(t, updateErr)
+ assert.Contains(t, updateErr.Error(), `tag name "service_name" is duplicated`)
+
+ s := testStream()
+ _, createStreamErr := reg.CreateStream(ctx, s)
+ require.NoError(t, createStreamErr)
+ storedStream, getStreamErr := reg.GetStream(ctx, &commonv1.Metadata{Group: "test-group", Name: testStreamName})
+ require.NoError(t, getStreamErr)
+ storedStream.TagFamilies = append(storedStream.TagFamilies, duplicateTagFamily())
+ _, updateStreamErr := reg.UpdateStream(ctx, storedStream)
+ require.Error(t, updateStreamErr)
+ assert.Contains(t, updateStreamErr.Error(), `tag name "service_name" is duplicated`)
+}
diff --git a/pkg/query/logical/schema.go b/pkg/query/logical/schema.go
index d9ae7dccb..7f957d475 100644
--- a/pkg/query/logical/schema.go
+++ b/pkg/query/logical/schema.go
@@ -151,8 +151,11 @@ func (cs *CommonSchema) IndexRuleDefined(indexRuleName string) (bool, *databasev
}
// CreateRef create TagRef to the given tags.
-// The family name of the tag is actually not used
-// since the uniqueness of the tag names can be guaranteed across families.
+// The family name of the tag is actually not used, since tag names are
+// expected to be unique across families: api/validate.UniqueTagNames enforces
+// that on the registry's create and update paths. A schema registered before
+// that check still loads, so it may carry the same name in two families, in
+// which case this resolves to whichever entry the tag map retained.
func (cs *CommonSchema) CreateRef(tags ...[]*Tag) ([][]*TagRef, error) {
tagRefs := make([][]*TagRef, 0, len(tags))
for _, tagInFamily := range tags {
diff --git a/pkg/query/vectorized/measure/plan.go b/pkg/query/vectorized/measure/plan.go
index 38d217322..d63ceee3c 100644
--- a/pkg/query/vectorized/measure/plan.go
+++ b/pkg/query/vectorized/measure/plan.go
@@ -279,8 +279,10 @@ func lookupFieldColumnIndex(schema *vectorized.BatchSchema, name string) (int, e
// lookupAggInputColumnIndex resolves agg's target column: a tag sibling of
// lookupGroupByKeyIndices when agg.TagName is set (matched on RoleTag +
-// TagFamily + Name, since tag names are only unique within a family),
-// otherwise the existing field lookup.
+// TagFamily + Name, which disambiguates a schema carrying the same tag name in
+// two families; api/validate.UniqueTagNames rejects that shape on the
+// registry's create and update paths only, so an older persisted schema can
+// still carry it), otherwise the existing field lookup.
func lookupAggInputColumnIndex(schema *vectorized.BatchSchema, agg *model.MeasureAgg) (int, error) {
if agg.TagName == "" {
return lookupFieldColumnIndex(schema, agg.FieldName)
diff --git a/pkg/query/vectorized/measure/plan/analyzer.go b/pkg/query/vectorized/measure/plan/analyzer.go
index bebc6b995..55c697745 100644
--- a/pkg/query/vectorized/measure/plan/analyzer.go
+++ b/pkg/query/vectorized/measure/plan/analyzer.go
@@ -298,11 +298,12 @@ func validateBucketableOrdering(req *measurev1.QueryRequest) error {
// translateAgg builds the model Agg struct from the proto. Exactly one of
// field_name / tag_name must be set; a tag target additionally requires
-// tag_family (tag names are only unique within a family — design §5.1) and
-// is validated against the §6 semantics matrix for the five functions this
-// issue implements. COUNT_DISTINCT's matrix row is accepted here (its type
-// rules match COUNT's) but has no execution support yet: an unmapped
-// function is rejected downstream, at protoAggFuncToInternal.
+// tag_family, which is what keeps the target resolvable when a schema carries
+// the same tag name in two families. api/validate.UniqueTagNames rejects that
+// shape, but only on the registry's create and update paths, so a schema
+// registered before that check still loads and can reach here with a
+// duplicate: the family qualifier is load-bearing, not decoration. The target
+// is then checked against the §6 semantics matrix.
func translateAgg(req *measurev1.QueryRequest, measureSchema *databasev1.Measure) (*model.MeasureAgg, error) {
aggProto := req.GetAgg()
fieldName := aggProto.GetFieldName()