| 21 | const maxFileSize = 1024 * 1024 * 20 |
| 22 | |
| 23 | func (c *Client) Read(ctx context.Context, table *schema.Table, res chan<- arrow.RecordBatch) error { |
| 24 | if !c.spec.NoRotate { |
| 25 | return fmt.Errorf("reading is not supported when no_rotate is false. Table: %q", table.Name) |
| 26 | } |
| 27 | if c.spec.PathContainsUUID() { |
| 28 | return fmt.Errorf("reading is not supported when path contains uuid variable. Table: %q", table.Name) |
| 29 | } |
| 30 | |
| 31 | name := c.spec.ReplacePathVariables(table.Name, uuid.NewString(), time.Time{}, c.syncID) |
| 32 | writerAtBuffer := manager.NewWriteAtBuffer(make([]byte, 0, maxFileSize)) |
| 33 | _, err := c.transferManager.DownloadObject(ctx, |
| 34 | &transfermanager.DownloadObjectInput{ |
| 35 | Bucket: aws.String(c.spec.Bucket), |
| 36 | Key: aws.String(name), |
| 37 | WriterAt: writerAtBuffer, |
| 38 | }) |
| 39 | |
| 40 | if err != nil { |
| 41 | var smithyError *smithy.OperationError |
| 42 | if errors.As(err, &smithyError); smithyError != nil { |
| 43 | var httpError *awshttp.ResponseError |
| 44 | if errors.As(smithyError.Err, &httpError); httpError != nil { |
| 45 | if httpError.HTTPStatusCode() == http.StatusNotFound { |
| 46 | return nil |
| 47 | } |
| 48 | } |
| 49 | } |
| 50 | return err |
| 51 | } |
| 52 | r := bytes.NewReader(writerAtBuffer.Bytes()) |
| 53 | return c.Client.Read(r, table, res) |
| 54 | } |