Namespace fields in catalog

airbyte-api/build.gradle

@@ -26,6 +26,7 @@ task generateApiServer(type: GenerateTask) {
             'SourceConfiguration'               : 'com.fasterxml.jackson.databind.JsonNode',
             'DestinationDefinitionSpecification': 'com.fasterxml.jackson.databind.JsonNode',
             'DestinationConfiguration'          : 'com.fasterxml.jackson.databind.JsonNode',
+            'StreamJsonSchema'                  : 'com.fasterxml.jackson.databind.JsonNode',
     ]
 
     generateApiDocumentation = false
@@ -57,6 +58,7 @@ task generateApiClient(type: GenerateTask) {
             'SourceConfiguration'               : 'com.fasterxml.jackson.databind.JsonNode',
             'DestinationDefinitionSpecification': 'com.fasterxml.jackson.databind.JsonNode',
             'DestinationConfiguration'          : 'com.fasterxml.jackson.databind.JsonNode',
+            'StreamJsonSchema'                  : 'com.fasterxml.jackson.databind.JsonNode',
     ]
 
     library = "native"
@@ -89,6 +91,7 @@ task generateApiDocs(type: GenerateTask) {
             'SourceConfiguration'               : 'com.fasterxml.jackson.databind.JsonNode',
             'DestinationDefinitionSpecification': 'com.fasterxml.jackson.databind.JsonNode',
             'DestinationConfiguration'          : 'com.fasterxml.jackson.databind.JsonNode',
+            'StreamJsonSchema'                  : 'com.fasterxml.jackson.databind.JsonNode',
     ]
 
     generateApiDocumentation = false

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

@@ -386,7 +386,7 @@ paths:
     post:
       tags:
         - source
-      summary: Discover the schema of the source
+      summary: Discover the schema catalog of the source
       operationId: discoverSchemaForSource
       requestBody:
         content:
@@ -1366,13 +1366,13 @@ components:
           items:
             $ref: "#/components/schemas/SourceRead"
     SourceDiscoverSchemaRead:
-      description: Returns the results of a discover schema job. If the job was not successful, the schema field will not be present. jobInfo will aways be present and its status be used to determine if the job was successful or not.
+      description: Returns the results of a discover catalog job. If the job was not successful, the catalog field will not be present. jobInfo will aways be present and its status be used to determine if the job was successful or not.
       type: object
       required:
         - jobInfo
       properties:
-        schema:
-          $ref: "#/components/schemas/SourceSchema"
+        catalog:
+          $ref: "#/components/schemas/AirbyteCatalog"
         jobInfo:
           $ref: "#/components/schemas/JobInfoRead"
     # DESTINATION DEFINITION
@@ -1592,8 +1592,8 @@ components:
           $ref: "#/components/schemas/SourceId"
         destinationId:
           $ref: "#/components/schemas/DestinationId"
-        syncSchema:
-          $ref: "#/components/schemas/SourceSchema"
+        syncCatalog:
+          $ref: "#/components/schemas/AirbyteCatalog"
         schedule:
           $ref: "#/components/schemas/ConnectionSchedule"
         status:
@@ -1602,13 +1602,13 @@ components:
       type: object
       required:
         - connectionId
-        - syncSchema
+        - syncCatalog
         - status
       properties:
         connectionId:
           $ref: "#/components/schemas/ConnectionId"
-        syncSchema:
-          $ref: "#/components/schemas/SourceSchema"
+        syncCatalog:
+          $ref: "#/components/schemas/AirbyteCatalog"
         schedule:
           $ref: "#/components/schemas/ConnectionSchedule"
         status:
@@ -1627,7 +1627,7 @@ components:
         - name
         - sourceId
         - destinationId
-        - syncSchema
+        - syncCatalog
         - status
       properties:
         connectionId:
@@ -1638,8 +1638,8 @@ components:
           $ref: "#/components/schemas/SourceId"
         destinationId:
           $ref: "#/components/schemas/DestinationId"
-        syncSchema:
-          $ref: "#/components/schemas/SourceSchema"
+        syncCatalog:
+          $ref: "#/components/schemas/AirbyteCatalog"
         schedule:
           $ref: "#/components/schemas/ConnectionSchedule"
         status:
@@ -1692,65 +1692,78 @@ components:
       properties:
         logType:
           $ref: "#/components/schemas/LogType"
-    # SCHEMA
-    SourceSchema:
-      description: describes the available schema.
+    # SCHEMA CATALOG
+    AirbyteCatalog:
+      description: describes the available schema (catalog).
       type: object
       required:
         - streams
       properties:
         streams:
           type: array
           items:
-            $ref: "#/components/schemas/SourceSchemaStream"
-    SourceSchemaStream:
+            $ref: "#/components/schemas/AirbyteStreamAndConfiguration"
+    AirbyteStreamAndConfiguration:
+      description: each stream is split in two parts; the immutable schema from source and mutable configuration for destination
       type: object
+      additionalProperties: false
+      properties:
+        stream:
+          $ref: "#/components/schemas/AirbyteStream"
+        config:
+          $ref: "#/components/schemas/AirbyteStreamConfiguration"
+    AirbyteStream:
+      description: the immutable schema defined by the source
+      type: object
+      additionalProperties: false
       required:
         - name
-        - fields
-        - supportedSyncModes
+        - json_schema
+        # todo (cgardens) - make required once sources are migrated
+        # - supported_sync_modes
       properties:
-        # immutable
         name:
           type: string
-        cleanedName:
+          description: Stream's name.
+        namespace:
+          description: Optional if source handles namespaces concept such as DB schemas or folders
           type: string
-        fields:
-          type: array
-          items:
-            $ref: "#/components/schemas/SourceSchemaField"
+        jsonSchema:
+          description: Stream schema using Json Schema specs.
+          $ref: "#/components/schemas/StreamJsonSchema"
         supportedSyncModes:
           type: array
           items:
             $ref: "#/components/schemas/SyncMode"
         sourceDefinedCursor:
+          description: If the source defines the cursor field, then it does any other cursor field inputs will be ignored. If it does not either the user_provided one is used or as a backup the default one is used.
           type: boolean
         defaultCursorField:
+          description: Path to the field that will be used to determine if a record is new or modified since the last sync. If not provided by the source, the end user will have to specify the comparable themselves.
           type: array
           items:
             type: string
-        # configurable
-        selected:
-          type: boolean
+    StreamJsonSchema:
+      type: object
+    AirbyteStreamConfiguration:
+      description: the mutable part of the stream to configure the destination
+      type: object
+      additionalProperties: false
+      properties:
         syncMode:
           $ref: "#/components/schemas/SyncMode"
+          default: full_refresh
         cursorField:
+          description: Path to the field that will be used to determine if a record is new or modified since the last sync. This field is REQUIRED if `sync_mode` is `incremental`. Otherwise it is ignored.
           type: array
           items:
             type: string
-    SourceSchemaField:
-      type: object
-      required:
-        - name
-        - dataType
-        - selected
-      properties:
-        name:
+        aliasName:
+          description: Alias name to the stream to be used in the destination
           type: string
-        cleanedName:
+        targetNamespace:
+          description: Namespace to store the stream in the destination
           type: string
-        dataType:
-          $ref: "#/components/schemas/DataType"
         selected:
           type: boolean
     DataType:
@@ -1942,7 +1955,7 @@ components:
         - name
         - sourceId
         - destinationId
-        - syncSchema
+        - syncCatalog
         - status
         - source
         - destination
@@ -1956,8 +1969,8 @@ components:
           $ref: "#/components/schemas/SourceId"
         destinationId:
           $ref: "#/components/schemas/DestinationId"
-        syncSchema:
-          $ref: "#/components/schemas/SourceSchema"
+        syncCatalog:
+          $ref: "#/components/schemas/AirbyteCatalog"
         schedule:
           $ref: "#/components/schemas/ConnectionSchedule"
         status:

airbyte-commons/src/main/java/io/airbyte/commons/enums/Enums.java

@@ -28,6 +28,7 @@
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import java.util.Arrays;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.Set;
@@ -43,6 +44,13 @@ public static <T1 extends Enum<T1>, T2 extends Enum<T2>> T2 convertTo(T1 ie, Cla
     return Enum.valueOf(oe, ie.name());
   }
 
+  public static <T1 extends Enum<T1>, T2 extends Enum<T2>> List<T2> convertListTo(List<T1> ies, Class<T2> oe) {
+    return ies
+        .stream()
+        .map(ie -> convertTo(ie, oe))
+        .collect(Collectors.toList());
+  }
+
   public static <T1 extends Enum<T1>, T2 extends Enum<T2>> boolean isCompatible(Class<T1> c1,
                                                                                 Class<T2> c2) {
     Preconditions.checkArgument(c1.isEnum());

airbyte-config/models/build.gradle

@@ -6,6 +6,7 @@ plugins {
 
 dependencies {
     implementation project(':airbyte-protocol:models')
+    implementation 'org.apache.commons:commons-lang3:3.11'
 }
 
 jsonSchema2Pojo {