source-postgres: Feature flag use_schema_inference

source-postgres/.snapshots/TestGeneric-SpecResponse

@@ -104,6 +104,11 @@
             "type": "boolean",
             "title": "Read-Only Capture",
             "description": "When set the capture will operate in read-only mode and avoid operations such as watermark writes. This comes with some tradeoffs; consult the connector documentation for more information."
+          },
+          "feature_flags": {
+            "type": "string",
+            "title": "Feature Flags",
+            "description": "This property is intended for Estuary internal use. You should only modify this field as directed by Estuary support."
           }
         },
         "additionalProperties": false,

source-postgres/discovery.go

@@ -50,6 +50,9 @@ func (db *postgresDatabase) DiscoverTables(ctx context.Context) (map[sqlcapture.
 			// We want to exclude the watermarks table from the output bindings, but we still discover it
 			table.OmitBinding = true
 		}
+		if db.featureFlags["use_schema_inference"] {
+			table.UseSchemaInference = true
+		}
 		tableMap[streamID] = table
 	}
 	for _, column := range columns {

source-postgres/main.go

@@ -76,7 +76,15 @@ func connectPostgres(ctx context.Context, name string, cfg json.RawMessage) (sql
 		}
 	}
 
-	var db = &postgresDatabase{config: &config}
+	var featureFlags = boilerplate.ParseFeatureFlags(config.Advanced.FeatureFlags, featureFlagDefaults)
+	if config.Advanced.FeatureFlags != "" {
+		logrus.WithField("flags", featureFlags).Info("parsed feature flags")
+	}
+
+	var db = &postgresDatabase{
+		config:       &config,
+		featureFlags: featureFlags,
+	}
 	if err := db.connect(ctx); err != nil {
 		return nil, err
 	}
@@ -107,6 +115,13 @@ type advancedConfig struct {
 	MinimumBackfillXID    string   `json:"min_backfill_xid,omitempty" jsonschema:"title=Minimum Backfill XID,description=Only backfill rows with XMIN values greater (in a 32-bit modular comparison) than the specified XID. Helpful for reducing re-backfill data volume in certain edge cases." jsonschema_extras:"pattern=^[0-9]+$"`
 	MaximumBackfillXID    string   `json:"max_backfill_xid,omitempty" jsonschema:"title=Maximum Backfill XID,description=Only backfill rows with XMIN values smaller (in a 32-bit modular comparison) than the specified XID. Helpful for reducing re-backfill data volume in certain edge cases." jsonschema_extras:"pattern=^[0-9]+$"`
 	ReadOnlyCapture       bool     `json:"read_only_capture,omitempty" jsonschema:"title=Read-Only Capture,description=When set the capture will operate in read-only mode and avoid operations such as watermark writes. This comes with some tradeoffs; consult the connector documentation for more information."`
+	FeatureFlags          string   `json:"feature_flags,omitempty" jsonschema:"title=Feature Flags,description=This property is intended for Estuary internal use. You should only modify this field as directed by Estuary support."`
+}
+
+var featureFlagDefaults = map[string]bool{
+	// When set, discovered collection schemas will request that schema inference be
+	// used _in addition to_ the full column/types discovery we already do.
+	"use_schema_inference": false,
 }
 
 // Validate checks that the configuration possesses all required properties.
@@ -226,6 +241,8 @@ type postgresDatabase struct {
 	explained       map[sqlcapture.StreamID]struct{} // Tracks tables which have had an `EXPLAIN` run on them during this connector invocation
 	includeTxIDs    map[sqlcapture.StreamID]bool     // Tracks which tables should have XID properties in their replication metadata
 	tablesPublished map[sqlcapture.StreamID]bool     // Tracks which tables are part of the configured publication
+
+	featureFlags map[string]bool // Parsed feature flag settings with defaults applied
 }
 
 func (db *postgresDatabase) HistoryMode() bool {

source-postgres/main_test.go

@@ -30,6 +30,8 @@ var (
 	dbCapturePass    = flag.String("db_capture_pass", "secret1234", "The password for the capture user")
 
 	readOnlyCapture = flag.Bool("read_only_capture", false, "When true, run test captures in read-only mode")
+
+	testFeatureFlags = flag.String("feature_flags", "", "Feature flags to apply to all test captures.")
 )
 
 const testSchemaName = "test"
@@ -75,6 +77,7 @@ func postgresTestBackend(t testing.TB) *testBackend {
 		Password: *dbCapturePass,
 		Database: *dbName,
 	}
+	captureConfig.Advanced.FeatureFlags = *testFeatureFlags
 	captureConfig.Advanced.BackfillChunkSize = 16
 	if *readOnlyCapture {
 		captureConfig.Advanced.ReadOnlyCapture = true

sqlcapture/discovery.go

@@ -192,6 +192,9 @@ func DiscoverCatalog(ctx context.Context, db Database) ([]*pc.Response_Discovere
 				{Ref: "#" + anchor},
 			},
 		}
+		if table.UseSchemaInference {
+			schema.Extras = map[string]any{"x-infer-schema": true}
+		}
 
 		var rawSchema, err = schema.MarshalJSON()
 		if err != nil {

sqlcapture/interface.go

@@ -214,6 +214,8 @@ type DiscoveryInfo struct {
 	BaseTable   bool                  // True if the table type is 'BASE TABLE' and false for views or other not-physical-table entities.
 	OmitBinding bool                  // True if the table should be omitted from discovery catalog generation.
 
+	UseSchemaInference bool // True if generated JSON schemas for this table should request schema inference.
+
 	// UnpredictableKeyOrdering will be true when the connector is unable to guarantee
 	// (for a particular table) that serialized RowKey values will accurately reproduce
 	// (when compared bytewise lexicographically) the database sort ordering of the same