record

Documentation

Index

• Constants
• Variables
• type Change
• type Data
• type Metadata
  • func (m Metadata) GetConduitPluginName() (string, error)
  • func (m Metadata) GetConduitPluginVersion() (string, error)
  • func (m Metadata) GetCreatedAt() (time.Time, error)
  • func (m Metadata) GetOpenCDCVersion() (string, error)
  • func (m Metadata) GetReadAt() (time.Time, error)
  • func (m Metadata) SetConduitPluginName(name string)
  • func (m Metadata) SetConduitPluginVersion(version string)
  • func (m Metadata) SetCreatedAt(createdAt time.Time)
  • func (m Metadata) SetOpenCDCVersion()
  • func (m Metadata) SetReadAt(createdAt time.Time)
• type Operation
  • func (i Operation) MarshalText() ([]byte, error)
  • func (i Operation) String() string
  • func (i *Operation) UnmarshalText(b []byte) error
• type Position
  • func (p Position) String() string
• type RawData
  • func (d RawData) Bytes() []byte
  • func (d RawData) MarshalText() ([]byte, error)
  • func (d *RawData) UnmarshalText() ([]byte, error)
• type Record
  • func (r Record) Bytes() []byte
• type StructuredData
  • func (d StructuredData) Bytes() []byte

Constants

const (
	// OpenCDCVersion is a constant that should be used as the value in the
	// metadata field MetadataVersion. It ensures the OpenCDC format version can
	// be easily identified in case the record gets marshaled into a different
	// untyped format (e.g. JSON).
	OpenCDCVersion = "v1"

	// MetadataOpenCDCVersion is a Record.Metadata key for the version of the
	// OpenCDC format (e.g. "v1"). This field exists to ensure the OpenCDC
	// format version can be easily identified in case the record gets marshaled
	// into a different untyped format (e.g. JSON).
	MetadataOpenCDCVersion = "opencdc.version"
	// MetadataCreatedAt is a Record.Metadata key for the time when the record
	// was created in the 3rd party system. The expected format is a unix
	// timestamp in nanoseconds.
	MetadataCreatedAt = "opencdc.createdAt"
	// MetadataReadAt is a Record.Metadata key for the time when the record was
	// read from the 3rd party system. The expected format is a unix timestamp
	// in nanoseconds.
	MetadataReadAt = "opencdc.readAt"

	// MetadataConduitPluginName is a Record.Metadata key for the name of the
	// plugin that created this record.
	MetadataConduitPluginName = "conduit.plugin.name"
	// MetadataConduitPluginVersion is a Record.Metadata key for the version of
	// the plugin that created this record.
	MetadataConduitPluginVersion = "conduit.plugin.version"
)

Variables

var (
	// ErrMetadataFieldNotFound is returned in metadata utility functions when a
	// metadata field is not found.
	ErrMetadataFieldNotFound = cerrors.New("metadata field not found")
)

Types

type Change

type Change struct {
	// Before contains the data before the operation occurred. This field is
	// optional and should only be populated for operations OperationUpdate
	// OperationDelete (if the system supports fetching the data before the
	// operation).
	Before Data `json:"before"`
	// After contains the data after the operation occurred. This field should
	// be populated for all operations except OperationDelete.
	After Data `json:"after"`
}

type Data

type Data interface {
	Bytes() []byte
}

Data is a structure that contains some bytes. The only structs implementing Data are RawData and StructuredData.

type Metadata

type Metadata map[string]string

func (Metadata) GetConduitPluginName

func (m Metadata) GetConduitPluginName() (string, error)

GetConduitPluginName returns the value for key MetadataConduitPluginName. If the value is does not exist or is empty the function returns ErrMetadataFieldNotFound.

func (Metadata) GetConduitPluginVersion

func (m Metadata) GetConduitPluginVersion() (string, error)

GetConduitPluginVersion returns the value for key MetadataConduitPluginVersion. If the value is does not exist or is empty the function returns ErrMetadataFieldNotFound.

func (Metadata) GetCreatedAt

func (m Metadata) GetCreatedAt() (time.Time, error)

GetCreatedAt parses the value for key MetadataCreatedAt as a unix timestamp. If the value does not exist or the value is empty the function returns ErrMetadataFieldNotFound. If the value is not a valid unix timestamp in nanoseconds the function returns an error.

func (Metadata) GetOpenCDCVersion

func (m Metadata) GetOpenCDCVersion() (string, error)

GetOpenCDCVersion returns the value for key MetadataOpenCDCVersion. If the value is does not exist or is empty the function returns ErrMetadataFieldNotFound.

func (Metadata) GetReadAt

func (m Metadata) GetReadAt() (time.Time, error)

GetReadAt parses the value for key MetadataReadAt as a unix timestamp. If the value does not exist or the value is empty the function returns ErrMetadataFieldNotFound. If the value is not a valid unix timestamp in nanoseconds the function returns an error.

func (Metadata) SetConduitPluginName

func (m Metadata) SetConduitPluginName(name string)

SetConduitPluginName sets the metadata value for key MetadataConduitPluginName.

func (Metadata) SetConduitPluginVersion

func (m Metadata) SetConduitPluginVersion(version string)

SetConduitPluginVersion sets the metadata value for key MetadataConduitPluginVersion.

func (Metadata) SetCreatedAt

func (m Metadata) SetCreatedAt(createdAt time.Time)

SetCreatedAt sets the metadata value for key MetadataCreatedAt as a unix timestamp in nanoseconds.

func (Metadata) SetOpenCDCVersion

func (m Metadata) SetOpenCDCVersion()

SetOpenCDCVersion sets the metadata value for key MetadataVersion to the current version of OpenCDC used.

func (Metadata) SetReadAt

func (m Metadata) SetReadAt(createdAt time.Time)

SetReadAt sets the metadata value for key MetadataReadAt as a unix timestamp in nanoseconds.

type Operation

type Operation int

Operation defines what triggered the creation of a record.

const (
	OperationCreate   Operation = iota + 1 // create
	OperationUpdate                        // update
	OperationDelete                        // delete
	OperationSnapshot                      // snapshot
)

func (Operation) MarshalText

func (i Operation) MarshalText() ([]byte, error)

func (Operation) String

func (i Operation) String() string

func (*Operation) UnmarshalText

func (i *Operation) UnmarshalText(b []byte) error

type Position

type Position []byte

Position is a unique identifier for a record being process. It's a Source's responsibility to choose and assign record positions, as they will be used by the Source in subsequent pipeline runs.

func (Position) String

func (p Position) String() string

String is used when displaying the position in logs.

type RawData

type RawData struct {
	Raw    []byte
	Schema schema.Schema
}

RawData contains unstructured data in form of a byte slice.

func (RawData) Bytes

func (d RawData) Bytes() []byte

func (RawData) MarshalText

func (d RawData) MarshalText() ([]byte, error)

func (*RawData) UnmarshalText

func (d *RawData) UnmarshalText() ([]byte, error)

type Record

type Record struct {
	// Position uniquely represents the record.
	Position Position `json:"position"`
	// Operation defines what triggered the creation of a record. There are four
	// possibilities: create, update, delete or snapshot. The first three
	// operations are encountered during normal CDC operation, while "snapshot"
	// is meant to represent records during an initial load. Depending on the
	// operation, the record will contain either the payload before the change,
	// after the change, or both (see field Payload).
	Operation Operation `json:"operation"`
	// Metadata contains additional information regarding the record.
	Metadata Metadata `json:"metadata"`

	// Key represents a value that should identify the entity (e.g. database
	// row).
	Key Data `json:"key"`
	// Payload holds the payload change (data before and after the operation
	// occurred).
	Payload Change `json:"payload"`
}

Record represents a single data record produced by a source and/or consumed by a destination connector.

func (Record) Bytes

func (r Record) Bytes() []byte

Bytes returns the JSON encoding of the Record.

type StructuredData

type StructuredData map[string]interface{}

StructuredData contains data in form of a map with string keys and arbitrary values.

func (StructuredData) Bytes

func (d StructuredData) Bytes() []byte