feat: wip

Author: oxyno-zeta

internal/controller/postgresql/postgres/database.go

@@ -28,6 +28,7 @@ const (
 	GetTablesFromSchemaSQLTemplate = `SELECT tablename,tableowner FROM pg_tables WHERE schemaname = '%s'`
 	ChangeTableOwnerSQLTemplate    = `ALTER TABLE IF EXISTS "%s" OWNER TO "%s"`
 	ChangeTypeOwnerSQLTemplate     = `ALTER TYPE "%s"."%s" OWNER TO "%s"`
+	GetColumnsFromTableSQLTemplate = `SELECT column_name FROM information_schema.columns WHERE table_schema = '%s' AND table_name = '%s'`
 	// Got and edited from : https://stackoverflow.com/questions/3660787/how-to-list-custom-types-using-postgres-information-schema
 	GetTypesFromSchemaSQLTemplate = `SELECT      t.typname as type, pg_catalog.pg_get_userbyid(t.typowner) as owner
 FROM        pg_type t
@@ -38,6 +39,43 @@ AND     n.nspname = '%s';`
 	DuplicateDatabaseErrorCode = "42P04"
 )
 
+func (c *pg) GetColumnNamesFromTable(ctx context.Context, database string, schemaName string, tableName string) ([]string, error) {
+	err := c.connect(database)
+	if err != nil {
+		return nil, err
+	}
+
+	rows, err := c.db.QueryContext(ctx, fmt.Sprintf(GetColumnsFromTableSQLTemplate, schemaName, tableName))
+	if err != nil {
+		return nil, err
+	}
+
+	defer rows.Close()
+
+	res := []string{}
+
+	for rows.Next() {
+		it := ""
+		// Scan
+		err = rows.Scan(&it)
+		// Check error
+		if err != nil {
+			return nil, err
+		}
+		// Save
+		res = append(res, it)
+	}
+
+	// Rows error
+	err = rows.Err()
+	// Check error
+	if err != nil {
+		return nil, err
+	}
+
+	return res, nil
+}
+
 func (c *pg) GetDatabaseOwner(ctx context.Context, dbname string) (string, error) {
 	err := c.connect(c.defaultDatabase)
 	if err != nil {

internal/controller/postgresql/postgres/postgres.go

@@ -68,9 +68,11 @@ type PG interface { //nolint:interfacebloat // This is needed
 	CreatePublication(ctx context.Context, dbname string, builder *CreatePublicationBuilder) error
 	UpdatePublication(ctx context.Context, dbname, publicationName string, builder *UpdatePublicationBuilder) error
 	ChangePublicationOwner(ctx context.Context, dbname string, publicationName string, owner string) error
+	GetPublicationTablesDetails(ctx context.Context, db, publicationName string) ([]*PublicationTableDetail, error)
 	DropReplicationSlot(ctx context.Context, name string) error
 	CreateReplicationSlot(ctx context.Context, dbname, name, plugin string) error
 	GetReplicationSlot(ctx context.Context, name string) (*ReplicationSlotResult, error)
+	GetColumnNamesFromTable(ctx context.Context, database string, schemaName string, tableName string) ([]string, error)
 	GetUser() string
 	GetHost() string
 	GetPort() int

internal/controller/postgresql/postgres/publication.go

@@ -18,6 +18,7 @@ const (
   pg_catalog.pg_get_userbyid(pubowner), puballtables, pubinsert, pubupdate, pubdelete, pubtruncate, pubviaroot
 FROM pg_catalog.pg_publication
 WHERE pubname = '%s';`
+	GetPublicationTablesSQLTemplate  = `SELECT schemaname, tablename, attnames, rowfilter FROM pg_publication_tables WHERE pubname = '%s'`
 	GetReplicationSlotSQLTemplate    = `SELECT slot_name,plugin,database FROM pg_replication_slots WHERE slot_name = '%s'`
 	CreateReplicationSlotSQLTemplate = `SELECT pg_create_logical_replication_slot('%s', '%s')`
 	DropReplicationSlotSQLTemplate   = `SELECT pg_drop_replication_slot('%s')`
@@ -46,6 +47,48 @@ type ReplicationSlotResult struct {
 	Database string
 }
 
+func (c *pg) GetPublicationTablesDetails(ctx context.Context, db, publicationName string) ([]*PublicationTableDetail, error) {
+	err := c.connect(db)
+	if err != nil {
+		return nil, err
+	}
+
+	rows, err := c.db.QueryContext(ctx, fmt.Sprintf(GetPublicationTablesSQLTemplate, publicationName))
+	if err != nil {
+		return nil, err
+	}
+
+	defer rows.Close()
+
+	res := []*PublicationTableDetail{}
+
+	for rows.Next() {
+		var it PublicationTableDetail
+		var pqSA pq.StringArray
+		// Scan
+		err = rows.Scan(&it.SchemaName, &it.TableName, &pqSA, &it.AdditionalWhere)
+		// Check error
+		if err != nil {
+			return nil, err
+		}
+		// Save
+		// ? Note: getting a list of string from pg imply a decode
+		// ? See issue: https://github.com/cockroachdb/cockroach/issues/39770#issuecomment-576170805
+		it.Columns = pqSA
+		// Save
+		res = append(res, &it)
+	}
+
+	// Rows error
+	err = rows.Err()
+	// Check error
+	if err != nil {
+		return nil, err
+	}
+
+	return res, nil
+}
+
 func (c *pg) DropReplicationSlot(ctx context.Context, name string) error {
 	err := c.connect(c.defaultDatabase)
 	if err != nil {

internal/controller/postgresql/postgresqlpublication_controller.go

@@ -18,7 +18,9 @@ package postgresql
 
 import (
 	"context"
+	"fmt"
 	"reflect"
+	"strings"
 	"time"
 
 	"k8s.io/apimachinery/pkg/api/errors"
@@ -39,6 +41,9 @@ import (
 )
 
 const DefaultReplicationSlotPlugin = "pgoutput"
+const DefaultPGWithPublishParameter = "insert, update, delete, truncate"
+
+var DefaultPGWithPublishViaRootParameter = false
 
 // PostgresqlPublicationReconciler reconciles a PostgresqlPublication object.
 type PostgresqlPublicationReconciler struct {
@@ -257,6 +262,20 @@ func (r *PostgresqlPublicationReconciler) mainReconcile(
 			if err != nil {
 				return r.manageError(ctx, reqLogger, instance, originalPatch, err)
 			}
+		} else {
+			// Check if reconcile from PG state is necessary because spec haven't been changed
+			need, err := r.isReconcileOnPGNecessary(ctx, instance, pg, pgDB, pubRes, nameToSearch)
+			// Check error
+			if err != nil {
+				return r.manageError(ctx, reqLogger, instance, originalPatch, err)
+			}
+
+			// Check if it is needed
+			if need {
+				reqLogger.Info("PG state have been changed but not via operator, update need to be done")
+
+				err = r.manageUpdate(ctx, instance, pg, pgDB, pubRes, nameToSearch)
+			}
 		}
 
 		// Check if owner are aligned
@@ -316,6 +335,138 @@ func (r *PostgresqlPublicationReconciler) mainReconcile(
 	return r.manageSuccess(ctx, reqLogger, instance, originalPatch)
 }
 
+func (*PostgresqlPublicationReconciler) isReconcileOnPGNecessary(
+	ctx context.Context,
+	instance *v1alpha1.PostgresqlPublication,
+	pg postgres.PG,
+	pgDB *v1alpha1.PostgresqlDatabase,
+	pubRes *postgres.PublicationResult,
+	currentPublicationName string,
+) (bool, error) {
+	instanceSpec := instance.Spec
+
+	// Check with parameters
+	// nil spec case
+	if instanceSpec.WithParameters == nil && (pubRes.PublicationViaRoot || !pubRes.Delete || !pubRes.Insert || !pubRes.Truncate || !pubRes.Update) {
+		return true, nil
+	}
+	// Non nil spec case
+	if instanceSpec.WithParameters != nil {
+		// publication via root check
+		if (instanceSpec.WithParameters.PublishViaPartitionRoot == nil && pubRes.PublicationViaRoot) ||
+			(instanceSpec.WithParameters.PublishViaPartitionRoot != nil && *instanceSpec.WithParameters.PublishViaPartitionRoot != pubRes.PublicationViaRoot) {
+			return true, nil
+		}
+
+		// Now check publish parameters
+		// Save
+		publish := strings.ToLower(instanceSpec.WithParameters.Publish)
+		// Empty spec case
+		if publish == "" && (!pubRes.Delete || !pubRes.Insert || !pubRes.Truncate || !pubRes.Update) {
+			return true, nil
+		}
+		// Not empty case
+		if publish != "" &&
+			(strings.Contains(publish, "insert") != pubRes.Insert ||
+				strings.Contains(publish, "update") != pubRes.Update ||
+				strings.Contains(publish, "delete") != pubRes.Delete ||
+				strings.Contains(publish, "truncate") != pubRes.Truncate) {
+			return true, nil
+		}
+	}
+
+	// Check if it is a all tables
+	if instanceSpec.AllTables {
+		// This state cannot be updated so.. Ignoring it
+		return false, nil
+	}
+
+	// Get publication details
+	details, err := pg.GetPublicationTablesDetails(ctx, pgDB.Status.Database, currentPublicationName)
+	if err != nil {
+		return false, err
+	}
+
+	// Check if we are in the all tables in schema case
+	if len(instanceSpec.TablesInSchema) != 0 {
+		// Compute list of schema coming from publication tables and compare list length.
+		// The computed list must be <= with the desired list
+		// Why <= ? Because we can list a schema without any tables in
+		// So we need to check > to go out quickly
+		// After that, we need to check that computed list is included in the desired list
+		// Compute list of schema
+		computedSchemaList := lo.Uniq(lo.Map(details, func(it *postgres.PublicationTableDetail, _ int) string { return it.SchemaName }))
+
+		// Check length
+		if len(computedSchemaList) > len(instanceSpec.TablesInSchema) {
+			return true, nil
+		}
+
+		// Check include/subset
+		if !lo.Every(instanceSpec.TablesInSchema, computedSchemaList) {
+			return true, nil
+		}
+	} else {
+		// Need to check with tables
+		// Loop over spec table list
+		for _, st := range instanceSpec.Tables {
+			// Check if table isn't in the current list
+			detail, found := lo.Find(details, func(it *postgres.PublicationTableDetail) bool {
+				return st.TableName == it.TableName || st.TableName == fmt.Sprintf("%s.%s", it.SchemaName, it.TableName)
+			})
+			if !found {
+				return true, nil
+			}
+
+			// Check if additional where  aren't identical (nil case)
+			if st.AdditionalWhere != detail.AdditionalWhere {
+				return true, nil
+			}
+			// Check if additional where  aren't identical (phase 2)
+			if st.AdditionalWhere != nil && detail.AdditionalWhere != nil && fmt.Sprintf("(%s)", *st.AdditionalWhere) != *detail.AdditionalWhere {
+				return true, nil
+			}
+
+			// Now need to check columns
+
+			columnNamesToCheck := []string{}
+
+			// Check if columns aren't set in spec
+			if st.Columns == nil {
+				// If so, get real columns from table and check if list aren't identical
+				// Split spec table name
+				spl := strings.Split(st.TableName, ".")
+				var schemaName, tableName string
+
+				// Check split size
+				if len(spl) == 1 {
+					schemaName = defaultPGPublicSchemaName
+					tableName = spl[0]
+				} else {
+					schemaName = spl[0]
+					tableName = spl[1]
+				}
+
+				columnNamesToCheck, err = pg.GetColumnNamesFromTable(ctx, pgDB.Status.Database, schemaName, tableName)
+				if err != nil {
+					return false, err
+				}
+			} else {
+				columnNamesToCheck = *st.Columns
+			}
+
+			// Check difference
+			r1, r2 := lo.Difference(columnNamesToCheck, detail.Columns)
+			if len(r1) != 0 || len(r2) != 0 {
+				return true, nil
+			}
+		}
+	}
+
+	// Default
+	return false, nil
+}
+
 func (*PostgresqlPublicationReconciler) manageUpdate(
 	ctx context.Context,
 	instance *v1alpha1.PostgresqlPublication,
@@ -350,6 +501,11 @@ func (*PostgresqlPublicationReconciler) manageUpdate(
 	if instance.Spec.WithParameters != nil {
 		// Change with
 		builder = builder.SetWith(instance.Spec.WithParameters.Publish, instance.Spec.WithParameters.PublishViaPartitionRoot)
+	} else {
+		// Potential reconcile case to manage
+		if pubRes.PublicationViaRoot || !pubRes.Delete || !pubRes.Insert || !pubRes.Truncate || !pubRes.Update {
+			builder = builder.SetWith(DefaultPGWithPublishParameter, &DefaultPGWithPublishViaRootParameter)
+		}
 	}
 
 	// Perform update

internal/controller/postgresql/postgresqlpublication_controller_test.go

@@ -4185,5 +4185,69 @@ var _ = Describe("PostgresqlPublication tests", func() {
 				}))
 			}
 		})
+
+		It("should be ok to reconcile publication via root changed and not in spec", func() {
+			// Setup pgec
+			setupPGEC("30s", false)
+			// Create pgdb
+			pgdb := setupPGDB(false)
+
+			// Create tables
+			err := create2KnownTablesWithColumnsInPublicSchema()
+			Expect(err).NotTo(HaveOccurred())
+
+			// Setup a pg publication
+			item := setupPGPublicationWithPartialSpec(postgresqlv1alpha1.PostgresqlPublicationSpec{
+				AllTables: true,
+			})
+
+			// Checks
+			Expect(item.Status.Ready).To(BeTrue())
+			Expect(item.Status.Phase).To(Equal(postgresqlv1alpha1.PublicationCreatedPhase))
+
+			data, err := getPublication(item.Status.Name)
+			Expect(err).NotTo(HaveOccurred())
+
+			// Alter publication
+			err = rawSQLQuery("ALTER PUBLICATION " + pgpublicationPublicationName1 + " SET (publish_via_partition_root=true)")
+			Expect(err).NotTo(HaveOccurred())
+
+			Eventually(
+				func() error {
+					data, err = getPublication(item.Status.Name)
+					// Check error
+					if err != nil {
+						return err
+					}
+
+					// Check if status hasn't been updated
+					if data.PublicationViaRoot {
+						return gerrors.New("hasn't been updated by operator")
+					}
+
+					return nil
+				},
+				generalEventuallyTimeout,
+				generalEventuallyInterval,
+			).
+				Should(Succeed())
+
+			if Expect(err).NotTo(HaveOccurred()) {
+				// Assert
+				Expect(data).To(Equal(&PublicationResult{
+					Owner:              pgdb.Status.Roles.Owner,
+					AllTables:          true,
+					Insert:             true,
+					Update:             true,
+					Delete:             true,
+					Truncate:           true,
+					PublicationViaRoot: false,
+				}))
+			}
+		})
+
+		// It("should be ok to reconcile publication via root changed and in spec", func() {})
+		// It("should be ok to reconcile with parameter publish and not in spec", func() {})
+		// It("should be ok to reconcile with parameter publish and in spec", func() {})
 	})
 })