fix(ingest/bigquery): Querying table metadata details in batch properly #7429
Conversation
github-actions
bot
added
the
ingestion
| @@ -80,10 +80,10 @@ class BigQueryV2Config( | |||
| number_of_datasets_process_in_batch: int = Field( | |||
| hidden_from_schema=True, | |||
| default=500, | |||
There was a problem hiding this comment.
In the description you said 200 without profiling, did you mean 500?
| stored_table_identifier.raw_table_name() | ||
| ) | ||
| # When table is none, we use dataset_name as table_name | ||
| assert stored_shard |
There was a problem hiding this comment.
Do we have exception handling for this assert? What does this statement do to the execution flow of ingestion?
There was a problem hiding this comment.
This should not happen and it is for mypy
There was a problem hiding this comment.
I see. I guess if we wanted to be more type safe we could make sharded_tables store mapping table_name -> (table, shard) but doesn't seem necessary
There was a problem hiding this comment.
Perhaps we should wrap with a try catch and log an error just in case we later make a code change that adds a non-sharded table to this dict
| ) -> Dict[str, TableListItem]: | ||
| table_items: Dict[str, TableListItem] = {} | ||
| # Dict to store sharded table and the last seen max shard id | ||
| sharded_tables: Dict[str, TableListItem] = defaultdict() |
There was a problem hiding this comment.
What does this being a defaultdict give us? The only default usage I see is sharded_tables[table_name].table_id but idk how this would work because the default_factory is None right? Just seems like it complicates the logic because we have to check:
if not sharded_tables.get(table_name): rather than if table_name not in sharded_tables
There was a problem hiding this comment.
makes sense, I can change it
| table_items = self.get_core_table_details(conn, dataset_name, project_id) | ||
|
|
||
| items_to_get: Dict[str, TableListItem] = {} | ||
| for table_item in table_items.keys(): |
There was a problem hiding this comment.
So ultimately we're querying INFORMATION_SCHEMA.TABLES (perhaps joined with INFORMATION_SCHEMA.PARTITIONS) with some filter and t.table_name in .... Is there any way to order the tables so we're fetching data most efficiently, or not really?
I guess I'm a bit confused how accessing the PARTITIONS table causes us to query more tables, unless each partition counts as its own table (or maybe a % of a table? lol)
There was a problem hiding this comment.
On Partitions table has this limitation the other information schemas don't and that's why only this table is affected. We don't query more tables here than in any other information schema table.
There was a problem hiding this comment.
Is this documented on https://cloud.google.com/bigquery/docs/information-schema-partitions? Or just something you have to find out lol
| if ( | ||
| len(table_items) % self.config.number_of_datasets_process_in_batch | ||
| == 0 |
There was a problem hiding this comment.
We were already batching this even if there was no partitioning right? What is this PR changing that will fix the issue?
There was a problem hiding this comment.
Earlier, we added the list of sharded tables to the last batch and because of this the last batch could be way more than the max batch size. This is now fixed
| # For example some_dataset.20220110 will be turned to some_dataset.some_dataset | ||
| # It seems like there are some bigquery user who uses this non-standard way of sharding the tables. | ||
| if shard: | ||
| if table_name in sharded_tables: |
There was a problem hiding this comment.
Think you need to change this to not in !!
Adding test for the case when a table name is a shard id only
Even though we queried table details earlier in batch as well but there were multiple problems with that:
Unfortunately, in the INFORMATION_SCHEAM.PARTITIONS table (which we query to get table stats for table profile info + to get the latest partition to profile) contains the non-partitioned tables as well so even if we don't have any partitioned tables the if filter on tables in that table it will raise the too many table was queried exception.
In this fix I changed to basically use a higher batch size (200) if profiling is not enabled and a lower batch size (80) if it is enabled.
Checklist