🐛 Source Shopify: fixed store redirection, add bulk checkpointing

[bazarnov]

What

Resolving:

• [source-shopify] product_variants sync can be "endless" #39499
• https://github.com/airbytehq/oncall/issues/5850
• https://github.com/airbytehq/oncall/issues/5866
• https://github.com/airbytehq/oncall/issues/5881
• https://github.com/airbytehq/oncall/issues/5604

How

• added new transient Exception --> BulkJobRedirectToOtherShopError, which occurs when the store is operated from the manager account and the responses are redirected to the original shop instead of the aliased one
• removed session and stream_name input arguments for job_manager for BULK streams, in favor of the http_client setup
• added checkpointing ability to all BULK streams:
  • added new config field --> job_checkpoint_interval with the default value of 200,000 (200k lines) to be collected, before the single Bulk Job is checkpointed.
  • once there are more than the threshold lines collected for the single BULK Job, it's Canceled and the intermediate results are processed, instead of complete cancelation and retry using smaller job_size, which is also there, but will be used as a safety net for long-running jobs. The primary method to overcome data-density issues for heavy-data-loaded streams is checkpointing.
• added default sorting by ASC for all BULK streams, to guarantee the correct order of records for CHECKPOINTING
• Improved the FAILED scenario, when the BULK Job has FAILED status and we've collected some results already, we want them to be caught, instead of raising the exception and retry, thus the source should continue from the checkpoint, for streams that support it.
• added/updated unit tests to cover the updates

The checkpointing for the Shopify Bulk Operation looks like this:

• Bulk Job is running
• Bulk Job satisfies the condition of job_should_checkpoint (once the server has already collected some results, based on the objectCount field, within the response for the RUNNING job)
• Bulk Job is CANCELED (intentionally, expected)
• Intermediate results are processed, due to the URL provided for the already collected results, and the fact the Job is canceled
• The intermediate results are processed and the checkpoint is saved for each record observed.
• The next slice will start from the checkpointed cursor value, the end is also adjusted according to the bulk_window_in_days step.

Streams that don't support BULK CHECKPOINTING currently, typically, because the API doesn't support sortKey: UPDATED_AT:

• Metafield Location
• Metafield Product Variants
• Inventory Items
• Inventory Levels
• Product Variants

For such streams, the Cancel > Reduce Slice Size > Retry strategy is applied.

User Impact

No impact is expected, except that the stuck syncs that were not able to make any progress due to Canceled / Retried Bulk Jobs should be fetched normally, and the progress should take place now.

Can this PR be safely reverted and rolled back?

• YES 💚
• NO ❌

[vercel]

The latest updates on your projects. Learn more about Vercel for Git ↗︎

Name	Status	Preview	Comments	Updated (UTC)
airbyte-docs	✅ Ready (Inspect)	Visit Preview	💬 Add feedback	Jul 29, 2024 0:48am

[bazarnov]

The new pre-release tag was published: 2.4.14-dev.f03bf1112.

This release contains:

• store redirection to the original store
• adds bulk checkpointing at the 200k records (adjusted in the config)

[brianjlai]

Things look good, nice work on all these OC issues. One recommendation in the future would be to break these up into individual PRs. It was kind of hard to sort through things to understand what was being fixed for each OC issues especially since I'm not that familiar with the shopify code. But not a problem this time around.

I did add a few comments that we can discuss and also the http_client.name property should be in CDK 3.9.3 if you want to uptake that so we don't have to reference a private _name attribute.

And do we have any regression tests run against this new version of shopify?

[bazarnov]

And do we have any regression tests run against this new version of shopify?

Yes, @aldogonzalez8 helped with this, they are identical. No changes or regressions are expected.

[aldogonzalez8]

Acceptance testing is passing in combination with my PR branch.

Testing CAT steps to run in my local:

git checkout baz/source/shopify/fix-base-url-mismatch
git merge aldogonzalez8/fix_state_not_updating_for_stream_with_records_desc_order
airbyte-ci connectors --name=source-shopify --use-local-cdk test --only-step="acceptance"

Now, I will run a Regression locally

[aldogonzalez8]

Regression tests results

airbyte-ci connectors --name=source-shopify --use-local-cdk build --architecture=linux/arm64

poetry run pytest src/live_tests/regression_tests --connector-image=airbyte/source-shopify --pr-url=https://github.com/airbytehq/airbyte/pull/42095 --connection-id=auto

...
run with state = yes
[?] Which connection would you like to retrieve?: 
 > https://cloud.airbyte.com/workspaces/112412bb-78aa-4ca6-aace-ae0b83be2215/connections/5f1d6da1-87bc-46b4-8de7-1b3fe1a2be90 - Internal: True - Stream count: 38 - Last attempt duration: 4 minutes
...

product_variants example record for control version

{
  "admin_graphql_api_id": "gid://shopify/ProductVariant/43315735396541",
  "available_for_sale": true,
  "barcode": "123",
  "compare_at_price": "1.00",
  "created_at": "2023-12-11T10:37:41+00:00",
  "display_name": "8 Ounce Soy Candle - Test Variant",
  "fulfillment_service": "manual",
  "grams": 127,
  "id": 43315735396541,
  "image_id": null,
  "inventory_item_id": 45419395743933,
  "inventory_management": "SHOPIFY",
  "inventory_policy": "CONTINUE",
  "inventory_quantity": 1,
  "old_inventory_quantity": 1,
  "options": [
    {
      "name": "Title",
      "option_value": {
        "has_variants": true,
        "id": "gid://shopify/ProductOptionValue/1646832976061",
        "name": "Test Variant",
        "swatch": null
      },
      "value": "Test Variant"
    }
  ],
  "position": 4,
  "presentment_prices": [
    {
      "compare_at_price": {
        "amount": 1.0,
        "currency_code": "USD"
      },
      "price": {
        "amount": 113.0,
        "currency_code": "USD"
      }
    }
  ],
  "price": 113.0,
  "product_id": 6796229509309,
  "requires_shipping": true,
  "shop_url": "airbyte-integration-test",
  "sku": "123",
  "tax_code": "",
  "taxable": true,
  "title": "Test Variant",
  "updated_at": "2023-12-11T10:37:41+00:00",
  "weight": 127.0,
  "weight_unit": "GRAMS"
}

product_variants example record for target version

{
  "admin_graphql_api_id": "gid://shopify/ProductVariant/43315735396541",
  "available_for_sale": true,
  "barcode": "123",
  "compare_at_price": "1.00",
  "created_at": "2023-12-11T10:37:41+00:00",
  "display_name": "8 Ounce Soy Candle - Test Variant",
  "fulfillment_service": "manual",
  "grams": 127,
  "id": 43315735396541,
  "image_id": null,
  "inventory_item_id": 45419395743933,
  "inventory_management": "SHOPIFY",
  "inventory_policy": "CONTINUE",
  "inventory_quantity": 1,
  "old_inventory_quantity": 1,
  "options": [
    {
      "name": "Title",
      "option_value": {
        "has_variants": true,
        "id": "gid://shopify/ProductOptionValue/1646832976061",
        "name": "Test Variant",
        "swatch": null
      },
      "value": "Test Variant"
    }
  ],
  "position": 4,
  "presentment_prices": [],
  "price": 113.0,
  "product_id": 6796229509309,
  "requires_shipping": true,
  "shop_url": "airbyte-integration-test",
  "sku": "123",
  "tax_code": "",
  "taxable": true,
  "title": "Test Variant",
  "updated_at": "2023-12-11T10:37:41+00:00",
  "weight": 127.0,
  "weight_unit": "GRAMS"
}

[aldogonzalez8]

Could this one be an issue? @bazarnov

[bazarnov]

@aldogonzalez8 It was a question of the new config property:

• job_product_variants_include_pres_prices: true, which was missing for the config used for the latest regression / live tests. Fixed and updated the branch.

I've checked the data locally, and the presentment_prices and the value is there, after falling back to the default behavior, for this record id 43315735396541 :

"presentment_prices":[{"price":{"amount":113.0,"currency_code":"USD"},"compare_at_price":{"amount":1.0,"currency_code":"USD"}}]

We can repeat the tests, if needed.

[aldogonzalez8]

Test repeated from scratch, everything is 1:1

airbyte-ci connectors --name=source-shopify --use-local-cdk build --architecture=linux/arm64

poetry run pytest src/live_tests/regression_tests --connector-image=airbyte/source-shopify --pr-url=https://github.com/airbytehq/airbyte/pull/42095 --connection-id=auto

...
run with state = yes
[?] Which connection would you like to retrieve?: 
 > https://cloud.airbyte.com/workspaces/112412bb-78aa-4ca6-aace-ae0b83be2215/connections/5f1d6da1-87bc-46b4-8de7-1b3fe1a2be90 - Internal: True - Stream count: 38 - Last attempt duration: 4 minutes
...

[aldogonzalez8]

/approve-regression-tests

Check job output.

✅ Approving regression tests

[bazarnov]

It will be released on Monday, July 29th.

[aldogonzalez8]

/approve-regression-tests

Check job output.

✅ Approving regression tests

[jblakeman]

Curious, is there a specific reason the minimum needs to be 3600 or greater than 1? Adding this requirement means large jobs significantly increase the time to which they arrive at a small enough slice that can run without failure.

EDIT: My guess is that < 3600 is thought to be obviated with the introduction of job_checkpoint_interval, though given the variation across shop windows in rows collected per second, if a job of 15000 (min) rows always fails with a new flavor of retryable error after 3600 seconds and would otherwise succeed in some smaller threshold, we have added a failure requirement here without a workaround until a fix is in place.

[bazarnov]

@jblakeman Thank you for raising this; we selected the values from what we got on our side and are happy to adjust them now that we have your feedback. I'll add this to the next source-shopify update.