feat: add file transfer metadata (#14307)

airbyte-api/server-api/src/main/openapi/config.yaml

@@ -9620,6 +9620,7 @@ components:
         - supportsRefreshes
         - supportState
         - isVersionOverrideApplied
+        - supportsFileTransfer
       properties:
         dockerRepository:
           type: string
@@ -9641,13 +9642,16 @@ components:
         cdkVersion:
           description: "The version of the CDK that the connector was built with. e.g. python:0.1.0, java:0.1.0"
           type: string
+        supportsFileTransfer:
+          type: boolean
     ResolveActorDefinitionVersionResponse:
       type: object
       required:
         - versionId
         - dockerRepository
         - dockerImageTag
         - supportRefreshes
+        - supportFileTransfer
       properties:
         versionId:
           $ref: "#/components/schemas/ActorDefinitionVersionId"
@@ -9657,6 +9661,8 @@ components:
           type: string
         supportRefreshes:
           type: boolean
+        supportFileTransfers:
+          type: boolean
     ActorDefinitionBreakingChange:
       type: object
       required:

airbyte-bootloader/src/test-integration/java/io/airbyte/bootloader/BootloaderTest.java

@@ -97,7 +97,7 @@ class BootloaderTest {
 
   // ⚠️ This line should change with every new migration to show that you meant to make a new
   // migration to the prod database
-  private static final String CURRENT_CONFIGS_MIGRATION_VERSION = "0.64.4.002";
+  private static final String CURRENT_CONFIGS_MIGRATION_VERSION = "1.1.0.000";
   private static final String CURRENT_JOBS_MIGRATION_VERSION = "0.64.7.002";
 
   @BeforeEach

airbyte-commons-server/src/main/java/io/airbyte/commons/server/handlers/ActorDefinitionVersionHandler.java

@@ -142,7 +142,8 @@ ActorDefinitionVersionRead createActorDefinitionVersionRead(final ActorDefinitio
         .supportLevel(toApiSupportLevel(actorDefinitionVersion.getSupportLevel()))
         .cdkVersion(actorDefinitionVersion.getCdkVersion())
         .lastPublished(ApiPojoConverters.toOffsetDateTime(actorDefinitionVersion.getLastPublished()))
-        .isVersionOverrideApplied(versionWithOverrideStatus.isOverrideApplied());
+        .isVersionOverrideApplied(versionWithOverrideStatus.isOverrideApplied())
+        .supportsFileTransfer(actorDefinitionVersion.getSupportsFileTransfer());
 
     final Optional<ActorDefinitionVersionBreakingChanges> breakingChanges =
         actorDefinitionHandlerHelper.getVersionBreakingChanges(actorDefinitionVersion);

airbyte-commons-server/src/test/java/io/airbyte/commons/server/handlers/ActorDefinitionVersionHandlerTest.java

@@ -124,7 +124,8 @@ void testGetActorDefinitionVersionForSource(final boolean isVersionOverrideAppli
         .supportState(io.airbyte.api.model.generated.SupportState.SUPPORTED)
         .dockerRepository(actorDefinitionVersion.getDockerRepository())
         .dockerImageTag(actorDefinitionVersion.getDockerImageTag())
-        .supportsRefreshes(false);
+        .supportsRefreshes(false)
+        .supportsFileTransfer(false);
 
     assertEquals(expectedRead, actorDefinitionVersionRead);
     verify(mSourceService).getSourceConnection(sourceId);
@@ -163,7 +164,8 @@ void testGetActorDefinitionVersionForDestination(final boolean isVersionOverride
         .supportState(io.airbyte.api.model.generated.SupportState.SUPPORTED)
         .dockerRepository(actorDefinitionVersion.getDockerRepository())
         .dockerImageTag(actorDefinitionVersion.getDockerImageTag())
-        .supportsRefreshes(false);
+        .supportsRefreshes(false)
+        .supportsFileTransfer(false);
 
     assertEquals(expectedRead, actorDefinitionVersionRead);
     verify(mDestinationService).getDestinationConnection(destinationId);
@@ -201,7 +203,8 @@ void testGetActorDefinitionVersionForDestinationWithNormalization(final boolean
         .supportState(io.airbyte.api.model.generated.SupportState.SUPPORTED)
         .dockerRepository(actorDefinitionVersion.getDockerRepository())
         .dockerImageTag(actorDefinitionVersion.getDockerImageTag())
-        .supportsRefreshes(actorDefinitionVersion.getSupportsRefreshes());
+        .supportsRefreshes(actorDefinitionVersion.getSupportsRefreshes())
+        .supportsFileTransfer(false);
 
     assertEquals(expectedRead, actorDefinitionVersionRead);
     verify(mDestinationService).getDestinationConnection(destinationId);
@@ -233,7 +236,8 @@ void testCreateActorDefinitionVersionReadWithBreakingChange() throws IOException
         .dockerRepository(actorDefinitionVersion.getDockerRepository())
         .dockerImageTag(actorDefinitionVersion.getDockerImageTag())
         .supportsRefreshes(false)
-        .breakingChanges(breakingChanges);
+        .breakingChanges(breakingChanges)
+        .supportsFileTransfer(false);
 
     assertEquals(expectedRead, actorDefinitionVersionRead);
     verify(mActorDefinitionHandlerHelper).getVersionBreakingChanges(actorDefinitionVersion);

airbyte-commons-worker/src/test/java/io/airbyte/workers/ReplicationInputHydratorTest.java

@@ -435,7 +435,8 @@ private void mockRefresh() throws IOException {
         UUID.randomUUID(),
         "dockerRepo",
         "dockerTag",
-        true));
+        true,
+        false));
   }
 
   private void mockNonRefresh() throws IOException {
@@ -447,6 +448,7 @@ private void mockNonRefresh() throws IOException {
         UUID.randomUUID(),
         "dockerRepo",
         "dockerTag",
+        false,
         false));
   }
 

airbyte-commons-worker/src/test/java/io/airbyte/workers/general/ReplicationWorkerHelperTest.java

@@ -371,7 +371,8 @@ private void mockSupportRefreshes(final boolean supportsRefreshes) throws IOExce
             UUID.randomUUID(),
             "dockerRepository",
             "dockerImageTag",
-            supportsRefreshes));
+            supportsRefreshes,
+            false));
   }
 
   private ConfiguredAirbyteCatalog buildConfiguredAirbyteCatalog() {

airbyte-commons-worker/src/test/java/io/airbyte/workers/general/ReplicationWorkerTest.java

@@ -283,6 +283,7 @@ void setup() throws Exception {
     var resolveActorDefinitionVersionResponse = new ResolveActorDefinitionVersionResponse(UUID.randomUUID(),
         "dockerRepository",
         "dockerImageTag",
+        false,
         false);
     when(actorDefinitionVersionApi.resolveActorDefinitionVersionByTag(any())).thenReturn(resolveActorDefinitionVersionResponse);
     when(airbyteApiClient.getActorDefinitionVersionApi()).thenReturn(actorDefinitionVersionApi);

airbyte-config/config-models/src/main/java/io/airbyte/config/helpers/ConnectorRegistryConverters.java

@@ -170,7 +170,8 @@ public static ActorDefinitionVersion toActorDefinitionVersion(@Nullable final Co
         .withLastPublished(lastModified)
         .withCdkVersion(cdkVersion)
         .withSupportsRefreshes(def.getSupportsRefreshes() != null && def.getSupportsRefreshes())
-        .withLanguage(def.getLanguage());
+        .withLanguage(def.getLanguage())
+        .withSupportsFileTransfer(def.getSupportsFileTransfer());
   }
 
   /**

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

@@ -63,3 +63,7 @@ properties:
   language:
     description: The language that the connector is written in.
     type: string
+  supportsFileTransfer:
+    description: an optional flag indicating whether a destination connector supports file transfer.
+    type: boolean
+    default: false

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

@@ -77,3 +77,7 @@ properties:
   language:
     description: The language that the connector is written in.
     type: string
+  supportsFileTransfer:
+    type: boolean
+    description: flag indicating whether the file transfer mode is available for this destination.
+    default: false

airbyte-config/config-models/src/test/java/io/airbyte/config/helpers/ConnectorRegistryConvertersTest.java

@@ -191,7 +191,8 @@ void testConvertRegistryDestinationToInternalTypes() {
         .withAllowedHosts(ALLOWED_HOSTS)
         .withResourceRequirements(RESOURCE_REQUIREMENTS)
         .withReleases(new ConnectorReleasesDestination().withBreakingChanges(destinationBreakingChanges))
-        .withLanguage(LANGUAGE);
+        .withLanguage(LANGUAGE)
+        .withSupportsFileTransfer(true);
 
     final StandardDestinationDefinition stdDestinationDef = new StandardDestinationDefinition()
         .withDestinationDefinitionId(DEF_ID)
@@ -213,7 +214,8 @@ void testConvertRegistryDestinationToInternalTypes() {
         .withReleaseDate(RELEASE_DATE)
         .withProtocolVersion(PROTOCOL_VERSION)
         .withAllowedHosts(ALLOWED_HOSTS)
-        .withLanguage(LANGUAGE);
+        .withLanguage(LANGUAGE)
+        .withSupportsFileTransfer(true);
 
     assertEquals(stdDestinationDef, ConnectorRegistryConverters.toStandardDestinationDefinition(registryDestinationDef));
     assertEquals(actorDefinitionVersion, ConnectorRegistryConverters.toActorDefinitionVersion(registryDestinationDef));

airbyte-connector-rollout-worker/src/test/kotlin/io/airbyte/connector/rollout/worker/activities/VerifyDefaultVersionActivityImplTest.kt

@@ -44,6 +44,7 @@ class VerifyDefaultVersionActivityImplTest {
         isVersionOverrideApplied = true,
         supportState = SupportState.SUPPORTED,
         supportsRefreshes = true,
+        supportsFileTransfer = false,
       )
 
     // Test without "-rc" suffix in the input dockerImageTag
@@ -85,6 +86,7 @@ class VerifyDefaultVersionActivityImplTest {
         isVersionOverrideApplied = true,
         supportState = SupportState.SUPPORTED,
         supportsRefreshes = true,
+        supportsFileTransfer = false,
       )
 
     val input =

airbyte-data/src/main/java/io/airbyte/data/services/impls/jooq/ConnectorMetadataJooqHelper.java

@@ -87,6 +87,7 @@ public static ActorDefinitionVersion writeActorDefinitionVersion(final ActorDefi
           .set(ACTOR_DEFINITION_VERSION.CDK_VERSION, actorDefinitionVersion.getCdkVersion())
           .set(ACTOR_DEFINITION_VERSION.INTERNAL_SUPPORT_LEVEL, actorDefinitionVersion.getInternalSupportLevel())
           .set(ACTOR_DEFINITION_VERSION.LANGUAGE, actorDefinitionVersion.getLanguage())
+          .set(ACTOR_DEFINITION_VERSION.SUPPORTS_FILE_TRANSFER, actorDefinitionVersion.getSupportsFileTransfer())
           .where(ACTOR_DEFINITION_VERSION.ID.eq(versionId))
           .execute();
     } else {
@@ -127,6 +128,7 @@ public static ActorDefinitionVersion writeActorDefinitionVersion(final ActorDefi
                   .orElseThrow())
           .set(ACTOR_DEFINITION_VERSION.INTERNAL_SUPPORT_LEVEL, actorDefinitionVersion.getInternalSupportLevel())
           .set(ACTOR_DEFINITION_VERSION.LANGUAGE, actorDefinitionVersion.getLanguage())
+          .set(ACTOR_DEFINITION_VERSION.SUPPORTS_FILE_TRANSFER, actorDefinitionVersion.getSupportsFileTransfer())
           .execute();
     }
 

airbyte-data/src/main/java/io/airbyte/data/services/impls/jooq/DbConverter.java

@@ -524,7 +524,8 @@ public static ActorDefinitionVersion buildActorDefinitionVersion(final Record re
         .withSupportsRefreshes(record.get(ACTOR_DEFINITION_VERSION.SUPPORTS_REFRESHES))
         .withSupportState(Enums.toEnum(record.get(ACTOR_DEFINITION_VERSION.SUPPORT_STATE, String.class), SupportState.class).orElseThrow())
         .withInternalSupportLevel(record.get(ACTOR_DEFINITION_VERSION.INTERNAL_SUPPORT_LEVEL, Long.class))
-        .withLanguage(record.get(ACTOR_DEFINITION_VERSION.LANGUAGE));
+        .withLanguage(record.get(ACTOR_DEFINITION_VERSION.LANGUAGE))
+        .withSupportsFileTransfer(record.get(ACTOR_DEFINITION_VERSION.SUPPORTS_FILE_TRANSFER));
   }
 
   public static SecretPersistenceCoordinate buildSecretPersistenceCoordinate(final Record record) {

airbyte-db/db-lib/src/main/java/io/airbyte/db/instance/configs/migrations/V1_1_0_000__AddSupportFileTransferToActorDefVersion.java

@@ -0,0 +1,34 @@
+/*
+ * Copyright (c) 2020-2024 Airbyte, Inc., all rights reserved.
+ */
+
+package io.airbyte.db.instance.configs.migrations;
+
+import org.flywaydb.core.api.migration.BaseJavaMigration;
+import org.flywaydb.core.api.migration.Context;
+import org.jooq.DSLContext;
+import org.jooq.impl.DSL;
+import org.jooq.impl.SQLDataType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+// TODO: update migration description in the class name
+public class V1_1_0_000__AddSupportFileTransferToActorDefVersion extends BaseJavaMigration {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(V1_1_0_000__AddSupportFileTransferToActorDefVersion.class);
+
+  @Override
+  public void migrate(final Context context) throws Exception {
+    LOGGER.info("Running migration: {}", this.getClass().getSimpleName());
+
+    // Warning: please do not use any jOOQ generated code to write a migration.
+    // As database schema changes, the generated jOOQ code can be deprecated. So
+    // old migration may not compile if there is any generated code.
+    final DSLContext ctx = DSL.using(context.getConnection());
+
+    ctx.alterTable("actor_definition_version")
+        .addColumn("supports_file_transfer", SQLDataType.BOOLEAN.nullable(false).defaultValue(false))
+        .execute();
+  }
+
+}

airbyte-db/db-lib/src/main/resources/configs_database/schema_dump.txt

@@ -101,6 +101,7 @@ create table "public"."actor_definition_version" (
   "last_published" timestamp(6) with time zone,
   "internal_support_level" bigint not null,
   "language" varchar(256),
+  "supports_file_transfer" boolean not null default false,
   constraint "actor_definition_version_pkey" primary key ("id"),
   constraint "actor_definition_version_actor_definition_id_version_key" unique ("actor_definition_id", "docker_image_tag")
 );

airbyte-webapp/src/test-utils/mock-data/mockDestination.ts

@@ -24,6 +24,7 @@ export const mockDestinationDefinitionVersion: ActorDefinitionVersionRead = {
   isVersionOverrideApplied: false,
   supportState: SupportState.supported,
   supportLevel: "certified",
+  supportsFileTransfer: false,
 };
 
 export const mockDestinationDefinitionSpecification: DestinationDefinitionSpecificationRead = {

airbyte-webapp/src/test-utils/mock-data/mockSource.ts

@@ -27,6 +27,7 @@ export const mockSourceDefinitionVersion: ActorDefinitionVersionRead = {
   isVersionOverrideApplied: false,
   supportState: SupportState.supported,
   supportLevel: "certified",
+  supportsFileTransfer: false,
 };
 
 export const mockSourceDefinitionSpecification: SourceDefinitionSpecificationRead = {