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 2024/11/22 01:00:21 UTC
(skywalking-banyandb) 01/02: Don't write tstable in the index mode
This is an automated email from the ASF dual-hosted git repository.
hanahmily pushed a commit to branch index
in repository https://gitbox.apache.org/repos/asf/skywalking-banyandb.git
commit 7b5553e4474fb76b85ba26b4251aff2da468990b
Author: Gao Hongtao <ha...@gmail.com>
AuthorDate: Fri Nov 22 06:23:16 2024 +0800
Don't write tstable in the index mode
Signed-off-by: Gao Hongtao <ha...@gmail.com>
---
banyand/measure/write.go | 51 +++++++++++++++-------
pkg/test/measure/testdata/groups/index_mode.json | 18 ++++++++
.../index_rule_bindings/service_traffic.json | 2 +-
.../index_rules/{layer.json => im_service_id.json} | 8 ++--
pkg/test/measure/testdata/index_rules/layer.json | 2 +-
.../measure/testdata/measures/service_traffic.json | 2 +-
test/cases/init.go | 4 +-
test/cases/measure/data/input/entity.yaml | 2 +-
test/cases/measure/data/input/entity_in.yaml | 2 +-
test/cases/measure/data/input/entity_service.yaml | 2 +-
test/cases/measure/data/input/index_mode_all.yaml | 2 +-
test/cases/measure/data/input/index_mode_none.yaml | 2 +-
.../measure/data/input/index_mode_order_desc.yaml | 2 +-
.../cases/measure/data/input/index_mode_range.yaml | 2 +-
test/cases/measure/data/input/no_field.yaml | 2 +-
test/cases/measure/data/input/tag_filter_int.yaml | 2 +-
16 files changed, 72 insertions(+), 33 deletions(-)
diff --git a/banyand/measure/write.go b/banyand/measure/write.go
index ab0879a2..485adc71 100644
--- a/banyand/measure/write.go
+++ b/banyand/measure/write.go
@@ -83,14 +83,6 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
break
}
}
- shardID := common.ShardID(writeEvent.ShardId)
- if dpt == nil {
- if dpt, err = w.newDpt(tsdb, dpg, t, ts, shardID); err != nil {
- return nil, fmt.Errorf("cannot create data points in table: %w", err)
- }
- }
- dpt.dataPoints.timestamps = append(dpt.dataPoints.timestamps, ts)
- dpt.dataPoints.versions = append(dpt.dataPoints.versions, req.DataPoint.Version)
stm, ok := w.schemaRepo.loadMeasure(req.GetMetadata())
if !ok {
return nil, fmt.Errorf("cannot find measure definition: %s", req.GetMetadata())
@@ -102,6 +94,14 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
if fLen > len(stm.schema.GetTagFamilies()) {
return nil, fmt.Errorf("%s has more tag families than %s", req.Metadata, stm.schema)
}
+
+ shardID := common.ShardID(writeEvent.ShardId)
+ if dpt == nil {
+ if dpt, err = w.newDpt(tsdb, dpg, t, ts, shardID, stm.schema.IndexMode); err != nil {
+ return nil, fmt.Errorf("cannot create data points in table: %w", err)
+ }
+ }
+
series := &pbv1.Series{
Subject: req.Metadata.Name,
EntityValues: writeEvent.EntityValues,
@@ -109,7 +109,24 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
if err := series.Marshal(); err != nil {
return nil, fmt.Errorf("cannot marshal series: %w", err)
}
+
+ tagFamily, fields := w.handleTagFamily(stm, req)
+ if stm.schema.IndexMode {
+ doc := index.Document{
+ DocID: uint64(series.ID),
+ EntityValues: series.Buffer,
+ Fields: fields,
+ }
+ doc.Timestamp = ts
+ dpg.docs = append(dpg.docs, doc)
+ return dst, nil
+ }
+
+ dpt.dataPoints.tagFamilies = append(dpt.dataPoints.tagFamilies, tagFamily)
+ dpt.dataPoints.timestamps = append(dpt.dataPoints.timestamps, ts)
+ dpt.dataPoints.versions = append(dpt.dataPoints.versions, req.DataPoint.Version)
dpt.dataPoints.seriesIDs = append(dpt.dataPoints.seriesIDs, series.ID)
+
field := nameValues{}
for i := range stm.GetSchema().GetFields() {
var v *modelv1.FieldValue
@@ -125,8 +142,6 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
))
}
dpt.dataPoints.fields = append(dpt.dataPoints.fields, field)
- tagFamily, fields := w.handleTagFamily(stm, req)
- dpt.dataPoints.tagFamilies = append(dpt.dataPoints.tagFamilies, tagFamily)
if stm.processorManager != nil {
stm.processorManager.onMeasureWrite(uint64(series.ID), &measurev1.InternalWriteRequest{
@@ -144,15 +159,13 @@ func (w *writeCallback) handle(dst map[string]*dataPointsInGroup, writeEvent *me
EntityValues: series.Buffer,
Fields: fields,
}
- if stm.schema.IndexMode {
- doc.Timestamp = ts
- }
dpg.docs = append(dpg.docs, doc)
return dst, nil
}
-func (w *writeCallback) newDpt(tsdb storage.TSDB[*tsTable, option], dpg *dataPointsInGroup, t time.Time, ts int64, shardID common.ShardID) (*dataPointsInTable, error) {
+func (w *writeCallback) newDpt(tsdb storage.TSDB[*tsTable, option], dpg *dataPointsInGroup,
+ t time.Time, ts int64, shardID common.ShardID, indexMode bool) (*dataPointsInTable, error) {
var segment storage.Segment[*tsTable, option]
for _, seg := range dpg.segments {
if seg.GetTimeRange().Contains(ts) {
@@ -167,6 +180,12 @@ func (w *writeCallback) newDpt(tsdb storage.TSDB[*tsTable, option], dpg *dataPoi
}
dpg.segments = append(dpg.segments, segment)
}
+ if indexMode {
+ return &dataPointsInTable{
+ timeRange: segment.GetTimeRange(),
+ }, nil
+ }
+
tstb, err := segment.CreateTSTableIfNotExist(shardID)
if err != nil {
return nil, fmt.Errorf("cannot create ts table: %w", err)
@@ -295,7 +314,9 @@ func (w *writeCallback) Rev(_ context.Context, message bus.Message) (resp bus.Me
g := groups[i]
for j := range g.tables {
dps := g.tables[j]
- dps.tsTable.mustAddDataPoints(&dps.dataPoints)
+ if dps.tsTable != nil {
+ dps.tsTable.mustAddDataPoints(&dps.dataPoints)
+ }
}
for _, segment := range g.segments {
if err := segment.IndexDB().Write(g.docs); err != nil {
diff --git a/pkg/test/measure/testdata/groups/index_mode.json b/pkg/test/measure/testdata/groups/index_mode.json
new file mode 100644
index 00000000..fb0d9b4b
--- /dev/null
+++ b/pkg/test/measure/testdata/groups/index_mode.json
@@ -0,0 +1,18 @@
+{
+ "metadata": {
+ "name": "index_mode"
+ },
+ "catalog": "CATALOG_MEASURE",
+ "resource_opts": {
+ "shard_num": 2,
+ "segment_interval": {
+ "unit": "UNIT_DAY",
+ "num": 1
+ },
+ "ttl": {
+ "unit": "UNIT_DAY",
+ "num": 7
+ }
+ },
+ "updated_at": "2021-04-15T01:30:15.01Z"
+}
\ No newline at end of file
diff --git a/pkg/test/measure/testdata/index_rule_bindings/service_traffic.json b/pkg/test/measure/testdata/index_rule_bindings/service_traffic.json
index 6a1bf9bc..08b65e0d 100644
--- a/pkg/test/measure/testdata/index_rule_bindings/service_traffic.json
+++ b/pkg/test/measure/testdata/index_rule_bindings/service_traffic.json
@@ -1,7 +1,7 @@
{
"metadata": {
"name": "service_traffic_rule_binding",
- "group": "sw_metric"
+ "group": "index_mode"
},
"rules": [
"service_id",
diff --git a/pkg/test/measure/testdata/index_rules/layer.json b/pkg/test/measure/testdata/index_rules/im_service_id.json
similarity index 57%
copy from pkg/test/measure/testdata/index_rules/layer.json
copy to pkg/test/measure/testdata/index_rules/im_service_id.json
index 32d73999..549a4037 100644
--- a/pkg/test/measure/testdata/index_rules/layer.json
+++ b/pkg/test/measure/testdata/index_rules/im_service_id.json
@@ -1,11 +1,11 @@
{
"metadata": {
- "id": 3,
- "name": "layer",
- "group": "sw_metric"
+ "id": 1,
+ "name": "service_id",
+ "group": "index_mode"
},
"tags": [
- "layer"
+ "service_id"
],
"type": "TYPE_INVERTED",
"updated_at": "2021-04-15T01:30:15.01Z"
diff --git a/pkg/test/measure/testdata/index_rules/layer.json b/pkg/test/measure/testdata/index_rules/layer.json
index 32d73999..dbf70d7a 100644
--- a/pkg/test/measure/testdata/index_rules/layer.json
+++ b/pkg/test/measure/testdata/index_rules/layer.json
@@ -2,7 +2,7 @@
"metadata": {
"id": 3,
"name": "layer",
- "group": "sw_metric"
+ "group": "index_mode"
},
"tags": [
"layer"
diff --git a/pkg/test/measure/testdata/measures/service_traffic.json b/pkg/test/measure/testdata/measures/service_traffic.json
index 887d1adb..1cf0608b 100644
--- a/pkg/test/measure/testdata/measures/service_traffic.json
+++ b/pkg/test/measure/testdata/measures/service_traffic.json
@@ -1,7 +1,7 @@
{
"metadata": {
"name": "service_traffic",
- "group": "sw_metric"
+ "group": "index_mode"
},
"tag_families": [
{
diff --git a/test/cases/init.go b/test/cases/init.go
index 64b60fae..c09946fc 100644
--- a/test/cases/init.go
+++ b/test/cases/init.go
@@ -41,8 +41,8 @@ func Initialize(addr string, now time.Time) {
casesstreamdata.Write(conn, "duplicated", now, 0)
// // measure
interval = time.Minute
- casesmeasuredata.Write(conn, "service_traffic", "sw_metric", "service_traffic_data_old.json", now.AddDate(0, 0, -1), interval)
- casesmeasuredata.Write(conn, "service_traffic", "sw_metric", "service_traffic_data.json", now, interval)
+ casesmeasuredata.Write(conn, "service_traffic", "index_mode", "service_traffic_data_old.json", now.AddDate(0, 0, -1), interval)
+ casesmeasuredata.Write(conn, "service_traffic", "index_mode", "service_traffic_data.json", now, interval)
casesmeasuredata.Write(conn, "service_instance_traffic", "sw_metric", "service_instance_traffic_data.json", now, interval)
casesmeasuredata.Write(conn, "service_cpm_minute", "sw_metric", "service_cpm_minute_data.json", now, interval)
casesmeasuredata.Write(conn, "instance_clr_cpu_minute", "sw_metric", "instance_clr_cpu_minute_data.json", now, interval)
diff --git a/test/cases/measure/data/input/entity.yaml b/test/cases/measure/data/input/entity.yaml
index 8e37f7b8..58d7a100 100644
--- a/test/cases/measure/data/input/entity.yaml
+++ b/test/cases/measure/data/input/entity.yaml
@@ -16,7 +16,7 @@
# under the License.
name: "service_traffic"
-groups: ["sw_metric"]
+groups: ["index_mode"]
tagProjection:
tagFamilies:
- name: "default"
diff --git a/test/cases/measure/data/input/entity_in.yaml b/test/cases/measure/data/input/entity_in.yaml
index b872a268..0bdedbe3 100644
--- a/test/cases/measure/data/input/entity_in.yaml
+++ b/test/cases/measure/data/input/entity_in.yaml
@@ -16,7 +16,7 @@
# under the License.
name: "service_traffic"
-groups: ["sw_metric"]
+groups: ["index_mode"]
tagProjection:
tagFamilies:
- name: "default"
diff --git a/test/cases/measure/data/input/entity_service.yaml b/test/cases/measure/data/input/entity_service.yaml
index b85da375..bf6e8a7c 100644
--- a/test/cases/measure/data/input/entity_service.yaml
+++ b/test/cases/measure/data/input/entity_service.yaml
@@ -16,7 +16,7 @@
# under the License.
name: "service_traffic"
-groups: ["sw_metric"]
+groups: ["index_mode"]
tagProjection:
tagFamilies:
- name: "default"
diff --git a/test/cases/measure/data/input/index_mode_all.yaml b/test/cases/measure/data/input/index_mode_all.yaml
index d0c01450..be5acb7d 100644
--- a/test/cases/measure/data/input/index_mode_all.yaml
+++ b/test/cases/measure/data/input/index_mode_all.yaml
@@ -16,7 +16,7 @@
# under the License.
name: "service_traffic"
-groups: [ "sw_metric" ]
+groups: [ "index_mode" ]
tagProjection:
tagFamilies:
- name: "default"
diff --git a/test/cases/measure/data/input/index_mode_none.yaml b/test/cases/measure/data/input/index_mode_none.yaml
index 0b5e5761..f5575a35 100644
--- a/test/cases/measure/data/input/index_mode_none.yaml
+++ b/test/cases/measure/data/input/index_mode_none.yaml
@@ -16,7 +16,7 @@
# under the License.
name: "service_traffic"
-groups: [ "sw_metric" ]
+groups: [ "index_mode" ]
tagProjection:
tagFamilies:
- name: "default"
diff --git a/test/cases/measure/data/input/index_mode_order_desc.yaml b/test/cases/measure/data/input/index_mode_order_desc.yaml
index 0dd0063d..7184f8a7 100644
--- a/test/cases/measure/data/input/index_mode_order_desc.yaml
+++ b/test/cases/measure/data/input/index_mode_order_desc.yaml
@@ -16,7 +16,7 @@
# under the License.
name: "service_traffic"
-groups: [ "sw_metric" ]
+groups: [ "index_mode" ]
tagProjection:
tagFamilies:
- name: "default"
diff --git a/test/cases/measure/data/input/index_mode_range.yaml b/test/cases/measure/data/input/index_mode_range.yaml
index c881b11c..6e696e22 100644
--- a/test/cases/measure/data/input/index_mode_range.yaml
+++ b/test/cases/measure/data/input/index_mode_range.yaml
@@ -16,7 +16,7 @@
# under the License.
name: "service_traffic"
-groups: [ "sw_metric" ]
+groups: [ "index_mode" ]
tagProjection:
tagFamilies:
- name: "default"
diff --git a/test/cases/measure/data/input/no_field.yaml b/test/cases/measure/data/input/no_field.yaml
index 8743c88c..eadd500b 100644
--- a/test/cases/measure/data/input/no_field.yaml
+++ b/test/cases/measure/data/input/no_field.yaml
@@ -16,7 +16,7 @@
# under the License.
name: "service_traffic"
-groups: ["sw_metric"]
+groups: ["index_mode"]
tagProjection:
tagFamilies:
- name: "default"
diff --git a/test/cases/measure/data/input/tag_filter_int.yaml b/test/cases/measure/data/input/tag_filter_int.yaml
index d8c7d1a0..4658f2e7 100644
--- a/test/cases/measure/data/input/tag_filter_int.yaml
+++ b/test/cases/measure/data/input/tag_filter_int.yaml
@@ -16,7 +16,7 @@
# under the License.
name: "service_traffic"
-groups: ["sw_metric"]
+groups: ["index_mode"]
tagProjection:
tagFamilies:
- name: "default"