Conversation
vanitabhagwat
changed the title
vanitabhagwat
changed the title
| # Leaving one core for operating system and other background processes | ||
| num_processes = num_spark_driver_cores - 1 | ||
|
|
||
| if table.num_rows < num_processes: | ||
| num_processes = table.num_rows | ||
|
|
||
| # Input table is split into smaller chunks and processed in parallel | ||
| chunks = self.split_table(num_processes, table) | ||
|
|
There was a problem hiding this comment.
Are you intentionally deleting the comments here? There are some comment deletions throughout.
There was a problem hiding this comment.
It wasn't intentional. I ran the black and ruff formatting which probably did that. Added them back.
sdk/python/feast/rate_limiter.py
Outdated
| backoff = 0.005 # initial minimal sleep | ||
| while not self.acquire(): | ||
| # Compute estimated sleep until oldest timestamp exits window. | ||
| # We use the current index position as the next candidate slot. | ||
| now = time.time() | ||
| with self._lock: | ||
| oldest_ts = self.timestamps[self.index] | ||
| remaining = oldest_ts + self.period - now | ||
| if remaining <= 0: | ||
| continue | ||
| # Sleep the smaller of remaining and a capped value to re-check frequently. | ||
| time.sleep(min(remaining, 0.05)) | ||
| # Optional exponential backoff (bounded) if still not free. | ||
| backoff = min(backoff * 2, 0.05) |
There was a problem hiding this comment.
It doesn't look like backoff is actually used anywhere, it's only recalculated. Is this meant to be part of the minimum sleep instead of the hardcoded 0.05?
| entities_to_keep: Sequence[Entity], | ||
| partial: bool, | ||
| ): | ||
| # Call update only if there is an online store |
There was a problem hiding this comment.
Can you recomment this as well.
| out, err := cmd.CombinedOutput() | ||
| if err != nil { | ||
| log.Printf("Repo init error: %s", string(out)) | ||
| log.Fatal(err) |
There was a problem hiding this comment.
log.Fatal() calls os.exit(1) which will exit the test process. We need to use log.Printf()
| percent_usage = 0.6 | ||
| interval = 1.0 # seconds |
There was a problem hiding this comment.
percent_usage -> Can we define this based on number of processors?
interval -> Good. We set is based on previous logic so no confusion here.
Co-authored-by: Bhargav Dodla <13788369+EXPEbdodla@users.noreply.github.com>
Removed unnecessary blank lines in the integration test utility functions.
vanitabhagwat
force-pushed
the
ratelimiter
branch
from
5daad13 to
e2067c0
Compare
Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
4 participants
This PR is to create a more generic solution for write rate limiting that can be utilized by all online stores.