source-mysql-batch: Require row_id only when keyless_row_id

estuary · Feb 26, 2025 · 90ebe2d · 90ebe2d
1 parent e88e311
commit 90ebe2d
Show file tree

Hide file tree

Showing 23 changed files with 35 additions and 58 deletions.
diff --git a/source-mysql-batch/.snapshots/TestAsyncCapture-Discovery b/source-mysql-batch/.snapshots/TestAsyncCapture-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestBinaryTypes-Discovery b/source-mysql-batch/.snapshots/TestBinaryTypes-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestCaptureFromView-DiscoveryWithViews b/source-mysql-batch/.snapshots/TestCaptureFromView-DiscoveryWithViews
@@ -52,8 +52,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {
@@ -122,8 +121,7 @@ Binding 1:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           }
         },

diff --git a/source-mysql-batch/.snapshots/TestCaptureFromView-DiscoveryWithoutViews b/source-mysql-batch/.snapshots/TestCaptureFromView-DiscoveryWithoutViews
@@ -52,8 +52,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestCaptureWithEmptyPoll-Discovery b/source-mysql-batch/.snapshots/TestCaptureWithEmptyPoll-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestCaptureWithModifications-Discovery b/source-mysql-batch/.snapshots/TestCaptureWithModifications-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestCaptureWithNullCursor-Discovery b/source-mysql-batch/.snapshots/TestCaptureWithNullCursor-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestCaptureWithTwoColumnCursor-Discovery b/source-mysql-batch/.snapshots/TestCaptureWithTwoColumnCursor-Discovery
@@ -56,8 +56,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestCaptureWithUpdatedAtCursor-Discovery b/source-mysql-batch/.snapshots/TestCaptureWithUpdatedAtCursor-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestDateAndTimeTypes-Discovery b/source-mysql-batch/.snapshots/TestDateAndTimeTypes-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestEnumAndSetTypes-Discovery b/source-mysql-batch/.snapshots/TestEnumAndSetTypes-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestFeatureFlagKeylessRowID-Disabled-Discovery b/source-mysql-batch/.snapshots/TestFeatureFlagKeylessRowID-Disabled-Discovery
@@ -51,8 +51,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           }
         },

diff --git a/source-mysql-batch/.snapshots/TestFullRefresh-Discovery b/source-mysql-batch/.snapshots/TestFullRefresh-Discovery
@@ -52,8 +52,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestIntegerTypes-Discovery b/source-mysql-batch/.snapshots/TestIntegerTypes-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestJSONType-Discovery b/source-mysql-batch/.snapshots/TestJSONType-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestKeylessCapture-Discovery b/source-mysql-batch/.snapshots/TestKeylessCapture-Discovery
@@ -54,8 +54,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           }
         },

diff --git a/source-mysql-batch/.snapshots/TestKeylessFullRefreshCapture-Discovery b/source-mysql-batch/.snapshots/TestKeylessFullRefreshCapture-Discovery
@@ -51,8 +51,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           }
         },

diff --git a/source-mysql-batch/.snapshots/TestNumericTypes-Discovery b/source-mysql-batch/.snapshots/TestNumericTypes-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestQueryTemplateOverride-Discovery b/source-mysql-batch/.snapshots/TestQueryTemplateOverride-Discovery
@@ -54,8 +54,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestSimpleCapture-Discovery b/source-mysql-batch/.snapshots/TestSimpleCapture-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestSpatialTypes-Discovery b/source-mysql-batch/.snapshots/TestSpatialTypes-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/.snapshots/TestStringTypes-Discovery b/source-mysql-batch/.snapshots/TestStringTypes-Discovery
@@ -55,8 +55,7 @@ Binding 0:
             "type": "object",
             "required": [
               "polled",
-              "index",
-              "row_id"
+              "index"
             ]
           },
           "id": {

diff --git a/source-mysql-batch/driver.go b/source-mysql-batch/driver.go
@@ -126,13 +126,16 @@ var (
 	fallbackKeyOld = []string{"/_meta/polled", "/_meta/index"}
 )
 
-func generateCollectionSchema(keyColumns []string, columnTypes map[string]*jsonschema.Schema) (json.RawMessage, error) {
+func generateCollectionSchema(cfg *Config, keyColumns []string, columnTypes map[string]*jsonschema.Schema) (json.RawMessage, error) {
 	// Generate schema for the metadata via reflection
 	var reflector = jsonschema.Reflector{
 		ExpandedStruct: true,
 		DoNotReference: true,
 	}
 	var metadataSchema = reflector.ReflectFromType(reflect.TypeOf(documentMetadata{}))
+	if !cfg.Advanced.parsedFeatureFlags["keyless_row_id"] { // Don't include row_id as required on old captures with keyless_row_id off
+		metadataSchema.Required = slices.DeleteFunc(metadataSchema.Required, func(s string) bool { return s == "row_id" })
+	}
 	metadataSchema.Definitions = nil
 	metadataSchema.AdditionalProperties = nil
 
@@ -168,13 +171,6 @@ func generateCollectionSchema(keyColumns []string, columnTypes map[string]*jsons
 	return json.RawMessage(bs), nil
 }
 
-var minimalSchema = func() json.RawMessage {
-	var schema, err = generateCollectionSchema(nil, nil)
-	if err != nil {
-		panic(err)
-	}
-	return schema
-}()
 
 // Spec returns metadata about the capture connector.
 func (drv *BatchSQLDriver) Spec(ctx context.Context, req *pc.Request_Spec) (*pc.Response_Spec, error) {
@@ -253,15 +249,19 @@ func (drv *BatchSQLDriver) Discover(ctx context.Context, req *pc.Request_Discove
 			return nil, fmt.Errorf("error serializing resource spec: %w", err)
 		}
 
-		// Try to generate a useful collection schema, but on error fall back to the
-		// minimal schema with a fallback collection key which is always present.
-		var collectionSchema = minimalSchema
+		// Start with a minimal schema and a fallback collection key, which will be
+		// replaced with more useful versions if we have sufficient information.
+		collectionSchema, err := generateCollectionSchema(&cfg, nil, nil)
+		if err != nil {
+			return nil, fmt.Errorf("error generating minimal collection schema: %w", err)
+		}
 		var collectionKey = fallbackKey
 		if !cfg.Advanced.parsedFeatureFlags["keyless_row_id"] {
 			collectionKey = fallbackKeyOld
 		}
+
 		if tableKey, ok := keysByTable[tableID]; ok {
-			if generatedSchema, err := generateCollectionSchema(tableKey.Columns, tableKey.ColumnTypes); err == nil {
+			if generatedSchema, err := generateCollectionSchema(&cfg, tableKey.Columns, tableKey.ColumnTypes); err == nil {
 				collectionSchema = generatedSchema
 				collectionKey = nil
 				for _, colName := range tableKey.Columns {