(store *sqlOfflineStore, def TrainingSetDef, tableName string, labelName string, isUpdate bool)
| 1501 | } |
| 1502 | |
| 1503 | func (q defaultOfflineSQLQueries) trainingSetQuery(store *sqlOfflineStore, def TrainingSetDef, tableName string, labelName string, isUpdate bool) error { |
| 1504 | columns := make([]string, 0) |
| 1505 | query := "" |
| 1506 | for i, feature := range def.Features { |
| 1507 | tableName, err := store.getResourceTableName(feature) |
| 1508 | santizedName := sanitize(tableName) |
| 1509 | if err != nil { |
| 1510 | return err |
| 1511 | } |
| 1512 | tableJoinAlias := fmt.Sprintf("t%d", i+1) |
| 1513 | columns = append(columns, santizedName) |
| 1514 | query = fmt.Sprintf("%s LEFT OUTER JOIN (SELECT entity, value as %s, ts FROM %s ORDER BY ts desc) as %s ON (%s.entity=t0.entity AND %s.ts <= t0.ts)", |
| 1515 | query, santizedName, santizedName, tableJoinAlias, tableJoinAlias, tableJoinAlias) |
| 1516 | |
| 1517 | } |
| 1518 | for i, lagFeature := range def.LagFeatures { |
| 1519 | lagFeaturesOffset := len(def.Features) |
| 1520 | tableName, err := store.getResourceTableName(ResourceID{lagFeature.FeatureName, lagFeature.FeatureVariant, Feature}) |
| 1521 | if err != nil { |
| 1522 | return err |
| 1523 | } |
| 1524 | lagColumnName := sanitize(lagFeature.LagName) |
| 1525 | if lagFeature.LagName == "" { |
| 1526 | lagColumnName = sanitize(fmt.Sprintf("%s_lag_%s", tableName, lagFeature.LagDelta)) |
| 1527 | } |
| 1528 | columns = append(columns, lagColumnName) |
| 1529 | sanitizedName := sanitize(tableName) |
| 1530 | tableJoinAlias := fmt.Sprintf("t%d", lagFeaturesOffset+i+1) |
| 1531 | timeDeltaSeconds := lagFeature.LagDelta.Seconds() |
| 1532 | query = fmt.Sprintf("%s LEFT OUTER JOIN (SELECT entity, value as %s, ts FROM %s ORDER BY ts desc) as %s ON (%s.entity=t0.entity AND (%s.ts + INTERVAL '%f') <= t0.ts)", |
| 1533 | query, lagColumnName, sanitizedName, tableJoinAlias, tableJoinAlias, tableJoinAlias, timeDeltaSeconds) |
| 1534 | } |
| 1535 | |
| 1536 | query = fmt.Sprintf("%s )) WHERE rn=1", query) |
| 1537 | columnStr := strings.Join(columns, ", ") |
| 1538 | if !isUpdate { |
| 1539 | fullQuery := fmt.Sprintf( |
| 1540 | "CREATE TABLE %s AS (SELECT %s, label FROM ("+ |
| 1541 | "SELECT *, row_number() over(PARTITION BY e, label, time ORDER BY time desc) as rn FROM ( "+ |
| 1542 | "SELECT t0.entity as e, t0.value as label, t0.ts as time, %s from %s as t0 %s )", |
| 1543 | sanitize(tableName), columnStr, columnStr, sanitize(labelName), query) |
| 1544 | if _, err := store.db.Exec(fullQuery); err != nil { |
| 1545 | return err |
| 1546 | } |
| 1547 | } else { |
| 1548 | tempTable := sanitize(fmt.Sprintf("tmp_%s", tableName)) |
| 1549 | fullQuery := fmt.Sprintf( |
| 1550 | "CREATE TABLE %s AS (SELECT %s, label FROM ("+ |
| 1551 | "SELECT *, row_number() over(PARTITION BY e, label, time ORDER BY time desc) as rn FROM ( "+ |
| 1552 | "SELECT t0.entity as e, t0.value as label, t0.ts as time, %s from %s as t0 %s )", |
| 1553 | tempTable, columnStr, columnStr, sanitize(labelName), query) |
| 1554 | err := q.atomicUpdate(store.db, tableName, tempTable, fullQuery) |
| 1555 | return err |
| 1556 | } |
| 1557 | return nil |
| 1558 | } |
| 1559 | |
| 1560 | func (q defaultOfflineSQLQueries) atomicUpdate(db *sql.DB, tableName string, tempName string, query string) error { |
no test coverage detected