[Schemas] Add subject, version to metadata, remove ID

opencdc/errors.go

@@ -25,7 +25,6 @@ var (
 	// ErrUnknownOperation is returned when trying to parse an Operation string
 	// and encountering an unknown operation.
 	ErrUnknownOperation = errors.New("unknown operation")
-
 	// ErrInvalidProtoDataType is returned when trying to convert a proto data
 	// type to raw or structured data.
 	ErrInvalidProtoDataType = errors.New("invalid proto data type")

opencdc/metadata.go

@@ -43,12 +43,26 @@ const (
 	// MetadataCollection is a Record.Metadata key for the name of the collection
 	// where the record originated from and/or where it should be stored.
 	MetadataCollection = "opencdc.collection"
-	// MetadataSchemaID is a Record.Metadata key for the ID of the schema of
-	// the record's .Payload.After field.
-	MetadataSchemaID = "opencdc.schema.id"
-	// MetadataSchemaType is a Record.Metadata key for the type of the schema of
-	// the record's .Payload.After field.
-	MetadataSchemaType = "opencdc.schema.type"
+
+	// MetadataKeySchemaName is a Record.Metadata key for the name of the schema of
+	// the record's .Key field.
+	MetadataKeySchemaName = "opencdc.key.schema.name"
+	// MetadataKeySchemaVersion is a Record.Metadata key for the version of the schema of
+	// the record's .Key field.
+	MetadataKeySchemaVersion = "opencdc.key.schema.version"
+	// MetadataKeySchemaType is a Record.Metadata key for the type of the schema of
+	// the record's .Key field.
+	MetadataKeySchemaType = "opencdc.key.schema.type"
+
+	// MetadataPayloadSchemaName is a Record.Metadata key for the name of the schema of
+	// the record's .Payload field.
+	MetadataPayloadSchemaName = "opencdc.payload.schema.name"
+	// MetadataPayloadSchemaVersion is a Record.Metadata key for the version of the schema of
+	// the record's .Payload field.
+	MetadataPayloadSchemaVersion = "opencdc.payload.schema.version"
+	// MetadataPayloadSchemaType is a Record.Metadata key for the type of the schema of
+	// the record's .Payload field.
+	MetadataPayloadSchemaType = "opencdc.payload.schema.type"
 
 	// MetadataConduitSourcePluginName is a Record.Metadata key for the name of
 	// the source plugin that created this record.
@@ -241,26 +255,90 @@ func (m Metadata) SetConduitDLQNackNodeID(id string) {
 	m[MetadataConduitDLQNackNodeID] = id
 }
 
-// GetSchemaID returns the value for key MetadataSchemaID.
+// GetKeySchemaName returns the value for key MetadataKeySchemaName.
+// If the value does not exist or is empty the function returns ErrMetadataFieldNotFound.
+func (m Metadata) GetKeySchemaName() (string, error) {
+	return m.getValue(MetadataKeySchemaName)
+}
+
+// SetKeySchemaName sets the metadata value for key MetadataKeySchemaName.
+func (m Metadata) SetKeySchemaName(name string) {
+	m[MetadataKeySchemaName] = name
+}
+
+// GetKeySchemaVersion returns the value for key MetadataKeySchemaVersion.
+// If the value does not exist or is empty the function returns ErrMetadataFieldNotFound.
+func (m Metadata) GetKeySchemaVersion() (int, error) {
+	vs, err := m.getValue(MetadataKeySchemaVersion)
+	if err != nil {
+		return 0, err
+	}
+
+	v, err := strconv.Atoi(vs)
+	if err != nil {
+		return 0, fmt.Errorf("invalid version %q: %w", vs, err)
+	}
+
+	return v, nil
+}
+
+// SetKeySchemaVersion sets the metadata value for key MetadataKeySchemaVersion.
+func (m Metadata) SetKeySchemaVersion(version int) {
+	m[MetadataKeySchemaVersion] = strconv.Itoa(version)
+}
+
+// GetKeySchemaType returns the value for key MetadataKeySchemaType.
 // If the value does not exist or is empty the function returns ErrMetadataFieldNotFound.
-func (m Metadata) GetSchemaID() (string, error) {
-	return m.getValue(MetadataSchemaID)
+func (m Metadata) GetKeySchemaType() (string, error) {
+	return m.getValue(MetadataKeySchemaType)
+}
+
+// SetKeySchemaType sets the metadata value for key MetadataKeySchemaType.
+func (m Metadata) SetKeySchemaType(t string) {
+	m[MetadataKeySchemaType] = t
+}
+
+// GetPayloadSchemaName returns the value for key MetadataPayloadSchemaName.
+// If the value does not exist or is empty the function returns ErrMetadataFieldNotFound.
+func (m Metadata) GetPayloadSchemaName() (string, error) {
+	return m.getValue(MetadataPayloadSchemaName)
+}
+
+// SetPayloadSchemaName sets the metadata value for key MetadataPayloadSchemaName.
+func (m Metadata) SetPayloadSchemaName(name string) {
+	m[MetadataPayloadSchemaName] = name
+}
+
+// GetPayloadSchemaVersion returns the value for key MetadataPayloadSchemaVersion.
+// If the value does not exist or is empty the function returns ErrMetadataFieldNotFound.
+func (m Metadata) GetPayloadSchemaVersion() (int, error) {
+	vs, err := m.getValue(MetadataPayloadSchemaVersion)
+	if err != nil {
+		return 0, err
+	}
+
+	v, err := strconv.Atoi(vs)
+	if err != nil {
+		return 0, fmt.Errorf("invalid version %q: %w", vs, err)
+	}
+
+	return v, nil
 }
 
-// SetSchemaID sets the metadata value for key MetadataSchemaID.
-func (m Metadata) SetSchemaID(id string) {
-	m[MetadataSchemaID] = id
+// SetPayloadSchemaVersion sets the metadata value for key MetadataPayloadSchemaVersion.
+func (m Metadata) SetPayloadSchemaVersion(version int) {
+	m[MetadataPayloadSchemaVersion] = strconv.Itoa(version)
 }
 
-// GetSchemaType returns the value for key MetadataSchemaType.
+// GetPayloadSchemaType returns the value for key MetadataPayloadSchemaType.
 // If the value does not exist or is empty the function returns ErrMetadataFieldNotFound.
-func (m Metadata) GetSchemaType() (string, error) {
-	return m.getValue(MetadataSchemaType)
+func (m Metadata) GetPayloadSchemaType() (string, error) {
+	return m.getValue(MetadataPayloadSchemaType)
 }
 
-// SetSchemaType sets the metadata value for key MetadataSchemaType.
-func (m Metadata) SetSchemaType(typeStr string) {
-	m[MetadataSchemaType] = typeStr
+// SetPayloadSchemaType sets the metadata value for key MetadataPayloadSchemaType.
+func (m Metadata) SetPayloadSchemaType(t string) {
+	m[MetadataPayloadSchemaType] = t
 }
 
 // getValue returns the value for a specific key. If the value does not exist or

opencdc/metadata_test.go

@@ -25,13 +25,17 @@ import (
 
 func TestMetadataConstants(t *testing.T) {
 	wantMapping := map[string]*protoimpl.ExtensionInfo{
-		OpenCDCVersion:         opencdcv1.E_OpencdcVersion,
-		MetadataOpenCDCVersion: opencdcv1.E_MetadataVersion,
-		MetadataCreatedAt:      opencdcv1.E_MetadataCreatedAt,
-		MetadataReadAt:         opencdcv1.E_MetadataReadAt,
-		MetadataCollection:     opencdcv1.E_MetadataCollection,
-		MetadataSchemaID:       opencdcv1.E_MetadataSchemaId,
-		MetadataSchemaType:     opencdcv1.E_MetadataSchemaType,
+		OpenCDCVersion:               opencdcv1.E_OpencdcVersion,
+		MetadataOpenCDCVersion:       opencdcv1.E_MetadataVersion,
+		MetadataCreatedAt:            opencdcv1.E_MetadataCreatedAt,
+		MetadataReadAt:               opencdcv1.E_MetadataReadAt,
+		MetadataCollection:           opencdcv1.E_MetadataCollection,
+		MetadataKeySchemaName:        opencdcv1.E_MetadataKeySchemaName,
+		MetadataKeySchemaVersion:     opencdcv1.E_MetadataKeySchemaVersion,
+		MetadataKeySchemaType:        opencdcv1.E_MetadataKeySchemaType,
+		MetadataPayloadSchemaName:    opencdcv1.E_MetadataPayloadSchemaName,
+		MetadataPayloadSchemaVersion: opencdcv1.E_MetadataPayloadSchemaVersion,
+		MetadataPayloadSchemaType:    opencdcv1.E_MetadataPayloadSchemaType,
 
 		MetadataConduitSourcePluginName:         metadatav1.E_MetadataConduitSourcePluginName,
 		MetadataConduitSourcePluginVersion:      metadatav1.E_MetadataConduitSourcePluginVersion,