refactor: remove unused SingleStoreSchemaHashWatcher interface

The SingleStoreSchemaHashWatcher interface and all its implementations
were only used for testing and not in production code. This commit
removes:

- SingleStoreSchemaHashWatcher interface and SchemaWatchCallback type
- All datastore schemawatch.go implementation files (postgres, mysql,
  crdb, spanner, memdb)
- SchemaHashWatcherForTesting methods from all datastores and proxies
- UnifiedSchemaHashWatchTest test function

Changes made:
- Inlined readSchemaHash logic into SchemaHashReaderForTesting methods
- Added missing SchemaModeForTesting to observable proxy and validating
  datastore wrapper
- Fixed test files to pass schema hash parameter to SnapshotReader
- Fixed linter issues (perfsprint, testifylint)

All datastore tests pass (postgres, mysql, crdb).

Author: josephschorr

internal/datastore/crdb/crdb.go

@@ -537,26 +537,29 @@ func (cds *crdbDatastore) SchemaHashReaderForTesting() interface {
 	return &crdbSchemaHashReaderForTesting{query: cds.readPool}
 }
 
-// SchemaHashWatcherForTesting returns a test-only interface for watching schema hash changes.
-func (cds *crdbDatastore) SchemaHashWatcherForTesting() datastore.SingleStoreSchemaHashWatcher {
-	// Determine the schema changefeed query format based on the relationship changefeed query
-	schemaChangefeedFormat := querySchemaChangefeed
-	switch cds.beginChangefeedQuery {
-	case queryChangefeedPreV22:
-		schemaChangefeedFormat = querySchemaChangefeedPreV22
-	case queryChangefeedPreV25:
-		schemaChangefeedFormat = querySchemaChangefeedPreV25
-	}
-	return newCRDBSchemaHashWatcher(cds.readPool, schemaChangefeedFormat)
+// SchemaModeForTesting returns the current schema mode for testing purposes.
+func (cds *crdbDatastore) SchemaModeForTesting() (options.SchemaMode, error) {
+	return cds.schemaMode, nil
 }
 
 type crdbSchemaHashReaderForTesting struct {
 	query *pool.RetryPool
 }
 
 func (r *crdbSchemaHashReaderForTesting) ReadSchemaHash(ctx context.Context) (string, error) {
-	watcher := &crdbSchemaHashWatcher{query: r.query}
-	return watcher.readSchemaHash(ctx)
+	var hashBytes []byte
+
+	err := r.query.QueryRowFunc(ctx, func(ctx context.Context, row pgx.Row) error {
+		return row.Scan(&hashBytes)
+	}, "SELECT hash FROM schema_revision WHERE name = 'current'")
+	if err != nil {
+		if errors.Is(err, pgx.ErrNoRows) {
+			return "", datastore.ErrSchemaNotFound
+		}
+		return "", fmt.Errorf("failed to query schema hash: %w", err)
+	}
+
+	return string(hashBytes), nil
 }
 
 func (cds *crdbDatastore) HeadRevision(ctx context.Context) (datastore.Revision, datastore.SchemaHash, error) {

internal/datastore/crdb/schemawatch.go

@@ -64,16 +64,16 @@ func (vds validatingDatastore) SchemaHashReaderForTesting() interface {
 	return nil
 }
 
-// SchemaHashWatcherForTesting delegates to the underlying datastore if it implements the test interface
-func (vds validatingDatastore) SchemaHashWatcherForTesting() datastore.SingleStoreSchemaHashWatcher {
-	type schemaHashWatcherProvider interface {
-		SchemaHashWatcherForTesting() datastore.SingleStoreSchemaHashWatcher
+// SchemaModeForTesting delegates to the underlying datastore if it implements the test interface
+func (vds validatingDatastore) SchemaModeForTesting() (options.SchemaMode, error) {
+	type schemaModeProvider interface {
+		SchemaModeForTesting() (options.SchemaMode, error)
 	}
 
-	if hashWatcher, ok := vds.Datastore.(schemaHashWatcherProvider); ok {
-		return hashWatcher.SchemaHashWatcherForTesting()
+	if provider, ok := vds.Datastore.(schemaModeProvider); ok {
+		return provider.SchemaModeForTesting()
 	}
-	return nil
+	return options.SchemaModeReadLegacyWriteLegacy, errors.New("delegate datastore does not implement SchemaModeForTesting()")
 }
 
 type validatingReader struct {

internal/datastore/dsfortesting/dsfortesting.go

@@ -421,18 +421,38 @@ func (mdb *memdbDatastore) SchemaHashReaderForTesting() interface {
 	return &memdbSchemaHashReaderForTesting{db: mdb}
 }
 
-// SchemaHashWatcherForTesting returns a test-only interface for watching schema hash changes.
-func (mdb *memdbDatastore) SchemaHashWatcherForTesting() datastore.SingleStoreSchemaHashWatcher {
-	return newMemdbSchemaHashWatcher(mdb)
+// SchemaModeForTesting returns the current schema mode for testing purposes.
+// MemDB always operates in a mode equivalent to ReadNewWriteNew.
+func (mdb *memdbDatastore) SchemaModeForTesting() (options.SchemaMode, error) {
+	return options.SchemaModeReadNewWriteNew, nil
 }
 
 type memdbSchemaHashReaderForTesting struct {
 	db *memdbDatastore
 }
 
 func (r *memdbSchemaHashReaderForTesting) ReadSchemaHash(ctx context.Context) (string, error) {
-	watcher := &memdbSchemaHashWatcher{db: r.db}
-	return watcher.readSchemaHash()
+	r.db.RLock()
+	defer r.db.RUnlock()
+
+	tx := r.db.db.Txn(false)
+	defer tx.Abort()
+
+	raw, err := tx.First(tableSchemaRevision, indexID, "current")
+	if err != nil {
+		return "", fmt.Errorf("failed to query schema hash: %w", err)
+	}
+
+	if raw == nil {
+		return "", datastore.ErrSchemaNotFound
+	}
+
+	revisionData, ok := raw.(*schemaRevisionData)
+	if !ok {
+		return "", errors.New("invalid schema revision data type")
+	}
+
+	return string(revisionData.hash), nil
 }
 
 // This code assumes that the RWMutex has been acquired.