Skip to content

fix: Arrow Retain and Release fixes #795

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
hermanschaaf merged 10 commits into from
Apr 19, 2023
Merged

fix: Arrow Retain and Release fixes #795

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
a4d19d4
Fix memory leak in writeAll
hermanschaaf Apr 18, 2023
2a29439
Merge remote-tracking branch 'upstream/main' into fix-memory-leak
hermanschaaf Apr 18, 2023
2aa29af
Merge branch 'main' into fix-memory-leak
hermanschaaf Apr 18, 2023
bfb6f63
Retain and release fixes
hermanschaaf Apr 18, 2023
0da0cbd
Merge branch 'fix-memory-leak' of github.com:cloudquery/plugin-sdk in…
hermanschaaf Apr 18, 2023
c0f42f4
Lint
hermanschaaf Apr 18, 2023
6d1595b
Remove extra retains
hermanschaaf Apr 18, 2023
c8ef24e
Undo convertedResource.Release() for now
hermanschaaf Apr 18, 2023
b5ca844
Updates
hermanschaaf Apr 19, 2023
1a667c9
Merge branch 'main' into fix-memory-leak
hermanschaaf Apr 19, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions internal/servers/destination/v0/destinations.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,7 @@ func (s *Server) Write2(msg pb.Destination_Write2Server) error {
select {
case resources <- convertedResource:
case <-ctx.Done():
convertedResource.Release()
close(resources)
if err := eg.Wait(); err != nil {
return status.Errorf(codes.Internal, "Context done: %v and failed to wait for plugin: %v", ctx.Err(), err)
Expand Down
14 changes: 9 additions & 5 deletions plugins/destination/managed_writer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -70,6 +70,9 @@ func (p *Plugin) flush(ctx context.Context, metrics *Metrics, table *arrow.Schem
p.logger.Info().Str("table", tableName).Int("len", batchSize).Dur("duration", time.Since(start)).Msg("batch written successfully")
atomic.AddUint64(&metrics.Writes, uint64(batchSize))
}
for _, r := range resources {
r.Release()
}
}

func (*Plugin) removeDuplicatesByPK(table *arrow.Schema, resources []arrow.Record) []arrow.Record {
Expand All @@ -81,18 +84,19 @@ func (*Plugin) removeDuplicatesByPK(table *arrow.Schema, resources []arrow.Recor

pks := make(map[string]struct{}, len(resources))
res := make([]arrow.Record, 0, len(resources))
var reported bool
for _, r := range resources {
if r.NumRows() > 1 {
panic(fmt.Sprintf("record with more than 1 row: %d", r.NumRows()))
Copy link
Member Author

@hermanschaaf hermanschaaf Apr 18, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I also added this panic because this function's logic will break down if a record ever contains more than one row.

I wonder if we may want to add a special type that guarantees we're dealing with a arrow.Record with only a single row.

}
key := pk.String(r)
_, ok := pks[key]
switch {
case !ok:
if !ok {
pks[key] = struct{}{}
res = append(res, r)
continue
case reported:
Copy link
Member Author

@hermanschaaf hermanschaaf Apr 18, 2023

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

the reported variable was not being used

continue
}
// duplicate, release early
r.Release()
}

return res
Expand Down