Method schemaTables

plugins/destination/mssql/client/schema.go:12–56

(ctx context.Context, messages message.WriteMigrateTables)

)

func (c *Client) schemaTables(ctx context.Context, messages message.WriteMigrateTables) (schema.Tables, error) {
	query, params := queries.AllTables(c.spec.Schema)
	rows, err := c.db.QueryContext(ctx, query, params...)
	if err != nil {
		c.logErr(err)
		return nil, err
	}

	names := make([]string, 0)
	if err := processRows(rows, func(row *sql.Rows) error {
		var tableCatalog string
		var tableName string
		var tableType string
		var schemaType string

		if err := row.Scan(&tableCatalog, &tableType, &tableName, &schemaType); err != nil {
			return err
		}
		if !messages.Exists(tableName) {
			return nil
		}
		names = append(names, tableName)
		return nil
	}); err != nil {
		c.logErr(err)
		return nil, err
	}

	result := make(schema.Tables, len(names))
	for i, tableName := range names {
		pks, err := c.getTablePK(ctx, tableName)
		if err != nil {
			return nil, err
		}

		columns, err := c.getTableColumns(ctx, tableName, pks)
		if err != nil {
			return nil, err
		}

		result[i] = &schema.Table{Name: tableName, Columns: columns}
	}

	return result, nil
}

func normalizedTables(messages message.WriteMigrateTables) schema.Tables {
	normalized := make(schema.Tables, len(messages))