Update airbyte protocol migration #20745

gosusnp · 2022-12-21T01:40:50Z

What

Addresses a few gaps in the current protocol migration:

AirbyteMessage migration need the ConfiguredAirbyteCatalog to be able to downgrade messages
ConfiguredAirbyteCatalog is a top level object that has a separate lifecycle from the AirbyteMessage, add the necessary classes to support its migration
ConfiguredAirbyteCatalog is passed as part of the connector inputs that has a separate serialization, we need to add versioning support in that code path.

Closes #20695

How

Add ConfiguredAirbyteCatalog migrations
Pass ConfiguredAirbyteCatalog to AirbyteMessage migrations
Add ProtocolSerializer to wrap migrations that are not directly no AirbyteMessages
Some refactoring to share the migration selection logic across AirbyteMessageMIgrator and ConfiguredAirbyteCatologMigrator. As a result, MigrationContainer has been extracted

🚨 User Impact 🚨

There should be no user impact.
This will enable the migration of the ConfiguredAirbyteCatalog.

edgao

lgtm, I only really reviewed the migration interface stuff (i.e. skipped over all the files that actually call into migrations). Did a quick merge into my upgrade/downgrade branches locally and they seem happy, so and I'll do the merges correctly + extract the catalog migration

edgao · 2023-01-06T19:35:52Z

@gosusnp is this good to merge (into #20036)?

…susnp/20695-update-airbyte-protocol-migration

github-actions · 2023-01-06T22:20:35Z

Affected Connector Report

NOTE ⚠️ Changes in this PR affect the following connectors. Make sure to do the following as needed:

Run integration tests
Bump connector or module version
Add changelog
Publish the new version

✅ Sources (0)

Connector	Version	Changelog	Publish

See "Actionable Items" below for how to resolve warnings and errors.

❌ Destinations (16)

Connector	Version	Changelog	Publish
`destination-bigquery`	`1.2.9`	✅	✅
`destination-bigquery-denormalized`	`1.2.10`	✅	✅
`destination-clickhouse`	`0.2.1`	✅	✅
`destination-clickhouse-strict-encrypt`	`0.2.1`	🔵 (ignored)	🔵 (ignored)
`destination-jdbc`	`0.3.14`	🔵 (ignored)	🔵 (ignored)
`destination-mssql`	`0.1.22`	✅	✅
`destination-mssql-strict-encrypt`	`0.1.22`	🔵 (ignored)	🔵 (ignored)
`destination-mysql`	`0.1.20`	✅	✅
`destination-mysql-strict-encrypt`	❌ `0.1.21` (mismatch: `0.1.20`)	🔵 (ignored)	🔵 (ignored)
`destination-oracle`	`0.1.19`	✅	✅
`destination-oracle-strict-encrypt`	`0.1.19`	🔵 (ignored)	🔵 (ignored)
`destination-postgres`	`0.3.26`	✅	✅
`destination-postgres-strict-encrypt`	`0.3.26`	🔵 (ignored)	🔵 (ignored)
`destination-redshift`	`0.3.53`	✅	✅
`destination-snowflake`	`0.4.41`	✅	✅
`destination-tidb`	`0.1.0`	✅	✅

See "Actionable Items" below for how to resolve warnings and errors.

👀 Other Modules (1)

base-normalization

Actionable Items

(click to expand)

Category	Status	Actionable Item
Version	❌ mismatch	The version of the connector is different from its normal variant. Please bump the version of the connector.
	⚠ doc not found	The connector does not seem to have a documentation file. This can be normal (e.g. basic connector like `source-jdbc` is not published or documented). Please double-check to make sure that it is not a bug.
Changelog	⚠ doc not found	The connector does not seem to have a documentation file. This can be normal (e.g. basic connector like `source-jdbc` is not published or documented). Please double-check to make sure that it is not a bug.
	❌ changelog missing	There is no chnagelog for the current version of the connector. If you are the author of the current version, please add a changelog.
Publish	⚠ not in seed	The connector is not in the seed file (e.g. `source_definitions.yaml`), so its publication status cannot be checked. This can be normal (e.g. some connectors are cloud-specific, and only listed in the cloud seed file). Please double-check to make sure that it is not a bug.
	❌ diff seed version	The connector exists in the seed file, but the latest version is not listed there. This usually means that the latest version is not published. Please use the `/publish` command to publish the latest version.

benmoriceau

LGTM with a small comment about creation a micronaut factory

benmoriceau · 2023-01-10T00:10:49Z

...tocol/src/main/java/io/airbyte/commons/protocol/AirbyteProtocolVersionedMigratorFactory.java

+ * Factory to build AirbyteMessageVersionedMigrator
+ */
+@Singleton
+public class AirbyteProtocolVersionedMigratorFactory {


Should this be an actual micronaut factory (the @Factory annotation)?

Given the current structure, we inject the factory so that the worker can get a version dependent instance. Because this is per request, I'd expect a bigger refactoring to be needed for this.
Trying to limit complexity of the changes at the moment.

colesnodgrass · 2023-01-10T00:18:23Z

...tocol/src/main/java/io/airbyte/commons/protocol/AirbyteProtocolVersionedMigratorFactory.java

+  }
+
+  public <T> AirbyteMessageVersionedMigrator<T> getAirbyteMessageMigrator(final Version version) {
+    return new AirbyteMessageVersionedMigrator<>(this.airbyteMessageMigrator, version);


is this qualifier necessary here?

colesnodgrass · 2023-01-10T00:19:39Z

...ons-protocol/src/main/java/io/airbyte/commons/protocol/ConfiguredAirbyteCatalogMigrator.java

+  public <PreviousVersion, CurrentVersion> PreviousVersion downgrade(final CurrentVersion message, final Version target) {
+    return migrationContainer.downgrade(message, target, ConfiguredAirbyteCatalogMigrator::applyDowngrade);
+  }


Do we need to validate that this version is indeed a downgrade?

I don't think so, current design enforces a one-way look up from/to the current version. If the version provided isn't a valid downgrade or upgrade, we either fail because the version isn't supported (in the list of known migrations) or a mismatch in the type conversion.

colesnodgrass · 2023-01-10T00:19:50Z

...ons-protocol/src/main/java/io/airbyte/commons/protocol/ConfiguredAirbyteCatalogMigrator.java

+  public <PreviousVersion, CurrentVersion> CurrentVersion upgrade(final PreviousVersion message, final Version source) {
+    return migrationContainer.upgrade(message, source, ConfiguredAirbyteCatalogMigrator::applyUpgrade);
+  }


Do we need to validate that this version is indeed an upgrade?

answered above.

colesnodgrass · 2023-01-10T00:21:17Z

...ons-protocol/src/main/java/io/airbyte/commons/protocol/ConfiguredAirbyteCatalogMigrator.java

+  private static <PreviousVersion, CurrentVersion> PreviousVersion applyDowngrade(final ConfiguredAirbyteCatalogMigration<PreviousVersion, CurrentVersion> migration,
+                                                                                  final Object message) {


As this method is generic, can you have final CurrentVersion message instead of final Object message?

colesnodgrass · 2023-01-10T00:21:36Z

...ons-protocol/src/main/java/io/airbyte/commons/protocol/ConfiguredAirbyteCatalogMigrator.java

+  }
+
+  // Helper function to work around type casting
+  private static <PreviousVersion, CurrentVersion> CurrentVersion applyUpgrade(final ConfiguredAirbyteCatalogMigration<PreviousVersion, CurrentVersion> migration,


As this method is generic, can you have final PreviousVersion message instead of final Object message?

I was not able to make it work because of the MigrationContainer.
More specifically, eventually, we apply a chain of transforms f(A -> B -> C) which is falling appart when passing in final BiFunction<T, Object, Object> applyDowngrade. Type end up being casted dynamically from object because of this.

colesnodgrass · 2023-01-10T00:24:06Z

.../src/main/java/io/airbyte/commons/protocol/migrations/ConfiguredAirbyteCatalogMigration.java

+public interface ConfiguredAirbyteCatalogMigration<PreviousVersion, CurrentVersion> extends Migration {
+
+  /**
+   * Downgrades a ConfiguredAirbyteCatalog to from the new version to the old version


typo in this javadoc, the first to doesn't apply

colesnodgrass · 2023-01-10T00:25:51Z

...ommons-protocol/src/main/java/io/airbyte/commons/protocol/migrations/MigrationContainer.java

+   */
+  public <PreviousVersion, CurrentVersion> PreviousVersion downgrade(final CurrentVersion message,
+                                                                     final Version target,
+                                                                     final BiFunction<T, Object, Object> applyDowngrade) {


Could this be BiFunction<T, PreviousVersion, CurrentVersion> applyDowngrade?

Same question applies to the next method as well.

Similar answer as a previous question, reposting here for context:

We apply a chain of transforms f(A -> B -> C) which is falling appart when passing in final BiFunction<T, Object, Object> applyDowngrade. Type end up being casted dynamically from object because of this.

colesnodgrass · 2023-01-10T00:27:03Z

...ommons-protocol/src/main/java/io/airbyte/commons/protocol/migrations/MigrationContainer.java

+      return (PreviousVersion) message;
+    }
+
+    Object result = message;


Could this be defined as PreviousVersion results?

No because of the chaining of transforms, if we end up having more than one migration, the type of results will change.

colesnodgrass · 2023-01-10T00:29:21Z

...ommons-protocol/src/main/java/io/airbyte/commons/protocol/migrations/MigrationContainer.java

+    }
+  }
+
+  public Set<String> getMigrationKeys() {


Should this be public or @VisibleForTesting?

This should be public, once the bootloader is migrated to micronaut, we should use this to check that we actually have all the migrations required to support the range of protocol versions from the config.

colesnodgrass · 2023-01-10T00:31:59Z

airbyte-commons-protocol/src/main/java/io/airbyte/commons/protocol/AirbyteMessageMigrator.java

+  public AirbyteMessageMigrator(final List<AirbyteMessageMigration<?, ?>> migrations) {
+    migrationContainer = new MigrationContainer<>(migrations);


Should the MigrationContainer be injected here instead of the List<>?

I don't think so, the intent is to inject the list of migrations. We have different migrations intended for the AirbyteMessageMigrator or the ConfiguredAirbyteCatalogMigrator.
The MigrationContainer<> is a shared container for abstracting the logic of how we fetch the migrations. Injecting the MigrationContainer would involve an extra layer of subclassing which looks more like overhead than solving a problem at the moment.

* Add Airbyte Protocol V1 support. * Fix VersionedAirbyteStreamFactoryTest * Remove AirbyteMessageMigrationV0 example * Add Protocol Version constants * 🎉Updated normalization to handle new datatypes (#19721) * Updated normalization simple stream processing to handle new datatypes * Updated normalization nested stream processing to handle new datatypes * Updated normalization nested stream processing to handle new datatypes * Updated normalization drop_scd_catalog processing to handle new datatypes * Updated normalization ephemeral test processing to handle new datatypes * fixed more tests for normalization * fixed more tests for normalization * fixed more tests for normalization * fixed more tests for normalization * fixed more issues * fixed more issues (clickhouse) * fixed more issues * fixed more issues * fixed more issues * added binary type processing for some DBs * cleared commented code and moved some hardcodes to processing as macro * fixed codestyle and cleared commented code * minor refactor * minor refactor * minor refactor * fixed bool cast error * fixed dict->str cast error * fixed is_combining_node cast py check * removed commented code * removed commented code * committed autogenerated normalization_test_output files * committed autogenerated normalization_test_output files (new files) * refactored utils.py * Updated utils.py to use Callable functions and get rid of property_type in is_number and is_bool functions * committed autogenerated normalization_test_output files (new files) * fixed typo in TIMESTAMP_WITH_TIMEZONE_TYPE * updated stream_processor to handle string type first as a wider type * fixed arrays normalization by updating is_simple_property method as per new approaches * format Co-authored-by: Edward Gao <[email protected]> * Update airbyte protocol migration (#20745) * Extract MigrationContainer from AirbyteMessageMigrator * Add ConfiguredAirbyteCatalogMigrations * Add ConfiguredAirbyteCatalog to AirbyteMessageMigrations * Enable ConfiguredAirbyteCatalog migration * Fix tests * Remove extra this. * Add missing docs * Typo Co-authored-by: Edward Gao <[email protected]> * Data types update: Implement protocol message migrations (#19240) * Extract MigrationContainer from AirbyteMessageMigrator * Add ConfiguredAirbyteCatalogMigrations * Add ConfiguredAirbyteCatalog to AirbyteMessageMigrations * Enable ConfiguredAirbyteCatalog migration * set up scaffolding * [wip] more scaffolding, basic unit test * minimal green code * [wip] add failing test for other primitive types * correct version number * handle basic primitive type decls * add implicit cases * add recursive schema * formatting * comment * support not * fix indentation * handle all nested schema cases * handle boolean schemas * verify empty schema handling * cleanup * extract map * code organization * extract method * reformat * [wip] more tests, minor fix type array handling * corrected test * cleanup * reformat * switch to v1 * add support for multityped fields * missed test case * nested test class * basic record upgrade * implement record upgrades * slight refactor * comments+clarificationso * extract constants * (partly) correct model classes * add de/ser * formatting * extract constants * fix json reference * update docs * switch to v1 models * fix compile+test * add base64 handling * use vnull * Data types update: Implement protocol message downgrade path (#19909) * rough skeleton for passing catalog into migration * basic test * more scaffolding * basic implementation * add primitives test * add in other tests (nested fields currently failing) * add formats * impleent oneOf handling * formatting * oneOf handling * better tests * comments + organization * progress * basic test case * downgrade objects, ish * basic array implementation * handle numeric failure * test for new type * handle array items * empty schema handling * first pass at oneof handling * add more tests+handling * more tests * comments * add empty oneof test case * format + reorganize * more reorganize * fix name * also downgrade binary data * only import vnull * move migrations into v1 package * extract schema mutation code * comment * extract schema migration to new class * extract record downgrade logic for future use * format * fix build after rebase * rename private method for consistency * also implement configuredcatalog migrations >.> * quick and dirty tests * slight cleanup * fix tests * pmd * pmd test * null check on message objects * maybe fix acceptance tests? * fix name * extract constants * more fixes * tmp * meh * fix cdc acc tests * revert to master source-postgres * remove log messages * revert other misc hacks * integers are valid cursors * remove unrelated change * fix build * fix build more? * [MUST REVERT] use dev normalization * capture kube logs * also here? * no debug logs? * delete dup from merging * add final everywhere * revert test changes Co-authored-by: Jimmy Ma <[email protected]> * On-the-fly migrations of persisted catalogs (#21757) * On the fly catalog migration for normalization activity * On the fly catalog migration for job persistence * On the fly migration for standard sync persistence * On the fly migration for airbyte catalogs * Refactor code to share JsonSchema traversal * Add V0 Data type search function * PMD and Format * Fix getOrInsertActorCatalog and ConfigRepositoryE2E tests * Null-proofing CatalogMigrationV1Helper * More null checks * Fix test * Format * Add data type v1 support to the FE * Changes AC test check to check exited ps (#21672) some docker compose changes no longer show exited processes. this broke out test this change should fix master tested in a runner that failed * Move wellknown types mapping to the utility function * use protocolv1 normalization --------- Co-authored-by: Topher Lubaway <[email protected]> Co-authored-by: Edward Gao <[email protected]> * Update protocol support range (#21996) * bump normalization version to 0.3.0 * Add version check on normalization (#22048) * Add normalization min version check * Add visible for testing --------- Co-authored-by: Edward Gao <[email protected]> Co-authored-by: Eugene <[email protected]> Co-authored-by: Topher Lubaway <[email protected]>

gosusnp added 4 commits December 20, 2022 11:56

Extract MigrationContainer from AirbyteMessageMigrator

e2146b9

Add ConfiguredAirbyteCatalogMigrations

caa1692

Add ConfiguredAirbyteCatalog to AirbyteMessageMigrations

a374a62

Enable ConfiguredAirbyteCatalog migration

7adbe36

octavia-squidington-iv added area/platform issues related to the platform area/worker Related to worker labels Dec 21, 2022

gosusnp requested a review from edgao December 21, 2022 01:41

gosusnp temporarily deployed to more-secrets December 21, 2022 17:06 — with GitHub Actions Inactive

gosusnp marked this pull request as ready for review December 21, 2022 17:07

gosusnp changed the title ~~Gosusnp/20695 update airbyte protocol migration~~ Update airbyte protocol migration Dec 21, 2022

gosusnp requested review from benmoriceau, colesnodgrass, jdpgrailsdev and alovew December 21, 2022 17:08

gosusnp mentioned this pull request Dec 21, 2022

Add protocol migration for Inputs passed to the connectors #20688

Closed

edgao approved these changes Dec 21, 2022

View reviewed changes

edgao mentioned this pull request Dec 22, 2022

Data types update: Implement protocol message migrations #19240

Merged

Merge branch 'gosusnp/platform-use-protocol-v1-the-quick-way' into go…

dbad5bb

…susnp/20695-update-airbyte-protocol-migration

edgao temporarily deployed to more-secrets January 6, 2023 22:20 — with GitHub Actions Inactive

edgao temporarily deployed to more-secrets January 9, 2023 21:19 — with GitHub Actions Inactive

edgao temporarily deployed to more-secrets January 9, 2023 22:19 — with GitHub Actions Inactive

edgao temporarily deployed to more-secrets January 9, 2023 23:01 — with GitHub Actions Inactive

Fix tests

dce7362

gosusnp temporarily deployed to more-secrets January 9, 2023 23:59 — with GitHub Actions Inactive

gosusnp temporarily deployed to more-secrets January 10, 2023 00:00 — with GitHub Actions Inactive

benmoriceau reviewed Jan 10, 2023

View reviewed changes

colesnodgrass reviewed Jan 10, 2023

View reviewed changes

gosusnp added 3 commits January 9, 2023 17:09

Remove extra this.

c8275d2

Add missing docs

f9717cd

Typo

c083b64

gosusnp temporarily deployed to more-secrets January 10, 2023 01:39 — with GitHub Actions Inactive

gosusnp temporarily deployed to more-secrets January 10, 2023 01:40 — with GitHub Actions Inactive

gosusnp requested review from benmoriceau and colesnodgrass January 10, 2023 18:33

colesnodgrass approved these changes Jan 10, 2023

View reviewed changes

benmoriceau approved these changes Jan 10, 2023

View reviewed changes

gosusnp merged commit ee150e3 into gosusnp/platform-use-protocol-v1-the-quick-way Jan 10, 2023

gosusnp deleted the gosusnp/20695-update-airbyte-protocol-migration branch January 10, 2023 18:58

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Update airbyte protocol migration #20745

Update airbyte protocol migration #20745

gosusnp commented Dec 21, 2022

edgao left a comment

edgao commented Jan 6, 2023

github-actions bot commented Jan 6, 2023 •

edited

Loading

✅ Sources (0)

❌ Destinations (16)

👀 Other Modules (1)

Actionable Items

benmoriceau left a comment

benmoriceau Jan 10, 2023

gosusnp Jan 10, 2023

colesnodgrass Jan 10, 2023

gosusnp Jan 10, 2023

colesnodgrass Jan 10, 2023

gosusnp Jan 10, 2023

colesnodgrass Jan 10, 2023

gosusnp Jan 10, 2023

colesnodgrass Jan 10, 2023

colesnodgrass Jan 10, 2023

gosusnp Jan 10, 2023

colesnodgrass Jan 10, 2023

gosusnp Jan 10, 2023

colesnodgrass Jan 10, 2023

colesnodgrass Jan 10, 2023

gosusnp Jan 10, 2023

colesnodgrass Jan 10, 2023

gosusnp Jan 10, 2023

colesnodgrass Jan 10, 2023

gosusnp Jan 10, 2023

colesnodgrass Jan 10, 2023

gosusnp Jan 10, 2023

		private static <PreviousVersion, CurrentVersion> PreviousVersion applyDowngrade(final ConfiguredAirbyteCatalogMigration<PreviousVersion, CurrentVersion> migration,
		final Object message) {

		public AirbyteMessageMigrator(final List<AirbyteMessageMigration<?, ?>> migrations) {
		migrationContainer = new MigrationContainer<>(migrations);

Update airbyte protocol migration #20745

Update airbyte protocol migration #20745

Conversation

gosusnp commented Dec 21, 2022

What

How

Recommended reading order

🚨 User Impact 🚨

edgao left a comment

Choose a reason for hiding this comment

edgao commented Jan 6, 2023

github-actions bot commented Jan 6, 2023 • edited Loading

Affected Connector Report

✅ Sources (0)

❌ Destinations (16)

👀 Other Modules (1)

Actionable Items

benmoriceau left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

github-actions bot commented Jan 6, 2023 •

edited

Loading