Merge branch 'master' into etsybaev/9525-source-redshift-add-schema-selection

Author: etsybaev

.bumpversion.cfg

@@ -1,5 +1,5 @@
 [bumpversion]
-current_version = 0.35.9-alpha
+current_version = 0.35.10-alpha
 commit = False
 tag = False
 parse = (?P<major>\d+)\.(?P<minor>\d+)\.(?P<patch>\d+)(\-[a-z]+)?

.env

@@ -10,7 +10,7 @@
 
 
 ### SHARED ###
-VERSION=0.35.9-alpha
+VERSION=0.35.10-alpha
 
 # When using the airbyte-db via default docker image
 CONFIG_ROOT=/data

airbyte-bootloader/Dockerfile

@@ -5,6 +5,6 @@ ENV APPLICATION airbyte-bootloader
 
 WORKDIR /app
 
-ADD bin/${APPLICATION}-0.35.9-alpha.tar /app
+ADD bin/${APPLICATION}-0.35.10-alpha.tar /app
 
-ENTRYPOINT ["/bin/bash", "-c", "${APPLICATION}-0.35.9-alpha/bin/${APPLICATION}"]
+ENTRYPOINT ["/bin/bash", "-c", "${APPLICATION}-0.35.10-alpha/bin/${APPLICATION}"]

airbyte-config/init/src/main/java/io/airbyte/config/init/YamlSeedConfigPersistence.java

@@ -119,6 +119,12 @@ public <T> List<T> listConfigs(final AirbyteConfig configType, final Class<T> cl
     return configs.values().stream().map(json -> Jsons.object(json, clazz)).collect(Collectors.toList());
   }
 
+  @Override
+  public <T> ConfigWithMetadata<T> getConfigWithMetadata(final AirbyteConfig configType, final String configId, final Class<T> clazz)
+      throws ConfigNotFoundException, JsonValidationException, IOException {
+    throw new UnsupportedOperationException("Yaml Seed Config doesn't support metadata");
+  }
+
   @Override
   public <T> List<ConfigWithMetadata<T>> listConfigsWithMetadata(final AirbyteConfig configType, final Class<T> clazz)
       throws JsonValidationException, IOException {

airbyte-config/init/src/main/resources/config/STANDARD_SOURCE_DEFINITION/36c891d9-4bd9-43ac-bad2-10e12756272c.json

@@ -2,7 +2,7 @@
   "sourceDefinitionId": "36c891d9-4bd9-43ac-bad2-10e12756272c",
   "name": "HubSpot",
   "dockerRepository": "airbyte/source-hubspot",
-  "dockerImageTag": "0.1.35",
+  "dockerImageTag": "0.1.36",
   "documentationUrl": "https://docs.airbyte.io/integrations/sources/hubspot",
   "icon": "hubspot.svg"
 }

airbyte-config/init/src/main/resources/seed/source_definitions.yaml

@@ -252,7 +252,7 @@
 - name: Google Analytics
   sourceDefinitionId: eff3616a-f9c3-11eb-9a03-0242ac130003
   dockerRepository: airbyte/source-google-analytics-v4
-  dockerImageTag: 0.1.15
+  dockerImageTag: 0.1.16
   documentationUrl: https://docs.airbyte.io/integrations/sources/google-analytics-v4
   icon: google-analytics.svg
   sourceType: api
@@ -307,7 +307,7 @@
 - name: HubSpot
   sourceDefinitionId: 36c891d9-4bd9-43ac-bad2-10e12756272c
   dockerRepository: airbyte/source-hubspot
-  dockerImageTag: 0.1.35
+  dockerImageTag: 0.1.36
   documentationUrl: https://docs.airbyte.io/integrations/sources/hubspot
   icon: hubspot.svg
   sourceType: api

airbyte-config/init/src/main/resources/seed/source_specs.yaml

@@ -2449,7 +2449,7 @@
         oauthFlowOutputParameters:
         - - "access_token"
         - - "refresh_token"
-- dockerImage: "airbyte/source-google-analytics-v4:0.1.15"
+- dockerImage: "airbyte/source-google-analytics-v4:0.1.16"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/google-analytics-v4"
     connectionSpecification:
@@ -2476,6 +2476,20 @@
             \ will not be replicated."
           examples:
           - "2020-06-01"
+        window_in_days:
+          type: "integer"
+          title: "Window in days"
+          description: "The amount of days for each data-chunk beginning from start_date.\
+            \ Bigger the value - faster the fetch. (Min=1, as for a Day; Max=364,\
+            \ as for a Year)."
+          examples:
+          - 30
+          - 60
+          - 90
+          - 120
+          - 200
+          - 364
+          default: 1
         custom_reports:
           order: 3
           type: "string"
@@ -2487,6 +2501,7 @@
           order: 0
           type: "object"
           title: "Credentials"
+          description: "Credentials for the service"
           oneOf:
           - title: "Authenticate via Google (Oauth)"
             type: "object"
@@ -3067,7 +3082,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-hubspot:0.1.35"
+- dockerImage: "airbyte/source-hubspot:0.1.36"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/hubspot"
     connectionSpecification:
@@ -4217,7 +4232,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-mssql:0.3.13"
+- dockerImage: "airbyte/source-mssql:0.3.14"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/destinations/mssql"
     connectionSpecification:
@@ -5895,7 +5910,7 @@
     supportsNormalization: false
     supportsDBT: false
     supported_destination_sync_modes: []
-- dockerImage: "airbyte/source-postgres:0.4.2"
+- dockerImage: "airbyte/source-postgres:0.4.3"
   spec:
     documentationUrl: "https://docs.airbyte.io/integrations/sources/postgres"
     connectionSpecification:

airbyte-config/models/src/main/resources/types/AttemptFailureSummary.yaml

@@ -0,0 +1,18 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/AttemptFailureSummary.yaml
+title: AttemptFailureSummary
+description: Attempt-level summarization of failures that occurred during a sync workflow.
+type: object
+additionalProperties: false
+required:
+  - failures
+properties:
+  failures:
+    description: Ordered list of failures that occurred during the attempt.
+    type: array
+    items:
+      "$ref": FailureReason.yaml
+  partialSuccess:
+    description: True if the number of committed records for this attempt was greater than 0. False if 0 records were committed.
+    type: boolean

airbyte-config/models/src/main/resources/types/FailureReason.yaml

@@ -0,0 +1,44 @@
+---
+"$schema": http://json-schema.org/draft-07/schema#
+"$id": https://github.com/airbytehq/airbyte/blob/master/airbyte-config/models/src/main/resources/types/FailureReason.yaml
+title: FailureSummary
+type: object
+required:
+  - failureOrigin
+  - timestamp
+additionalProperties: false
+properties:
+  failureOrigin:
+    description: Indicates where the error originated. If not set, the origin of error is not well known.
+    type: string
+    enum:
+      - unknown
+      - source
+      - destination
+      - replicationWorker
+      - persistence
+      - normalization
+      - dbt
+  failureType:
+    description: Categorizes well known errors into types for programmatic handling. If not set, the type of error is not well known.
+    type: string
+    enum:
+      - unknown
+      - userError
+      - systemError
+      - transient
+  internalMessage:
+    description: Human readable failure description for consumption by technical system operators, like Airbyte engineers or OSS users.
+    type: string
+  externalMessage:
+    description: Human readable failure description for presentation in the UI to non-technical users.
+    type: string
+  metadata:
+    description: Key-value pairs of relevant data
+    type: object
+    additionalProperties: true
+  stacktrace:
+    description: Raw stacktrace associated with the failure.
+    type: string
+  timestamp:
+    type: integer

airbyte-config/models/src/main/resources/types/ReplicationOutput.yaml

@@ -16,3 +16,7 @@ properties:
     "$ref": State.yaml
   output_catalog:
     existingJavaType: io.airbyte.protocol.models.ConfiguredAirbyteCatalog
+  failures:
+    type: array
+    items:
+      "$ref": FailureReason.yaml

airbyte-config/models/src/main/resources/types/StandardSyncOutput.yaml

@@ -16,3 +16,7 @@ properties:
     "$ref": State.yaml
   output_catalog:
     existingJavaType: io.airbyte.protocol.models.ConfiguredAirbyteCatalog
+  failures:
+    type: array
+    items:
+      "$ref": FailureReason.yaml

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/ConfigPersistence.java

@@ -19,6 +19,9 @@ public interface ConfigPersistence {
 
   <T> List<T> listConfigs(AirbyteConfig configType, Class<T> clazz) throws JsonValidationException, IOException;
 
+  <T> ConfigWithMetadata<T> getConfigWithMetadata(AirbyteConfig configType, String configId, Class<T> clazz)
+      throws ConfigNotFoundException, JsonValidationException, IOException;
+
   <T> List<ConfigWithMetadata<T>> listConfigsWithMetadata(AirbyteConfig configType, Class<T> clazz) throws JsonValidationException, IOException;
 
   <T> void writeConfig(AirbyteConfig configType, String configId, T config) throws JsonValidationException, IOException;

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/DatabaseConfigPersistence.java

@@ -139,7 +139,7 @@ private StandardDestinationDefinition getStandardDestinationDefinition(final Str
     return result.get(0).getConfig();
   }
 
-  private SourceConnection getSourceConnection(String configId) throws IOException, ConfigNotFoundException {
+  private SourceConnection getSourceConnection(final String configId) throws IOException, ConfigNotFoundException {
     final List<ConfigWithMetadata<SourceConnection>> result = listSourceConnectionWithMetadata(Optional.of(UUID.fromString(configId)));
     validate(configId, result, ConfigSchema.SOURCE_CONNECTION);
     return result.get(0).getConfig();
@@ -188,7 +188,7 @@ private List<UUID> connectionOperationIds(final UUID connectionId) throws IOExce
         .fetch());
 
     final List<UUID> ids = new ArrayList<>();
-    for (Record record : result) {
+    for (final Record record : result) {
       ids.add(record.get(CONNECTION_OPERATION.OPERATION_ID));
     }
 
@@ -204,13 +204,50 @@ private <T> void validate(final String configId, final List<ConfigWithMetadata<T
     }
   }
 
+  private <T> ConfigWithMetadata<T> validateAndReturn(final String configId,
+                                                      final List<ConfigWithMetadata<T>> result,
+                                                      final AirbyteConfig airbyteConfig)
+      throws ConfigNotFoundException {
+    validate(configId, result, airbyteConfig);
+    return result.get(0);
+  }
+
   @Override
   public <T> List<T> listConfigs(final AirbyteConfig configType, final Class<T> clazz) throws JsonValidationException, IOException {
     final List<T> config = new ArrayList<>();
     listConfigsWithMetadata(configType, clazz).forEach(c -> config.add(c.getConfig()));
     return config;
   }
 
+  @Override
+  public <T> ConfigWithMetadata<T> getConfigWithMetadata(final AirbyteConfig configType, final String configId, final Class<T> clazz)
+      throws ConfigNotFoundException, JsonValidationException, IOException {
+    final Optional<UUID> configIdOpt = Optional.of(UUID.fromString(configId));
+    if (configType == ConfigSchema.STANDARD_WORKSPACE) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listStandardWorkspaceWithMetadata(configIdOpt), configType);
+    } else if (configType == ConfigSchema.STANDARD_SOURCE_DEFINITION) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listStandardSourceDefinitionWithMetadata(configIdOpt), configType);
+    } else if (configType == ConfigSchema.STANDARD_DESTINATION_DEFINITION) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listStandardDestinationDefinitionWithMetadata(configIdOpt), configType);
+    } else if (configType == ConfigSchema.SOURCE_CONNECTION) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listSourceConnectionWithMetadata(configIdOpt), configType);
+    } else if (configType == ConfigSchema.DESTINATION_CONNECTION) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listDestinationConnectionWithMetadata(configIdOpt), configType);
+    } else if (configType == ConfigSchema.SOURCE_OAUTH_PARAM) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listSourceOauthParamWithMetadata(configIdOpt), configType);
+    } else if (configType == ConfigSchema.DESTINATION_OAUTH_PARAM) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listDestinationOauthParamWithMetadata(configIdOpt), configType);
+    } else if (configType == ConfigSchema.STANDARD_SYNC_OPERATION) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listStandardSyncOperationWithMetadata(configIdOpt), configType);
+    } else if (configType == ConfigSchema.STANDARD_SYNC) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listStandardSyncWithMetadata(configIdOpt), configType);
+    } else if (configType == ConfigSchema.STANDARD_SYNC_STATE) {
+      return (ConfigWithMetadata<T>) validateAndReturn(configId, listStandardSyncStateWithMetadata(configIdOpt), configType);
+    } else {
+      throw new IllegalArgumentException("Unknown Config Type " + configType);
+    }
+  }
+
   @Override
   public <T> List<ConfigWithMetadata<T>> listConfigsWithMetadata(final AirbyteConfig configType, final Class<T> clazz) throws IOException {
     final List<ConfigWithMetadata<T>> configWithMetadata = new ArrayList<>();
@@ -258,7 +295,7 @@ private List<ConfigWithMetadata<StandardWorkspace>> listStandardWorkspaceWithMet
     for (final Record record : result) {
       final List<Notification> notificationList = new ArrayList<>();
       final List fetchedNotifications = Jsons.deserialize(record.get(WORKSPACE.NOTIFICATIONS).data(), List.class);
-      for (Object notification : fetchedNotifications) {
+      for (final Object notification : fetchedNotifications) {
         notificationList.add(Jsons.convertValue(notification, Notification.class));
       }
       final StandardWorkspace workspace = buildStandardWorkspace(record, notificationList);

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/DeprecatedDatabaseConfigPersistence.java

@@ -132,6 +132,29 @@ public <T> List<T> listConfigs(final AirbyteConfig configType, final Class<T> cl
         .collect(Collectors.toList());
   }
 
+  @Override
+  public <T> ConfigWithMetadata<T> getConfigWithMetadata(final AirbyteConfig configType, final String configId, final Class<T> clazz)
+      throws ConfigNotFoundException, JsonValidationException, IOException {
+    final Result<Record> result = database.query(ctx -> ctx.select(asterisk())
+        .from(AIRBYTE_CONFIGS)
+        .where(AIRBYTE_CONFIGS.CONFIG_TYPE.eq(configType.name()), AIRBYTE_CONFIGS.CONFIG_ID.eq(configId))
+        .fetch());
+
+    if (result.isEmpty()) {
+      throw new ConfigNotFoundException(configType, configId);
+    } else if (result.size() > 1) {
+      throw new IllegalStateException(String.format("Multiple %s configs found for ID %s: %s", configType, configId, result));
+    }
+
+    final Record record = result.get(0);
+    return new ConfigWithMetadata<>(
+        record.get(AIRBYTE_CONFIGS.CONFIG_ID),
+        record.get(AIRBYTE_CONFIGS.CONFIG_TYPE),
+        record.get(AIRBYTE_CONFIGS.CREATED_AT).toInstant(),
+        record.get(AIRBYTE_CONFIGS.UPDATED_AT).toInstant(),
+        Jsons.deserialize(result.get(0).get(AIRBYTE_CONFIGS.CONFIG_BLOB).data(), clazz));
+  }
+
   @Override
   public <T> List<ConfigWithMetadata<T>> listConfigsWithMetadata(final AirbyteConfig configType, final Class<T> clazz) throws IOException {
     final Result<Record> results = database.query(ctx -> ctx.select(asterisk())

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/FileSystemConfigPersistence.java

@@ -79,6 +79,12 @@ public <T> List<T> listConfigs(final AirbyteConfig configType, final Class<T> cl
     }
   }
 
+  @Override
+  public <T> ConfigWithMetadata<T> getConfigWithMetadata(final AirbyteConfig configType, final String configId, final Class<T> clazz)
+      throws ConfigNotFoundException, JsonValidationException, IOException {
+    throw new UnsupportedOperationException("File Persistence doesn't support metadata");
+  }
+
   @Override
   public <T> List<ConfigWithMetadata<T>> listConfigsWithMetadata(final AirbyteConfig configType, final Class<T> clazz)
       throws JsonValidationException, IOException {

airbyte-config/persistence/src/main/java/io/airbyte/config/persistence/ValidatingConfigPersistence.java

@@ -48,6 +48,14 @@ public <T> List<T> listConfigs(final AirbyteConfig configType, final Class<T> cl
     return configs;
   }
 
+  @Override
+  public <T> ConfigWithMetadata<T> getConfigWithMetadata(final AirbyteConfig configType, final String configId, final Class<T> clazz)
+      throws ConfigNotFoundException, JsonValidationException, IOException {
+    final ConfigWithMetadata<T> config = decoratedPersistence.getConfigWithMetadata(configType, configId, clazz);
+    validateJson(config.getConfig(), configType);
+    return config;
+  }
+
   @Override
   public <T> List<ConfigWithMetadata<T>> listConfigsWithMetadata(final AirbyteConfig configType, final Class<T> clazz)
       throws JsonValidationException, IOException {

airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/DatabaseConfigPersistenceTest.java

@@ -86,6 +86,21 @@ public void testWriteAndGetConfig() throws Exception {
             .hasSameElementsAs(List.of(DESTINATION_SNOWFLAKE, DESTINATION_S3));
   }
 
+  @Test
+  public void testGetConfigWithMetadata() throws Exception {
+    final Instant now = Instant.now().minus(Duration.ofSeconds(1));
+    writeDestination(configPersistence, DESTINATION_S3);
+    final ConfigWithMetadata<StandardDestinationDefinition> configWithMetadata = configPersistence.getConfigWithMetadata(
+        STANDARD_DESTINATION_DEFINITION,
+        DESTINATION_S3.getDestinationDefinitionId().toString(),
+        StandardDestinationDefinition.class);
+    assertEquals("STANDARD_DESTINATION_DEFINITION", configWithMetadata.getConfigType());
+    assertTrue(configWithMetadata.getCreatedAt().isAfter(now));
+    assertTrue(configWithMetadata.getUpdatedAt().isAfter(now));
+    assertEquals(DESTINATION_S3.getDestinationDefinitionId().toString(), configWithMetadata.getConfigId());
+    assertEquals(DESTINATION_S3, configWithMetadata.getConfig());
+  }
+
   @Test
   public void testListConfigWithMetadata() throws Exception {
     final Instant now = Instant.now().minus(Duration.ofSeconds(1));

airbyte-config/persistence/src/test/java/io/airbyte/config/persistence/DeprecatedDatabaseConfigPersistenceTest.java

@@ -78,6 +78,21 @@ public void testWriteAndGetConfig() throws Exception {
         configPersistence.listConfigs(STANDARD_DESTINATION_DEFINITION, StandardDestinationDefinition.class));
   }
 
+  @Test
+  public void testGetConfigWithMetadata() throws Exception {
+    final Instant now = Instant.now().minus(Duration.ofSeconds(1));
+    writeDestination(configPersistence, DESTINATION_S3);
+    final ConfigWithMetadata<StandardDestinationDefinition> configWithMetadata = configPersistence.getConfigWithMetadata(
+        STANDARD_DESTINATION_DEFINITION,
+        DESTINATION_S3.getDestinationDefinitionId().toString(),
+        StandardDestinationDefinition.class);
+    assertEquals("STANDARD_DESTINATION_DEFINITION", configWithMetadata.getConfigType());
+    assertTrue(configWithMetadata.getCreatedAt().isAfter(now));
+    assertTrue(configWithMetadata.getUpdatedAt().isAfter(now));
+    assertEquals(DESTINATION_S3.getDestinationDefinitionId().toString(), configWithMetadata.getConfigId());
+    assertEquals(DESTINATION_S3, configWithMetadata.getConfig());
+  }
+
   @Test
   public void testListConfigWithMetadata() throws Exception {
     final Instant now = Instant.now().minus(Duration.ofSeconds(1));