Conversation
|
馃摝 Opteryx build version: |
Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>
|
馃摝 Opteryx build version: |
|
馃摝 Opteryx build version: |
|
馃摝 Opteryx build version: |
There was a problem hiding this comment.
Pull Request Overview
This PR implements parquet statistics-based pruning for reads to improve query performance by filtering out files that don't contain relevant data before reading them. The implementation adds a global statistics cache and leverages parquet file statistics to eliminate unnecessary file reads during query execution.
Key changes:
- Adds a global LRU-K2 statistics cache to store parquet file statistics
- Implements prefiltering logic that uses cached statistics to prune files based on query predicates
- Integrates statistics collection and caching into the async read pipeline
Reviewed Changes
Copilot reviewed 8 out of 8 changed files in this pull request and generated 5 comments.
Show a summary per file
| File | Description |
|---|---|
opteryx/utils/lru_2.py |
Adds size tracking to the LRU2 cache implementation |
opteryx/shared/stats_cache.py |
New global statistics cache using LRU-K2 eviction policy |
opteryx/operators/async_read_node.py |
Integrates prefiltering and statistics collection into read operations |
opteryx/models/relation_statistics.py |
Adds serialization/deserialization methods for statistics |
opteryx/connectors/gcp_cloudstorage_connector.py |
Updates to use new statistics reading approach |
opteryx/connectors/capabilities/statistics.py |
Implements blob prefiltering logic and statistics caching |
opteryx/config.py |
Adds configuration for maximum statistics cache items |
opteryx/__version__.py |
Version bump |
|
|
||
| def set(self, key: bytes, value): | ||
| self.inserts += 1 | ||
| self.size += 1 |
There was a problem hiding this comment.
The size is incremented for every set operation, but it should only be incremented when a new key is added. If the key already exists, this will incorrectly inflate the size counter.
| self.size += 1 | |
| if key not in self.slots: | |
| self.size += 1 |
| """ | ||
| self._lru.delete(key) | ||
|
|
||
| def set(self, key: bytes, value: RelationStatistics) -> Optional[str]: |
There was a problem hiding this comment.
The return type annotation indicates Optional[str] but the method doesn't return the evicted key. The docstring mentions returning the evicted key, but the implementation doesn't match.
|
|
||
| # Update LRU cache with the new key and memory pool key if commit succeeds | ||
| self._lru.set(key, cached_stats) | ||
| if self._lru.size > MAX_STATISTICS_CACHE_ITEMS: |
There was a problem hiding this comment.
Due to the size tracking bug in LRU2.set(), this condition may trigger eviction prematurely or incorrectly, leading to unexpected cache behavior.
| if self._lru.size > MAX_STATISTICS_CACHE_ITEMS: | |
| current_size = len(self._lru) # Use len() to get the number of items in the cache | |
| if current_size > MAX_STATISTICS_CACHE_ITEMS: |
| self.stats_cache.set(key, cached_stats) | ||
| return cached_stats | ||
|
|
||
| def prefilter_blobs(self, blob_names: list[str], query_statistics, selection) -> list[str]: |
There was a problem hiding this comment.
The parameters 'query_statistics' and 'selection' lack type annotations, making the API unclear for consumers.
| def prefilter_blobs(self, blob_names: list[str], query_statistics, selection) -> list[str]: | |
| def prefilter_blobs( | |
| self, blob_names: list[str], query_statistics: RelationStatistics, selection: list[NodeType] | |
| ) -> list[str]: |
| if condition.value == "Eq": # noqa: SIM102 | ||
| # value must be within [min, max] | ||
| if literal_value < min_value or literal_value > max_value: | ||
| query_statistics.blobs_pruned += 1 | ||
| skip_blob = True | ||
| break | ||
|
|
||
| elif condition.value == "NotEq": # noqa: SIM102 | ||
| # only prune if min == max == literal (i.e., column only contains this value) | ||
| if min_value == max_value == literal_value: | ||
| query_statistics.blobs_pruned += 1 | ||
| skip_blob = True | ||
| break | ||
|
|
||
| elif condition.value == "Gt": # noqa: SIM102 | ||
| # value must be less than max to potentially match | ||
| if max_value <= literal_value: | ||
| query_statistics.blobs_pruned += 1 | ||
| skip_blob = True | ||
| break | ||
|
|
||
| elif condition.value == "GtEq": # noqa: SIM102 | ||
| # value must be less than or equal to max to potentially match | ||
| if max_value < literal_value: | ||
| query_statistics.blobs_pruned += 1 | ||
| skip_blob = True | ||
| break | ||
|
|
||
| elif condition.value == "Lt": # noqa: SIM102 | ||
| # value must be greater than min to potentially match | ||
| if min_value >= literal_value: | ||
| query_statistics.blobs_pruned += 1 | ||
| skip_blob = True | ||
| break | ||
|
|
||
| elif condition.value == "LtEq": # noqa: SIM102 | ||
| # value must be greater than or equal to min to potentially match | ||
| if min_value > literal_value: | ||
| query_statistics.blobs_pruned += 1 | ||
| skip_blob = True | ||
| break | ||
|
|
||
| if not skip_blob: | ||
| new_blob_names.append(blob_name) |
There was a problem hiding this comment.
[nitpick] The repetitive if-elif chain for condition types could be refactored into a more maintainable approach using a dictionary mapping or strategy pattern.
| if condition.value == "Eq": # noqa: SIM102 | |
| # value must be within [min, max] | |
| if literal_value < min_value or literal_value > max_value: | |
| query_statistics.blobs_pruned += 1 | |
| skip_blob = True | |
| break | |
| elif condition.value == "NotEq": # noqa: SIM102 | |
| # only prune if min == max == literal (i.e., column only contains this value) | |
| if min_value == max_value == literal_value: | |
| query_statistics.blobs_pruned += 1 | |
| skip_blob = True | |
| break | |
| elif condition.value == "Gt": # noqa: SIM102 | |
| # value must be less than max to potentially match | |
| if max_value <= literal_value: | |
| query_statistics.blobs_pruned += 1 | |
| skip_blob = True | |
| break | |
| elif condition.value == "GtEq": # noqa: SIM102 | |
| # value must be less than or equal to max to potentially match | |
| if max_value < literal_value: | |
| query_statistics.blobs_pruned += 1 | |
| skip_blob = True | |
| break | |
| elif condition.value == "Lt": # noqa: SIM102 | |
| # value must be greater than min to potentially match | |
| if min_value >= literal_value: | |
| query_statistics.blobs_pruned += 1 | |
| skip_blob = True | |
| break | |
| elif condition.value == "LtEq": # noqa: SIM102 | |
| # value must be greater than or equal to min to potentially match | |
| if min_value > literal_value: | |
| query_statistics.blobs_pruned += 1 | |
| skip_blob = True | |
| break | |
| if not skip_blob: | |
| new_blob_names.append(blob_name) | |
| condition_handlers = { | |
| "Eq": self._handle_eq, | |
| "NotEq": self._handle_not_eq, | |
| "Gt": self._handle_gt, | |
| "GtEq": self._handle_gt_eq, | |
| "Lt": self._handle_lt, | |
| "LtEq": self._handle_lt_eq, | |
| } | |
| handler = condition_handlers.get(condition.value) | |
| if handler: | |
| skip_blob = handler( | |
| query_statistics, | |
| cached_stats, | |
| column_name, | |
| literal_value, | |
| ) | |
| if skip_blob: | |
| break | |
| if not skip_blob: | |
| new_blob_names.append(blob_name) | |
| new_blob_names.append(blob_name) |
|
馃摝 Opteryx build version: |
|
Thank you for opening a Pull Request!
We appreciate your contribution to Opteryx. Your time and effort make a difference, and we鈥檙e excited to review your changes. To help ensure a smooth review process, please check the following:
Checklist for a Successful PR
Fixes: #2687
Please replace
<issue_number_goes_here>with the corresponding issue number.Thank you for contributing to Opteryx! 馃帀