NEOS-1798: Adds passthrough as a new column addition strategy (#3454)

Author: alishakawaguchi

backend/gen/go/protos/mgmt/v1alpha1/job.pb.go

@@ -168,11 +168,15 @@ message PostgresSourceConnectionOptions {
       // If this doesn't exist, will fall back to configuring generators for supported datatypes.
       // If none of the criteria above can be met, the job run will fail to prevent leaking of PII.
       AutoMap auto_map = 2;
+      // pass the new column through as is.
+      Passthrough passthrough = 3;
     }
     // Configuration for the HaltJob strategy
     message HaltJob {}
     // Configuration for the AutoMap strategy
     message AutoMap {}
+    // Configuration for the Passthrough strategy
+    message Passthrough {}
   }
 
   message ColumnRemovalStrategy {
@@ -245,11 +249,15 @@ message MysqlSourceConnectionOptions {
       // If this doesn't exist, will fall back to configuring generators for supported datatypes.
       // If none of the criteria above can be met, the job run will fail to prevent leaking of PII.
       AutoMap auto_map = 2;
+      // pass the new column through as is.
+      Passthrough passthrough = 3;
     }
     // Configuration for the HaltJob strategy
     message HaltJob {}
     // Configuration for the AutoMap strategy
     message AutoMap {}
+    // Configuration for the Passthrough strategy
+    message Passthrough {}
   }
 }
 
@@ -269,7 +277,8 @@ message MysqlSourceTableOption {
 
 message MssqlSourceConnectionOptions {
   // Whether to halt the job if a new column is added
-  bool halt_on_new_column_addition = 1;
+  // Deprecated: Use new_column_addition_strategy instead
+  bool halt_on_new_column_addition = 1 [deprecated = true];
   // The list of schemas (and their tables) along with any configuration options that will be used.
   repeated MssqlSourceSchemaOption schemas = 2;
   // The unique connection id to a mssql connection configuration
@@ -278,6 +287,8 @@ message MssqlSourceConnectionOptions {
   bool subset_by_foreign_key_constraints = 4;
   // Provide a strategy of what to do in the event Neosync encounters a column that is removed from the source table.
   ColumnRemovalStrategy column_removal_strategy = 5;
+  // Provide a strategy of what to do in the event Neosync encounters an unmapped column for the job's mapped tables.
+  NewColumnAdditionStrategy new_column_addition_strategy = 6;
 
   message ColumnRemovalStrategy {
     oneof strategy {
@@ -293,6 +304,19 @@ message MssqlSourceConnectionOptions {
     // Configuration for the ContinueJob strategy
     message ContinueJob {}
   }
+
+  message NewColumnAdditionStrategy {
+    oneof strategy {
+      // halt job if a new column is detected.
+      HaltJob halt_job = 1;
+      // pass the new column through as is.
+      Passthrough passthrough = 2;
+    }
+    // Configuration for the HaltJob strategy
+    message HaltJob {}
+    // Configuration for the Passthrough strategy
+    message Passthrough {}
+  }
 }
 
 message MssqlSourceSchemaOption {

backend/gen/go/protos/mgmt/v1alpha1/job.pb.json.go

@@ -945,13 +945,23 @@ type JobSourceOptions struct {
 }
 
 type MssqlSourceOptions struct {
-	HaltOnNewColumnAddition       bool                        `json:"haltOnNewColumnAddition"`
-	SubsetByForeignKeyConstraints bool                        `json:"subsetByForeignKeyConstraints"`
-	Schemas                       []*MssqlSourceSchemaOption  `json:"schemas"`
-	ConnectionId                  string                      `json:"connectionId"`
-	ColumnRemovalStrategy         *MssqlColumnRemovalStrategy `json:"columnRemovalStrategy,omitempty"`
+	// @deprecated
+	HaltOnNewColumnAddition       bool                            `json:"haltOnNewColumnAddition"`
+	SubsetByForeignKeyConstraints bool                            `json:"subsetByForeignKeyConstraints"`
+	Schemas                       []*MssqlSourceSchemaOption      `json:"schemas"`
+	ConnectionId                  string                          `json:"connectionId"`
+	ColumnRemovalStrategy         *MssqlColumnRemovalStrategy     `json:"columnRemovalStrategy,omitempty"`
+	NewColumnAdditionStrategy     *MssqlNewColumnAdditionStrategy `json:"newColumnAdditionStrategy,omitempty"`
+}
+
+type MssqlNewColumnAdditionStrategy struct {
+	HaltJob     *MssqlHaltJobNewColumnAdditionStrategy     `json:"haltJob,omitempty"`
+	Passthrough *MssqlPassthroughNewColumnAdditionStrategy `json:"passthrough,omitempty"`
 }
 
+type MssqlHaltJobNewColumnAdditionStrategy struct{}
+type MssqlPassthroughNewColumnAdditionStrategy struct{}
+
 type MssqlColumnRemovalStrategy struct {
 	HaltJob     *MssqlHaltJobColumnRemovalStrategy     `json:"haltJob,omitempty"`
 	ContinueJob *MssqlContinueJobColumnRemovalStrategy `json:"continueJob,omitempty"`
@@ -993,8 +1003,7 @@ type MssqlContinueJobColumnRemovalStrategy struct{}
 
 func (m *MssqlSourceOptions) ToDto() *mgmtv1alpha1.MssqlSourceConnectionOptions {
 	dto := &mgmtv1alpha1.MssqlSourceConnectionOptions{
-		HaltOnNewColumnAddition: m.HaltOnNewColumnAddition,
-		ConnectionId:            m.ConnectionId,
+		ConnectionId: m.ConnectionId,
 		Schemas: make(
 			[]*mgmtv1alpha1.MssqlSourceSchemaOption,
 			len(m.Schemas),
@@ -1009,13 +1018,23 @@ func (m *MssqlSourceOptions) ToDto() *mgmtv1alpha1.MssqlSourceConnectionOptions
 		dto.ColumnRemovalStrategy = m.ColumnRemovalStrategy.ToDto()
 	}
 
+	if m.NewColumnAdditionStrategy != nil {
+		dto.NewColumnAdditionStrategy = m.NewColumnAdditionStrategy.ToDto()
+	} else if m.HaltOnNewColumnAddition {
+		dto.NewColumnAdditionStrategy = &mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy{
+			Strategy: &mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy_HaltJob_{
+				HaltJob: &mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy_HaltJob{},
+			},
+		}
+	}
+
 	return dto
 }
+
 func (m *MssqlSourceOptions) FromDto(dto *mgmtv1alpha1.MssqlSourceConnectionOptions) {
 	if dto == nil {
 		dto = &mgmtv1alpha1.MssqlSourceConnectionOptions{}
 	}
-	m.HaltOnNewColumnAddition = dto.GetHaltOnNewColumnAddition()
 	m.ConnectionId = dto.GetConnectionId()
 	m.SubsetByForeignKeyConstraints = dto.GetSubsetByForeignKeyConstraints()
 	m.Schemas = FromDtoMssqlSourceSchemaOptions(dto.GetSchemas())
@@ -1024,6 +1043,46 @@ func (m *MssqlSourceOptions) FromDto(dto *mgmtv1alpha1.MssqlSourceConnectionOpti
 		m.ColumnRemovalStrategy = &MssqlColumnRemovalStrategy{}
 		m.ColumnRemovalStrategy.FromDto(dto.GetColumnRemovalStrategy())
 	}
+
+	if dto.GetNewColumnAdditionStrategy().GetStrategy() != nil {
+		m.NewColumnAdditionStrategy = &MssqlNewColumnAdditionStrategy{}
+		m.NewColumnAdditionStrategy.FromDto(dto.GetNewColumnAdditionStrategy())
+	} else if m.HaltOnNewColumnAddition {
+		m.NewColumnAdditionStrategy = &MssqlNewColumnAdditionStrategy{
+			HaltJob: &MssqlHaltJobNewColumnAdditionStrategy{},
+		}
+	}
+}
+
+func (s *MssqlNewColumnAdditionStrategy) ToDto() *mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy {
+	if s.HaltJob != nil {
+		return &mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy{
+			Strategy: &mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy_HaltJob_{
+				HaltJob: &mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy_HaltJob{},
+			},
+		}
+	}
+	if s.Passthrough != nil {
+		return &mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy{
+			Strategy: &mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy_Passthrough_{
+				Passthrough: &mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy_Passthrough{},
+			},
+		}
+	}
+	return nil
+}
+
+func (s *MssqlNewColumnAdditionStrategy) FromDto(
+	dto *mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy,
+) {
+	if dto.GetStrategy() != nil {
+		switch dto.GetStrategy().(type) {
+		case *mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy_HaltJob_:
+			s.HaltJob = &MssqlHaltJobNewColumnAdditionStrategy{}
+		case *mgmtv1alpha1.MssqlSourceConnectionOptions_NewColumnAdditionStrategy_Passthrough_:
+			s.Passthrough = &MssqlPassthroughNewColumnAdditionStrategy{}
+		}
+	}
 }
 
 type MssqlSourceSchemaOption struct {
@@ -1260,13 +1319,14 @@ type MysqlSourceOptions struct {
 }
 
 type MysqlNewColumnAdditionStrategy struct {
-	HaltJob *MysqlHaltJobNewColumnAdditionStrategy `json:"haltJob,omitempty"`
-	AutoMap *MysqlAutoMapNewColumnAdditionStrategy `json:"autoMap,omitempty"`
+	HaltJob     *MysqlHaltJobNewColumnAdditionStrategy     `json:"haltJob,omitempty"`
+	AutoMap     *MysqlAutoMapNewColumnAdditionStrategy     `json:"autoMap,omitempty"`
+	Passthrough *MysqlPassthroughNewColumnAdditionStrategy `json:"passthrough,omitempty"`
 }
 
 type MysqlHaltJobNewColumnAdditionStrategy struct{}
 type MysqlAutoMapNewColumnAdditionStrategy struct{}
-
+type MysqlPassthroughNewColumnAdditionStrategy struct{}
 type MysqlColumnRemovalStrategy struct {
 	HaltJob     *MysqlHaltJobColumnRemovalStrategy     `json:"haltJob,omitempty"`
 	ContinueJob *MysqlContinueJobColumnRemovalStrategy `json:"continueJob,omitempty"`
@@ -1317,8 +1377,9 @@ type PostgresSourceOptions struct {
 }
 
 type PostgresNewColumnAdditionStrategy struct {
-	HaltJob *PostgresHaltJobStrategy `json:"haltJob,omitempty"`
-	AutoMap *PostgresAutoMapStrategy `json:"autoMap,omitempty"`
+	HaltJob     *PostgresHaltJobStrategy     `json:"haltJob,omitempty"`
+	AutoMap     *PostgresAutoMapStrategy     `json:"autoMap,omitempty"`
+	Passthrough *PostgresPassthroughStrategy `json:"passthrough,omitempty"`
 }
 
 func (p *PostgresNewColumnAdditionStrategy) ToDto() *mgmtv1alpha1.PostgresSourceConnectionOptions_NewColumnAdditionStrategy {
@@ -1334,6 +1395,12 @@ func (p *PostgresNewColumnAdditionStrategy) ToDto() *mgmtv1alpha1.PostgresSource
 				AutoMap: &mgmtv1alpha1.PostgresSourceConnectionOptions_NewColumnAdditionStrategy_AutoMap{},
 			},
 		}
+	} else if p.Passthrough != nil {
+		return &mgmtv1alpha1.PostgresSourceConnectionOptions_NewColumnAdditionStrategy{
+			Strategy: &mgmtv1alpha1.PostgresSourceConnectionOptions_NewColumnAdditionStrategy_Passthrough_{
+				Passthrough: &mgmtv1alpha1.PostgresSourceConnectionOptions_NewColumnAdditionStrategy_Passthrough{},
+			},
+		}
 	}
 	return nil
 }
@@ -1349,11 +1416,14 @@ func (p *PostgresNewColumnAdditionStrategy) FromDto(
 		p.AutoMap = &PostgresAutoMapStrategy{}
 	case *mgmtv1alpha1.PostgresSourceConnectionOptions_NewColumnAdditionStrategy_HaltJob_:
 		p.HaltJob = &PostgresHaltJobStrategy{}
+	case *mgmtv1alpha1.PostgresSourceConnectionOptions_NewColumnAdditionStrategy_Passthrough_:
+		p.Passthrough = &PostgresPassthroughStrategy{}
 	}
 }
 
 type PostgresHaltJobStrategy struct{}
 type PostgresAutoMapStrategy struct{}
+type PostgresPassthroughStrategy struct{}
 
 type PostgresColumnRemovalStrategy struct {
 	HaltJob     *PostgresHaltJobColumnRemovalStrategy     `json:"haltJob,omitempty"`
@@ -1504,7 +1574,6 @@ func FromDtoPostgresSourceSchemaOptions(
 
 func (s *MysqlSourceOptions) ToDto() *mgmtv1alpha1.MysqlSourceConnectionOptions {
 	dto := &mgmtv1alpha1.MysqlSourceConnectionOptions{
-		HaltOnNewColumnAddition:       s.HaltOnNewColumnAddition,
 		SubsetByForeignKeyConstraints: s.SubsetByForeignKeyConstraints,
 		ConnectionId:                  s.ConnectionId,
 	}
@@ -1536,7 +1605,6 @@ func (s *MysqlSourceOptions) ToDto() *mgmtv1alpha1.MysqlSourceConnectionOptions
 				HaltJob: &mgmtv1alpha1.MysqlSourceConnectionOptions_NewColumnAdditionStrategy_HaltJob{},
 			},
 		}
-		dto.HaltOnNewColumnAddition = true //nolint:staticcheck
 	}
 
 	return dto
@@ -1557,6 +1625,13 @@ func (s *MysqlNewColumnAdditionStrategy) ToDto() *mgmtv1alpha1.MysqlSourceConnec
 			},
 		}
 	}
+	if s.Passthrough != nil {
+		return &mgmtv1alpha1.MysqlSourceConnectionOptions_NewColumnAdditionStrategy{
+			Strategy: &mgmtv1alpha1.MysqlSourceConnectionOptions_NewColumnAdditionStrategy_Passthrough_{
+				Passthrough: &mgmtv1alpha1.MysqlSourceConnectionOptions_NewColumnAdditionStrategy_Passthrough{},
+			},
+		}
+	}
 	return nil
 }
 
@@ -1569,20 +1644,20 @@ func (s *MysqlNewColumnAdditionStrategy) FromDto(
 			s.HaltJob = &MysqlHaltJobNewColumnAdditionStrategy{}
 		case *mgmtv1alpha1.MysqlSourceConnectionOptions_NewColumnAdditionStrategy_AutoMap_:
 			s.AutoMap = &MysqlAutoMapNewColumnAdditionStrategy{}
+		case *mgmtv1alpha1.MysqlSourceConnectionOptions_NewColumnAdditionStrategy_Passthrough_:
+			s.Passthrough = &MysqlPassthroughNewColumnAdditionStrategy{}
 		}
 	}
 }
 
 func (s *MysqlSourceOptions) FromDto(dto *mgmtv1alpha1.MysqlSourceConnectionOptions) {
-	s.HaltOnNewColumnAddition = dto.HaltOnNewColumnAddition //nolint:staticcheck
 	if dto.GetNewColumnAdditionStrategy().GetStrategy() != nil {
 		s.NewColumnAdditionStrategy = &MysqlNewColumnAdditionStrategy{}
 		s.NewColumnAdditionStrategy.FromDto(dto.GetNewColumnAdditionStrategy())
 	} else if dto.HaltOnNewColumnAddition { //nolint:staticcheck
 		s.NewColumnAdditionStrategy = &MysqlNewColumnAdditionStrategy{
 			HaltJob: &MysqlHaltJobNewColumnAdditionStrategy{},
 		}
-		s.HaltOnNewColumnAddition = true
 	}
 	s.SubsetByForeignKeyConstraints = dto.SubsetByForeignKeyConstraints
 	s.Schemas = FromDtoMysqlSourceSchemaOptions(dto.Schemas)