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