| 149 | } |
| 150 | |
| 151 | func (c *Client) Read(ctx context.Context, table *schema.Table, res chan<- arrow.RecordBatch) error { |
| 152 | colNames := make([]string, len(table.Columns)) |
| 153 | for i, col := range table.Columns { |
| 154 | colNames[i] = identifier(col.Name) |
| 155 | } |
| 156 | cols := strings.Join(colNames, ", ") |
| 157 | rows, err := c.db.Query(fmt.Sprintf(readSQL, cols, identifier(table.Name))) |
| 158 | if err != nil { |
| 159 | return err |
| 160 | } |
| 161 | arrowSchema := table.ToArrowSchema() |
| 162 | for rows.Next() { |
| 163 | values := c.createResultsArray(arrowSchema) |
| 164 | if err := rows.Scan(values...); err != nil { |
| 165 | return fmt.Errorf("failed to read from table %s: %w", table.Name, err) |
| 166 | } |
| 167 | record, err := reverseTransform(arrowSchema, values) |
| 168 | if err != nil { |
| 169 | return err |
| 170 | } |
| 171 | res <- record |
| 172 | } |
| 173 | rows.Close() |
| 174 | return nil |
| 175 | } |