(destinationSpecs []specs.Destination, destinationsVersions [][]int)
| 22 | ) |
| 23 | |
| 24 | func getSourceV2DestV3DestinationsTransformers(destinationSpecs []specs.Destination, destinationsVersions [][]int) []*transformer.RecordTransformer { |
| 25 | destinationsTransformers := make([]*transformer.RecordTransformer, 0, len(destinationsVersions)) |
| 26 | for i := range destinationsVersions { |
| 27 | // We only need to transform to destinations that are v3 |
| 28 | if !slices.Contains(destinationsVersions[i], 3) { |
| 29 | destinationsTransformers = append(destinationsTransformers, nil) |
| 30 | continue |
| 31 | } |
| 32 | opts := []transformer.RecordTransformerOption{} |
| 33 | if destinationSpecs[i].WriteMode == specs.WriteModeAppend { |
| 34 | opts = append(opts, transformer.WithRemovePKs(), transformer.WithRemovePKs()) |
| 35 | opts = append(opts, transformer.WithRemoveUniqueConstraints()) |
| 36 | } else if destinationSpecs[i].PKMode == specs.PKModeCQID { |
| 37 | opts = append(opts, transformer.WithRemovePKs()) |
| 38 | opts = append(opts, transformer.WithCQIDPrimaryKey()) |
| 39 | } |
| 40 | destinationsTransformers = append(destinationsTransformers, transformer.NewRecordTransformer(opts...)) |
| 41 | } |
| 42 | return destinationsTransformers |
| 43 | } |
| 44 | |
| 45 | func transformSourceV2DestV3Schemas(originalSchemas [][]byte, recordTransformer *transformer.RecordTransformer) ([][]byte, error) { |
| 46 | if recordTransformer == nil { |
no outgoing calls
no test coverage detected