MCPcopy Create free account
hub / github.com/cloudquery/cloudquery / Read

Method Read

plugins/destination/s3/client/read.go:23–54  ·  view source on GitHub ↗
(ctx context.Context, table *schema.Table, res chan<- arrow.RecordBatch)

Source from the content-addressed store, hash-verified

21const maxFileSize = 1024 * 1024 * 20
22
23func (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}

Callers 1

readAllFunction · 0.45

Calls 4

ErrorfMethod · 0.80
PathContainsUUIDMethod · 0.45
ReplacePathVariablesMethod · 0.45
StringMethod · 0.45

Tested by 1

readAllFunction · 0.36