---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/go/examples/wait_click_join.md
  - https://ytsaurus.tech/docs/ru/flow/go/examples/wait_click_join.md
  - href: https://ytsaurus.tech/docs/en/flow/go/examples/wait_click_join.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65

<!-- source: en/_includes/flow/go/examples/wait_click_join.md -->
# Wait Click Join in YTsaurus Flow (Go)

An example of a join [pipeline](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#pipeline) that joins the [streams](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#stream-and-computation) of impressions (`hit`) and actions (`action`) by a shared key, using [timers](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#timer) and an external [state](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#state). This is a Go implementation of the similar [C++ example](https://ytsaurus.tech/docs/en/flow/cpp/examples/wait_click_join.md).

[Source code](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/go/wait_click_join)

## Structure {#structure}

The pipeline consists of three [computations](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#stream-and-computation):

- `hit_reader` and `action_reader` — native sources (`TSwiftPassthroughOrderedSourceComputation`) declared directly in the [spec](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#spec-and-dynamic-spec): they read two queues and publish the `hit` and `action` streams. They have no Go code. Each source declares a `watermark_strategy`, so the [watermark](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#timestamps-and-watermarks) of the pipeline is computed from the event time.
- `join` (`joinFunction`) — a transform computation served by the companion.

The `join` computation works as follows:

1. It receives messages from two input streams: `hit` and `action`.
2. It accumulates the window of one impression in the external state `/join-state`.
3. It sets a timer for the moment the waiting window closes.
4. When the timer fires, it publishes the join result to the `joined_action` stream or simply clears the state.

Both input streams are grouped by `hit_id` and `hit_time`, so everything that belongs to one impression falls into one key. The length of the window is set by the `wait_for_actions` computation parameter from the spec.

## `main.go` {#main-go}

The entry point: creating the pipeline, registering the only computation, and starting it.

[Source code: `main.go`](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/go/wait_click_join/main.go)

```go

type actionMessage struct {
	flow.YSONMessage
	HitID      string `yson:"hit_id"`
	HitTime    uint64 `yson:"hit_time"`
	IsClick    bool   `yson:"is_click"`
	ActionTime uint64 `yson:"action_time"`
}

type hitMessage struct {
	flow.YSONMessage
	HitID      string `yson:"hit_id"`
	HitTime    uint64 `yson:"hit_time"`
	HitPayload string `yson:"hit_payload"`
}

type joinedActionMessage struct {
	flow.YSONMessage
	HitID      string `yson:"hit_id"`
	HitTime    uint64 `yson:"hit_time"`
	IsClick    bool   `yson:"is_click"`
	ShowTime   uint64 `yson:"show_time"`
	ClickTime  uint64 `yson:"click_time"`
	HitPayload string `yson:"hit_payload"`
}

func main() {
	pipeline := flow.NewPipeline()
	pipeline.AddStreams(
		flow.NewYSONStream[actionMessage](actionStream),
		flow.NewYSONStream[hitMessage](hitStream),
		flow.NewYSONStream[joinedActionMessage](joinedStream),
	)
	pipeline.Add(flow.NewRowComputation("join", &joinFunction{}))

	if err := pipeline.Run(); err != nil {
		fmt.Fprintf(os.Stderr, "wait_click_join: %v\n", err)
		os.Exit(1)
	}
}
```

## `join_function.go` {#join-function-go}

The main logic of the join: the function handles two streams and uses timers for window aggregation.

### `OnMessage` {#on-message}

It drops late messages, appends its own field to the impression window, and sets the timer for closing the window:

```go

type joinFunction struct{}

var (
	_ flow.RowFunction      = (*joinFunction)(nil)
	_ flow.RowTimerFunction = (*joinFunction)(nil)
)

func (*joinFunction) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	var hit *hitMessage
	var action *actionMessage
	var hitTime uint64
	switch msg.StreamID {
	case hitStream:
		hit = &hitMessage{}
		if err := msg.ConvertTo(hit); err != nil {
			return err
		}
		hitTime = hit.HitTime
	case actionStream:
		action = &actionMessage{}
		if err := msg.ConvertTo(action); err != nil {
			return err
		}
		hitTime = action.HitTime
	default:
		return fmt.Errorf("unhandled stream %q", msg.StreamID)
	}

	wait, err := waitForActions(rt)
	if err != nil {
		return err
	}
	closeTime := hitTime + wait
	if msg.EventTimestamp >= closeTime || msg.EventTimestamp < rt.MinWatermark() {
		return nil
	}

	state, err := flow.OpenExternalState(rt, joinStateName, msg)
	if err != nil {
		return err
	}
	var window joinState
	if _, err := state.ConvertTo(&window); err != nil {
		return err
	}

	if hit != nil {
		window.HitPayload = &hit.HitPayload
	} else if action.IsClick {
		window.ClickTime = &action.ActionTime
	} else {
		window.ShowTime = &action.ActionTime
	}
	if err := state.ConvertFrom(&window); err != nil {
		return err
	}

	out.AddTimer(flow.TimerRequest{TriggerTimestamp: closeTime, EventTimestamp: hitTime})
	return nil
}
```

### `OnTimer` {#on-timer}

It closes the window: it publishes the join result if both sides arrived, and clears the state in any case:

```go

func (*joinFunction) OnTimer(
	ctx context.Context,
	rt flow.Runtime,
	timer flow.Timer,
	out flow.OutputCollector,
) error {
	if timer.StreamID != timerStream {
		return fmt.Errorf("unhandled timer stream %q", timer.StreamID)
	}

	state, err := flow.OpenExternalState(rt, joinStateName, timer)
	if err != nil {
		return err
	}
	var window joinState
	if _, err := state.ConvertTo(&window); err != nil {
		return err
	}

	if window.ShowTime != nil && *window.ShowTime != 0 && window.HitPayload != nil {
		var key joinKey
		if err := timer.Key.ConvertTo(&key); err != nil {
			return err
		}
		joined, err := joinedAction(rt, key, window)
		if err != nil {
			return err
		}
		out.AddMessage(joined)
	}

	return state.Clear()
}
```

## Key patterns {#key-patterns}

- External state through `flow.OpenExternalState(rt, joinStateName, msg)`: `ConvertTo` reads the window into `joinState`, the message changes the field it needs in the structure, and `ConvertFrom` saves the window.
- A timer with two timestamps: `out.AddTimer(flow.TimerRequest{TriggerTimestamp: closeTime, EventTimestamp: hitTime})` — `TriggerTimestamp` sets when the window will close, `EventTimestamp` ties the timer to the event. Messages from both streams set the timer, not only `hit`: an action can arrive earlier than the impression it belongs to, and the window still has to be closed. A message that already falls outside its window or behind the watermark returns early and sets no timer.
- Filtering out late data: `msg.EventTimestamp < rt.MinWatermark()` drops the messages of already closed windows — otherwise a second join result would be published for one impression. For details, see [watermarks](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#timestamps-and-watermarks).
- Computation parameters: `rt.Parameters().Get(waitForActionsParameter, &spelled)` reads the window length from the spec instead of hardcoding it.
- Handling several input streams: the branch on `msg.StreamID` splits the logic for `hit` and `action`, and an unknown stream returns an error instead of a silent skip.
- The key instead of the state: `timer.Key` carries `hit_id` and `hit_time` from `group_by_schema`, so the window holds only the data being joined — there is no need to duplicate the key in it.
<!-- endsource: en/_includes/flow/go/examples/wait_click_join.md -->

<!-- source: en/_includes/flow/go/examples/wait_click_join_also.md -->
## See also

- [Getting started (Go)](https://ytsaurus.tech/docs/en/flow/go/getting-started.md)
- [External State (Go)](https://ytsaurus.tech/docs/en/flow/go/external-state.md)
- [Timers](https://ytsaurus.tech/docs/en/flow/concepts/timers.md)
- [Watermarks](https://ytsaurus.tech/docs/en/flow/concepts/watermarks.md)
- [Similar example in C++](https://ytsaurus.tech/docs/en/flow/cpp/examples/wait_click_join.md)
<!-- endsource: en/_includes/flow/go/examples/wait_click_join_also.md -->