parquet

type Config[T any] struct {
	// Reader is the Parquet object. parquet.ReaderAtSeeker is
	// satisfied by *os.File and by aws.ReadSeekCloser bodies.
	// The Source does not own its lifecycle; close it externally
	// (or use NewSourceClosing to delegate).
	Reader parquet.ReaderAtSeeker

	// Name is a stable identifier for logging / metrics. Typically the
	// S3 key or the file path.
	Name string

	// Decode converts an Arrow row into T. Required — no
	// default schema-introspection exists; callers know their schema.
	Decode Decoder[T]

	// EventID derives the dedup key per record. Optional; see EventIDFn.
	EventID EventIDFn[T]

	// EventTime extracts the upstream event time. Optional; default
	// is time.Now at decode.
	EventTime EventTimeFn[T]

	// OnDecodeError, when non-nil, is invoked for rows that fail to
	// decode. Default behavior: drop silently and continue. Wire to a
	// DLQ producer or logger to surface poison rows.
	OnDecodeError func(rowOrdinal int, err error)

	// BatchSize is the number of rows materialized per arrow.Record
	// pulled out of Parquet. Defaults to 4096. Lower for large rows
	// (less peak memory), higher for tiny rows (less per-batch
	// overhead).
	BatchSize int64

	// Allocator is the Arrow memory allocator. Defaults to
	// memory.DefaultAllocator. Override for tracked allocators in
	// tests or for arena-style reuse.
	Allocator memory.Allocator

	// HandoffToken, when non-nil, is what CaptureHandoff returns.
	// Parquet snapshots have no inherent live-source resume position;
	// the caller's deployment system is responsible for capturing one
	// before the Parquet file was generated (typically a Kinesis shard
	// timestamp or Kafka offset).
	HandoffToken snapshot.HandoffToken
}

type Decoder[T any] func(rec arrow.Record, row int) (T, error)

type EventIDFn[T any] func(decoded T, rowOrdinal int) string

type EventTimeFn[T any] func(decoded T) time.Time

type Partition struct {
	// Key is the full S3 object key.
	Key string

	// Values maps each `name=value` segment found in the key path
	// (relative to Config.Prefix) to its value.
	Values map[string]string
}

type PartitionFilter func(p Partition) bool

type S3Config[T any] struct {
	// Client is an SDK v2 S3 client. The Source does not own its
	// lifecycle.
	Client *awss3.Client

	// Bucket is the S3 bucket to scan.
	Bucket string

	// Prefix narrows the scan, e.g. "events/". Empty means the entire
	// bucket — usually undesirable. Hive-style partition segments
	// (`year=...`) under this prefix are auto-discovered.
	Prefix string

	// Decode is the per-row Arrow decoder. Required; see
	// parquet.Decoder.
	Decode Decoder[T]

	// EventID derives the dedup key per record. Receives the object
	// key + row ordinal within that object. Default:
	// "<key>:<row-ordinal>".
	EventID func(decoded T, key string, rowOrdinal int) string

	// EventTime extracts the upstream event time from a decoded row.
	// Optional; default is processing time at decode.
	EventTime EventTimeFn[T]

	// PartitionFilter, when non-nil, is invoked once per discovered
	// object. Return false to skip — no GetObject, no decode. Use
	// for time-range pruning under a static prefix.
	PartitionFilter PartitionFilter

	// KeyFilter, when non-nil, is a low-level pre-PartitionFilter
	// hook. Receives the raw key and must return true to keep it.
	// Use this to drop control objects like `_SUCCESS` markers or
	// `_committed_*` files that Spark writes alongside the data.
	// Defaults to skipping anything that doesn't end in `.parquet`.
	KeyFilter func(key string) bool

	// OnDecodeError is forwarded to the underlying Parquet source for
	// each object. Receives the object key + row ordinal + error.
	OnDecodeError func(key string, rowOrdinal int, err error)

	// MaxConcurrency caps the number of concurrent object reads.
	// Default 4. Capped at len(matched keys).
	MaxConcurrency int

	// BatchSize is forwarded to the per-file source as the Arrow
	// record-batch size. Defaults to 4096.
	BatchSize int64

	// HandoffToken is what CaptureHandoff returns. Same pattern as
	// the other snapshot sources: caller is responsible for capturing
	// it before the Parquet dataset was generated.
	HandoffToken snapshot.HandoffToken

	// OpenObject, when non-nil, overrides the default GetObject path.
	// Use for tests that want to inject deterministic file bodies
	// without a live S3 client. The returned bytes must be the
	// complete Parquet file (the Arrow library needs ReaderAt + seek,
	// so partial / streaming bodies aren't supported).
	OpenObject func(ctx context.Context, key string) ([]byte, error)
}

type S3Source[T any] struct {
	// contains filtered or unexported fields
}

type Source[T any] struct {
	// contains filtered or unexported fields
}

type Stats struct {
	RowsScanned int64
	RowsDecoded int64
}