(store *sqlOfflineStore, def TrainingSetDef, tableName string, labelName string, isUpdate bool)
| 138 | } |
| 139 | |
| 140 | func (q mySQLQueries) trainingSetQuery(store *sqlOfflineStore, def TrainingSetDef, tableName string, labelName string, isUpdate bool) error { |
| 141 | columns := make([]string, 0) |
| 142 | query := fmt.Sprintf("(SELECT entity, value , ts from %s ) l", sanitize(labelName)) |
| 143 | for i, feature := range def.Features { |
| 144 | tableName, err := store.getResourceTableName(feature) |
| 145 | if err != nil { |
| 146 | return err |
| 147 | } |
| 148 | santizedName := sanitize(tableName) |
| 149 | tableJoinAlias := fmt.Sprintf("t%d", i) |
| 150 | columns = append(columns, santizedName) |
| 151 | query = fmt.Sprintf("%s LEFT JOIN (SELECT entity, value AS %s, ts FROM %s "+ |
| 152 | "WHERE entity=l.entity AND ts <= l.ts ORDER BY ts DESC LIMIT 1) AS %s ON %s.entity=l.entity", |
| 153 | query, santizedName, santizedName, tableJoinAlias, tableJoinAlias) |
| 154 | if i == len(def.Features)-1 { |
| 155 | query = fmt.Sprintf("%s )", query) |
| 156 | } |
| 157 | } |
| 158 | columnStr := strings.Join(columns, ", ") |
| 159 | |
| 160 | if isUpdate { |
| 161 | tempName := sanitize(fmt.Sprintf("tmp_%s", tableName)) |
| 162 | fullQuery := fmt.Sprintf("CREATE TABLE %s AS (SELECT %s, l.value as label FROM %s ", tempName, columnStr, query) |
| 163 | err := q.atomicUpdate(store.db, tableName, tempName, fullQuery) |
| 164 | if err != nil { |
| 165 | return err |
| 166 | } |
| 167 | } else { |
| 168 | fullQuery := fmt.Sprintf("CREATE TABLE %s AS (SELECT %s, l.value as label FROM %s ", sanitize(tableName), columnStr, query) |
| 169 | if _, err := store.db.Exec(fullQuery); err != nil { |
| 170 | return err |
| 171 | } |
| 172 | } |
| 173 | return nil |
| 174 | } |
| 175 | |
| 176 | func (q mySQLQueries) castTableItemType(v interface{}, t interface{}) interface{} { |
| 177 | if v == nil { |
no test coverage detected