materializations: fail validation on nullable collection keys with no…

… default value

Materializations do not support collections with nullable keys, and will generally fail if a null
value is every encountered for a collection key.

This changes validation to enforce that a nullable collection key not be allowed, unless it has a
default value annotation. If there is a default value annotation, the connector is assured that it
will never actually see a null value, and it will not try to drop NOT NULL constraints from the
materialized field in future `Apply` calls.

materialize-boilerplate/testdata/validate/nullable-key.flow.yaml

@@ -0,0 +1,24 @@
+collections:
+  key/value:
+    schema:
+      type: object
+      properties:
+        key:
+          type: string
+          default: "value"
+    key: [/key]
+
+materializations:
+  test/sqlite:
+    endpoint:
+      connector:
+        image: ghcr.io/estuary/materialize-sqlite:dev
+        config: {}
+    bindings:
+      - source: key/value
+        resource: { table: key_value }
+        fields:
+          recommended: true
+
+storageMappings:
+  "": { stores: [{ provider: S3, bucket: a-bucket }] }

materialize-boilerplate/validate.go

@@ -64,6 +64,22 @@ func (v Validator) ValidateBinding(
 		)
 	}
 
+	for _, p := range boundCollection.Projections {
+		// Don't allow collection keys to be nullable unless they have a default value set. If a
+		// default value is set, there will always be a value provided for the field from the
+		// runtime.
+		if !p.IsPrimaryKey {
+			continue
+		}
+
+		mustExist := p.Inference.Exists == pf.Inference_MUST && !slices.Contains(p.Inference.Types, "null")
+		hasDefault := p.Inference.DefaultJson != nil
+
+		if !mustExist && !hasDefault {
+			return nil, fmt.Errorf("cannot materialize collection with nullable key field '%s' unless it has a default value annotation", p.Field)
+		}
+	}
+
 	var constraints map[string]*pm.Response_Validated_Constraint
 	if !v.is.HasResource(path) || (existingBinding != nil && backfill != existingBinding.Backfill) {
 		// Always validate as a new table if the existing table doesn't exist, since there is no

materialize-boilerplate/validate_test.go

@@ -20,6 +20,7 @@ import (
 //go:generate ./testdata/generate-spec-proto.sh testdata/validate/alternate-root.flow.yaml
 //go:generate ./testdata/generate-spec-proto.sh testdata/validate/increment-backfill.flow.yaml
 //go:generate ./testdata/generate-spec-proto.sh testdata/validate/ambiguous-fields.flow.yaml
+//go:generate ./testdata/generate-spec-proto.sh testdata/validate/nullable-key.flow.yaml
 
 //go:embed testdata/validate/generated_specs
 var validateFS embed.FS
@@ -262,6 +263,45 @@ func TestValidate(t *testing.T) {
 
 		require.ErrorContains(t, err, "cannot add a new binding to materialize collection 'other' to '[key_value]' because an existing binding for collection 'key/value' is already materializing to '[key_value]'")
 	})
+
+	t.Run("can't materialize a nullable collection key with no default value", func(t *testing.T) {
+		proposed := loadValidateSpec(t, "nullable-key.flow.proto")
+
+		require.Equal(t, "key", proposed.Bindings[0].Collection.Projections[3].Field)
+		proposed.Bindings[0].Collection.Projections[3].Inference.DefaultJson = nil
+
+		is := testInfoSchemaFromSpec(t, nil, simpleTestTransform)
+		validator := NewValidator(testConstrainter{}, is)
+
+		_, err := validator.ValidateBinding(
+			[]string{"key_value"},
+			false,
+			proposed.Bindings[0].Backfill,
+			proposed.Bindings[0].Collection,
+			proposed.Bindings[0].FieldSelection.FieldConfigJsonMap,
+			nil,
+		)
+
+		require.ErrorContains(t, err, "cannot materialize collection with nullable key field 'key' unless it has a default value annotation")
+	})
+
+	t.Run("can materialize a nullable collection key with a default value", func(t *testing.T) {
+		proposed := loadValidateSpec(t, "nullable-key.flow.proto")
+
+		is := testInfoSchemaFromSpec(t, nil, simpleTestTransform)
+		validator := NewValidator(testConstrainter{}, is)
+
+		_, err := validator.ValidateBinding(
+			[]string{"key_value"},
+			false,
+			proposed.Bindings[0].Backfill,
+			proposed.Bindings[0].Collection,
+			proposed.Bindings[0].FieldSelection.FieldConfigJsonMap,
+			nil,
+		)
+
+		require.NoError(t, err)
+	})
 }
 
 func TestAsFormattedNumeric(t *testing.T) {