| 602 | } |
| 603 | |
| 604 | func (q defaultPythonOfflineQueries) trainingSetCreate(def TrainingSetDef, featureSchemas []ResourceSchema, labelSchema ResourceSchema) string { |
| 605 | columns := make([]string, 0) |
| 606 | joinQueries := make([]string, 0) |
| 607 | feature_timestamps := make([]string, 0) |
| 608 | for i, feature := range def.Features { |
| 609 | featureColumnName := createQuotedIdentifier(feature) |
| 610 | columns = append(columns, featureColumnName) |
| 611 | var featureWindowQuery string |
| 612 | // if no timestamp column, set to default generated by resource registration |
| 613 | if featureSchemas[i].TS == "" { |
| 614 | featureWindowQuery = fmt.Sprintf("SELECT * FROM (SELECT %s as t%d_entity, %s as %s, CAST(0 AS TIMESTAMP) as t%d_ts FROM source_%d) ORDER BY t%d_ts ASC", featureSchemas[i].Entity, i+1, featureSchemas[i].Value, featureColumnName, i+1, i+1, i+1) |
| 615 | } else { |
| 616 | featureWindowQuery = fmt.Sprintf("SELECT * FROM (SELECT %s as t%d_entity, %s as %s, %s as t%d_ts FROM source_%d) ORDER BY t%d_ts ASC", featureSchemas[i].Entity, i+1, featureSchemas[i].Value, featureColumnName, featureSchemas[i].TS, i+1, i+1, i+1) |
| 617 | } |
| 618 | featureJoinQuery := fmt.Sprintf("LEFT OUTER JOIN (%s) t%d ON (t%d_entity = entity AND t%d_ts <= label_ts)", featureWindowQuery, i+1, i+1, i+1) |
| 619 | joinQueries = append(joinQueries, featureJoinQuery) |
| 620 | feature_timestamps = append(feature_timestamps, fmt.Sprintf("t%d_ts", i+1)) |
| 621 | } |
| 622 | for i, lagFeature := range def.LagFeatures { |
| 623 | lagFeaturesOffset := len(def.Features) |
| 624 | idx := slices.IndexFunc(def.Features, func(id ResourceID) bool { |
| 625 | return id.Name == lagFeature.FeatureName && id.Variant == lagFeature.FeatureVariant |
| 626 | }) |
| 627 | lagSource := fmt.Sprintf("source_%d", idx+1) |
| 628 | lagColumnName := sanitize(lagFeature.LagName) |
| 629 | if lagFeature.LagName == "" { |
| 630 | lagColumnName = fmt.Sprintf("`%s_%s_lag_%s`", lagFeature.FeatureName, lagFeature.FeatureVariant, lagFeature.LagDelta) |
| 631 | } |
| 632 | columns = append(columns, lagColumnName) |
| 633 | timeDeltaSeconds := lagFeature.LagDelta.Seconds() //parquet stores time as microseconds |
| 634 | curIdx := lagFeaturesOffset + i + 1 |
| 635 | var lagWindowQuery string |
| 636 | if featureSchemas[idx].TS == "" { |
| 637 | lagWindowQuery = fmt.Sprintf("SELECT * FROM (SELECT %s as t%d_entity, %s as %s, CAST(0 AS TIMESTAMP) as t%d_ts FROM %s) ORDER BY t%d_ts ASC", featureSchemas[idx].Entity, curIdx, featureSchemas[idx].Value, lagColumnName, curIdx, lagSource, curIdx) |
| 638 | } else { |
| 639 | lagWindowQuery = fmt.Sprintf("SELECT * FROM (SELECT %s as t%d_entity, %s as %s, %s as t%d_ts FROM %s) ORDER BY t%d_ts ASC", featureSchemas[idx].Entity, curIdx, featureSchemas[idx].Value, lagColumnName, featureSchemas[idx].TS, curIdx, lagSource, curIdx) |
| 640 | } |
| 641 | lagJoinQuery := fmt.Sprintf("LEFT OUTER JOIN (%s) t%d ON (t%d_entity = entity AND (t%d_ts + INTERVAL %f SECOND) <= label_ts)", lagWindowQuery, curIdx, curIdx, curIdx, timeDeltaSeconds) |
| 642 | joinQueries = append(joinQueries, lagJoinQuery) |
| 643 | feature_timestamps = append(feature_timestamps, fmt.Sprintf("t%d_ts", curIdx)) |
| 644 | } |
| 645 | columnStr := strings.Join(columns, ", ") |
| 646 | joinQueryString := strings.Join(joinQueries, " ") |
| 647 | var labelWindowQuery string |
| 648 | if labelSchema.TS == "" { |
| 649 | labelWindowQuery = fmt.Sprintf("SELECT %s AS entity, %s AS value, CAST(0 AS TIMESTAMP) AS label_ts FROM source_0", labelSchema.Entity, labelSchema.Value) |
| 650 | } else { |
| 651 | labelWindowQuery = fmt.Sprintf("SELECT %s AS entity, %s AS value, %s AS label_ts FROM source_0", labelSchema.Entity, labelSchema.Value, labelSchema.TS) |
| 652 | } |
| 653 | labelPartitionQuery := fmt.Sprintf("(SELECT * FROM (SELECT entity, value, label_ts FROM (%s) t ) t0)", labelWindowQuery) |
| 654 | labelJoinQuery := fmt.Sprintf("%s %s", labelPartitionQuery, joinQueryString) |
| 655 | |
| 656 | timeStamps := strings.Join(feature_timestamps, ", ") |
| 657 | timeStampsDesc := strings.Join(feature_timestamps, " DESC,") |
| 658 | fullQuery := fmt.Sprintf("SELECT %s, value AS %s, entity, label_ts, %s, ROW_NUMBER() over (PARTITION BY entity, value, label_ts ORDER BY label_ts DESC, %s DESC) as row_number FROM (%s) tt", columnStr, createQuotedIdentifier(def.Label), timeStamps, timeStampsDesc, labelJoinQuery) |
| 659 | finalQuery := fmt.Sprintf("SELECT %s, %s FROM (SELECT * FROM (SELECT *, row_number FROM (%s) WHERE row_number=1 )) ORDER BY label_ts", columnStr, createQuotedIdentifier(def.Label), fullQuery) |
| 660 | return finalQuery |
| 661 | } |