fix: Log warning instead of erroring out of PK component validation failure (#2039)

#### Summary

Follow up to #2037.
Gave it another thought and instead of erroring out might be better to
log a warning (for starts) so we can fix the issues instead of breaking
tables in the AWS plugin.

Even when a PK component has a null value CQ ID will still be calculated
with the null value, so the data can be inserted (opposed to a missing
PK where in the case of DB destinations the insertion will fail)

---

Use the following steps to ensure your PR is ready to be reviewed

- [ ] Read the [contribution guidelines](../blob/main/CONTRIBUTING.md)
🧑‍🎓
- [ ] Run `go fmt` to format your code 🖊
- [ ] Lint your changes via `golangci-lint run` 🚨 (install golangci-lint
[here](https://golangci-lint.run/usage/install/#local-installation))
- [ ] Update or add tests 🧪
- [ ] Ensure the status checks below are successful ✅

Author: erezrokah

scheduler/queue/worker.go

@@ -148,10 +148,15 @@ func (w *worker) resolveResource(ctx context.Context, table *schema.Table, clien
 					return
 				}
 				if err := resolvedResource.Validate(); err != nil {
-					tableMetrics := w.metrics.TableClient[table.Name][client.ID()]
-					w.logger.Error().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("resource resolver finished with validation error")
-					atomic.AddUint64(&tableMetrics.Errors, 1)
-					return
+					switch err.(type) {
+					case *schema.PKError:
+						tableMetrics := w.metrics.TableClient[table.Name][client.ID()]
+						w.logger.Error().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("resource resolver finished with validation error")
+						atomic.AddUint64(&tableMetrics.Errors, 1)
+						return
+					case *schema.PKComponentError:
+						w.logger.Warn().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("resource resolver finished with validation warning")
+					}
 				}
 				select {
 				case resourcesChan <- resolvedResource:

scheduler/scheduler_dfs.go

@@ -189,10 +189,15 @@ func (s *syncClient) resolveResourcesDfs(ctx context.Context, table *schema.Tabl
 					return
 				}
 				if err := resolvedResource.Validate(); err != nil {
-					tableMetrics := s.metrics.TableClient[table.Name][client.ID()]
-					s.logger.Error().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("resource resolver finished with validation error")
-					atomic.AddUint64(&tableMetrics.Errors, 1)
-					return
+					switch err.(type) {
+					case *schema.PKError:
+						tableMetrics := s.metrics.TableClient[table.Name][client.ID()]
+						s.logger.Error().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("resource resolver finished with validation error")
+						atomic.AddUint64(&tableMetrics.Errors, 1)
+						return
+					case *schema.PKComponentError:
+						s.logger.Warn().Err(err).Str("table", table.Name).Str("client", client.ID()).Msg("resource resolver finished with validation warning")
+					}
 				}
 				select {
 				case resourcesChan <- resolvedResource:

schema/resource.go

@@ -123,6 +123,22 @@ func (r *Resource) storeCQID(value uuid.UUID) error {
 	return r.Set(CqIDColumn.Name, b)
 }
 
+type PKError struct {
+	MissingPKs []string
+}
+
+func (e *PKError) Error() string {
+	return fmt.Sprintf("missing primary key on columns: %v", e.MissingPKs)
+}
+
+type PKComponentError struct {
+	MissingPKComponents []string
+}
+
+func (e *PKComponentError) Error() string {
+	return fmt.Sprintf("missing primary key component on columns: %v", e.MissingPKComponents)
+}
+
 // Validates that all primary keys have values.
 func (r *Resource) Validate() error {
 	var missingPks []string
@@ -136,10 +152,10 @@ func (r *Resource) Validate() error {
 		}
 	}
 	if len(missingPks) > 0 {
-		return fmt.Errorf("missing primary key on columns: %v", missingPks)
+		return &PKError{MissingPKs: missingPks}
 	}
 	if len(missingPKComponents) > 0 {
-		return fmt.Errorf("missing primary key component on columns: %v", missingPKComponents)
+		return &PKComponentError{MissingPKComponents: missingPKComponents}
 	}
 	return nil
 }

schema/resource_test.go

@@ -1,7 +1,6 @@
 package schema
 
 import (
-	"errors"
 	"testing"
 
 	"github.com/apache/arrow-go/v18/arrow"
@@ -39,12 +38,12 @@ func TestResource_Validate(t *testing.T) {
 		{
 			name:     "invalid resource with primary keys",
 			resource: NewResourceData(&Table{Name: "test", Columns: ColumnList{{Name: "col1", Type: arrow.BinaryTypes.String, PrimaryKey: true}}}, nil, nil),
-			err:      errors.New(`missing primary key on columns: [col1]`),
+			err:      &PKError{MissingPKs: []string{"col1"}},
 		},
 		{
 			name:     "invalid resource with primary key components",
 			resource: NewResourceData(&Table{Name: "test", Columns: ColumnList{{Name: "col1", Type: arrow.BinaryTypes.String, PrimaryKeyComponent: true}}}, nil, nil),
-			err:      errors.New(`missing primary key component on columns: [col1]`),
+			err:      &PKComponentError{MissingPKComponents: []string{"col1"}},
 		},
 	}
 	for _, tt := range tests {
@@ -53,11 +52,7 @@ func TestResource_Validate(t *testing.T) {
 				require.NoError(t, tt.valueSetter(tt.resource))
 			}
 			validationError := tt.resource.Validate()
-			if tt.err == nil {
-				require.NoError(t, validationError)
-			} else {
-				require.ErrorContains(t, validationError, tt.err.Error())
-			}
+			require.Equal(t, tt.err, validationError)
 		})
 	}
 }