conformance

type PipelineSubject struct {
	// Integration is the integrations.yml id, e.g. "pipeline.stateful".
	Integration string

	// KeepsState says whether this configuration commits durable state and
	// offsets alongside the source position. A configuration that keeps none
	// cannot prove the invariants about committing them together, and the
	// registry must exempt it.
	KeepsState bool

	// Options builds the pipeline's options from the harness's recorder. A
	// stateful configuration returns core.WithStateStore(r.Offsets(), r.Tx()).
	Options func(r *Recorder) []core.TurbineOption

	// NewSink supplies the destination the pipeline writes to. Nil means the
	// harness discards the rows and judges the event order alone.
	//
	// A subject that supplies one, with ReadBack and Break, is judged on what
	// the destination actually holds. That is the difference between asserting
	// the loop emitted three events in an order and asserting it never
	// committed offsets for a row the sink never wrote.
	NewSink func(t *testing.T) core.Sink

	// ReadBack returns every row the destination holds. Required with NewSink.
	ReadBack func(t *testing.T) []Row

	// Break makes the destination stop accepting writes, and Heal reverses it.
	// With a real sink these replace the harness's synthetic flush failure, so
	// the flush fails the way it would in production.
	Break func(t *testing.T)
	Heal  func(t *testing.T)
}

type Proxy struct {
	// Addr is the host:port a sink dials instead of the destination.
	Addr string
	// contains filtered or unexported fields
}

type Recorder struct {
	// contains filtered or unexported fields
}

type RecordingOffsets struct {
	// contains filtered or unexported fields
}

type RecordingTx struct {
	// contains filtered or unexported fields
}

type Row map[string]any

type SinkSubject struct {
	// Integration is the integrations.yml id, e.g. "sink.clickhouse".
	Integration string

	// New builds the sink under test. Called once per sequence.
	New func(t *testing.T) core.Sink

	// Break makes the destination stop answering.
	//
	// Nil means the integration has nothing that can break, and the
	// network-shaped invariants are skipped. integrations.yml must then
	// exempt it with a reason, or the cell stays missing: the harness
	// skipping and the registry excusing are two statements that must agree.
	Break func(t *testing.T)

	// Heal reverses Break.
	Heal func(t *testing.T)

	// ReadBack returns every row the destination holds. It must not go through
	// whatever Break breaks.
	ReadBack func(t *testing.T) []Row

	// OrderedReadBack says ReadBack returns rows in the order they arrived.
	//
	// False is the default because a destination that cannot report arrival
	// order is common and the failure is silent: an Iceberg scan returns rows
	// in data-file order, and a ClickHouse MergeTree returns them in its
	// ORDER BY key order, so both would satisfy preserves_order while
	// preserving nothing. The harness skips that claim rather than reading a
	// sorted list as evidence, and integrations.yml must carry the exemption.
	OrderedReadBack bool

	// Table returns a one-row table with an int64 "id" column that the
	// destination accepts. The subject owns the schema because a ClickHouse
	// table has the columns its DDL declared, and the harness has no DDL.
	Table func(t *testing.T, id int64) arrow.Table
}

type Trigger string

type TypeDestination struct {
	// Sink writes to a destination holding a single column named "v".
	Sink core.Sink

	// ReadBack returns the one value that column holds, nil when it is null.
	// It must not go through the sink: a destination read through the thing
	// under test cannot contradict it.
	ReadBack func(t *testing.T) (any, error)

	// ReadBackNullElement reports whether the second element of the single
	// list row is null.
	//
	// It exists because nil-ness cannot answer the question. A list holding a
	// null element is not itself null, so ReadBack returns a value either way
	// and [1, 0, 3] is indistinguishable from [1, NULL, 3] through it. Only
	// the destination can say which it holds.
	//
	// Required of a subject whose table declares any list row.
	ReadBackNullElement func(t *testing.T) (bool, error)
}

type TypeSubject struct {
	// Integration is the integrations.yml id, e.g. "sink.clickhouse".
	Integration string

	// Declared is the integration's type table. Passed in rather than read
	// here, so the harness's own tests can drive the runner with a table of
	// one row and a known-wrong outcome.
	Declared []coverage.TypeDecl

	// Nulls is what the integration does with a null. It is one statement
	// about every type: ClickHouse stores a null in a non-Nullable column as
	// the column type's zero value, whatever that type is, and asserting that
	// a null survives would assert against the destination's own semantics.
	Nulls coverage.NullRule

	// ListElementNulls is what the integration does with a null held inside a
	// non-null list. It is a separate statement from Nulls because the answer
	// differs: a ClickHouse column can be Nullable and its Array(T) elements
	// cannot, so the column-level rule says nothing about what a list holds.
	ListElementNulls coverage.NullRule

	// Prepare creates a destination with a single column "v" of columnType,
	// and a sink writing to it.
	//
	// columnType is empty for an unsupported row: no column type accepts the
	// value, and the batch must fail before any destination matters.
	Prepare func(t *testing.T, key, columnType string) TypeDestination
}