From 8679bfe82add59b590e834f9bf20575dbd18711a Mon Sep 17 00:00:00 2001 From: douxu Date: Tue, 28 Jul 2026 16:24:29 +0800 Subject: [PATCH] feat: bootstrap parameter data objects from PostgreSQL to Redis - resolve component and dynamic parameter attributes through project mappings - generate full and local parameter token hashes during startup - validate parameter routes, descriptions, duplicate records, and token ambiguity - track initialized Redis keys for stale cache cleanup - centralize supported parameter attribute groups - distinguish CL3611 phasor and sampled-data identifiers - add parameter initialization and SQL mapping tests --- constants/parameter_table.go | 28 ++- constants/redis.go | 4 + database/query_parameter_initialization.go | 141 +++++++++++ .../query_parameter_initialization_test.go | 158 +++++++++++++ deploy/redis-test-data/util/rand.go | 3 +- main.go | 12 + model/data_object_token.go | 14 +- model/measurement_protol_model.go | 35 ++- model/parameter_data_object_init.go | 223 ++++++++++++++++++ model/parameter_data_object_init_test.go | 124 ++++++++++ sql/data_object_measurement.go | 2 +- sql/data_object_parameter.go | 2 +- sql/data_object_parameter_initialization.go | 124 ++++++++++ sql/topologic.go | 2 +- 14 files changed, 851 insertions(+), 21 deletions(-) create mode 100644 database/query_parameter_initialization.go create mode 100644 database/query_parameter_initialization_test.go create mode 100644 model/parameter_data_object_init.go create mode 100644 model/parameter_data_object_init_test.go create mode 100644 sql/data_object_parameter_initialization.go diff --git a/constants/parameter_table.go b/constants/parameter_table.go index bcbb12c..2ab4702 100644 --- a/constants/parameter_table.go +++ b/constants/parameter_table.go @@ -3,7 +3,9 @@ package constants import "strings" -var supportedParameterTableSuffixes = [...]string{ +const ComponentParameterAttributeGroup = "component" + +var supportedDynamicParameterAttributeGroups = [...]string{ "base_extend", "rated", "setup", @@ -14,10 +16,32 @@ var supportedParameterTableSuffixes = [...]string{ "behavior", } +// IsSupportedParameterAttributeGroup reports whether token6 identifies a +// parameter attribute group supported by the data-object APIs. +func IsSupportedParameterAttributeGroup(group string) bool { + if group == ComponentParameterAttributeGroup { + return true + } + for _, supportedGroup := range supportedDynamicParameterAttributeGroups { + if group == supportedGroup { + return true + } + } + return false +} + +// SupportedDynamicParameterAttributeGroups returns the token6 values backed by +// project_manager dynamic tables. +func SupportedDynamicParameterAttributeGroups() []string { + groups := make([]string, len(supportedDynamicParameterAttributeGroups)) + copy(groups, supportedDynamicParameterAttributeGroups[:]) + return groups +} + // IsSupportedParameterTableName reports whether a dynamic parameter table has // one of the supported attribute-group suffixes. func IsSupportedParameterTableName(tableName string) bool { - for _, suffix := range supportedParameterTableSuffixes { + for _, suffix := range supportedDynamicParameterAttributeGroups { if strings.HasSuffix(tableName, "_"+suffix) { return true } diff --git a/constants/redis.go b/constants/redis.go index 3ddbcef..6a3adb0 100644 --- a/constants/redis.go +++ b/constants/redis.go @@ -4,4 +4,8 @@ package constants const ( // RedisSearchDictName define redis search dictionary name RedisSearchDictName = "search_suggestions_dict" + + // RedisParameterDataObjectKeySet tracks parameter hashes created during + // startup so stale parameter data-object keys can be removed safely. + RedisParameterDataObjectKeySet = "modelrt:parameter-data-object:keys" ) diff --git a/database/query_parameter_initialization.go b/database/query_parameter_initialization.go new file mode 100644 index 0000000..f256070 --- /dev/null +++ b/database/query_parameter_initialization.go @@ -0,0 +1,141 @@ +// Package database define database operation functions +package database + +import ( + "context" + "fmt" + + "modelRT/constants" + "modelRT/model" + "modelRT/sql" + + "gorm.io/gorm" +) + +type parameterInitializationRoute struct { + TableName string `gorm:"column:name"` + ModelName string `gorm:"column:tag"` + AttributeGroup string `gorm:"column:group_name"` +} + +// QueryParameterInitializationRecords loads every parameter accepted by the +// data-object query API. Dynamic parameters are resolved through +// project_manager; component parameters are read directly from component. +func QueryParameterInitializationRecords(ctx context.Context, db *gorm.DB) ([]model.ParameterInitializationRecord, error) { + if db == nil { + return nil, fmt.Errorf("postgres client is nil") + } + + var routes []parameterInitializationRoute + if err := db.WithContext(ctx). + Raw( + compactParameterSQL(sql.ParameterInitializationRoutes), + constants.SupportedDynamicParameterAttributeGroups(), + ). + Scan(&routes).Error; err != nil { + return nil, fmt.Errorf("query parameter initialization routes: %w", err) + } + if err := validateParameterInitializationRoutes(routes); err != nil { + return nil, err + } + + records := make([]model.ParameterInitializationRecord, 0) + for _, route := range routes { + quotedTableName := `"` + route.TableName + `"` + query := compactParameterSQL( + fmt.Sprintf(sql.DynamicParameterInitializationRows, quotedTableName), + ) + + var tableRecords []model.ParameterInitializationRecord + if err := db.WithContext(ctx). + Raw(query, route.TableName, route.ModelName, route.AttributeGroup). + Scan(&tableRecords).Error; err != nil { + return nil, fmt.Errorf( + "query parameter initialization table %q for model %q group %q: %w", + route.TableName, + route.ModelName, + route.AttributeGroup, + err, + ) + } + if err := validateParameterInitializationRecords(tableRecords); err != nil { + return nil, fmt.Errorf("validate parameter initialization table %q: %w", route.TableName, err) + } + records = append(records, tableRecords...) + } + + var componentRecords []model.ParameterInitializationRecord + if err := db.WithContext(ctx). + Raw(compactParameterSQL(sql.ComponentParameterInitializationRows)). + Scan(&componentRecords).Error; err != nil { + return nil, fmt.Errorf("query component parameter initialization records: %w", err) + } + if err := validateParameterInitializationRecords(componentRecords); err != nil { + return nil, fmt.Errorf("validate component parameter initialization records: %w", err) + } + records = append(records, componentRecords...) + return records, nil +} + +func validateParameterInitializationRoutes(routes []parameterInitializationRoute) error { + seen := make(map[string]struct{}, len(routes)) + for _, route := range routes { + if !validParameterTableName(route.TableName) { + return fmt.Errorf("project_manager contains unsupported parameter table name %q", route.TableName) + } + if !constants.IsSupportedParameterAttributeGroup(route.AttributeGroup) || + route.AttributeGroup == constants.ComponentParameterAttributeGroup { + return fmt.Errorf("project_manager contains unsupported dynamic attribute group %q", route.AttributeGroup) + } + + key := route.ModelName + "\x00" + route.AttributeGroup + if _, exists := seen[key]; exists { + return fmt.Errorf( + "model %q and attribute group %q match more than one project_manager record", + route.ModelName, + route.AttributeGroup, + ) + } + seen[key] = struct{}{} + } + return nil +} + +func validateParameterInitializationRecords(records []model.ParameterInitializationRecord) error { + for _, record := range records { + switch record.DynamicRecordCount { + case 1: + case 0: + return fmt.Errorf( + "component %q has no %q parameter record", + record.ComponentTag, + record.AttributeGroup, + ) + default: + return fmt.Errorf( + "component %q has %d %q parameter records", + record.ComponentTag, + record.DynamicRecordCount, + record.AttributeGroup, + ) + } + if record.AttributeName == "" || record.AttributeType == "" { + return fmt.Errorf( + "component %q group %q contains an invalid parameter column", + record.ComponentTag, + record.AttributeGroup, + ) + } + switch record.DescriptionCount { + case 0: + return fmt.Errorf("parameter description not found for attribute %q", record.AttributeName) + case 1: + if !record.Description.Valid { + return fmt.Errorf("parameter description is null for attribute %q", record.AttributeName) + } + default: + return fmt.Errorf("ambiguous parameter description for attribute %q", record.AttributeName) + } + } + return nil +} diff --git a/database/query_parameter_initialization_test.go b/database/query_parameter_initialization_test.go new file mode 100644 index 0000000..63eb250 --- /dev/null +++ b/database/query_parameter_initialization_test.go @@ -0,0 +1,158 @@ +// Package database define database operation functions +package database + +import ( + "context" + "database/sql" + "database/sql/driver" + "strings" + "testing" + + "modelRT/constants" + "modelRT/model" + modelsql "modelRT/sql" + + "github.com/DATA-DOG/go-sqlmock" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gorm.io/driver/postgres" + "gorm.io/gorm" +) + +func TestParameterInitializationSQLUsesStationIDAndExcludesItFromComponentAttributes(t *testing.T) { + dynamicSQL := compactParameterSQL(modelsql.DynamicParameterInitializationRows) + componentSQL := compactParameterSQL(modelsql.ComponentParameterInitializationRows) + + assert.Contains(t, dynamicSQL, "component.station_id = station.id") + assert.Contains(t, componentSQL, "component.station_id = station.id") + assert.Contains(t, componentSQL, "to_jsonb(component) - 'station_id'") + assert.Contains(t, dynamicSQL, "component.nspath <> ''") + assert.Contains(t, dynamicSQL, "component.tag <> ''") + assert.Contains(t, componentSQL, "component.nspath <> ''") + assert.Contains(t, componentSQL, "component.tag <> ''") + assert.NotContains(t, strings.ToLower(componentSQL), "component.station = station.tagname") +} + +func TestQueryParameterInitializationRecordsJoinsHierarchyAndDynamicTable(t *testing.T) { + sqlDB, mock, err := sqlmock.New() + require.NoError(t, err) + t.Cleanup(func() { _ = sqlDB.Close() }) + + db, err := gorm.Open(postgres.New(postgres.Config{Conn: sqlDB}), &gorm.Config{}) + require.NoError(t, err) + + groups := constants.SupportedDynamicParameterAttributeGroups() + routeArgs := make([]driver.Value, len(groups)) + for index, group := range groups { + routeArgs[index] = group + } + mock.ExpectQuery(`(?s)SELECT name, tag, group_name.*FROM project_manager.*WHERE group_name IN`). + WithArgs(routeArgs...). + WillReturnRows(sqlmock.NewRows([]string{"name", "tag", "group_name"}). + AddRow("cable_cable_demo_base_extend", "cable_demo", "base_extend")) + + mock.ExpectQuery(`(?s)WITH dynamic_rows AS.*FROM public\."cable_cable_demo_base_extend".*FROM public\.grid.*INNER JOIN public\.zone.*INNER JOIN public\.station.*INNER JOIN public\.component.*INNER JOIN public\.project_manager.*jsonb_each`). + WithArgs("cable_cable_demo_base_extend", "cable_demo", "base_extend"). + WillReturnRows(parameterInitializationRows(). + AddRow( + "grid000", + "zone000", + "station000", + true, + "component-uuid", + "nspath", + "component", + "base_extend", + "vnom_kv", + "220.0", + "DOUBLE PRECISION", + "额定电压", + int64(1), + int64(1), + )) + + mock.ExpectQuery(`(?s)SELECT.*FROM public\.grid.*INNER JOIN public\.zone.*INNER JOIN public\.station.*INNER JOIN public\.component.*jsonb_each`). + WillReturnRows(parameterInitializationRows(). + AddRow( + "grid000", + "zone000", + "station000", + true, + "component-uuid", + "nspath", + "component", + "component", + "description", + `"组件"`, + "CHARACTER VARYING(512)", + "组件名称", + int64(1), + int64(1), + )) + + records, err := QueryParameterInitializationRecords(context.Background(), db) + require.NoError(t, err) + require.Len(t, records, 2) + assert.Equal(t, "vnom_kv", records[0].AttributeName) + assert.Equal(t, "description", records[1].AttributeName) + require.NoError(t, mock.ExpectationsWereMet()) +} + +func TestValidateParameterInitializationRoutesRejectsAmbiguousMapping(t *testing.T) { + routes := []parameterInitializationRoute{ + {TableName: "cable_demo_stable", ModelName: "cable_demo", AttributeGroup: "stable"}, + {TableName: "cable_other_stable", ModelName: "cable_demo", AttributeGroup: "stable"}, + } + + err := validateParameterInitializationRoutes(routes) + require.Error(t, err) + assert.Contains(t, err.Error(), "more than one project_manager record") +} + +func TestValidateParameterInitializationRecordsEnforcesDescriptionAndRowUniqueness(t *testing.T) { + validRecord := modelParameterInitializationRecordForTest() + + missingDescription := validRecord + missingDescription.Description = sql.NullString{} + missingDescription.DescriptionCount = 0 + err := validateParameterInitializationRecords([]model.ParameterInitializationRecord{missingDescription}) + require.Error(t, err) + assert.Contains(t, err.Error(), "description not found") + + duplicateRow := validRecord + duplicateRow.DynamicRecordCount = 2 + err = validateParameterInitializationRecords([]model.ParameterInitializationRecord{duplicateRow}) + require.Error(t, err) + assert.Contains(t, err.Error(), "has 2") +} + +func parameterInitializationRows() *sqlmock.Rows { + return sqlmock.NewRows([]string{ + "grid_tag", + "zone_tag", + "station_tag", + "station_is_local", + "component_uuid", + "component_nspath", + "component_tag", + "attribute_group", + "attribute_name", + "attribute_value", + "attribute_type", + "description", + "description_count", + "dynamic_record_count", + }) +} + +func modelParameterInitializationRecordForTest() model.ParameterInitializationRecord { + return model.ParameterInitializationRecord{ + ComponentTag: "component", + AttributeGroup: "stable", + AttributeName: "attribute", + AttributeType: "INTEGER", + Description: sql.NullString{String: "属性", Valid: true}, + DescriptionCount: 1, + DynamicRecordCount: 1, + } +} diff --git a/deploy/redis-test-data/util/rand.go b/deploy/redis-test-data/util/rand.go index 2e31302..6a80b17 100644 --- a/deploy/redis-test-data/util/rand.go +++ b/deploy/redis-test-data/util/rand.go @@ -61,8 +61,9 @@ func ProcessMeasurements(measurements []orm.Measurement) map[string]CalculationR station, _ := ioAddress["station"].(string) device, _ := ioAddress["device"].(string) channel, _ := ioAddress["channel"].(string) + option, _ := ioAddress["option"].(string) - result := strings.ToLower(fmt.Sprintf("%s:%s:phasor:%s", station, device, channel)) + result := strings.ToLower(fmt.Sprintf("%s:%s:phasor:%s:%s", station, device, channel, option)) if measurement.EventPlan == nil { continue } diff --git a/main.go b/main.go index b9c38e3..95754b7 100644 --- a/main.go +++ b/main.go @@ -247,6 +247,18 @@ func main() { panic(err) } + parameterRecords, err := database.QueryParameterInitializationRecords(ctx, tx) + if err != nil { + logger.Error(ctx, "load parameter data objects from postgres failed", "error", err) + panic(err) + } + + err = model.InitializeParameterDataObjects(ctx, parameterRecords) + if err != nil { + logger.Error(ctx, "initialize parameter data objects failed", "error", err) + panic(err) + } + allMeasurement, err := database.GetAllMeasurements(ctx, tx) if err != nil { logger.Error(ctx, "load topologic info from postgres failed", "error", err) diff --git a/model/data_object_token.go b/model/data_object_token.go index 32b2f04..cd1aa9d 100644 --- a/model/data_object_token.go +++ b/model/data_object_token.go @@ -9,18 +9,6 @@ import ( "modelRT/constants" ) -var parameterAttributeGroups = map[string]struct{}{ - "component": {}, - "base_extend": {}, - "rated": {}, - "setup": {}, - "model": {}, - "stable": {}, - "craft": {}, - "integrity": {}, - "behavior": {}, -} - // ClassifyDataObjectToken determines whether token identifies a parameter or a // measurement. Seven-part and four-part tokens are classified by token6, while // two-part tokens are treated as measurements at the current stage. @@ -40,7 +28,7 @@ func ClassifyDataObjectToken(token string) (constants.DataObjectType, error) { } token6 := parts[token6Index] - if _, ok := parameterAttributeGroups[token6]; ok { + if constants.IsSupportedParameterAttributeGroup(token6) { return constants.DataObjectTypeParameter, nil } if token6 == "bay" { diff --git a/model/measurement_protol_model.go b/model/measurement_protol_model.go index cfc4a1f..95b9e89 100644 --- a/model/measurement_protol_model.go +++ b/model/measurement_protol_model.go @@ -11,6 +11,14 @@ import ( "modelRT/constants" ) +const ( + // CL3611DataSourceTypePhasor define identifies CL3611 phasor data source. + CL3611DataSourceTypePhasor = 1 + + // CL3611DataSourceTypeSample define identifies CL3611 sampled data source. + CL3611DataSourceTypeSample = 2 +) + // MeasurementDataSource define measurement data source struct type MeasurementDataSource struct { Type int `json:"type"` @@ -222,6 +230,11 @@ func GenerateMeasureIdentifier(source map[string]any) (string, error) { switch regType { case constants.DataSourceTypeCL3611: + rawDtype, ok := ioAddress["dtype"].(float64) + if !ok { + return "", fmt.Errorf("CL3611:invalid or missing dtype field") + } + station, ok := ioAddress["station"].(string) if !ok { return "", fmt.Errorf("CL3611:invalid or missing station field") @@ -235,7 +248,21 @@ func GenerateMeasureIdentifier(source map[string]any) (string, error) { if !ok { return "", fmt.Errorf("CL3611:invalid or missing channel field") } - return concatCL361WithPlus(station, device, channel), nil + + optinon, ok := ioAddress["option"].(string) + if !ok { + return "", fmt.Errorf("CL3611:invalid or missing optinon field") + } + dtype := int(rawDtype) + switch dtype { + case CL3611DataSourceTypePhasor: + return buildCL3611PhasorIdentifier(station, device, channel, optinon), nil + case CL3611DataSourceTypeSample: + return buildCL3611SampleIdentifier(station, device, channel), nil + default: + return "", fmt.Errorf("CL3611:unsupported dtype %d", dtype) + } + case constants.DataSourceTypePower104: station, ok := ioAddress["station"].(string) if !ok { @@ -270,6 +297,10 @@ func concatP104WithPlus(station string, packet int, offset int) string { return strings.ToLower(station + ":104:" + packetStr + ":" + offsetStr) } -func concatCL361WithPlus(station, device, channel string) string { +func buildCL3611SampleIdentifier(station, device, channel string) string { return strings.ToLower(station + ":" + device + ":" + "phasor" + ":" + channel) } + +func buildCL3611PhasorIdentifier(station, device, channel, option string) string { + return strings.ToLower(station + ":" + device + ":" + "phasor" + ":" + channel + ":" + option) +} diff --git a/model/parameter_data_object_init.go b/model/parameter_data_object_init.go new file mode 100644 index 0000000..4819685 --- /dev/null +++ b/model/parameter_data_object_init.go @@ -0,0 +1,223 @@ +package model + +import ( + "bytes" + "context" + "database/sql" + "encoding/json" + "fmt" + "strings" + + "modelRT/constants" + "modelRT/diagram" + "modelRT/logger" + + "github.com/redis/go-redis/v9" +) + +const parameterDataObjectPipelineSize = 500 + +type parameterDataObjectHash struct { + Key string + Fields map[string]any +} + +// ParameterInitializationRecord contains one parameter attribute together with +// the hierarchy and metadata needed to create its Redis data-object hashes. +type ParameterInitializationRecord struct { + GridTag string `gorm:"column:grid_tag"` + ZoneTag string `gorm:"column:zone_tag"` + StationTag string `gorm:"column:station_tag"` + StationIsLocal bool `gorm:"column:station_is_local"` + ComponentUUID string `gorm:"column:component_uuid"` + ComponentNSPath string `gorm:"column:component_nspath"` + ComponentTag string `gorm:"column:component_tag"` + AttributeGroup string `gorm:"column:attribute_group"` + AttributeName string `gorm:"column:attribute_name"` + AttributeValue string `gorm:"column:attribute_value"` + AttributeType string `gorm:"column:attribute_type"` + Description sql.NullString `gorm:"column:description"` + DescriptionCount int64 `gorm:"column:description_count"` + DynamicRecordCount int64 `gorm:"column:dynamic_record_count"` +} + +// InitializeParameterDataObjects creates full and local-short Redis hashes for +// parameter attributes previously loaded from PostgreSQL. +func InitializeParameterDataObjects(ctx context.Context, records []ParameterInitializationRecord) error { + hashes, err := buildParameterDataObjectHashes(records) + if err != nil { + return fmt.Errorf("build parameter data-object hashes: %w", err) + } + if err := storeParameterDataObjectHashes(ctx, diagram.GetRedisClientInstance(), hashes); err != nil { + return fmt.Errorf("store parameter data-object hashes in redis: %w", err) + } + + logger.Info(ctx, "initialize parameter data objects completed", + "postgres_record_count", len(records), + "redis_hash_count", len(hashes), + ) + return nil +} + +func buildParameterDataObjectHashes(records []ParameterInitializationRecord) ([]parameterDataObjectHash, error) { + hashes := make([]parameterDataObjectHash, 0, len(records)*2) + seenKeys := make(map[string]string, len(records)*2) + + for _, record := range records { + value, err := parameterRedisValue(record.AttributeValue) + if err != nil { + return nil, fmt.Errorf( + "decode value for component %q group %q attribute %q: %w", + record.ComponentTag, + record.AttributeGroup, + record.AttributeName, + err, + ) + } + + fullToken := strings.Join([]string{ + record.GridTag, + record.ZoneTag, + record.StationTag, + record.ComponentNSPath, + record.ComponentTag, + record.AttributeGroup, + record.AttributeName, + }, ".") + shortToken := strings.Join([]string{ + record.ComponentNSPath, + record.ComponentTag, + record.AttributeGroup, + record.AttributeName, + }, ".") + + if err := validateInitializedParameterToken(fullToken); err != nil { + return nil, err + } + fields := map[string]any{ + "value": value, + "meta": "PARAM", + "type": record.AttributeType, + "name": shortToken, + "description": record.Description.String, + "id": fullToken, + } + owner := strings.Join([]string{ + record.ComponentUUID, + record.AttributeGroup, + record.AttributeName, + }, "/") + if err := appendParameterDataObjectHash(&hashes, seenKeys, fullToken, owner, fields); err != nil { + return nil, err + } + + if !record.StationIsLocal { + continue + } + if err := validateInitializedParameterToken(shortToken); err != nil { + return nil, err + } + if err := appendParameterDataObjectHash(&hashes, seenKeys, shortToken, owner, fields); err != nil { + return nil, err + } + } + return hashes, nil +} + +func validateInitializedParameterToken(token string) error { + dataObjectType, err := ClassifyDataObjectToken(token) + if err != nil { + return fmt.Errorf("generated invalid parameter token %q: %w", token, err) + } + if dataObjectType != constants.DataObjectTypeParameter { + return fmt.Errorf("generated token %q is not a parameter", token) + } + return nil +} + +func appendParameterDataObjectHash( + hashes *[]parameterDataObjectHash, + seenKeys map[string]string, + key string, + owner string, + fields map[string]any, +) error { + if existingOwner, exists := seenKeys[key]; exists { + return fmt.Errorf( + "ambiguous parameter token %q is produced by %q and %q", + key, + existingOwner, + owner, + ) + } + seenKeys[key] = owner + *hashes = append(*hashes, parameterDataObjectHash{Key: key, Fields: fields}) + return nil +} + +func parameterRedisValue(rawJSON string) (any, error) { + decoder := json.NewDecoder(bytes.NewBufferString(rawJSON)) + decoder.UseNumber() + + var value any + if err := decoder.Decode(&value); err != nil { + return nil, err + } + switch typedValue := value.(type) { + case nil: + return "null", nil + case string: + return typedValue, nil + case json.Number: + return typedValue.String(), nil + case bool: + return typedValue, nil + default: + encoded, err := json.Marshal(typedValue) + if err != nil { + return nil, err + } + return string(encoded), nil + } +} + +func storeParameterDataObjectHashes( + ctx context.Context, + rdb *redis.Client, + hashes []parameterDataObjectHash, +) error { + if rdb == nil { + return fmt.Errorf("redis client is nil") + } + + oldKeys, err := rdb.SMembers(ctx, constants.RedisParameterDataObjectKeySet).Result() + if err != nil { + return fmt.Errorf("query previously initialized parameter keys: %w", err) + } + cleanupPipeline := rdb.TxPipeline() + for start := 0; start < len(oldKeys); start += parameterDataObjectPipelineSize { + end := min(start+parameterDataObjectPipelineSize, len(oldKeys)) + cleanupPipeline.Del(ctx, oldKeys[start:end]...) + } + cleanupPipeline.Del(ctx, constants.RedisParameterDataObjectKeySet) + if _, err := cleanupPipeline.Exec(ctx); err != nil { + return fmt.Errorf("remove stale parameter data-object hashes: %w", err) + } + + for start := 0; start < len(hashes); start += parameterDataObjectPipelineSize { + end := min(start+parameterDataObjectPipelineSize, len(hashes)) + pipeline := rdb.TxPipeline() + keyMembers := make([]any, 0, end-start) + for _, hash := range hashes[start:end] { + pipeline.HSet(ctx, hash.Key, hash.Fields) + keyMembers = append(keyMembers, hash.Key) + } + if len(keyMembers) > 0 { + pipeline.SAdd(ctx, constants.RedisParameterDataObjectKeySet, keyMembers...) + } + if _, err := pipeline.Exec(ctx); err != nil { + return fmt.Errorf("write parameter data-object hash batch starting at %d: %w", start, err) + } + } + return nil +} diff --git a/model/parameter_data_object_init_test.go b/model/parameter_data_object_init_test.go new file mode 100644 index 0000000..0f3a600 --- /dev/null +++ b/model/parameter_data_object_init_test.go @@ -0,0 +1,124 @@ +package model + +import ( + "database/sql" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +func TestBuildParameterDataObjectHashesCreatesFullAndLocalKeys(t *testing.T) { + records := []ParameterInitializationRecord{ + { + GridTag: "grid000", + ZoneTag: "zone000", + StationTag: "station000", + StationIsLocal: true, + ComponentUUID: "component-uuid", + ComponentNSPath: "220kV_xuefulu1", + ComponentTag: "cable_22", + AttributeGroup: "base_extend", + AttributeName: "vnom_kv", + AttributeValue: "7800.00", + AttributeType: "DOUBLE PRECISION", + Description: sql.NullString{String: "额定电压", Valid: true}, + DescriptionCount: 1, + DynamicRecordCount: 1, + }, + } + + hashes, err := buildParameterDataObjectHashes(records) + require.NoError(t, err) + require.Len(t, hashes, 2) + + fullToken := "grid000.zone000.station000.220kV_xuefulu1.cable_22.base_extend.vnom_kv" + shortToken := "220kV_xuefulu1.cable_22.base_extend.vnom_kv" + assert.Equal(t, fullToken, hashes[0].Key) + assert.Equal(t, shortToken, hashes[1].Key) + assert.Equal(t, "7800.00", hashes[0].Fields["value"]) + assert.Equal(t, "PARAM", hashes[0].Fields["meta"]) + assert.Equal(t, "DOUBLE PRECISION", hashes[0].Fields["type"]) + assert.Equal(t, shortToken, hashes[0].Fields["name"]) + assert.Equal(t, "额定电压", hashes[0].Fields["description"]) + assert.Equal(t, fullToken, hashes[0].Fields["id"]) + assert.Equal(t, hashes[0].Fields, hashes[1].Fields) +} + +func TestBuildParameterDataObjectHashesSkipsShortKeyForNonLocalStation(t *testing.T) { + records := []ParameterInitializationRecord{ + { + GridTag: "grid", + ZoneTag: "zone", + StationTag: "station", + StationIsLocal: false, + ComponentUUID: "component-uuid", + ComponentNSPath: "nspath", + ComponentTag: "component", + AttributeGroup: "stable", + AttributeName: "attribute", + AttributeValue: "true", + AttributeType: "BOOLEAN", + Description: sql.NullString{String: "属性", Valid: true}, + DescriptionCount: 1, + DynamicRecordCount: 1, + }, + } + + hashes, err := buildParameterDataObjectHashes(records) + require.NoError(t, err) + require.Len(t, hashes, 1) + assert.Equal(t, "grid.zone.station.nspath.component.stable.attribute", hashes[0].Key) + assert.Equal(t, true, hashes[0].Fields["value"]) +} + +func TestBuildParameterDataObjectHashesRejectsAmbiguousShortToken(t *testing.T) { + baseRecord := ParameterInitializationRecord{ + GridTag: "grid1", + ZoneTag: "zone1", + StationTag: "station1", + StationIsLocal: true, + ComponentUUID: "component-uuid-1", + ComponentNSPath: "nspath", + ComponentTag: "component", + AttributeGroup: "stable", + AttributeName: "attribute", + AttributeValue: "1", + AttributeType: "INTEGER", + Description: sql.NullString{String: "属性", Valid: true}, + DescriptionCount: 1, + DynamicRecordCount: 1, + } + otherRecord := baseRecord + otherRecord.GridTag = "grid2" + otherRecord.ZoneTag = "zone2" + otherRecord.StationTag = "station2" + otherRecord.ComponentUUID = "component-uuid-2" + + _, err := buildParameterDataObjectHashes([]ParameterInitializationRecord{baseRecord, otherRecord}) + require.Error(t, err) + assert.Contains(t, err.Error(), "ambiguous parameter token") + assert.Contains(t, err.Error(), "nspath.component.stable.attribute") +} + +func TestParameterRedisValuePreservesHashRepresentations(t *testing.T) { + tests := []struct { + name string + rawValue string + expected any + }{ + {name: "null", rawValue: "null", expected: "null"}, + {name: "string", rawValue: `"text"`, expected: "text"}, + {name: "number precision", rawValue: "1234567890.123456789", expected: "1234567890.123456789"}, + {name: "boolean", rawValue: "true", expected: true}, + {name: "object", rawValue: `{"key":"value"}`, expected: `{"key":"value"}`}, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + actual, err := parameterRedisValue(test.rawValue) + require.NoError(t, err) + assert.Equal(t, test.expected, actual) + }) + } +} diff --git a/sql/data_object_measurement.go b/sql/data_object_measurement.go index bd734f6..833cb51 100644 --- a/sql/data_object_measurement.go +++ b/sql/data_object_measurement.go @@ -1,4 +1,4 @@ -// Package sql defines reusable database SQL statements. +// Package sql defines reusable database SQL statements package sql const ( diff --git a/sql/data_object_parameter.go b/sql/data_object_parameter.go index f3e2ba2..8951df6 100644 --- a/sql/data_object_parameter.go +++ b/sql/data_object_parameter.go @@ -1,4 +1,4 @@ -// Package sql defines reusable database SQL statements. +// Package sql defines reusable database SQL statements package sql const ( diff --git a/sql/data_object_parameter_initialization.go b/sql/data_object_parameter_initialization.go new file mode 100644 index 0000000..11cfd27 --- /dev/null +++ b/sql/data_object_parameter_initialization.go @@ -0,0 +1,124 @@ +// Package sql defines reusable database SQL statements +package sql + +const ( + // ParameterInitializationRoutes returns the dynamic-table mappings used by + // supported parameter attribute groups. + ParameterInitializationRoutes = `SELECT name, tag, group_name + FROM project_manager + WHERE group_name IN ?` + + // DynamicParameterInitializationRows joins a dynamic parameter table to its + // component hierarchy and project_manager route. The table identifier is + // inserted only after application-level identifier and allowlist checks. + DynamicParameterInitializationRows = `WITH dynamic_rows AS ( + SELECT dynamic_record.*, + COUNT(*) OVER ( + PARTITION BY dynamic_record.global_uuid, dynamic_record.attribute_group + ) AS initialization_record_count + FROM public.%[1]s AS dynamic_record + ) + SELECT + grid.tagname AS grid_tag, + zone.tagname AS zone_tag, + station.tagname AS station_tag, + station.is_local AS station_is_local, + component.global_uuid::text AS component_uuid, + component.nspath AS component_nspath, + component.tag AS component_tag, + project.group_name AS attribute_group, + attribute.key AS attribute_name, + attribute.value::text AS attribute_value, + UPPER(pg_catalog.format_type(column_attribute.atttypid, column_attribute.atttypmod)) AS attribute_type, + attribute_description.description, + attribute_description.description_count, + dynamic_row.initialization_record_count AS dynamic_record_count + FROM public.grid AS grid + INNER JOIN public.zone AS zone ON zone.grid_id = grid.id + INNER JOIN public.station AS station ON station.zone_id = zone.id + INNER JOIN public.component AS component ON component.station_id = station.id + INNER JOIN public.project_manager AS project + ON project.tag = component.model_name + INNER JOIN dynamic_rows AS dynamic_row + ON dynamic_row.global_uuid = component.global_uuid + AND dynamic_row.attribute_group = project.group_name + CROSS JOIN LATERAL jsonb_each( + to_jsonb(dynamic_row) + - 'id' + - 'global_uuid' + - 'attribute_group' + - 'initialization_record_count' + ) AS attribute + INNER JOIN pg_catalog.pg_namespace AS table_namespace + ON table_namespace.nspname = 'public' + INNER JOIN pg_catalog.pg_class AS parameter_table + ON parameter_table.relnamespace = table_namespace.oid + AND parameter_table.relname = project.name + INNER JOIN pg_catalog.pg_attribute AS column_attribute + ON column_attribute.attrelid = parameter_table.oid + AND column_attribute.attname = attribute.key + AND column_attribute.attnum > 0 + AND NOT column_attribute.attisdropped + LEFT JOIN LATERAL ( + SELECT + MIN(basic_attribute.attribute_name) AS description, + COUNT(*) AS description_count + FROM basic.attribute AS basic_attribute + WHERE basic_attribute.attribute = attribute.key + ) AS attribute_description ON TRUE + WHERE project.name = ? + AND project.tag = ? + AND project.group_name = ? + AND grid.tagname <> '' + AND zone.tagname <> '' + AND station.tagname <> '' + AND component.nspath <> '' + AND component.tag <> ''` + + // ComponentParameterInitializationRows expands the component table into one + // row per queryable component attribute while retaining the full hierarchy. + ComponentParameterInitializationRows = `SELECT + grid.tagname AS grid_tag, + zone.tagname AS zone_tag, + station.tagname AS station_tag, + station.is_local AS station_is_local, + component.global_uuid::text AS component_uuid, + component.nspath AS component_nspath, + component.tag AS component_tag, + 'component' AS attribute_group, + attribute.key AS attribute_name, + attribute.value::text AS attribute_value, + UPPER(pg_catalog.format_type(column_attribute.atttypid, column_attribute.atttypmod)) AS attribute_type, + attribute_description.description, + attribute_description.description_count, + 1::bigint AS dynamic_record_count + FROM public.grid AS grid + INNER JOIN public.zone AS zone ON zone.grid_id = grid.id + INNER JOIN public.station AS station ON station.zone_id = zone.id + INNER JOIN public.component AS component ON component.station_id = station.id + CROSS JOIN LATERAL jsonb_each( + to_jsonb(component) - 'station_id' + ) AS attribute + INNER JOIN pg_catalog.pg_namespace AS table_namespace + ON table_namespace.nspname = 'public' + INNER JOIN pg_catalog.pg_class AS component_table + ON component_table.relnamespace = table_namespace.oid + AND component_table.relname = 'component' + INNER JOIN pg_catalog.pg_attribute AS column_attribute + ON column_attribute.attrelid = component_table.oid + AND column_attribute.attname = attribute.key + AND column_attribute.attnum > 0 + AND NOT column_attribute.attisdropped + LEFT JOIN LATERAL ( + SELECT + MIN(basic_attribute.attribute_name) AS description, + COUNT(*) AS description_count + FROM basic.attribute AS basic_attribute + WHERE basic_attribute.attribute = attribute.key + ) AS attribute_description ON TRUE + WHERE grid.tagname <> '' + AND zone.tagname <> '' + AND station.tagname <> '' + AND component.nspath <> '' + AND component.tag <> ''` +) diff --git a/sql/topologic.go b/sql/topologic.go index 9c38eae..ee48e7d 100644 --- a/sql/topologic.go +++ b/sql/topologic.go @@ -1,4 +1,4 @@ -// Package sql define database sql statement +// Package sql defines reusable database SQL statements package sql // RecursiveSQL define topologic table recursive query statement