Migrate migrates to the latest schema
(ctx context.Context, msgs message.WriteMigrateTables)
| 101 | |
| 102 | // Migrate migrates to the latest schema |
| 103 | func (c *Client) MigrateTables(ctx context.Context, msgs message.WriteMigrateTables) error { |
| 104 | tables := make(schema.Tables, len(msgs)) |
| 105 | for i, msg := range msgs { |
| 106 | tables[i] = msg.Table |
| 107 | } |
| 108 | |
| 109 | duckdbTables := make(schema.Tables, 0, len(tables)) |
| 110 | for _, table := range tables { |
| 111 | t, err := c.getTableInfo(ctx, table.Name) |
| 112 | if err != nil { |
| 113 | return err |
| 114 | } |
| 115 | if t != nil { |
| 116 | duckdbTables = append(duckdbTables, t) |
| 117 | } |
| 118 | } |
| 119 | |
| 120 | normalizedTables := c.normalizeColumns(tables) |
| 121 | normalizedTablesSafeMode := make(schema.Tables, 0, len(normalizedTables)) |
| 122 | for _, table := range normalizedTables { |
| 123 | msg := msgs.GetMessageByTable(table.Name) |
| 124 | if msg == nil { |
| 125 | continue |
| 126 | } |
| 127 | if !msg.MigrateForce { |
| 128 | normalizedTablesSafeMode = append(normalizedTablesSafeMode, table) |
| 129 | } |
| 130 | } |
| 131 | |
| 132 | nonAutoMigratableTables := c.nonAutoMigratableTables(normalizedTablesSafeMode, duckdbTables) |
| 133 | if len(nonAutoMigratableTables) > 0 { |
| 134 | return fmt.Errorf("\nCan't migrate tables automatically, migrate manually or consider using 'migrate_mode: forced'. Non auto migratable tables changes:\n\n%s", schema.GetChangesSummary(nonAutoMigratableTables)) |
| 135 | } |
| 136 | |
| 137 | for _, table := range normalizedTables { |
| 138 | c.logger.Info().Str("table", table.Name).Msg("Migrating table") |
| 139 | if len(table.Columns) == 0 { |
| 140 | c.logger.Info().Str("table", table.Name).Msg("Table with no columns, skipping") |
| 141 | continue |
| 142 | } |
| 143 | duckdb := duckdbTables.Get(table.Name) |
| 144 | if duckdb == nil { |
| 145 | c.logger.Debug().Str("table", table.Name).Msg("Table doesn't exist, creating") |
| 146 | if err := c.createTableIfNotExist(ctx, table.Name, table, false); err != nil { |
| 147 | return err |
| 148 | } |
| 149 | continue |
| 150 | } |
| 151 | |
| 152 | changes := table.GetChanges(duckdb) |
| 153 | if c.canAutoMigrate(changes) { |
| 154 | c.logger.Info().Str("table", table.Name).Msg("Table exists, auto-migrating") |
| 155 | if err := c.autoMigrateTable(ctx, table, changes); err != nil { |
| 156 | return err |
| 157 | } |
| 158 | } else { |
| 159 | c.logger.Info().Str("table", table.Name).Msg("Table exists, force migration required") |
| 160 | if err := c.recreateTable(ctx, table); err != nil { |
nothing calls this directly
no test coverage detected