Temporarily allocate *nbytes* from the given *pool*.
(pool, nbytes)
| 44 | |
| 45 | @contextlib.contextmanager |
| 46 | def allocate_bytes(pool, nbytes): |
| 47 | """ |
| 48 | Temporarily allocate *nbytes* from the given *pool*. |
| 49 | """ |
| 50 | arr = pa.array([b"x" * nbytes], type=pa.binary(), memory_pool=pool) |
| 51 | # Fetch the values buffer from the varbinary array and release the rest, |
| 52 | # to get the desired allocation amount |
| 53 | buf = arr.buffers()[2] |
| 54 | arr = None |
| 55 | assert len(buf) == nbytes |
| 56 | try: |
| 57 | yield |
| 58 | finally: |
| 59 | buf = None |
| 60 | |
| 61 | |
| 62 | def check_allocated_bytes(pool): |
no test coverage detected