Method Read

plugins/destination/mysql/client/read.go:226–251

(ctx context.Context, table *schema.Table, res chan<- arrow.RecordBatch)

}

func (c *Client) Read(ctx context.Context, table *schema.Table, res chan<- arrow.RecordBatch) error {
	colNames := make([]string, len(table.Columns))
	for i, col := range table.Columns {
		colNames[i] = identifier(col.Name)
	}
	cols := strings.Join(colNames, ", ")
	read := fmt.Sprintf(readSQL, cols, table.Name)
	rows, err := c.db.QueryContext(ctx, read)
	if err != nil {
		return err
	}
	defer rows.Close()
	arrowSchemaTable := table.ToArrowSchema()
	for rows.Next() {
		values := c.createResultsArray(arrowSchemaTable)
		if err := rows.Scan(values...); err != nil {
			return fmt.Errorf("failed to read from table %s: %w", table.Name, err)
		}
		record, err := reverseTransform(arrowSchemaTable, values)
		if err != nil {
			return err
		}
		res <- record
	}
	return nil
}