diff --git a/packages/interloper-assets/src/interloper_assets/facebook_ads/schemas/ads_stats.py b/packages/interloper-assets/src/interloper_assets/facebook_ads/schemas/ads_stats.py index c04a2e464..a5b1804c7 100644 --- a/packages/interloper-assets/src/interloper_assets/facebook_ads/schemas/ads_stats.py +++ b/packages/interloper-assets/src/interloper_assets/facebook_ads/schemas/ads_stats.py @@ -10,35 +10,138 @@ class AdsStats(Schema): account_currency: str | None = Field(default=None, description="The currency used for the account") account_id: str | None = Field(default=None, description="Identifier for the account") account_name: str | None = Field(default=None, description="Name of the account") + action_values_add_payment_info: float | None = Field(default=None, description="Action values for adding payment info") + action_values_add_to_cart: float | None = Field(default=None, description="Action values for adding to cart") + action_values_initiate_checkout: float | None = Field(default=None, description="Action values for initiating checkout") + action_values_offsite_conversion_fb_pixel_add_payment_info: float | None = Field(default=None, description="Action values for offsite conversion (Facebook Pixel) - Add Payment Info") + action_values_offsite_conversion_fb_pixel_add_to_cart: float | None = Field(default=None, description="Action values for offsite conversion (Facebook Pixel) - Add to Cart") + action_values_offsite_conversion_fb_pixel_initiate_checkout: float | None = Field(default=None, description="Action values for offsite conversion (Facebook Pixel) - Initiate Checkout") action_values_offsite_conversion_fb_pixel_purchase: float | None = Field(default=None, description="Value of offsite conversions tracked through the Facebook pixel for purchases") + action_values_offsite_conversion_fb_pixel_view_content: float | None = Field(default=None, description="Action values for offsite conversion (Facebook Pixel) - View Content") + action_values_omni_add_to_cart: float | None = Field(default=None, description="Action values for adding to cart (omni-channel)") + action_values_omni_initiated_checkout: float | None = Field(default=None, description="Action values for initiating checkout (omni-channel)") + action_values_omni_purchase: float | None = Field(default=None, description="Action values for purchase (omni-channel)") + action_values_omni_view_content: float | None = Field(default=None, description="Action values for viewing content (omni-channel)") + action_values_onsite_app_purchase: float | None = Field(default=None, description="Action values for onsite app purchase") + action_values_onsite_conversion_purchase: float | None = Field(default=None, description="Action values for onsite conversion purchase") + action_values_onsite_web_add_to_cart: float | None = Field(default=None, description="Action values for onsite web add to cart") + action_values_onsite_web_app_add_to_cart: float | None = Field(default=None, description="Action values for onsite web app add to cart") + action_values_onsite_web_app_purchase: float | None = Field(default=None, description="Action values for onsite web app purchase") + action_values_onsite_web_app_view_content: float | None = Field(default=None, description="Action values for onsite web app view content") + action_values_onsite_web_purchase: float | None = Field(default=None, description="Action values for onsite web purchase") + action_values_onsite_web_view_content: float | None = Field(default=None, description="Action values for onsite web view content") + action_values_purchase: float | None = Field(default=None, description="Action values for purchase") + action_values_view_content: float | None = Field(default=None, description="Action values for viewing content") + actions_add_payment_info: int | None = Field(default=None, description="Number of actions to add payment info") + actions_add_to_cart: int | None = Field(default=None, description="Number of actions to add to cart") actions_comment: int | None = Field(default=None, description="Number of comments on posts or ads") + actions_complete_registration: int | None = Field(default=None, description="Number of actions for completing registration") + actions_initiate_checkout: int | None = Field(default=None, description="Number of actions for initiating checkout") actions_landing_page_view: int | None = Field(default=None, description="Number of times people clicked on a link that led to the advertiser's chosen destination") + actions_like: int | None = Field(default=None, description="Number of actions for liking") actions_link_click: int | None = Field(default=None, description="Number of clicks on links within the ad") + actions_offsite_conversion_fb_pixel_add_payment_info: int | None = Field(default=None, description="Number of offsite conversion actions (Facebook Pixel) - Add Payment Info") + actions_offsite_conversion_fb_pixel_add_to_cart: int | None = Field(default=None, description="Number of offsite conversion actions (Facebook Pixel) - Add to Cart") + actions_offsite_conversion_fb_pixel_complete_registration: int | None = Field(default=None, description="Number of offsite conversion actions (Facebook Pixel) - Complete Registration") + actions_offsite_conversion_fb_pixel_initiate_checkout: int | None = Field(default=None, description="Number of offsite conversion actions (Facebook Pixel) - Initiate Checkout") actions_offsite_conversion_fb_pixel_lead: int | None = Field(default=None, description="Number of offsite conversions tracked through the Facebook pixel for leads") actions_offsite_conversion_fb_pixel_purchase: int | None = Field(default=None, description="Number of offsite conversions tracked through the Facebook pixel for purchases") + actions_offsite_conversion_fb_pixel_search: int | None = Field(default=None, description="Number of offsite conversion actions (Facebook Pixel) - Search") + actions_offsite_conversion_fb_pixel_view_content: int | None = Field(default=None, description="Number of offsite conversion actions (Facebook Pixel) - View Content") + actions_omni_add_to_cart: int | None = Field(default=None, description="Number of omni-channel actions for adding to cart") + actions_omni_add_to_wishlist: int | None = Field(default=None, description="Number of omni-channel actions for adding to wishlist") + actions_omni_complete_registration: int | None = Field(default=None, description="Number of omni-channel actions for completing registration") + actions_omni_initiated_checkout: int | None = Field(default=None, description="Number of omni-channel actions for initiating checkout") + actions_omni_purchase: int | None = Field(default=None, description="Number of omni-channel actions for purchase") + actions_omni_search: int | None = Field(default=None, description="Number of omni-channel actions for search") + actions_omni_view_content: int | None = Field(default=None, description="Number of omni-channel actions for viewing content") + actions_onsite_app_add_to_cart: int | None = Field(default=None, description="Number of onsite app actions for adding to cart") + actions_onsite_app_purchase: int | None = Field(default=None, description="Number of onsite app actions for purchase") + actions_onsite_app_view_content: int | None = Field(default=None, description="Number of onsite app actions for viewing content") + actions_onsite_conversion_add_to_cart: int | None = Field(default=None, description="Number of onsite conversion actions for adding to cart") + actions_onsite_conversion_add_to_wishlist: int | None = Field(default=None, description="Number of onsite conversion actions for adding to wishlist") + actions_onsite_conversion_initiate_checkout: int | None = Field(default=None, description="Number of onsite conversion actions for initiating checkout") actions_onsite_conversion_post_save: int | None = Field(default=None, description="Number of onsite conversions tracked after a post is saved") + actions_onsite_conversion_purchase: int | None = Field(default=None, description="Number of onsite conversion actions for purchase") + actions_onsite_conversion_view_content: int | None = Field(default=None, description="Number of onsite conversion actions for viewing content") + actions_onsite_web_add_to_cart: int | None = Field(default=None, description="Number of onsite web actions for adding to cart") + actions_onsite_web_app_add_to_cart: int | None = Field(default=None, description="Number of onsite web app actions for adding to cart") + actions_onsite_web_app_purchase: int | None = Field(default=None, description="Number of onsite web app actions for purchase") + actions_onsite_web_app_view_content: int | None = Field(default=None, description="Number of onsite web app actions for viewing content") + actions_onsite_web_purchase: int | None = Field(default=None, description="Number of onsite web actions for purchase") + actions_onsite_web_view_content: int | None = Field(default=None, description="Number of onsite web actions for viewing content") actions_page_engagement: int | None = Field(default=None, description="Number of engagements with a Facebook Page") actions_post: int | None = Field(default=None, description="Number of actions taken on a post") actions_post_engagement: int | None = Field(default=None, description="Number of engagements with posts") actions_post_reaction: int | None = Field(default=None, description="Number of reactions on posts") + actions_purchase: int | None = Field(default=None, description="Number of purchase actions") + actions_search: int | None = Field(default=None, description="Number of actions for search") actions_video_view: int | None = Field(default=None, description="Number of views on videos") + actions_view_content: int | None = Field(default=None, description="Number of actions for viewing content") ad_id: str | None = Field(default=None, description="Identifier for the ad") ad_name: str | None = Field(default=None, description="Name of the ad") adset_id: str | None = Field(default=None, description="Identifier for the ad set") adset_name: str | None = Field(default=None, description="Name of the ad set") + attribution_setting: str | None = Field(default=None, description="The attribution setting applied to the conversion metrics") buying_type: str | None = Field(default=None, description="Type of buying, such as auction or fixed price") campaign_id: str | None = Field(default=None, description="Identifier for the campaign") campaign_name: str | None = Field(default=None, description="Name of the campaign") clicks: int | None = Field(default=None, description="Total number of clicks on the ad") + converted_product_quantity: float | None = Field(default=None, description="Number of products purchased as a result of the ad") + cost_per_action_type_add_payment_info: float | None = Field(default=None, description="Cost per action type for adding payment info") + cost_per_action_type_add_to_cart: float | None = Field(default=None, description="Cost per action type for adding to cart") + cost_per_action_type_complete_registration: float | None = Field(default=None, description="Cost per action type for completing registration") + cost_per_action_type_initiate_checkout: float | None = Field(default=None, description="Cost per action type for initiating checkout") + cost_per_action_type_landing_page_view: float | None = Field(default=None, description="Cost per action type for landing page view") + cost_per_action_type_like: float | None = Field(default=None, description="Cost per action type for liking") cost_per_action_type_link_click: float | None = Field(default=None, description="Cost per action type for link clicks") + cost_per_action_type_omni_add_to_cart: float | None = Field(default=None, description="Cost per action type for omni-channel adding to cart") + cost_per_action_type_omni_add_to_wishlist: float | None = Field(default=None, description="Cost per action type for omni-channel adding to wishlist") + cost_per_action_type_omni_complete_registration: float | None = Field(default=None, description="Cost per action type for omni-channel completing registration") + cost_per_action_type_omni_initiated_checkout: float | None = Field(default=None, description="Cost per action type for omni-channel initiating checkout") + cost_per_action_type_omni_purchase: float | None = Field(default=None, description="Cost per action type for omni-channel purchase") + cost_per_action_type_omni_search: float | None = Field(default=None, description="Cost per action type for omni-channel search") + cost_per_action_type_omni_view_content: float | None = Field(default=None, description="Cost per action type for omni-channel viewing content") + cost_per_action_type_onsite_app_add_to_cart: float | None = Field(default=None, description="Cost per action type for onsite app adding to cart") + cost_per_action_type_onsite_app_purchase: float | None = Field(default=None, description="Cost per action type for onsite app purchase") + cost_per_action_type_onsite_app_view_content: float | None = Field(default=None, description="Cost per action type for onsite app viewing content") + cost_per_action_type_onsite_conversion_add_to_wishlist: float | None = Field(default=None, description="Cost per action type for onsite conversion adding to wishlist") + cost_per_action_type_onsite_conversion_initiate_checkout: float | None = Field(default=None, description="Cost per action type for onsite conversion initiating checkout") cost_per_action_type_onsite_conversion_post_save: float | None = Field(default=None, description="Cost per action type for onsite conversions after post save") + cost_per_action_type_onsite_web_add_to_cart: float | None = Field(default=None, description="Cost per action type for onsite web adding to cart") + cost_per_action_type_onsite_web_app_add_to_cart: float | None = Field(default=None, description="Cost per action type for onsite web app adding to cart") + cost_per_action_type_onsite_web_app_purchase: float | None = Field(default=None, description="Cost per action type for onsite web app purchase") + cost_per_action_type_onsite_web_app_view_content: float | None = Field(default=None, description="Cost per action type for onsite web app viewing content") + cost_per_action_type_onsite_web_purchase: float | None = Field(default=None, description="Cost per action type for onsite web purchase") + cost_per_action_type_onsite_web_view_content: float | None = Field(default=None, description="Cost per action type for onsite web viewing content") cost_per_action_type_page_engagement: float | None = Field(default=None, description="Cost per action type for page engagements") cost_per_action_type_post_engagement: float | None = Field(default=None, description="Cost per action type for post engagements") + cost_per_action_type_purchase: float | None = Field(default=None, description="Cost per action type for purchase") + cost_per_action_type_search: float | None = Field(default=None, description="Cost per action type for search") cost_per_action_type_video_view: float | None = Field(default=None, description="Cost per action type for video views") + cost_per_action_type_view_content: float | None = Field(default=None, description="Cost per action type for viewing content") cost_per_inline_link_click: float | None = Field(default=None, description="Cost per inline link click") cost_per_inline_post_engagement: float | None = Field(default=None, description="Cost per inline post engagement") cost_per_outbound_click_outbound_click: float | None = Field(default=None, description="Cost per outbound click") + cost_per_unique_action_type_commerce_event: float | None = Field(default=None, description="Cost per unique action type for commerce event") + cost_per_unique_action_type_like: float | None = Field(default=None, description="Cost per unique action type for liking") cost_per_unique_action_type_link_click: float | None = Field(default=None, description="Cost per unique action type for link clicks") + cost_per_unique_action_type_omni_add_to_cart: float | None = Field(default=None, description="Cost per unique action type for omni-channel adding to cart") + cost_per_unique_action_type_omni_add_to_wishlist: float | None = Field(default=None, description="Cost per unique action type for omni-channel adding to wishlist") + cost_per_unique_action_type_omni_initiated_checkout: float | None = Field(default=None, description="Cost per unique action type for omni-channel initiating checkout") + cost_per_unique_action_type_omni_purchase: float | None = Field(default=None, description="Cost per unique action type for omni-channel purchase") + cost_per_unique_action_type_omni_view_content: float | None = Field(default=None, description="Cost per unique action type for omni-channel viewing content") + cost_per_unique_action_type_onsite_app_add_to_cart: float | None = Field(default=None, description="Cost per unique action type for onsite app adding to cart") + cost_per_unique_action_type_onsite_app_purchase: float | None = Field(default=None, description="Cost per unique action type for onsite app purchase") + cost_per_unique_action_type_onsite_app_view_content: float | None = Field(default=None, description="Cost per unique action type for onsite app viewing content") + cost_per_unique_action_type_onsite_conversion_add_to_wishlist: float | None = Field(default=None, description="Cost per unique action type for onsite conversion adding to wishlist") + cost_per_unique_action_type_onsite_conversion_initiate_checkout: float | None = Field(default=None, description="Cost per unique action type for onsite conversion initiating checkout") + cost_per_unique_action_type_onsite_web_add_to_cart: float | None = Field(default=None, description="Cost per unique action type for onsite web adding to cart") + cost_per_unique_action_type_onsite_web_app_add_to_cart: float | None = Field(default=None, description="Cost per unique action type for onsite web app adding to cart") + cost_per_unique_action_type_onsite_web_app_purchase: float | None = Field(default=None, description="Cost per unique action type for onsite web app purchase") + cost_per_unique_action_type_onsite_web_app_view_content: float | None = Field(default=None, description="Cost per unique action type for onsite web app viewing content") + cost_per_unique_action_type_onsite_web_purchase: float | None = Field(default=None, description="Cost per unique action type for onsite web purchase") + cost_per_unique_action_type_onsite_web_view_content: float | None = Field(default=None, description="Cost per unique action type for onsite web viewing content") cost_per_unique_action_type_page_engagement: float | None = Field(default=None, description="Cost per unique action type for page engagements") cost_per_unique_action_type_post_engagement: float | None = Field(default=None, description="Cost per unique action type for post engagements") cost_per_unique_action_type_video_view: float | None = Field(default=None, description="Cost per unique action type for video views") @@ -64,12 +167,34 @@ class AdsStats(Schema): outbound_clicks_outbound_click: float | None = Field(default=None, description="Number of outbound clicks") platform_position: str | None = Field(default=None, description="Position of the ad on the platform") publisher_platform: str | None = Field(default=None, description="Platform where the ad was published") + quality_ranking: str | None = Field(default=None, description="Ranking of the ad's perceived quality against ads competing for the same audience") reach: int | None = Field(default=None, description="Total number of people who saw the ad") social_spend: float | None = Field(default=None, description="Spend specifically on social media") spend: float | None = Field(default=None, description="Total spend on the ad") unique_actions_comment: int | None = Field(default=None, description="Number of unique actions (comments) on posts or ads") + unique_actions_commerce_event: int | None = Field(default=None, description="Unique actions for commerce event") + unique_actions_like: int | None = Field(default=None, description="Unique actions for liking") unique_actions_link_click: int | None = Field(default=None, description="Number of unique actions (link clicks) on the ad") + unique_actions_omni_add_to_cart: int | None = Field(default=None, description="Unique actions for omni-channel adding to cart") + unique_actions_omni_add_to_wishlist: int | None = Field(default=None, description="Unique actions for omni-channel adding to wishlist") + unique_actions_omni_initiated_checkout: int | None = Field(default=None, description="Unique actions for omni-channel initiating checkout") + unique_actions_omni_purchase: int | None = Field(default=None, description="Unique actions for omni-channel purchase") + unique_actions_omni_view_content: int | None = Field(default=None, description="Unique actions for omni-channel viewing content") + unique_actions_onsite_app_add_to_cart: int | None = Field(default=None, description="Unique actions for onsite app adding to cart") + unique_actions_onsite_app_purchase: int | None = Field(default=None, description="Unique actions for onsite app purchase") + unique_actions_onsite_app_view_content: int | None = Field(default=None, description="Unique actions for onsite app viewing content") + unique_actions_onsite_conversion_add_to_cart: int | None = Field(default=None, description="Unique actions for onsite conversion adding to cart") + unique_actions_onsite_conversion_add_to_wishlist: int | None = Field(default=None, description="Unique actions for onsite conversion adding to wishlist") + unique_actions_onsite_conversion_initiate_checkout: int | None = Field(default=None, description="Unique actions for onsite conversion initiating checkout") unique_actions_onsite_conversion_post_save: int | None = Field(default=None, description="Number of unique actions (onsite conversions post save)") + unique_actions_onsite_conversion_purchase: int | None = Field(default=None, description="Unique actions for onsite conversion purchase") + unique_actions_onsite_conversion_view_content: int | None = Field(default=None, description="Unique actions for onsite conversion viewing content") + unique_actions_onsite_web_add_to_cart: int | None = Field(default=None, description="Unique actions for onsite web adding to cart") + unique_actions_onsite_web_app_add_to_cart: int | None = Field(default=None, description="Unique actions for onsite web app adding to cart") + unique_actions_onsite_web_app_purchase: int | None = Field(default=None, description="Unique actions for onsite web app purchase") + unique_actions_onsite_web_app_view_content: int | None = Field(default=None, description="Unique actions for onsite web app viewing content") + unique_actions_onsite_web_purchase: int | None = Field(default=None, description="Unique actions for onsite web purchase") + unique_actions_onsite_web_view_content: int | None = Field(default=None, description="Unique actions for onsite web viewing content") unique_actions_page_engagement: int | None = Field(default=None, description="Number of unique actions (page engagements)") unique_actions_post: int | None = Field(default=None, description="Number of unique actions (post actions)") unique_actions_post_engagement: int | None = Field(default=None, description="Number of unique actions (post engagements)") diff --git a/packages/interloper-assets/src/interloper_assets/facebook_ads/schemas/campaigns_stats.py b/packages/interloper-assets/src/interloper_assets/facebook_ads/schemas/campaigns_stats.py index a392d5588..a6239cead 100644 --- a/packages/interloper-assets/src/interloper_assets/facebook_ads/schemas/campaigns_stats.py +++ b/packages/interloper-assets/src/interloper_assets/facebook_ads/schemas/campaigns_stats.py @@ -77,10 +77,12 @@ class CampaignsStats(Schema): actions_search: int | None = Field(default=None, description="Number of actions for search") actions_video_view: int | None = Field(default=None, description="Number of actions for video views") actions_view_content: int | None = Field(default=None, description="Number of actions for viewing content") + attribution_setting: str | None = Field(default=None, description="The attribution setting applied to the conversion metrics") buying_type: str | None = Field(default=None, description="The type of buying") campaign_id: int | None = Field(default=None, description="The ID of the campaign") campaign_name: str | None = Field(default=None, description="The name of the campaign") clicks: int | None = Field(default=None, description="The number of clicks") + converted_product_quantity: float | None = Field(default=None, description="Number of products purchased as a result of the ad") cost_per_action_type_add_payment_info: float | None = Field(default=None, description="Cost per action type for adding payment info") cost_per_action_type_add_to_cart: float | None = Field(default=None, description="Cost per action type for adding to cart") cost_per_action_type_complete_registration: float | None = Field(default=None, description="Cost per action type for completing registration") @@ -162,6 +164,7 @@ class CampaignsStats(Schema): outbound_clicks_outbound_click: int | None = Field(default=None, description="Outbound clicks") platform_position: str | None = Field(default=None, description="The position of the ad on the platform") publisher_platform: str | None = Field(default=None, description="The platform where the ad was published") + quality_ranking: str | None = Field(default=None, description="Ranking of the ad's perceived quality against ads competing for the same audience") reach: int | None = Field(default=None, description="The reach of the campaign") spend: float | None = Field(default=None, description="The amount spent on the campaign") unique_actions_comment: int | None = Field(default=None, description="Unique actions for commenting") diff --git a/packages/interloper-assets/tests/test_facebook_ads.py b/packages/interloper-assets/tests/test_facebook_ads.py index eb201398c..e592bd7fb 100644 --- a/packages/interloper-assets/tests/test_facebook_ads.py +++ b/packages/interloper-assets/tests/test_facebook_ads.py @@ -19,8 +19,8 @@ from interloper.representation import Representation from interloper_pandas import DataFrameNormalizer -from interloper_assets.facebook_ads import schemas -from interloper_assets.facebook_ads.source import FacebookActionsNormalizer, FacebookAds +from interloper_assets.facebook_ads import constants, schemas +from interloper_assets.facebook_ads.source import PIVOT_COLUMNS, FacebookActionsNormalizer, FacebookAds def _source() -> Any: @@ -103,7 +103,7 @@ def test_insights_row_reconciles_against_ads_schema(self): def test_sparse_row_passes_validation(self): """A sparse insights frame still passes validation. - Only a few of the 75 fields arrive, and every schema field is + Only a few of the 200 fields arrive, and every schema field is optional, so absent action types are not 'Field required' errors. """ df = pd.DataFrame([{"date_start": "2026-06-10", "account_id": "123", "actions_link_click": 7}]) @@ -117,6 +117,45 @@ def test_entity_row_flattens_creative_and_reconciles(self): Representation.of(normalized).reconcile(schemas.Ads) # must not raise +class TestSchemaParity: + """Both stats assets fetch the same action arrays; the schemas must not drop them unevenly.""" + + _ACTION_FAMILIES = ( + "actions_", + "action_values_", + "cost_per_action_type_", + "cost_per_unique_action_type_", + "unique_actions_", + ) + + def test_ads_stats_declares_every_campaign_action_column(self): + campaigns = schemas.CampaignsStats.model_fields + ads = schemas.AdsStats.model_fields + expected = {name: info for name, info in campaigns.items() if name.startswith(self._ACTION_FAMILIES)} + + missing = sorted(set(expected) - set(ads)) + assert not missing, f"AdsStats lacks {len(missing)} campaign action columns: {missing}" + + mismatched = { + name: (str(info.annotation), str(ads[name].annotation)) + for name, info in expected.items() + if info.annotation != ads[name].annotation + } + assert not mismatched, mismatched + + def test_ads_stats_declares_every_requested_scalar(self): + # Breakdown dimensions come back as columns too, though they are not in the fields list. + breakdowns = {"publisher_platform", "platform_position", "impression_device"} + requested = {field for field in constants.ADS_INSIGHT_FIELDS if field not in PIVOT_COLUMNS} | breakdowns + + missing = sorted(requested - set(schemas.AdsStats.model_fields)) + assert not missing, f"requested but undeclared on AdsStats (silently dropped): {missing}" + + def test_campaigns_stats_declares_the_shared_scalars(self): + for name in ("attribution_setting", "quality_ranking", "converted_product_quantity"): + assert name in schemas.CampaignsStats.model_fields, name + + def test_isinstance_of_dataframe_normalizer(): # The custom normalizer is a DataFrameNormalizer, so it inherits its config. assert isinstance(FacebookActionsNormalizer(), DataFrameNormalizer) diff --git a/packages/interloper-google-cloud/src/interloper_google_cloud/bigquery/destination.py b/packages/interloper-google-cloud/src/interloper_google_cloud/bigquery/destination.py index 86f1401c1..fca062dc7 100644 --- a/packages/interloper-google-cloud/src/interloper_google_cloud/bigquery/destination.py +++ b/packages/interloper-google-cloud/src/interloper_google_cloud/bigquery/destination.py @@ -259,7 +259,13 @@ def insert(self, table: str, dataset: str | None, data: Any, context: IOContext) self._ensure_dataset(dataset) time_partitioning = _time_partitioning(partitioning, None) - self._load(self._table_ref(table, dataset), frame, bq_schema, time_partitioning=time_partitioning) + self._load( + self._table_ref(table, dataset), + frame, + bq_schema, + time_partitioning=time_partitioning, + allow_field_addition=context.asset.schema is not None, + ) def _load( self, @@ -267,13 +273,19 @@ def _load( df: pd.DataFrame, bq_schema: list[bigquery.SchemaField] | None, time_partitioning: bigquery.TimePartitioning | None = None, + *, + allow_field_addition: bool = False, ) -> None: """Load a DataFrame via a Parquet load job. When a schema is available, columns are aligned to it: extra columns are dropped (with a warning) and the load job receives explicit field types, so pyarrow casts values (including ``NaN`` → ``NULL``) instead - of relying on dtype autodetection. + of relying on dtype autodetection. A load behind a declared schema may + add the columns that schema gained since the table was created: the + asset's contract changed, and conform already shaped the data to it. + An inferred schema never grows a table, since a column BigQuery does + not know is then drift in the data. Args: ref: Fully-qualified table reference. @@ -281,10 +293,15 @@ def _load( bq_schema: BigQuery field definitions, or ``None`` to autodetect. time_partitioning: Partitioning spec for the table the load job is about to create; ``None`` when the table already exists. + allow_field_addition: Whether the asset declares the schema the + load carries, so new declared columns may be added to the + table; defaults to ``False``. """ job_config = bigquery.LoadJobConfig(write_disposition=bigquery.WriteDisposition.WRITE_APPEND) if time_partitioning is not None: job_config.time_partitioning = time_partitioning + if allow_field_addition and bq_schema is not None: + job_config.schema_update_options = [bigquery.SchemaUpdateOption.ALLOW_FIELD_ADDITION] if bq_schema is not None: schema_columns = [field.name for field in bq_schema] diff --git a/packages/interloper-google-cloud/tests/test_bigquery.py b/packages/interloper-google-cloud/tests/test_bigquery.py index defc9ba65..3de921451 100644 --- a/packages/interloper-google-cloud/tests/test_bigquery.py +++ b/packages/interloper-google-cloud/tests/test_bigquery.py @@ -184,6 +184,11 @@ def _plain_asset() -> list: return [] +@il.asset(schema=_RowSchema) +def _declared_asset() -> list: + return [] + + @il.asset(partitioning=il.TimePartitionConfig(column="day")) def _partitioned_asset() -> list: """Daily rows.""" # noqa: DOC201 — an asset docstring is its table description, not dev docs @@ -295,6 +300,46 @@ def test_rows_load_carries_schema(self): # -- Partitioning and table metadata ------------------------------------------- +class TestSchemaUpdateOptions: + """A load may add the columns a declared schema gained; inferred or absent schemas never grow a table.""" + + def _job_config(self, mock_client: MagicMock) -> bigquery.LoadJobConfig: + return mock_client.load_table_from_dataframe.call_args.kwargs["job_config"] + + def test_a_declared_schema_lets_the_load_add_its_columns(self): + import pandas as pd + + dest, mock_client = _make_destination(dataset="ds") + df = pd.DataFrame([{"id": 1, "cost": 1.0, "day": datetime.date(2024, 1, 1)}]) + + dest.insert("tbl", "ds", df, _ctx(_declared_asset(), _RowSchema)) + + assert self._job_config(mock_client).schema_update_options == [bigquery.SchemaUpdateOption.ALLOW_FIELD_ADDITION] + + def test_rows_behind_a_declared_schema_get_the_same_allowance(self): + dest, mock_client = _make_destination(dataset="ds") + + dest.insert("tbl", "ds", [{"id": 1, "cost": 1.0, "day": None}], _ctx(_declared_asset(), _RowSchema)) + + assert self._job_config(mock_client).schema_update_options == [bigquery.SchemaUpdateOption.ALLOW_FIELD_ADDITION] + + def test_an_inferred_schema_never_grows_the_table(self): + # Conform infers an effective schema for a schemaless asset; a column + # BigQuery does not know is then drift in the data, not a contract change. + dest, mock_client = _make_destination(dataset="ds") + + dest.insert("tbl", "ds", [{"id": 1, "cost": 1.0, "day": None}], _ctx(_plain_asset(), _RowSchema)) + + assert self._job_config(mock_client).schema_update_options is None + + def test_no_schema_never_grows_the_table(self): + dest, mock_client = _make_destination(dataset="ds") + + dest.insert("tbl", "ds", [{"id": 1}], _ctx(_plain_asset(), None)) + + assert self._job_config(mock_client).schema_update_options is None + + class TestTimePartitioning: """Partition config → BigQuery time partitioning spec."""