feat(ingest/iceberg): Iceberg performance improvement #11182
Conversation
|
Important Review skippedAuto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the You can disable this status message by setting the Thank you for using CodeRabbit. We offer it for free to the OSS community and would appreciate your support in helping us grow. If you find it useful, would you consider giving us a shout-out on your favorite social media? 🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
Note: Be mindful of the bot's finite context window. It's strongly recommended to break down tasks such as reading entire modules into smaller chunks. For a focused discussion, use review comments to chat about specific files and their changes, instead of using the PR comments. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
github-actions
bot
added
the
ingestion
verbose code from the ingestor code
| ) | ||
| tables_count = 0 | ||
| for namespace in namespaces: | ||
| tables = catalog.list_tables(namespace) |
There was a problem hiding this comment.
Can this fail? If it can maybe we should handle it to not fail all the namespaces.
There was a problem hiding this comment.
Added handling of exceptions there.
| tables_count = 0 | ||
| for namespace in namespaces: | ||
| tables = catalog.list_tables(namespace) | ||
| tables_count += len(tables) |
There was a problem hiding this comment.
Don't we want to add this to report?
| tables = catalog.list_tables(namespace) | ||
| tables_count += len(tables) | ||
| LOGGER.debug( | ||
| f"Retrieved {len(tables)} tables for namespace: {namespace}, in total retrieved {tables_count}, first 10: {tables[:10]}" |
There was a problem hiding this comment.
I think this should go as well to report
| LOGGER.error("Failed to get catalog", exc_info=True) | ||
| self.report.report_failure("get-catalog", f"Failed to get catalog: {e}") | ||
| return | ||
| thread_local = threading.local() |
There was a problem hiding this comment.
Is this needed? Why not just pass catalog to _process_dataset?
def _process_dataset(dataset_path, catalog):
There was a problem hiding this comment.
Catalog object can not be used across threads (at least not for every catalog type).
|
|
||
| for dataset_path in self._get_datasets(catalog): | ||
| def _process_dataset(dataset_path): |
| ) | ||
| if row_count: | ||
| # Iterating through fieldPaths introduces unwanted stats for list element fields... | ||
| for field_path, field_id in table.schema()._name_to_id.items(): |
There was a problem hiding this comment.
table.schema can't be null right?
There was a problem hiding this comment.
According to the definition of the function - not, but of course it might happen. If it happens we will simply report failure and not send the datasetProfile aspect - then continue with the next table. I think it is reasonable behavior.
| time_taken = timer.elapsed_seconds() | ||
| self.report.report_table_profiling_time(time_taken) | ||
| LOGGER.debug( | ||
| "Finished profiling of dataset: %s in %s", dataset_name, time_taken |
There was a problem hiding this comment.
please us fstring instead of string formatting
| profiling_table_timings: TimingClass = field(default_factory=TimingClass) | ||
| listed_namespaces: int = 0 | ||
| total_listed_tables: int = 0 | ||
| tables_listed_per_namespace: Dict = field(default_factory=dict) |
There was a problem hiding this comment.
Can you use TopKDict here?
This will keep the top 10 namespaces only.
treff7es
changed the title
Checklist