(ctx context.Context, table *schema.Table, res chan<- arrow.RecordBatch)
| 26 | ) |
| 27 | |
| 28 | func (c *Client) Read(ctx context.Context, table *schema.Table, res chan<- arrow.RecordBatch) error { |
| 29 | colSQL := "`" + strings.Join(table.Columns.Names(), "`, `") + "`" |
| 30 | stmt := fmt.Sprintf(readSQL, colSQL, c.spec.ProjectID, c.spec.DatasetID, table.Name) |
| 31 | q := c.client.Query(stmt) |
| 32 | q.Parameters = []bigquery.QueryParameter{} |
| 33 | q.Location = c.client.Location |
| 34 | it, err := q.Read(ctx) |
| 35 | if err != nil { |
| 36 | return fmt.Errorf("failed to read table %s: %w", table.Name, err) |
| 37 | } |
| 38 | arrowSchema := table.ToArrowSchema() |
| 39 | for { |
| 40 | values := make([]bigquery.Value, len(table.Columns)) |
| 41 | err := it.Next(&values) |
| 42 | if err != nil { |
| 43 | if errors.Is(err, iterator.Done) { |
| 44 | break |
| 45 | } |
| 46 | return fmt.Errorf("failed to read from table %s: %w", table.Name, err) |
| 47 | } |
| 48 | |
| 49 | rb := array.NewRecordBuilder(memory.DefaultAllocator, arrowSchema) |
| 50 | for i := range values { |
| 51 | err := appendValue(rb.Field(i), values[i]) |
| 52 | if err != nil { |
| 53 | return fmt.Errorf("failed to read from table %s: %w", table.Name, err) |
| 54 | } |
| 55 | } |
| 56 | res <- rb.NewRecordBatch() |
| 57 | } |
| 58 | return nil |
| 59 | } |
| 60 | |
| 61 | func parseRat(v *big.Rat) (str string, precision int32, scale int32) { |
| 62 | str = v.FloatString(10) |
nothing calls this directly
no test coverage detected