Implementation:Risingwavelabs Risingwave Ad CTR Generator

Metadata

Property	Value
File	integration_tests/datagen/ad_ctr/ad_ctr.go
Package	ad_ctr
Language	Go
Lines	127
Category	Integration Test Data Generator
Repository	https://github.com/risingwavelabs/risingwave

Overview

The Ad CTR Generator produces synthetic ad impression and click event data for the ad click-through rate (CTR) integration test demo in RisingWave. It generates correlated impression and click events with a simulated per-ad click-through rate, sending the events to separate Kafka topics (ad_impression and ad_click).

Each generated cycle creates an ad impression event with a random bid ID and ad ID. A corresponding click event is conditionally generated based on the ad's simulated CTR, which is a random value between 0 and 1 that is lazily assigned and cached per ad ID. Click events include a random delay of 1 to 10 seconds after the impression timestamp to simulate real-world user behavior.

The generator uses a deterministic random seed (seed=1) via the gofakeit library to ensure reproducible output across runs.

Code Reference

Source Location

integration_tests/datagen/ad_ctr/ad_ctr.go

Signature

func NewAdCtrGen() gen.LoadGenerator

Constructor that returns a new adCtrGen instance implementing the gen.LoadGenerator interface. Initializes the internal CTR map and a deterministic faker with seed 1.

Key Types

type adImpressionEvent struct {
	sink.BaseSinkRecord
	BidId               int64  `json:"bid_id"`
	AdId                int64  `json:"ad_id"`
	ImpressionTimestamp string `json:"impression_timestamp"`
}

type adClickEvent struct {
	sink.BaseSinkRecord
	BidId          int64  `json:"bid_id"`
	ClickTimestamp string `json:"click_timestamp"`
}

type adCtrGen struct {
	faker *gofakeit.Faker
	ctr   map[int64]float64
}

Import

import (
	"context"
	"datagen/gen"
	"datagen/sink"
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"github.com/brianvoe/gofakeit/v6"
)

I/O Contract

Inputs

• Context (context.Context): Used for cancellation signaling. When the context is done, the Load method returns.
• Output Channel (chan<- sink.SinkRecord): A channel to which generated SinkRecord events are pushed.

Outputs

The generator produces two types of sink.SinkRecord:

• adImpressionEvent: Always generated per cycle, sent to the ad_impression topic. Contains bid_id, ad_id, and impression_timestamp.
• adClickEvent: Conditionally generated based on the ad's CTR probability, sent to the ad_click topic. Contains bid_id and click_timestamp (impression time plus a random 1-10 second delay).

Kafka Topics

func (g *adCtrGen) KafkaTopics() []string {
	return []string{"ad_impression", "ad_click"}
}

Serialization Formats

Both event types support:

• JSON via ToJson(): Marshals the struct to JSON using encoding/json.
• PostgreSQL via ToPostgresSql(): Returns an INSERT INTO SQL statement string.

Usage Examples

Creating and Using the Generator

package main

import (
	"context"
	"datagen/ad_ctr"
	"datagen/sink"
	"fmt"
)

func main() {
	gen := ad_ctr.NewAdCtrGen()

	// Print the Kafka topics this generator writes to
	fmt.Println(gen.KafkaTopics()) // ["ad_impression", "ad_click"]

	// Create a channel and context, then start generating
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	outCh := make(chan sink.SinkRecord, 100)
	go gen.Load(ctx, outCh)

	// Consume generated records
	for record := range outCh {
		fmt.Println(record.Topic(), string(record.ToJson()))
	}
}

CLI Invocation

The generator is selected via the --mode ad-ctr flag in the datagen CLI:

// In load_gen.go newGen function:
if cfg.Mode == "ad-ctr" {
	return ad_ctr.NewAdCtrGen(), nil
}

Related Pages

• Risingwavelabs_Risingwave_LoadGenerator_Interface -- The LoadGenerator interface that this generator implements
• Risingwavelabs_Risingwave_Datagen_Sink_Interface -- The SinkRecord and Sink interfaces used for event output
• Risingwavelabs_Risingwave_Load_Generation_Loop -- The load generation loop that orchestrates sink creation, generator spawning, and rate-limited writing
• Risingwavelabs_Risingwave_Datagen_CLI -- The CLI entry point that dispatches to this generator