(ctx context.Context, table *schema.Table, res chan<- arrow.RecordBatch)
| 148 | } |
| 149 | |
| 150 | func (c *Client) Read(ctx context.Context, table *schema.Table, res chan<- arrow.RecordBatch) error { |
| 151 | tableName := table.Name |
| 152 | cur, err := c.client.Database(c.spec.Database).Collection(tableName).Find( |
| 153 | ctx, |
| 154 | bson.D{}) |
| 155 | if err != nil { |
| 156 | return fmt.Errorf("failed to read table %s: %w", tableName, err) |
| 157 | } |
| 158 | for cur.Next(ctx) { |
| 159 | var result bson.M |
| 160 | err := cur.Decode(&result) |
| 161 | if err != nil { |
| 162 | return fmt.Errorf("failed to read from table %s: %w", tableName, err) |
| 163 | } |
| 164 | rec, err := c.reverseTransformer(table, result) |
| 165 | if err != nil { |
| 166 | return fmt.Errorf("failed to read from table %s: %w", tableName, err) |
| 167 | } |
| 168 | res <- rec |
| 169 | } |
| 170 | return nil |
| 171 | } |
no test coverage detected