(store *sqlOfflineStore, def TrainingSetDef, tableName string, labelName string, isUpdate bool)
| 158 | } |
| 159 | |
| 160 | func (q redshiftSQLQueries) trainingSetQuery(store *sqlOfflineStore, def TrainingSetDef, tableName string, labelName string, isUpdate bool) error { |
| 161 | columns := make([]string, 0) |
| 162 | selectColumns := make([]string, 0) |
| 163 | query := "" |
| 164 | for i, feature := range def.Features { |
| 165 | tableName, err := store.getResourceTableName(feature) |
| 166 | santizedName := sanitize(tableName) |
| 167 | if err != nil { |
| 168 | return err |
| 169 | } |
| 170 | tableJoinAlias := fmt.Sprintf("t%d", i+1) |
| 171 | selectColumns = append(selectColumns, fmt.Sprintf("%s_rnk", tableJoinAlias)) |
| 172 | columns = append(columns, santizedName) |
| 173 | query = fmt.Sprintf("%s LEFT OUTER JOIN (SELECT entity, value AS %s, ts, RANK() OVER (ORDER BY ts DESC) AS %s_rnk FROM %s ORDER BY ts desc) AS %s ON (%s.entity=t0.entity AND %s.ts <= t0.ts)", |
| 174 | query, santizedName, tableJoinAlias, santizedName, tableJoinAlias, tableJoinAlias, tableJoinAlias) |
| 175 | if i == len(def.Features)-1 { |
| 176 | query = fmt.Sprintf("%s )) WHERE rn=1", query) |
| 177 | } |
| 178 | } |
| 179 | columnStr := strings.Join(columns, ", ") |
| 180 | selectColumnStr := strings.Join(selectColumns, ", ") |
| 181 | |
| 182 | if !isUpdate { |
| 183 | fullQuery := fmt.Sprintf( |
| 184 | "CREATE TABLE %s AS (SELECT %s, label FROM ("+ |
| 185 | "SELECT *, row_number() over(PARTITION BY e, label, time ORDER BY \"time\", %s DESC) AS rn FROM ( "+ |
| 186 | "SELECT t0.entity AS e, t0.value AS label, t0.ts AS time, %s, %s FROM %s AS t0 %s )", |
| 187 | sanitize(tableName), columnStr, selectColumnStr, columnStr, selectColumnStr, sanitize(labelName), query) |
| 188 | if _, err := store.db.Exec(fullQuery); err != nil { |
| 189 | return err |
| 190 | } |
| 191 | } else { |
| 192 | tempTable := sanitize(fmt.Sprintf("tmp_%s", tableName)) |
| 193 | fullQuery := fmt.Sprintf( |
| 194 | "CREATE TABLE %s AS (SELECT %s, label FROM ("+ |
| 195 | "SELECT *, row_number() over(PARTITION BY e, label, time ORDER BY \"time\", %s desc) AS rn FROM ( "+ |
| 196 | "SELECT t0.entity AS e, t0.value AS label, t0.ts AS time, %s, %s FROM %s AS t0 %s )", |
| 197 | tempTable, columnStr, selectColumnStr, columnStr, selectColumnStr, sanitize(labelName), query) |
| 198 | |
| 199 | err := q.atomicUpdate(store.db, tableName, tempTable, fullQuery) |
| 200 | return err |
| 201 | } |
| 202 | return nil |
| 203 | } |
| 204 | |
| 205 | func (q redshiftSQLQueries) castTableItemType(v interface{}, t interface{}) interface{} { |
| 206 | if v == nil { |
no test coverage detected