feat: Expose transformer option for Skipping PK Validation (#2312)

This adds an option `WithSkipPrimaryKeyValidation()` that will allow users to skip validation by specifying the name of the field

Author: bbernays

transformers/options.go

@@ -85,6 +85,12 @@ func WithPrimaryKeyComponents(fields ...string) StructTransformerOption {
 	}
 }
 
+func WithSkipPrimaryKeyValidation(fields ...string) StructTransformerOption {
+	return func(t *structTransformer) {
+		t.skipPKValidationFields = fields
+	}
+}
+
 // WithMaxJSONTypeSchemaDepth allows to specify the maximum depth of JSON type schema
 func WithMaxJSONTypeSchemaDepth(maxDepth int) StructTransformerOption {
 	return func(t *structTransformer) {

transformers/struct.go

@@ -30,6 +30,8 @@ type structTransformer struct {
 	pkFieldsFound                 []string
 	pkComponentFields             []string
 	pkComponentFieldsFound        []string
+	skipPKValidationFields        []string
+	skipPKValidationFieldsFound   []string
 	jsonSchemaNameTransformer     NameTransformer
 
 	maxJSONTypeSchemaDepth int
@@ -190,6 +192,13 @@ func (t *structTransformer) addColumnFromField(field reflect.StructField, parent
 		}
 	}
 
+	for _, skipPK := range t.skipPKValidationFields {
+		if skipPK == path {
+			column.SkipPKValidation = true
+			t.skipPKValidationFieldsFound = append(t.skipPKValidationFieldsFound, skipPK)
+		}
+	}
+
 	t.table.Columns = append(t.table.Columns, column)
 
 	return nil
@@ -244,6 +253,10 @@ func TransformWithStruct(st any, opts ...StructTransformerOption) schema.Transfo
 		if diff := funk.SubtractString(t.pkComponentFields, t.pkComponentFieldsFound); len(diff) > 0 {
 			return fmt.Errorf("failed to find all of the desired primary key components: %v", diff)
 		}
+
+		if diff := funk.SubtractString(t.skipPKValidationFields, t.skipPKValidationFieldsFound); len(diff) > 0 {
+			return fmt.Errorf("failed to find all of the desired skip primary key validation fields: %v", diff)
+		}
 		return nil
 	}
 }

transformers/struct_test.go

@@ -322,6 +322,39 @@ var (
 			},
 		},
 	}
+
+	expectedTableWithSkipPKValidation = schema.Table{
+		Name: "test_pk_struct",
+		Columns: schema.ColumnList{
+			{
+				Name:             "parent",
+				Type:             arrow.BinaryTypes.String,
+				SkipPKValidation: true,
+			},
+			{
+				Name: "name",
+				Type: arrow.BinaryTypes.String,
+			},
+			{
+				Name: "version",
+				Type: arrow.PrimitiveTypes.Int64,
+			},
+		},
+	}
+
+	expectedTestTableNonEmbeddedStructWithSkipPKValidation = schema.Table{
+		Name: "test_struct",
+		Columns: schema.ColumnList{
+			schema.Column{Name: "int_col", Type: arrow.PrimitiveTypes.Int64},
+			schema.Column{Name: "test_struct", Type: types.ExtensionTypes.JSON},
+			schema.Column{
+				Name:             "non_embedded_embedded_string",
+				Type:             arrow.BinaryTypes.String,
+				SkipPKValidation: true,
+			},
+			schema.Column{Name: "non_embedded_int_col", Type: arrow.PrimitiveTypes.Int64},
+		},
+	}
 )
 
 func TestTableFromGoStruct(t *testing.T) {
@@ -476,6 +509,38 @@ func TestTableFromGoStruct(t *testing.T) {
 			},
 			want: expectedTestTableStructNonNullableFields,
 		},
+		{
+			name: "Should configure skip PK validation when option is set",
+			args: args{
+				testStruct: testPKStruct{},
+				options: []StructTransformerOption{
+					WithSkipPrimaryKeyValidation("Parent"),
+				},
+			},
+			want: expectedTableWithSkipPKValidation,
+		},
+		{
+			name: "Should return an error when a skip PK validation field is not found",
+			args: args{
+				testStruct: testPKStruct{},
+				options: []StructTransformerOption{
+					WithSkipPrimaryKeyValidation("Parent", "InvalidColumn"),
+				},
+			},
+			want:    expectedTableWithSkipPKValidation,
+			wantErr: true,
+		},
+		{
+			name: "Should configure skip PK validation for unwrapped struct fields",
+			args: args{
+				testStruct: testStructWithNonEmbeddedStruct{},
+				options: []StructTransformerOption{
+					WithUnwrapStructFields("NonEmbedded"),
+					WithSkipPrimaryKeyValidation("NonEmbedded.EmbeddedString"),
+				},
+			},
+			want: expectedTestTableNonEmbeddedStructWithSkipPKValidation,
+		},
 	}
 
 	for _, tt := range tests {
@@ -512,6 +577,19 @@ func TestTableFromGoStruct(t *testing.T) {
 			if diff := cmp.Diff(table.PrimaryKeys(), tt.want.PrimaryKeys()); diff != "" {
 				t.Fatalf("table does not match expected. diff (-got, +want): %v", diff)
 			}
+
+			// Check SkipPKValidation field for columns that have it set in expected
+			for _, wantCol := range tt.want.Columns {
+				if wantCol.SkipPKValidation {
+					gotCol := table.Column(wantCol.Name)
+					if gotCol == nil {
+						t.Fatalf("column %q not found", wantCol.Name)
+					}
+					if !gotCol.SkipPKValidation {
+						t.Fatalf("column %q expected SkipPKValidation=true, got false", wantCol.Name)
+					}
+				}
+			}
 		})
 	}
 }