diff --git a/backend/modules/evaluation/domain/component/conf.go b/backend/modules/evaluation/domain/component/conf.go index cde437b52..be5675eda 100644 --- a/backend/modules/evaluation/domain/component/conf.go +++ b/backend/modules/evaluation/domain/component/conf.go @@ -18,4 +18,4 @@ type IConfiger interface { GetExptTurnResultFilterBmqProducerCfg(ctx context.Context) *entity.BmqProducerCfg GetCKDBName(ctx context.Context) *entity.CKDBConfig GetExptExportWhiteList(ctx context.Context) *entity.ExptExportWhiteList -} +} \ No newline at end of file diff --git a/backend/modules/evaluation/infra/repo/experiment/ck/convertor/expt_turn_result_filter.go b/backend/modules/evaluation/infra/repo/experiment/ck/convertor/expt_turn_result_filter.go index 637dee928..842a9853b 100644 --- a/backend/modules/evaluation/infra/repo/experiment/ck/convertor/expt_turn_result_filter.go +++ b/backend/modules/evaluation/infra/repo/experiment/ck/convertor/expt_turn_result_filter.go @@ -26,11 +26,11 @@ func ExptTurnResultFilterEntity2PO(filterEntity *entity.ExptTurnResultFilterEnti } return &model.ExptTurnResultFilter{ - SpaceID: stringifyInt64(filterEntity.SpaceID), - ExptID: stringifyInt64(filterEntity.ExptID), - ItemID: stringifyInt64(filterEntity.ItemID), + SpaceID: strconv.FormatInt(filterEntity.SpaceID, 10), + ExptID: strconv.FormatInt(filterEntity.ExptID, 10), + ItemID: strconv.FormatInt(filterEntity.ItemID, 10), ItemIdx: filterEntity.ItemIdx, - TurnID: stringifyInt64(filterEntity.TurnID), + TurnID: strconv.FormatInt(filterEntity.TurnID, 10), Status: int32(filterEntity.Status), EvalTargetData: filterEntity.EvalTargetData, EvaluatorScore: filterEntity.EvaluatorScore, @@ -39,6 +39,7 @@ func ExptTurnResultFilterEntity2PO(filterEntity *entity.ExptTurnResultFilterEnti AnnotationString: filterEntity.AnnotationString, CreatedDate: filterEntity.CreatedDate, EvalSetVersionID: strconv.FormatInt(filterEntity.EvalSetVersionID, 10), + UpdatedAt: filterEntity.UpdatedAt, } } @@ -70,11 +71,6 @@ func ExptTurnResultFilterPO2Entity(filterPO *model.ExptTurnResultFilter) *entity } } -// stringifyInt64 将 int64 转换为 string -func stringifyInt64(i int64) string { - return string(rune(i)) -} - // ParseStringToInt64 将 string 转换为 int64 func ParseStringToInt64(s string) int64 { if s == "" { diff --git a/backend/modules/evaluation/infra/repo/experiment/ck/expt_turn_result_filter.go b/backend/modules/evaluation/infra/repo/experiment/ck/expt_turn_result_filter.go index ebe74150b..43052f83c 100644 --- a/backend/modules/evaluation/infra/repo/experiment/ck/expt_turn_result_filter.go +++ b/backend/modules/evaluation/infra/repo/experiment/ck/expt_turn_result_filter.go @@ -6,6 +6,7 @@ package ck import ( "context" "fmt" + "os" "strconv" "strings" "time" @@ -116,9 +117,19 @@ func (d *exptTurnResultFilterDAOImpl) Save(ctx context.Context, filter []*model. // 定义浮点数比较的精度 const floatEpsilon = 1e-8 +// getClickHouseDatabaseName 从环境变量获取ClickHouse数据库名 +func getClickHouseDatabaseName() string { + dbName := os.Getenv("COZE_LOOP_CLICKHOUSE_DATABASE") + if dbName == "" { + // 默认值，保持向后兼容 + dbName = "cozeloop-clickhouse" + } + return "`" + dbName + "`" +} + func (d *exptTurnResultFilterDAOImpl) QueryItemIDStates(ctx context.Context, cond *ExptTurnResultFilterQueryCond) (map[string]int32, int64, error) { - joinSQL, whereSQL, keywordCond, args := d.buildQueryConditions(ctx, cond) - sql := d.buildBaseSQL(ctx, joinSQL, whereSQL, keywordCond, cond.EvalSetSyncCkDate, &args) + whereSQL, keywordCond, args := d.buildQueryConditions(ctx, cond) + sql := d.buildBaseSQL(ctx, whereSQL, keywordCond, &args) total, err := d.getTotalCount(ctx, sql, args) if err != nil { return nil, total, err @@ -129,18 +140,16 @@ func (d *exptTurnResultFilterDAOImpl) QueryItemIDStates(ctx context.Context, con } // buildQueryConditions 构建查询条件 -func (d *exptTurnResultFilterDAOImpl) buildQueryConditions(ctx context.Context, cond *ExptTurnResultFilterQueryCond) (string, string, string, []interface{}) { - joinSQL := "" +func (d *exptTurnResultFilterDAOImpl) buildQueryConditions(ctx context.Context, cond *ExptTurnResultFilterQueryCond) (string, string, []interface{}) { whereSQL := "" keywordCond := "" args := []interface{}{} d.buildMainTableConditions(cond, &whereSQL, &args) d.buildMapFieldConditions(cond, &whereSQL, &args) - d.buildItemSnapshotConditions(cond, &joinSQL, &args) d.buildKeywordSearchConditions(ctx, cond, &keywordCond, &args) - return joinSQL, whereSQL, keywordCond, args + return whereSQL, keywordCond, args } // buildMainTableConditions 构建主表字段条件 @@ -226,19 +235,19 @@ func (d *exptTurnResultFilterDAOImpl) buildMapFieldConditions(cond *ExptTurnResu switch f.Op { case "=": // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.eval_target_data{'%s'} = ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.eval_target_data['%s'] = ?", f.Key) *args = append(*args, f.Values[0]) case "LIKE": // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.eval_target_data{'%s'} LIKE ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.eval_target_data['%s'] LIKE ?", f.Key) *args = append(*args, "%"+escapeSpecialChars(fmt.Sprintf("%v", f.Values[0]))+"%") case "NOT LIKE": // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.eval_target_data{'%s'} NOT LIKE ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.eval_target_data['%s'] NOT LIKE ?", f.Key) *args = append(*args, "%"+escapeSpecialChars(fmt.Sprintf("%v", f.Values[0]))+"%") case "!=": // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.eval_target_data{'%s'}!=?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.eval_target_data['%s']!=?", f.Key) *args = append(*args, f.Values[0]) } } @@ -251,7 +260,7 @@ func (d *exptTurnResultFilterDAOImpl) buildMapFieldConditions(cond *ExptTurnResu continue } // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND abs(etrf.evaluator_score{'%s'} - ?) < %g", f.Key, floatEpsilon) + *whereSQL += fmt.Sprintf(" AND abs(etrf.evaluator_score['%s'] - ?) < %g", f.Key, floatEpsilon) *args = append(*args, floatValue) case ">", ">=", "<", "<=", "!=": floatValue, err := strconv.ParseFloat(fmt.Sprintf("%v", f.Values[0]), 64) @@ -260,7 +269,7 @@ func (d *exptTurnResultFilterDAOImpl) buildMapFieldConditions(cond *ExptTurnResu continue } // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.evaluator_score{'%s'} %s ?", f.Key, f.Op) + *whereSQL += fmt.Sprintf(" AND etrf.evaluator_score['%s'] %s ?", f.Key, f.Op) *args = append(*args, floatValue) case "BETWEEN": floatValue1, err1 := strconv.ParseFloat(fmt.Sprintf("%v", f.Values[0]), 64) @@ -270,7 +279,7 @@ func (d *exptTurnResultFilterDAOImpl) buildMapFieldConditions(cond *ExptTurnResu continue } // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.evaluator_score{'%s'} BETWEEN ? AND ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.evaluator_score['%s'] BETWEEN ? AND ?", f.Key) *args = append(*args, floatValue1, floatValue2) } } @@ -283,7 +292,7 @@ func (d *exptTurnResultFilterDAOImpl) buildMapFieldConditions(cond *ExptTurnResu continue } // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND abs(etrf.annotation_float{'%s'} - ?) < %g", f.Key, floatEpsilon) + *whereSQL += fmt.Sprintf(" AND abs(etrf.annotation_float['%s'] - ?) < %g", f.Key, floatEpsilon) *args = append(*args, floatValue) case ">", ">=", "<", "<=", "!=": floatValue, err := strconv.ParseFloat(fmt.Sprintf("%v", f.Values[0]), 64) @@ -292,7 +301,7 @@ func (d *exptTurnResultFilterDAOImpl) buildMapFieldConditions(cond *ExptTurnResu continue } // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.annotation_float{'%s'} %s ?", f.Key, f.Op) + *whereSQL += fmt.Sprintf(" AND etrf.annotation_float['%s'] %s ?", f.Key, f.Op) *args = append(*args, floatValue) case "BETWEEN": floatValue1, err1 := strconv.ParseFloat(fmt.Sprintf("%v", f.Values[0]), 64) @@ -302,7 +311,7 @@ func (d *exptTurnResultFilterDAOImpl) buildMapFieldConditions(cond *ExptTurnResu continue } // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.annotation_float{'%s'} BETWEEN ? AND ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.annotation_float['%s'] BETWEEN ? AND ?", f.Key) *args = append(*args, floatValue1, floatValue2) } } @@ -311,141 +320,34 @@ func (d *exptTurnResultFilterDAOImpl) buildMapFieldConditions(cond *ExptTurnResu switch f.Op { case "=": // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.annotation_string{'%s'} = ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.annotation_string['%s'] = ?", f.Key) *args = append(*args, f.Values[0]) case "LIKE": // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.annotation_string{'%s'} LIKE ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.annotation_string['%s'] LIKE ?", f.Key) *args = append(*args, "%"+escapeSpecialChars(fmt.Sprintf("%v", f.Values[0]))+"%") case "NOT LIKE": // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.annotation_string{'%s'} NOT LIKE ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.annotation_string['%s'] NOT LIKE ?", f.Key) *args = append(*args, "%"+escapeSpecialChars(fmt.Sprintf("%v", f.Values[0]))+"%") case "!=": // 删除 mapContains 条件 - *whereSQL += fmt.Sprintf(" AND etrf.annotation_string{'%s'}!=?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.annotation_string['%s']!=?", f.Key) *args = append(*args, f.Values[0]) // tag_value_id case "in", "IN": //*whereSQL += " AND etrf.annotation_string IN ?" - *whereSQL += fmt.Sprintf(" AND etrf.annotation_string{'%s'} IN ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.annotation_string['%s'] IN ?", f.Key) *args = append(*args, f.Values) case "NOT IN": //*whereSQL += " AND etrf.annotation_string NOT IN?" - *whereSQL += fmt.Sprintf(" AND etrf.annotation_string{'%s'} NOT IN ?", f.Key) + *whereSQL += fmt.Sprintf(" AND etrf.annotation_string['%s'] NOT IN ?", f.Key) *args = append(*args, f.Values) } } } -// buildItemSnapshotConditions 构建联表条件 -func (d *exptTurnResultFilterDAOImpl) buildItemSnapshotConditions(cond *ExptTurnResultFilterQueryCond, joinSQL *string, args *[]interface{}) { - if cond.ItemSnapshotCond == nil { - return - } - for _, f := range cond.ItemSnapshotCond.FloatMapFilters { - switch f.Op { - case "=": - floatValue, err := strconv.ParseFloat(fmt.Sprintf("%v", f.Values[0]), 64) - if err != nil { - logs.CtxError(context.Background(), "Parse float value failed: %v", err) - continue - } - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND abs(dis.float_map{'%s'} - ?) < %g", f.Key, floatEpsilon) - *args = append(*args, floatValue) - case ">", ">=", "<", "<=", "!=": - floatValue, err := strconv.ParseFloat(fmt.Sprintf("%v", f.Values[0]), 64) - if err != nil { - logs.CtxError(context.Background(), "Parse float value failed: %v", err) - continue - } - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.float_map{'%s'} %s ?", f.Key, f.Op) - *args = append(*args, floatValue) - case "BETWEEN": - floatValue1, err1 := strconv.ParseFloat(fmt.Sprintf("%v", f.Values[0]), 64) - floatValue2, err2 := strconv.ParseFloat(fmt.Sprintf("%v", f.Values[1]), 64) - if err1 != nil || err2 != nil { - logs.CtxError(context.Background(), "Parse float value failed: %v, %v", err1, err2) - continue - } - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.float_map{'%s'} BETWEEN ? AND ?", f.Key) - *args = append(*args, floatValue1, floatValue2) - } - } - // int_map - for _, f := range cond.ItemSnapshotCond.IntMapFilters { - switch f.Op { - case "=", ">", ">=", "<", "<=", "!=": - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.int_map{'%s'} %s ?", f.Key, f.Op) - *args = append(*args, f.Values[0]) - case "BETWEEN": - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.int_map{'%s'} BETWEEN ? AND ?", f.Key) - *args = append(*args, f.Values[0], f.Values[1]) - } - } - // 处理 BoolMapFilters - for _, f := range cond.ItemSnapshotCond.BoolMapFilters { - switch f.Op { - case "=": - boolValue, err := strconv.ParseBool(fmt.Sprintf("%v", f.Values[0])) - if err != nil { - logs.CtxError(context.Background(), "Parse bool value failed: %v", err) - continue - } - intBoolValue := 0 - if boolValue { - intBoolValue = 1 - } - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.bool_map{'%s'} = ?", f.Key) - *args = append(*args, intBoolValue) - case "!=": - boolValue, err := strconv.ParseBool(fmt.Sprintf("%v", f.Values[0])) - if err != nil { - logs.CtxError(context.Background(), "Parse bool value failed: %v", err) - continue - } - intBoolValue := 0 - if boolValue { - intBoolValue = 1 - } - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.bool_map{'%s'} != ?", f.Key) - *args = append(*args, intBoolValue) - default: - logs.CtxWarn(context.Background(), "Unsupported operator %s for BoolMapFilters", f.Op) - } - } - - // string_map - for _, f := range cond.ItemSnapshotCond.StringMapFilters { - switch f.Op { - case "LIKE": - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.string_map{'%s'} LIKE ?", f.Key) - *args = append(*args, "%"+escapeSpecialChars(fmt.Sprintf("%v", f.Values[0]))+"%") - case "=": - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.string_map{'%s'} = ?", f.Key) - *args = append(*args, f.Values[0]) - case "NOT LIKE": - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.string_map{'%s'} NOT LIKE ?", f.Key) - *args = append(*args, "%"+escapeSpecialChars(fmt.Sprintf("%v", f.Values[0]))+"%") - case "!=": - // 删除 mapContains 条件 - *joinSQL += fmt.Sprintf(" AND dis.string_map{'%s'}!=?", f.Key) - *args = append(*args, f.Values[0]) - } - } -} - // buildKeywordSearchConditions 构建全文搜索条件 func (d *exptTurnResultFilterDAOImpl) buildKeywordSearchConditions(ctx context.Context, cond *ExptTurnResultFilterQueryCond, keywordCond *string, args *[]interface{}) { if cond.KeywordSearch == nil || cond.KeywordSearch.Keyword == nil { @@ -463,87 +365,27 @@ func (d *exptTurnResultFilterDAOImpl) buildKeywordSearchConditions(ctx context.C for _, f := range cond.KeywordSearch.EvalTargetDataFilters { *keywordCond += " OR " // 删除 mapContains 条件 - *keywordCond += fmt.Sprintf("etrf.eval_target_data{'%s'} LIKE ?", f.Key) - *args = append(*args, "%"+escapeSpecialChars(kw)+"%") - } - } - - // 处理 ItemSnapshotFilter - if cond.KeywordSearch.ItemSnapshotFilter != nil { - // float_map - for _, f := range cond.KeywordSearch.ItemSnapshotFilter.FloatMapFilters { - floatValue, err := strconv.ParseFloat(kw, 64) - if err != nil { - logs.CtxInfo(ctx, "Parse float value failed in keyword search: %v", err) - continue - } - // 删除 mapContains 条件 - *keywordCond += " OR " - *keywordCond += fmt.Sprintf("abs(dis.float_map{'%s'} - ?) < %g", f.Key, floatEpsilon) - *args = append(*args, floatValue) - } - // int_map - for _, f := range cond.KeywordSearch.ItemSnapshotFilter.IntMapFilters { - intValue, err := strconv.ParseInt(kw, 10, 64) - if err != nil { - logs.CtxInfo(ctx, "Parse int value failed in keyword search: %v", err) - continue - } - // 删除 mapContains 条件 - *keywordCond += " OR " - *keywordCond += fmt.Sprintf("dis.int_map{'%s'} = ?", f.Key) - *args = append(*args, intValue) - } - // string_map - for _, f := range cond.KeywordSearch.ItemSnapshotFilter.StringMapFilters { - *keywordCond += " OR " - // 删除 mapContains 条件 - *keywordCond += fmt.Sprintf("dis.string_map{'%s'} LIKE ?", f.Key) + *keywordCond += fmt.Sprintf("etrf.eval_target_data['%s'] LIKE ?", f.Key) *args = append(*args, "%"+escapeSpecialChars(kw)+"%") } - // bool_map - boolVal := 0 - switch kw { - case "true": - boolVal = 1 - case "false": - boolVal = 0 - } - if kw == "true" || kw == "false" { - for _, f := range cond.KeywordSearch.ItemSnapshotFilter.BoolMapFilters { - *keywordCond += " OR " - // 删除 mapContains 条件 - *keywordCond += fmt.Sprintf("dis.bool_map{'%s'} = %d", f.Key, boolVal) - } - } } *keywordCond += ")" } // buildBaseSQL 构建基础SQL语句 -func (d *exptTurnResultFilterDAOImpl) buildBaseSQL(ctx context.Context, joinSQL, whereSQL, keywordCond, evalSetSyncCkDate string, args *[]interface{}) string { - sql := "SELECT etrf.item_id, etrf.status FROM " + d.configer.GetCKDBName(ctx).ExptTurnResultFilterDBName + ".expt_turn_result_filter etrf" - if joinSQL != "" || keywordCond != "" { - sql += " INNER JOIN " + d.configer.GetCKDBName(ctx).DatasetItemsSnapshotDBName + ".dataset_item_snapshot dis ON etrf.eval_set_version_id = dis.version_id AND etrf.item_id = dis.item_id" - } - - sql += " WHERE 1=1" - - if joinSQL != "" || keywordCond != "" { - sql += " And dis.sync_ck_date = ?" +func (d *exptTurnResultFilterDAOImpl) buildBaseSQL(ctx context.Context, whereSQL, keywordCond string, args *[]interface{}) string { + sql := "SELECT etrf.item_id, etrf.status FROM " + getClickHouseDatabaseName() + ".expt_turn_result_filter etrf" + sql += " FINAL WHERE 1=1" + if keywordCond != "" { // 将 evalSetSyncCkDate 插入到 args 切片的第一个位置 - newArgs := make([]interface{}, 0, len(*args)+1) - newArgs = append(newArgs, evalSetSyncCkDate) + newArgs := make([]interface{}, 0, len(*args)) newArgs = append(newArgs, *args...) *args = newArgs } if whereSQL != "" { sql += whereSQL } - if joinSQL != "" { - sql += joinSQL - } if keywordCond != "" { sql += keywordCond } @@ -552,7 +394,7 @@ func (d *exptTurnResultFilterDAOImpl) buildBaseSQL(ctx context.Context, joinSQL, // getTotalCount 获取总记录数 func (d *exptTurnResultFilterDAOImpl) getTotalCount(ctx context.Context, sql string, args []interface{}) (int64, error) { - countSQL := "SELECT COUNT(DISTINCT etrf.item_id) FROM (" + sql + ")" + countSQL := "SELECT COUNT(DISTINCT item_id) FROM (" + sql + ")" var total int64 logs.CtxInfo(ctx, "Query count sql: %v, args: %v", countSQL, args) if err := d.db.NewSession(ctx).Raw(countSQL, args...).Scan(&total).Error; err != nil { @@ -660,19 +502,19 @@ func (d *exptTurnResultFilterDAOImpl) buildGetByExptIDItemIDsSQL(ctx context.Con "etrf.status, " + "etrf.eval_set_version_id, " + "etrf.created_date, " + - "etrf.eval_target_data{'actual_output'} as actual_output, " + - "etrf.evaluator_score{'key1'} as evaluator_score_key_1, " + - "etrf.evaluator_score{'key2'} as evaluator_score_key_2, " + - "etrf.evaluator_score{'key3'} as evaluator_score_key_3, " + - "etrf.evaluator_score{'key4'} as evaluator_score_key_4, " + - "etrf.evaluator_score{'key5'} as evaluator_score_key_5, " + - "etrf.evaluator_score{'key6'} as evaluator_score_key_6, " + - "etrf.evaluator_score{'key7'} as evaluator_score_key_7, " + - "etrf.evaluator_score{'key8'} as evaluator_score_key_8, " + - "etrf.evaluator_score{'key9'} as evaluator_score_key_9, " + - "etrf.evaluator_score{'key10'} as evaluator_score_key_10, " + + "etrf.eval_target_data['actual_output'] as actual_output, " + + "etrf.evaluator_score['key1'] as evaluator_score_key_1, " + + "etrf.evaluator_score['key2'] as evaluator_score_key_2, " + + "etrf.evaluator_score['key3'] as evaluator_score_key_3, " + + "etrf.evaluator_score['key4'] as evaluator_score_key_4, " + + "etrf.evaluator_score['key5'] as evaluator_score_key_5, " + + "etrf.evaluator_score['key6'] as evaluator_score_key_6, " + + "etrf.evaluator_score['key7'] as evaluator_score_key_7, " + + "etrf.evaluator_score['key8'] as evaluator_score_key_8, " + + "etrf.evaluator_score['key9'] as evaluator_score_key_9, " + + "etrf.evaluator_score['key10'] as evaluator_score_key_10, " + "etrf.evaluator_score_corrected " + - "FROM " + d.configer.GetCKDBName(ctx).ExptTurnResultFilterDBName + ".expt_turn_result_filter etrf " + + "FROM " + getClickHouseDatabaseName() + ".expt_turn_result_filter" + " etrf " + "WHERE etrf.space_id = ? AND etrf.expt_id = ? AND etrf.created_date =?" if len(itemIDs) > 0 { sql += " AND etrf.item_id IN (?)" diff --git a/backend/modules/evaluation/infra/repo/experiment/ck/expt_turn_result_filter_test.go b/backend/modules/evaluation/infra/repo/experiment/ck/expt_turn_result_filter_test.go index 2359b07d4..3ecb98ac7 100644 --- a/backend/modules/evaluation/infra/repo/experiment/ck/expt_turn_result_filter_test.go +++ b/backend/modules/evaluation/infra/repo/experiment/ck/expt_turn_result_filter_test.go @@ -5,6 +5,7 @@ package ck import ( "context" + "fmt" "testing" "time" @@ -167,10 +168,9 @@ func TestExptTurnResultFilterDAOImpl_buildQueryConditions(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - gotSelectClause, gotWhereClause, gotOrderClause, gotArgs := d.buildQueryConditions(ctx, tt.cond) - assert.NotNil(t, gotSelectClause) - assert.NotNil(t, gotWhereClause) - assert.NotNil(t, gotOrderClause) + whereSQL, keywordCond, gotArgs := d.buildQueryConditions(ctx, tt.cond) + assert.NotNil(t, whereSQL) + assert.NotNil(t, keywordCond) assert.NotNil(t, gotArgs) }) } @@ -186,22 +186,18 @@ func TestExptTurnResultFilterDAOImpl_buildBaseSQL(t *testing.T) { ctx := context.Background() tests := []struct { - name string - joinSQL string - whereSQL string - keywordCond string - evalSetSyncCkDate string - args *[]interface{} - want string + name string + whereSQL string + keywordCond string + args *[]interface{} + want string }{ { - name: "empty_conditions", - joinSQL: "1", - whereSQL: "2", - keywordCond: "3", - evalSetSyncCkDate: "4", - args: &[]interface{}{}, - want: "生成的基础 SQL 预期值，需根据实际实现修改", + name: "empty_conditions", + whereSQL: "2", + keywordCond: "3", + args: &[]interface{}{}, + want: "SELECT etrf.item_id, etrf.status FROM `cozeloop-clickhouse`.expt_turn_result_filter etrf FINAL WHERE 1=123", }, } @@ -210,8 +206,8 @@ func TestExptTurnResultFilterDAOImpl_buildBaseSQL(t *testing.T) { mockConfig.EXPECT().GetCKDBName(gomock.Any()).Return(&entity.CKDBConfig{ ExptTurnResultFilterDBName: "ck", }).AnyTimes() - got := d.buildBaseSQL(ctx, tt.joinSQL, tt.whereSQL, tt.keywordCond, tt.evalSetSyncCkDate, tt.args) - assert.NotNil(t, got) + got := d.buildBaseSQL(ctx, tt.whereSQL, tt.keywordCond, tt.args) + assert.Equal(t, tt.want, got) }) } } @@ -238,15 +234,13 @@ func TestExptTurnResultFilterDAOImpl_appendPaginationArgs(t *testing.T) { }, }, args: []interface{}{}, - want: "生成的基础 SQL 预期值，需根据实际实现修改", + want: "LIMIT 10 OFFSET 0", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { args := d.appendPaginationArgs(tt.args, tt.cond) - if len(args) != 2 { - t.Errorf("appendPaginationArgs failed, args len not equal 2, args: %v", args) - } + assert.Equal(t, tt.want, fmt.Sprintf("LIMIT %d OFFSET %d", args[len(args)-2], args[len(args)-1])) }) } } @@ -278,11 +272,11 @@ func TestExptTurnResultFilterDAOImpl_buildGetByExptIDItemIDsSQL(t *testing.T) { t.Run(tt.name, func(t *testing.T) { mockConfig.EXPECT().GetCKDBName(gomock.Any()).Return(&entity.CKDBConfig{ ExptTurnResultFilterDBName: "ck", - }) + }).AnyTimes() got, args := d.buildGetByExptIDItemIDsSQL(ctx, tt.spaceID, tt.exptID, tt.createdDate, tt.itemIDs) assert.NotNil(t, got) if len(args) != 4 { - t.Errorf("buildGetByExptIDItemIDsSQL failed, args len not equal 3, args: %v", args) + t.Errorf("buildGetByExptIDItemIDsSQL failed, args len not equal 4, args: %v", args) } }) } diff --git a/backend/modules/evaluation/infra/repo/experiment/expt_turn_result_filter_repo_impl.go b/backend/modules/evaluation/infra/repo/experiment/expt_turn_result_filter_repo_impl.go index c14af9f6c..24b6ff8ed 100644 --- a/backend/modules/evaluation/infra/repo/experiment/expt_turn_result_filter_repo_impl.go +++ b/backend/modules/evaluation/infra/repo/experiment/expt_turn_result_filter_repo_impl.go @@ -6,6 +6,7 @@ package experiment import ( "context" "strconv" + "time" "github.com/coze-dev/coze-loop/backend/infra/db" "github.com/coze-dev/coze-loop/backend/modules/evaluation/domain/entity" @@ -40,6 +41,7 @@ func (e *ExptTurnResultFilterRepoImpl) Save(ctx context.Context, filter []*entit // 转换为 model.ExptTurnResultFilterAccelerator models := make([]*model.ExptTurnResultFilter, 0, len(filter)) for _, filterEntity := range filter { + filterEntity.UpdatedAt = time.Now() models = append(models, convertor.ExptTurnResultFilterEntity2PO(filterEntity)) } logs.CtxInfo(ctx, "ExptTurnResultFilterRepoImpl.Save: %v", json.Jsonify(models)) diff --git a/backend/modules/evaluation/pkg/conf/expt.go b/backend/modules/evaluation/pkg/conf/expt.go index 7d317424f..35b7d6400 100644 --- a/backend/modules/evaluation/pkg/conf/expt.go +++ b/backend/modules/evaluation/pkg/conf/expt.go @@ -54,7 +54,9 @@ func (c *configer) GetExptTurnResultFilterBmqProducerCfg(ctx context.Context) *e } func (c *configer) GetCKDBName(ctx context.Context) *entity.CKDBConfig { - return nil + const key = "clickhouse_config" + ckdb := &entity.CKDBConfig{} + return lo.Ternary(c.loader.UnmarshalKey(ctx, key, ckdb) == nil, ckdb, &entity.CKDBConfig{}) } func (c *configer) GetExptExportWhiteList(ctx context.Context) (eec *entity.ExptExportWhiteList) { diff --git a/frontend/packages/cozeloop/evaluate/src/pages/experiment/detail/components/experiment-detail-table/index.tsx b/frontend/packages/cozeloop/evaluate/src/pages/experiment/detail/components/experiment-detail-table/index.tsx index da3fc3ab3..d71e06600 100644 --- a/frontend/packages/cozeloop/evaluate/src/pages/experiment/detail/components/experiment-detail-table/index.tsx +++ b/frontend/packages/cozeloop/evaluate/src/pages/experiment/detail/components/experiment-detail-table/index.tsx @@ -11,7 +11,6 @@ import { type LogicField, type SemiTableSort, } from '@cozeloop/evaluate-components'; -import { IS_HIDDEN_EXPERIMENT_DETAIL_FILTER } from '@cozeloop/biz-hooks-adapter'; import { type Experiment, FieldType, @@ -423,12 +422,7 @@ export default function ({ service={service as Service} heightFull={true} - header={ - - } + header={} pageSizeStorageKey="experiment_detail_page_size" empty={tableEmpty} tableProps={tableProps} diff --git a/release/deployment/docker-compose/bootstrap/clickhouse-init/init-sql/evaluation.sql b/release/deployment/docker-compose/bootstrap/clickhouse-init/init-sql/evaluation.sql new file mode 100755 index 000000000..482518670 --- /dev/null +++ b/release/deployment/docker-compose/bootstrap/clickhouse-init/init-sql/evaluation.sql @@ -0,0 +1,31 @@ +-- Copyright (c) 2025 coze-dev Authors +-- SPDX-License-Identifier: Apache-2.0 + +-- Create expt_turn_result_filter table for docker environment +CREATE TABLE IF NOT EXISTS expt_turn_result_filter +( + `space_id` String, + `expt_id` String, + `item_id` String, + `item_idx` Int32, + `turn_id` String, + `status` Int32, + `eval_target_data` Map(String, String), + `evaluator_score` Map(String, Float64), + `annotation_float` Map(String, Float64), + `annotation_bool` Map(String, Int8), + `annotation_string` Map(String, String), + `evaluator_score_corrected` Int32, + `eval_set_version_id` String, + `created_date` Date, + `created_at` DateTime, + `updated_at` DateTime, + INDEX idx_space_id space_id TYPE bloom_filter() GRANULARITY 1, + INDEX idx_expt_id expt_id TYPE bloom_filter() GRANULARITY 1, + INDEX idx_item_id item_id TYPE bloom_filter() GRANULARITY 1, + INDEX idx_turn_id turn_id TYPE bloom_filter() GRANULARITY 1 +) +ENGINE = ReplacingMergeTree(updated_at) +PARTITION BY created_date +ORDER BY (expt_id, item_id, turn_id) +SETTINGS index_granularity = 8192; \ No newline at end of file diff --git a/release/deployment/docker-compose/conf/evaluation.yaml b/release/deployment/docker-compose/conf/evaluation.yaml index e3f8a0e70..997c59915 100644 --- a/release/deployment/docker-compose/conf/evaluation.yaml +++ b/release/deployment/docker-compose/conf/evaluation.yaml @@ -47,6 +47,10 @@ expt_export_csv_event_rmq: consumer_group: 'expt_export_csv_event_cg' producer_group: 'expt_export_csv_event_pg' +# ClickHouse table configuration +clickhouse_table_config: + expt_turn_result_filter_table_name: 'expt_turn_result_filter' + rate_limiter_conf: - key_expr: biz_key + string(space_id) limit: @@ -2195,4 +2199,7 @@ code_evaluator_template_conf: code_template_name: "包含性检查器" expt_export_white_list: - allow_all: true \ No newline at end of file + allow_all: true + +clickhouse_config: + expt_turn_result_filter_db_name: "cozeloop-clickhouse" diff --git a/release/deployment/helm-chart/charts/app/bootstrap/init/clickhouse/init-sql/evaluation.sql b/release/deployment/helm-chart/charts/app/bootstrap/init/clickhouse/init-sql/evaluation.sql new file mode 100755 index 000000000..6411cc376 --- /dev/null +++ b/release/deployment/helm-chart/charts/app/bootstrap/init/clickhouse/init-sql/evaluation.sql @@ -0,0 +1,31 @@ +-- Copyright (c) 2025 coze-dev Authors +-- SPDX-License-Identifier: Apache-2.0 + +-- Create expt_turn_result_filter table for docker environment +CREATE TABLE IF NOT EXISTS expt_turn_result_filter +( + `space_id` String, + `expt_id` String, + `item_id` String, + `item_idx` Int32, + `turn_id` String, + `status` Int32, + `eval_target_data` Map(String, String), + `evaluator_score` Map(String, Float64), + `annotation_float` Map(String, Float64), + `annotation_bool` Map(String, Int8), + `annotation_string` Map(String, String), + `evaluator_score_corrected` Int32, + `eval_set_version_id` String, + `created_date` Date, + `created_at` DateTime, + `updated_at` DateTime, + INDEX idx_space_id space_id TYPE bloom_filter() GRANULARITY 1, + INDEX idx_expt_id expt_id TYPE bloom_filter() GRANULARITY 1, + INDEX idx_item_id item_id TYPE bloom_filter() GRANULARITY 1, + INDEX idx_turn_id turn_id TYPE bloom_filter() GRANULARITY 1 + ) + ENGINE = ReplacingMergeTree(updated_at) + PARTITION BY created_date + ORDER BY (expt_id, item_id, turn_id) + SETTINGS index_granularity = 8192; \ No newline at end of file diff --git a/release/deployment/helm-chart/umbrella/conf/evaluation.yaml b/release/deployment/helm-chart/umbrella/conf/evaluation.yaml index 4627b008f..d65d0d4da 100644 --- a/release/deployment/helm-chart/umbrella/conf/evaluation.yaml +++ b/release/deployment/helm-chart/umbrella/conf/evaluation.yaml @@ -47,6 +47,10 @@ expt_export_csv_event_rmq: consumer_group: 'expt_export_csv_event_cg' producer_group: 'expt_export_csv_event_pg' +# ClickHouse table configuration +clickhouse_table_config: + expt_turn_result_filter_table_name: 'expt_turn_result_filter' + rate_limiter_conf: - key_expr: biz_key + string(space_id) limit: @@ -2195,4 +2199,7 @@ code_evaluator_template_conf: code_template_name: "包含性检查器" expt_export_white_list: - allow_all: true \ No newline at end of file + allow_all: true + +clickhouse_config: + expt_turn_result_filter_db_name: "cozeloop-clickhouse" \ No newline at end of file