Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

Bugfix/fix update schema to read current schema in target table #102

Closed
wants to merge 7 commits into from
Closed

Bugfix/fix update schema to read current schema in target table #102

Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
d02fe92
updated updat_schema function
loveeklund-osttra Oct 4, 2024
375cf99
adding space to trigger workflow
loveeklund-osttra Oct 4, 2024
dcd4273
removing space to trigger tests
loveeklund-osttra Oct 4, 2024
a078760
adding space to trigger workflow
loveeklund-osttra Oct 4, 2024
1477ff2
removing space to trigger workflow
loveeklund-osttra Oct 4, 2024
6f5b01c
updating update_schema and adding function to get current version of …
loveeklund-osttra Oct 4, 2024
c6ca811
fixing merge issues
loveeklund-osttra Oct 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
12 changes: 8 additions & 4 deletions target_bigquery/core.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -204,6 +204,9 @@ def create_table(
time.sleep(5)
return self._dataset, self._table

def get_current_schema(self) -> List[bigquery.SchemaField]:
return self._table.schema

def default_table_options(self) -> Dict[str, Any]:
"""Returns the default table options for this table."""
schema_dump = json.dumps(self.jsonschema)
Expand Down Expand Up @@ -587,10 +590,11 @@ class Denormalized:
def update_schema(self: BaseBigQuerySink) -> None: # type: ignore
"""Update the target schema."""
table = self.table.as_table()
current_schema = table.schema[:]
mut_schema = table.schema[:]
for expected_field in self.table.get_resolved_schema(self.apply_transforms):
if not any(field.name == expected_field.name for field in current_schema):
resolved_schema = self.table.get_resolved_schema(self.apply_transforms)
current_schema = self.table.get_current_schema()[:]
mut_schema = current_schema[:]
for expected_field in resolved_schema:
if not any(field.name == expected_field.name for field in mut_schema):
mut_schema.append(expected_field)
if len(mut_schema) > len(current_schema):
table.schema = mut_schema
Expand Down
Loading