Implementation:Risingwavelabs Risingwave Datagen Sink Interface

Metadata

Property	Value
File	integration_tests/datagen/sink/sink.go
Package	sink
Language	Go
Lines	78
Category	Integration Test Sink Abstraction Layer
Repository	https://github.com/risingwavelabs/risingwave

Overview

The Datagen Sink Interface module defines the foundation abstraction layer for the datagen tool's output system. It declares two core interfaces and a helper:

• SinkRecord interface: Represents a single data record that can be serialized to multiple formats (JSON, Protobuf, Avro, PostgreSQL SQL) and carries routing metadata (topic and key).
• Sink interface: Represents an output destination that can prepare topics, write records, flush buffered data, and close resources.
• BaseSinkRecord struct: A base implementation with panic defaults, intended to be embedded in concrete record types so they only need to override the methods they actually use.
• Encode helper function: A format dispatcher that serializes a SinkRecord to bytes based on a format string ("json", "protobuf", or "avro").

All sink implementations in the datagen tool (Kafka, Kinesis, Postgres, MySQL, Pulsar, S3, NATS) implement the Sink interface, and all data generator record types implement the SinkRecord interface (typically by embedding BaseSinkRecord).

Code Reference

Source Location

integration_tests/datagen/sink/sink.go

Signature

type SinkRecord interface {
	// Topic that the record belongs to.
	Topic() string

	// Record key for partitioning.
	Key() string

	// Convert the event to an INSERT INTO command.
	ToPostgresSql() string

	// Convert the event to a message in JSON format.
	// Also used for Pulsar and Kinesis.
	ToJson() []byte

	// Convert the event to a message in Protobuf format.
	// Also used for Pulsar and Kinesis.
	ToProtobuf() []byte

	// Convert the event to a message in Avro format.
	// Also used for Pulsar and Kinesis.
	ToAvro() []byte
}

type Sink interface {
	Prepare(topics []string) error
	WriteRecord(ctx context.Context, format string, record SinkRecord) error
	Flush(ctx context.Context) error
	Close() error
}

func Encode(r SinkRecord, format string) []byte

Import

import (
	"context"
)

I/O Contract

SinkRecord Interface Methods

Method	Return Type	Description
Topic()	string	Returns the topic name for routing (e.g., Kafka topic, Kinesis stream)
Key()	string	Returns a partitioning key for the record
ToPostgresSql()	string	Serializes the record as an INSERT INTO SQL statement
ToJson()	[]byte	Serializes the record as JSON bytes
ToProtobuf()	[]byte	Serializes the record as Protobuf bytes
ToAvro()	[]byte	Serializes the record as Avro bytes

Sink Interface Methods

Method	Parameters	Return	Description
Prepare	topics []string	error	Initializes the sink with the required topics (e.g., creates Kafka topics)
WriteRecord	ctx context.Context, format string, record SinkRecord	error	Writes a single record to the sink in the specified format
Flush	ctx context.Context	error	Flushes any buffered data to the underlying storage
Close	--	error	Releases all resources held by the sink

BaseSinkRecord Behavior

BaseSinkRecord is an empty struct that provides default implementations for all SinkRecord methods. Each default method calls panic("not implemented"), ensuring that concrete types that forget to override a required method fail fast rather than silently producing incorrect output.

type BaseSinkRecord struct{}

// All methods panic with "not implemented":
func (r BaseSinkRecord) Topic() string       { panic("not implemented") }
func (r BaseSinkRecord) Key() string         { panic("not implemented") }
func (r BaseSinkRecord) ToPostgresSql() string { panic("not implemented") }
func (r BaseSinkRecord) ToJson() []byte      { panic("not implemented") }
func (r BaseSinkRecord) ToProtobuf() []byte  { panic("not implemented") }
func (r BaseSinkRecord) ToAvro() []byte      { panic("not implemented") }

Encode Function

The Encode function dispatches to the appropriate serialization method based on the format string:

func Encode(r SinkRecord, format string) []byte {
	if format == "json" {
		return r.ToJson()
	} else if format == "protobuf" {
		return r.ToProtobuf()
	} else if format == "avro" {
		return r.ToAvro()
	} else {
		panic("unsupported format")
	}
}

Panics if an unsupported format string is provided.

Usage Examples

Implementing a SinkRecord

package mypackage

import (
	"datagen/sink"
	"encoding/json"
	"fmt"
)

type myEvent struct {
	sink.BaseSinkRecord // Embed for panic defaults on unimplemented methods

	UserId    int64  `json:"user_id"`
	EventType string `json:"event_type"`
}

func (r *myEvent) Topic() string {
	return "my_events"
}

func (r *myEvent) Key() string {
	return fmt.Sprint(r.UserId)
}

func (r *myEvent) ToJson() []byte {
	data, _ := json.Marshal(r)
	return data
}

func (r *myEvent) ToPostgresSql() string {
	return fmt.Sprintf("INSERT INTO my_events (user_id, event_type) VALUES ('%d', '%s')",
		r.UserId, r.EventType)
}

Using the Encode Helper

record := &myEvent{UserId: 42, EventType: "click"}
jsonBytes := sink.Encode(record, "json")
// jsonBytes contains the JSON-serialized event

Sink Lifecycle

// Typical sink usage pattern (from load_gen.go):
sinkImpl, err := createSink(ctx, cfg)
if err != nil {
	return err
}
defer sinkImpl.Close()

err = sinkImpl.Prepare(gen.KafkaTopics())
if err != nil {
	return err
}

// Write records in a loop...
err = sinkImpl.WriteRecord(ctx, cfg.Format, record)
// Periodically flush...
err = sinkImpl.Flush(ctx)

Related Pages

• Risingwavelabs_Risingwave_LoadGenerator_Interface -- The LoadGenerator interface whose Load method produces SinkRecord values
• Risingwavelabs_Risingwave_Load_Generation_Loop -- The orchestration loop that calls Sink.WriteRecord and Sink.Flush
• Risingwavelabs_Risingwave_Datagen_CLI -- The CLI that selects the sink type via subcommands
• Risingwavelabs_Risingwave_Ad_CTR_Generator -- Example generator that produces SinkRecord implementations embedding BaseSinkRecord