Consolidates Database Schema Row Structs (#2891)

.mockery.yml

@@ -95,7 +95,7 @@ packages:
     interfaces:
       ManagerInterface:
       ClientInterface:
-  github.com/nucleuscloud/neosync/worker/internal/connection-tunnel-manager:
+  github.com/nucleuscloud/neosync/internal/connection-tunnel-manager:
     interfaces:
       ConnectionProvider:
   github.com/nucleuscloud/neosync/worker/pkg/benthos/dynamodb:

backend/pkg/sqlmanager/mssql/mssql-manager.go

@@ -67,7 +67,7 @@ func (m *Manager) GetDatabaseSchema(ctx context.Context) ([]*sqlmanager_shared.D
 			ColumnName:             row.ColumnName,
 			DataType:               row.DataType,
 			ColumnDefault:          row.ColumnDefault, // todo: make sure this is valid for the other funcs
-			IsNullable:             row.IsNullable,
+			IsNullable:             row.IsNullable != "NO",
 			GeneratedType:          generatedType,
 			OrdinalPosition:        int(row.OrdinalPosition),
 			CharacterMaximumLength: charMaxLength,
@@ -80,7 +80,7 @@ func (m *Manager) GetDatabaseSchema(ctx context.Context) ([]*sqlmanager_shared.D
 	return output, nil
 }
 
-func (m *Manager) GetSchemaColumnMap(ctx context.Context) (map[string]map[string]*sqlmanager_shared.ColumnInfo, error) {
+func (m *Manager) GetSchemaColumnMap(ctx context.Context) (map[string]map[string]*sqlmanager_shared.DatabaseSchemaRow, error) {
 	dbSchemas, err := m.GetDatabaseSchema(ctx)
 	if err != nil {
 		return nil, err
@@ -309,7 +309,7 @@ func BuildMssqlSetIdentityInsertStatement(
 	return fmt.Sprintf("SET IDENTITY_INSERT %q.%q %s;", schema, table, enabledKeyword)
 }
 
-func GetMssqlColumnOverrideAndResetProperties(columnInfo *sqlmanager_shared.ColumnInfo) (needsOverride, needsReset bool) {
+func GetMssqlColumnOverrideAndResetProperties(columnInfo *sqlmanager_shared.DatabaseSchemaRow) (needsOverride, needsReset bool) {
 	needsOverride = false
 	needsReset = false
 

backend/pkg/sqlmanager/mssql/mssql-manager_integration_test.go

@@ -21,7 +21,7 @@ func (s *IntegrationTestSuite) Test_GetDatabaseSchema() {
 			ColumnName:             "id",
 			DataType:               "int",
 			ColumnDefault:          "",
-			IsNullable:             "NO",
+			IsNullable:             false,
 			CharacterMaximumLength: -1,
 			NumericPrecision:       10,
 			NumericScale:           0,

backend/pkg/sqlmanager/mysql/mysql-manager.go

@@ -85,7 +85,7 @@ func (m *MysqlManager) GetDatabaseSchema(ctx context.Context) ([]*sqlmanager_sha
 			DataType:               row.DataType,
 			ColumnDefault:          columnDefaultStr,
 			ColumnDefaultType:      columnDefaultType,
-			IsNullable:             row.IsNullable,
+			IsNullable:             row.IsNullable != "NO",
 			GeneratedType:          generatedType,
 			CharacterMaximumLength: charMaxLength,
 			NumericPrecision:       numericPrecision,
@@ -98,7 +98,7 @@ func (m *MysqlManager) GetDatabaseSchema(ctx context.Context) ([]*sqlmanager_sha
 }
 
 // returns: {public.users: { id: struct{}{}, created_at: struct{}{}}}
-func (m *MysqlManager) GetSchemaColumnMap(ctx context.Context) (map[string]map[string]*sqlmanager_shared.ColumnInfo, error) {
+func (m *MysqlManager) GetSchemaColumnMap(ctx context.Context) (map[string]map[string]*sqlmanager_shared.DatabaseSchemaRow, error) {
 	dbSchemas, err := m.GetDatabaseSchema(ctx)
 	if err != nil {
 		return nil, err
@@ -841,7 +841,7 @@ func EscapeMysqlDefaultColumn(defaultColumnValue string, defaultColumnType *stri
 	return fmt.Sprintf("(%s)", defaultColumnValue), fmt.Errorf("unsupported default column type: %s, currently supported types are: %v", *defaultColumnType, defaultColumnTypes)
 }
 
-func GetMysqlColumnOverrideAndResetProperties(columnInfo *sqlmanager_shared.ColumnInfo) (needsOverride, needsReset bool) {
+func GetMysqlColumnOverrideAndResetProperties(columnInfo *sqlmanager_shared.DatabaseSchemaRow) (needsOverride, needsReset bool) {
 	needsOverride = false
 	needsReset = false
 	return

backend/pkg/sqlmanager/postgres/postgres-manager.go

@@ -51,7 +51,7 @@ func (p *PostgresManager) GetDatabaseSchema(ctx context.Context) ([]*sqlmanager_
 			ColumnName:             row.ColumnName,
 			DataType:               row.DataType,
 			ColumnDefault:          row.ColumnDefault,
-			IsNullable:             row.IsNullable,
+			IsNullable:             row.IsNullable != "NO",
 			CharacterMaximumLength: int(row.CharacterMaximumLength),
 			NumericPrecision:       int(row.NumericPrecision),
 			NumericScale:           int(row.NumericScale),
@@ -64,7 +64,7 @@ func (p *PostgresManager) GetDatabaseSchema(ctx context.Context) ([]*sqlmanager_
 }
 
 // returns: {public.users: { id: struct{}{}, created_at: struct{}{}}}
-func (p *PostgresManager) GetSchemaColumnMap(ctx context.Context) (map[string]map[string]*sqlmanager_shared.ColumnInfo, error) {
+func (p *PostgresManager) GetSchemaColumnMap(ctx context.Context) (map[string]map[string]*sqlmanager_shared.DatabaseSchemaRow, error) {
 	dbSchemas, err := p.GetDatabaseSchema(ctx)
 	if err != nil {
 		return nil, err
@@ -933,7 +933,7 @@ func BuildPgResetSequenceSql(sequenceName string) string {
 	return fmt.Sprintf("ALTER SEQUENCE %s RESTART;", sequenceName)
 }
 
-func GetPostgresColumnOverrideAndResetProperties(columnInfo *sqlmanager_shared.ColumnInfo) (needsOverride, needsReset bool) {
+func GetPostgresColumnOverrideAndResetProperties(columnInfo *sqlmanager_shared.DatabaseSchemaRow) (needsOverride, needsReset bool) {
 	needsOverride = false
 	needsReset = false
 

backend/pkg/sqlmanager/postgres/postgres-manager_integration_test.go

@@ -22,7 +22,7 @@ func (s *IntegrationTestSuite) Test_GetDatabaseSchema() {
 			ColumnName:             "id",
 			DataType:               "text",
 			ColumnDefault:          "",
-			IsNullable:             "NO",
+			IsNullable:             false,
 			CharacterMaximumLength: -1,
 			NumericPrecision:       -1,
 			NumericScale:           -1,
@@ -45,7 +45,7 @@ func (s *IntegrationTestSuite) Test_GetDatabaseSchema_With_Identity() {
 			ColumnName:             "id",
 			DataType:               "integer",
 			ColumnDefault:          "",
-			IsNullable:             "NO",
+			IsNullable:             false,
 			CharacterMaximumLength: -1,
 			NumericPrecision:       32,
 			NumericScale:           0,

backend/pkg/sqlmanager/shared/types.go

@@ -15,7 +15,7 @@ type DatabaseSchemaRow struct {
 	DataType               string
 	ColumnDefault          string
 	ColumnDefaultType      *string
-	IsNullable             string
+	IsNullable             bool
 	CharacterMaximumLength int
 	NumericPrecision       int
 	NumericScale           int
@@ -24,6 +24,13 @@ type DatabaseSchemaRow struct {
 	IdentityGeneration     *string
 }
 
+func (d *DatabaseSchemaRow) NullableString() string {
+	if d.IsNullable {
+		return "YES"
+	}
+	return "NO"
+}
+
 type ForeignKeyConstraintsRow struct {
 	ConstraintName    string
 	SchemaName        string
@@ -117,17 +124,6 @@ type TableConstraints struct {
 	UniqueConstraints     map[string][][]string
 }
 
-type ColumnInfo struct {
-	OrdinalPosition        int     // Specifies the sequence or order in which each column is defined within the table. Starts at 1 for the first column.
-	ColumnDefault          string  // Specifies the default value for a column, if any is set.
-	IsNullable             bool    // Specifies if the column is nullable or not.
-	DataType               string  // Specifies the data type of the column, i.e., bool, varchar, int, etc.
-	CharacterMaximumLength *int    // Specifies the maximum allowable length of the column for character-based data types. For datatypes such as integers, boolean, dates etc. this is NULL.
-	NumericPrecision       *int    // Specifies the precision for numeric data types. It represents the TOTAL count of significant digits in the whole number, that is, the number of digits to BOTH sides of the decimal point. Null for non-numeric data types.
-	NumericScale           *int    // Specifies the scale of the column for numeric data types, specifically non-integers. It represents the number of digits to the RIGHT of the decimal point. Null for non-numeric data types and integers.
-	IdentityGeneration     *string // Specifies the identity generation strategy for the column, if applicable.
-}
-
 type DataType struct {
 	Schema     string
 	Name       string

backend/pkg/sqlmanager/shared/utils.go

@@ -7,34 +7,21 @@ import (
 
 func GetUniqueSchemaColMappings(
 	schemas []*DatabaseSchemaRow,
-) map[string]map[string]*ColumnInfo {
-	groupedSchemas := map[string]map[string]*ColumnInfo{} // ex: {public.users: { id: struct{}{}, created_at: struct{}{}}}
+) map[string]map[string]*DatabaseSchemaRow {
+	groupedSchemas := map[string]map[string]*DatabaseSchemaRow{} // ex: {public.users: { id: struct{}{}, created_at: struct{}{}}}
 	for _, record := range schemas {
 		key := BuildTable(record.TableSchema, record.TableName)
 		if _, ok := groupedSchemas[key]; ok {
-			groupedSchemas[key][record.ColumnName] = toColumnInfo(record)
+			groupedSchemas[key][record.ColumnName] = record
 		} else {
-			groupedSchemas[key] = map[string]*ColumnInfo{
-				record.ColumnName: toColumnInfo(record),
+			groupedSchemas[key] = map[string]*DatabaseSchemaRow{
+				record.ColumnName: record,
 			}
 		}
 	}
 	return groupedSchemas
 }
 
-func toColumnInfo(row *DatabaseSchemaRow) *ColumnInfo {
-	return &ColumnInfo{
-		OrdinalPosition:        row.OrdinalPosition,
-		ColumnDefault:          row.ColumnDefault,
-		IsNullable:             ConvertNullableTextToBool(row.IsNullable),
-		DataType:               row.DataType,
-		CharacterMaximumLength: Ptr(row.CharacterMaximumLength),
-		NumericPrecision:       Ptr(row.NumericPrecision),
-		NumericScale:           Ptr(row.NumericScale),
-		IdentityGeneration:     row.IdentityGeneration,
-	}
-}
-
 func ConvertNullableTextToBool(isNullableStr string) bool {
 	return isNullableStr != "NO"
 }

backend/pkg/sqlmanager/sql-manager.go

@@ -25,7 +25,7 @@ import (
 
 type SqlDatabase interface {
 	GetDatabaseSchema(ctx context.Context) ([]*sqlmanager_shared.DatabaseSchemaRow, error)
-	GetSchemaColumnMap(ctx context.Context) (map[string]map[string]*sqlmanager_shared.ColumnInfo, error) // ex: {public.users: { id: struct{}{}, created_at: struct{}{}}}
+	GetSchemaColumnMap(ctx context.Context) (map[string]map[string]*sqlmanager_shared.DatabaseSchemaRow, error) // ex: {public.users: { id: struct{}{}, created_at: struct{}{}}}
 	GetTableConstraintsBySchema(ctx context.Context, schemas []string) (*sqlmanager_shared.TableConstraints, error)
 	GetCreateTableStatement(ctx context.Context, schema, table string) (string, error)
 	GetTableInitStatements(ctx context.Context, tables []*sqlmanager_shared.SchemaTable) ([]*sqlmanager_shared.TableInitStatement, error)
@@ -362,7 +362,7 @@ func (s *SqlManager) NewSqlDbFromUrl(
 	}, nil
 }
 
-func GetColumnOverrideAndResetProperties(driver string, cInfo *sqlmanager_shared.ColumnInfo) (needsOverride, needsReset bool, err error) {
+func GetColumnOverrideAndResetProperties(driver string, cInfo *sqlmanager_shared.DatabaseSchemaRow) (needsOverride, needsReset bool, err error) {
 	switch driver {
 	case sqlmanager_shared.PostgresDriver, "postgres":
 		needsOverride, needsReset := sqlmanager_postgres.GetPostgresColumnOverrideAndResetProperties(cInfo)

backend/services/mgmt/v1alpha1/connection-data-service/connection-data.go

@@ -513,7 +513,7 @@ func (s *Service) GetConnectionSchema(
 				Table:              col.TableName,
 				Column:             col.ColumnName,
 				DataType:           col.DataType,
-				IsNullable:         col.IsNullable,
+				IsNullable:         col.NullableString(),
 				ColumnDefault:      defaultColumn,
 				GeneratedType:      col.GeneratedType,
 				IdentityGeneration: col.IdentityGeneration,

backend/services/mgmt/v1alpha1/connection-data-service/connection-data_test.go

@@ -151,10 +151,11 @@ func Test_GetConnectionSchema_Postgres(t *testing.T) {
 	expected := []*mgmtv1alpha1.DatabaseColumn{}
 	for _, col := range mockColumns {
 		expected = append(expected, &mgmtv1alpha1.DatabaseColumn{
-			Schema:   col.TableSchema,
-			Table:    col.TableName,
-			Column:   col.ColumnName,
-			DataType: col.DataType,
+			Schema:     col.TableSchema,
+			Table:      col.TableName,
+			Column:     col.ColumnName,
+			DataType:   col.DataType,
+			IsNullable: "NO",
 		})
 	}
 

backend/services/mgmt/v1alpha1/job-service/jobs.go

@@ -1819,7 +1819,7 @@ func validateVirtualForeignKeys(
 	virtualForeignKeys []*mgmtv1alpha1.VirtualForeignConstraint,
 	jobColMappings map[string]map[string]*mgmtv1alpha1.JobMapping,
 	tc *sqlmanager_shared.TableConstraints,
-	colMap map[string]map[string]*sqlmanager_shared.ColumnInfo,
+	colMap map[string]map[string]*sqlmanager_shared.DatabaseSchemaRow,
 ) *validateVirtualForeignKeysResponse {
 	dbErrors := []string{}
 	colErrorsMap := map[string]map[string][]string{}

backend/services/mgmt/v1alpha1/job-service/jobs_test.go

@@ -1788,11 +1788,11 @@ func Test_ValidateJobMappings_NoValidationErrors(t *testing.T) {
 
 	m.SqlManagerMock.On("NewSqlDb", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&sql_manager.SqlConnection{Db: m.SqlDbMock, Driver: sqlmanager_shared.PostgresDriver}, nil)
 	m.SqlDbMock.On("Close").Return(nil)
-	m.SqlDbMock.On("GetSchemaColumnMap", mock.Anything).Return(map[string]map[string]*sqlmanager_shared.ColumnInfo{
-		"public.users":   {"id": &sqlmanager_shared.ColumnInfo{}, "name": &sqlmanager_shared.ColumnInfo{IsNullable: true}},
-		"public.orders":  {"id": &sqlmanager_shared.ColumnInfo{}, "buyer_id": &sqlmanager_shared.ColumnInfo{IsNullable: true}},
-		"circle.table_1": {"id": &sqlmanager_shared.ColumnInfo{}, "table2_id": &sqlmanager_shared.ColumnInfo{IsNullable: true}},
-		"circle.table_2": {"id": &sqlmanager_shared.ColumnInfo{}, "table1_id": &sqlmanager_shared.ColumnInfo{}},
+	m.SqlDbMock.On("GetSchemaColumnMap", mock.Anything).Return(map[string]map[string]*sqlmanager_shared.DatabaseSchemaRow{
+		"public.users":   {"id": &sqlmanager_shared.DatabaseSchemaRow{}, "name": &sqlmanager_shared.DatabaseSchemaRow{IsNullable: true}},
+		"public.orders":  {"id": &sqlmanager_shared.DatabaseSchemaRow{}, "buyer_id": &sqlmanager_shared.DatabaseSchemaRow{IsNullable: true}},
+		"circle.table_1": {"id": &sqlmanager_shared.DatabaseSchemaRow{}, "table2_id": &sqlmanager_shared.DatabaseSchemaRow{IsNullable: true}},
+		"circle.table_2": {"id": &sqlmanager_shared.DatabaseSchemaRow{}, "table1_id": &sqlmanager_shared.DatabaseSchemaRow{}},
 	}, nil)
 	m.SqlDbMock.On("GetTableConstraintsBySchema", mock.Anything, mock.Anything).Return(&sqlmanager_shared.TableConstraints{
 		ForeignKeyConstraints: map[string][]*sqlmanager_shared.ForeignConstraint{
@@ -1857,11 +1857,11 @@ func Test_ValidateJobMappings_ValidationErrors(t *testing.T) {
 
 	m.SqlManagerMock.On("NewSqlDb", mock.Anything, mock.Anything, mock.Anything, mock.Anything).Return(&sql_manager.SqlConnection{Db: m.SqlDbMock, Driver: sqlmanager_shared.PostgresDriver}, nil)
 	m.SqlDbMock.On("Close").Return(nil)
-	m.SqlDbMock.On("GetSchemaColumnMap", mock.Anything).Return(map[string]map[string]*sqlmanager_shared.ColumnInfo{
-		"public.users":   {"id": &sqlmanager_shared.ColumnInfo{}, "name": &sqlmanager_shared.ColumnInfo{}},
-		"public.orders":  {"id": &sqlmanager_shared.ColumnInfo{}, "buyer_id": &sqlmanager_shared.ColumnInfo{}},
-		"circle.table_1": {"id": &sqlmanager_shared.ColumnInfo{}, "table2_id": &sqlmanager_shared.ColumnInfo{}},
-		"circle.table_2": {"id": &sqlmanager_shared.ColumnInfo{}, "table1_id": &sqlmanager_shared.ColumnInfo{}},
+	m.SqlDbMock.On("GetSchemaColumnMap", mock.Anything).Return(map[string]map[string]*sqlmanager_shared.DatabaseSchemaRow{
+		"public.users":   {"id": &sqlmanager_shared.DatabaseSchemaRow{}, "name": &sqlmanager_shared.DatabaseSchemaRow{}},
+		"public.orders":  {"id": &sqlmanager_shared.DatabaseSchemaRow{}, "buyer_id": &sqlmanager_shared.DatabaseSchemaRow{}},
+		"circle.table_1": {"id": &sqlmanager_shared.DatabaseSchemaRow{}, "table2_id": &sqlmanager_shared.DatabaseSchemaRow{}},
+		"circle.table_2": {"id": &sqlmanager_shared.DatabaseSchemaRow{}, "table1_id": &sqlmanager_shared.DatabaseSchemaRow{}},
 	}, nil)
 	m.SqlDbMock.On("GetTableConstraintsBySchema", mock.Anything, mock.Anything).Return(&sqlmanager_shared.TableConstraints{
 		ForeignKeyConstraints: map[string][]*sqlmanager_shared.ForeignConstraint{

cli/internal/cmds/neosync/sync/sync.go

@@ -1131,7 +1131,7 @@ func (c *clisync) getDestinationSchemas() ([]*mgmtv1alpha1.DatabaseColumn, error
 			Table:              col.TableName,
 			Column:             col.ColumnName,
 			DataType:           col.DataType,
-			IsNullable:         col.IsNullable,
+			IsNullable:         col.NullableString(),
 			ColumnDefault:      defaultColumn,
 			GeneratedType:      col.GeneratedType,
 			IdentityGeneration: col.IdentityGeneration,