Skip to content
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

Jump to bottom

Feat: Threaded MutationsBatcher #722

Merged
merged 24 commits into from
May 10, 2023
Merged

Feat: Threaded MutationsBatcher #722

Changes from 1 commit
Commits
Show all changes
24 commits
Select commit Hold shift + click to select a range
3010cd5
Feat: Threaded MutationsBatcher
Mariatta Jan 12, 2023
dbf80c3
Make _MutationsBatchQueue private
Mariatta Jan 13, 2023
9ef3ec5
Fix the tests.
Mariatta Jan 16, 2023
148d1c0
Unit test adjustments
Mariatta Jan 17, 2023
f312659
Add flush_interval.
Mariatta Jan 18, 2023
e657321
Add test for the batcher time interval
Mariatta Jan 18, 2023
00e03e5
Adjust parameters
Mariatta Jan 31, 2023
649cd9b
Raise errors from mutations.
Mariatta Feb 1, 2023
1a5ff05
test coverage
Mariatta Feb 2, 2023
70f1cc8
Use `from_grpc_status`
Mariatta Feb 10, 2023
fd06c2f
Update MutationsBatchError
Mariatta Feb 16, 2023
b912792
Add flow control to control inflight requests
Mariatta Mar 9, 2023
cc334ec
Improve test coverage
Mariatta Mar 10, 2023
177414f
Submit the batch when queue's empty
Mariatta Mar 10, 2023
d907180
Update google/cloud/bigtable/batcher.py
Mariatta Mar 23, 2023
d0a6ba3
Changes based on PR review
Mariatta Mar 29, 2023
06bc2c1
Update google/cloud/bigtable/batcher.py
Mariatta Apr 5, 2023
c0584b5
Update google/cloud/bigtable/batcher.py
Mariatta Apr 5, 2023
9899e45
Changes based on PR review
Mariatta Apr 5, 2023
51b063b
Docstring updates
Mariatta Apr 5, 2023
1e42c1b
Remove unused variable, is_open
Mariatta Apr 6, 2023
c1c50ea
Adjust the logic for when the batch is larger than available resources.
Mariatta May 5, 2023
414fe87
Make `_flush_async` private
Mariatta May 9, 2023
c2943ad
Update test case for flush async
Mariatta May 9, 2023
File filter

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
Adjust parameters
  • Loading branch information
@Mariatta
Mariatta committed Apr 5, 2023
commit 00e03e5512267ab293e009218d6f5a1168be3417
12 changes: 7 additions & 5 deletions google/cloud/bigtable/batcher.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,10 +18,10 @@
import concurrent.futures
import atexit

FLUSH_COUNT = 1000
FLUSH_COUNT = 100
MAX_MUTATIONS = 100000
MAX_ROW_BYTES = 5242880 # 5MB

MAX_ROW_BYTES = 20 * 1024 * 1024 # 20MB
MAX_BATCH_SIZE = 100 * 1024 * 1024

class MaxMutationsError(ValueError):
"""The number of mutations for bulk request is too big."""
Expand Down Expand Up @@ -60,11 +60,13 @@ def put(self, item, block=True, timeout=None):
item.row_key, mutation_count
)
)
mutianf marked this conversation as resolved.
Show resolved Hide resolved
self.total_size += item.get_mutations_size()
self.total_mutation_count += mutation_count


self._queue.put(item, block=block, timeout=timeout)

self.total_size += item.get_mutations_size()
self.total_mutation_count += mutation_count

def full(self):
"""Check if the queue is full."""
if (
Expand Down