-
Notifications
You must be signed in to change notification settings - Fork 279
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
revert protocol v1 migration to noop (#5678)
- Loading branch information
Showing
16 changed files
with
2,336 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
219 changes: 219 additions & 0 deletions
219
...col/src/main/java/io/airbyte/commons/protocol/migrations/v1/CatalogMigrationV1Helper.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,219 @@ | ||
| /* | ||
| * Copyright (c) 2023 Airbyte, Inc., all rights reserved. | ||
| */ | ||
|
|
||
| package io.airbyte.commons.protocol.migrations.v1; | ||
|
|
||
| import com.fasterxml.jackson.databind.JsonNode; | ||
| import io.airbyte.commons.protocol.migrations.util.SchemaMigrations; | ||
| import io.airbyte.protocol.models.AirbyteCatalog; | ||
| import io.airbyte.protocol.models.AirbyteStream; | ||
| import io.airbyte.protocol.models.ConfiguredAirbyteCatalog; | ||
| import io.airbyte.protocol.models.ConfiguredAirbyteStream; | ||
|
|
||
| /** | ||
| * For the v0 to v1 migration, it appears that we are persisting some protocol objects without | ||
| * version. Until this gets addressed more properly, this class contains the helper functions used | ||
| * to handle this on the fly migration. | ||
| * | ||
| * Once persisted objects are versioned, this code should be deleted. | ||
| */ | ||
| public class CatalogMigrationV1Helper { | ||
|
|
||
| /** | ||
| * Performs an in-place migration of the schema from v0 to v1 if v0 data types are detected. | ||
| * | ||
| * @param configuredAirbyteCatalog to migrate | ||
| */ | ||
| public static void upgradeSchemaIfNeeded(final ConfiguredAirbyteCatalog configuredAirbyteCatalog) { | ||
| if (containsV0DataTypes(configuredAirbyteCatalog)) { | ||
| upgradeSchema(configuredAirbyteCatalog); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Performs an in-place migration of the schema from v0 to v1 if v0 data types are detected. | ||
| * | ||
| * @param airbyteCatalog to migrate | ||
| */ | ||
| public static void upgradeSchemaIfNeeded(final AirbyteCatalog airbyteCatalog) { | ||
| if (containsV0DataTypes(airbyteCatalog)) { | ||
| upgradeSchema(airbyteCatalog); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Performs an in-place migration of the schema from v0 to v1. | ||
| * | ||
| * @param configuredAirbyteCatalog to migrate | ||
| */ | ||
| private static void upgradeSchema(final ConfiguredAirbyteCatalog configuredAirbyteCatalog) { | ||
| for (final var stream : configuredAirbyteCatalog.getStreams()) { | ||
| SchemaMigrationV1.upgradeSchema(stream.getStream().getJsonSchema()); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Performs an in-place migration of the schema from v0 to v1. | ||
| * | ||
| * @param airbyteCatalog to migrate | ||
| */ | ||
| private static void upgradeSchema(final AirbyteCatalog airbyteCatalog) { | ||
| for (final var stream : airbyteCatalog.getStreams()) { | ||
| SchemaMigrationV1.upgradeSchema(stream.getJsonSchema()); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Returns true if catalog contains v0 data types. | ||
| */ | ||
| private static boolean containsV0DataTypes(final ConfiguredAirbyteCatalog configuredAirbyteCatalog) { | ||
| if (configuredAirbyteCatalog == null) { | ||
| return false; | ||
| } | ||
|
|
||
| return configuredAirbyteCatalog | ||
| .getStreams() | ||
| .stream().findFirst() | ||
| .map(ConfiguredAirbyteStream::getStream) | ||
| .map(CatalogMigrationV1Helper::streamContainsV0DataTypes) | ||
| .orElse(false); | ||
| } | ||
|
|
||
| /** | ||
| * Returns true if catalog contains v0 data types. | ||
| */ | ||
| private static boolean containsV0DataTypes(final AirbyteCatalog airbyteCatalog) { | ||
| if (airbyteCatalog == null) { | ||
| return false; | ||
| } | ||
|
|
||
| return airbyteCatalog | ||
| .getStreams() | ||
| .stream().findFirst() | ||
| .map(CatalogMigrationV1Helper::streamContainsV0DataTypes) | ||
| .orElse(false); | ||
| } | ||
|
|
||
| private static boolean streamContainsV0DataTypes(final AirbyteStream airbyteStream) { | ||
| if (airbyteStream == null || airbyteStream.getJsonSchema() == null) { | ||
| return false; | ||
| } | ||
| return hasV0DataType(airbyteStream.getJsonSchema()); | ||
| } | ||
|
|
||
| /** | ||
| * Performs of search of a v0 data type node, returns true at the first node found. | ||
| */ | ||
| private static boolean hasV0DataType(final JsonNode schema) { | ||
| if (SchemaMigrationV1.isPrimitiveTypeDeclaration(schema)) { | ||
| return true; | ||
| } | ||
|
|
||
| for (final JsonNode subSchema : SchemaMigrations.findSubschemas(schema)) { | ||
| if (hasV0DataType(subSchema)) { | ||
| return true; | ||
| } | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
| /** | ||
| * Performs an in-place migration of the schema from v1 to v0 if v1 data types are detected. | ||
| * | ||
| * @param configuredAirbyteCatalog to migrate | ||
| */ | ||
| public static void downgradeSchemaIfNeeded(final ConfiguredAirbyteCatalog configuredAirbyteCatalog) { | ||
| if (containsV1DataTypes(configuredAirbyteCatalog)) { | ||
| downgradeSchema(configuredAirbyteCatalog); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Performs an in-place migration of the schema from v1 to v0 if v1 data types are detected. | ||
| * | ||
| * @param airbyteCatalog to migrate | ||
| */ | ||
| public static void downgradeSchemaIfNeeded(final AirbyteCatalog airbyteCatalog) { | ||
| if (containsV1DataTypes(airbyteCatalog)) { | ||
| downgradeSchema(airbyteCatalog); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Performs an in-place migration of the schema from v1 to v0. | ||
| * | ||
| * @param configuredAirbyteCatalog to migrate | ||
| */ | ||
| private static void downgradeSchema(final ConfiguredAirbyteCatalog configuredAirbyteCatalog) { | ||
| for (final var stream : configuredAirbyteCatalog.getStreams()) { | ||
| SchemaMigrationV1.downgradeSchema(stream.getStream().getJsonSchema()); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Performs an in-place migration of the schema from v1 to v0. | ||
| * | ||
| * @param airbyteCatalog to migrate | ||
| */ | ||
| private static void downgradeSchema(final AirbyteCatalog airbyteCatalog) { | ||
| for (final var stream : airbyteCatalog.getStreams()) { | ||
| SchemaMigrationV1.downgradeSchema(stream.getJsonSchema()); | ||
| } | ||
| } | ||
|
|
||
| /** | ||
| * Returns true if catalog contains v1 data types. | ||
| */ | ||
| private static boolean containsV1DataTypes(final ConfiguredAirbyteCatalog configuredAirbyteCatalog) { | ||
| if (configuredAirbyteCatalog == null) { | ||
| return false; | ||
| } | ||
|
|
||
| return configuredAirbyteCatalog | ||
| .getStreams() | ||
| .stream().findFirst() | ||
| .map(ConfiguredAirbyteStream::getStream) | ||
| .map(CatalogMigrationV1Helper::streamContainsV1DataTypes) | ||
| .orElse(false); | ||
| } | ||
|
|
||
| /** | ||
| * Returns true if catalog contains v1 data types. | ||
| */ | ||
| private static boolean containsV1DataTypes(final AirbyteCatalog airbyteCatalog) { | ||
| if (airbyteCatalog == null) { | ||
| return false; | ||
| } | ||
|
|
||
| return airbyteCatalog | ||
| .getStreams() | ||
| .stream().findFirst() | ||
| .map(CatalogMigrationV1Helper::streamContainsV1DataTypes) | ||
| .orElse(false); | ||
| } | ||
|
|
||
| private static boolean streamContainsV1DataTypes(final AirbyteStream airbyteStream) { | ||
| if (airbyteStream == null || airbyteStream.getJsonSchema() == null) { | ||
| return false; | ||
| } | ||
| return hasV1DataType(airbyteStream.getJsonSchema()); | ||
| } | ||
|
|
||
| /** | ||
| * Performs of search of a v0 data type node, returns true at the first node found. | ||
| */ | ||
| private static boolean hasV1DataType(final JsonNode schema) { | ||
| if (SchemaMigrationV1.isPrimitiveReferenceTypeDeclaration(schema)) { | ||
| return true; | ||
| } | ||
|
|
||
| for (final JsonNode subSchema : SchemaMigrations.findSubschemas(schema)) { | ||
| if (hasV1DataType(subSchema)) { | ||
| return true; | ||
| } | ||
| } | ||
| return false; | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.