athena

Documentation

Overview

Package athena provides a database/sql-style executor backed by github.com/influxdata/athenadriver/v2. The driver speaks to Amazon Athena via the AWS Athena API (StartQueryExecution / GetQueryExecution / GetQueryResults) but exposes a database/sql-compatible surface, which lets us reuse the stdsql helpers other scrappers already use.

Index

• func NewQuerier[T any](conn *AthenaExecutor) querier.Querier[T]
• type AthenaConf
• type AthenaExecutor
  • func NewAthenaExecutor(ctx context.Context, conf *AthenaConf) (*AthenaExecutor, error)
  • func (e *AthenaExecutor) AccountID() string
  • func (e *AthenaExecutor) AthenaClient() *awsathena.Client
  • func (e *AthenaExecutor) Catalog() string
  • func (e *AthenaExecutor) Close() error
  • func (e *AthenaExecutor) Exec(ctx context.Context, query string, args ...any) error
  • func (e *AthenaExecutor) GetDb() *sqlx.DB
  • func (e *AthenaExecutor) GlueClient() *awsglue.Client
  • func (e *AthenaExecutor) Instance() string
  • func (e *AthenaExecutor) QueryRows(ctx context.Context, sql string, args ...interface{}) (*sqlx.Rows, error)
  • func (e *AthenaExecutor) Region() string
  • func (e *AthenaExecutor) Select(ctx context.Context, dest any, query string, args ...any) error
  • func (e *AthenaExecutor) Workgroup() string
• type Executor
• type Querier
  • func (q *Querier[T]) Close() error
  • func (q *Querier[T]) Exec(ctx context.Context, sql string) error
  • func (q *Querier[T]) QueryAndProcessMany(ctx context.Context, sql string, ...) error
  • func (q *Querier[T]) QueryMany(ctx context.Context, sql string, opts ...exec.QueryManyOpt[T]) ([]*T, error)
  • func (q *Querier[T]) QueryMaps(ctx context.Context, sql string) ([]exec.QueryMapResult, error)

Functions

func NewQuerier

func NewQuerier[T any](conn *AthenaExecutor) querier.Querier[T]

Types

type AthenaConf

type AthenaConf struct {
	Region    string
	Workgroup string // empty defaults to "primary"
	Catalog   string // empty defaults to "AwsDataCatalog"

	// Static credentials. Highest priority.
	AccessKeyID     string
	SecretAccessKey string
	SessionToken    string

	// Named AWS shared-config profile. Used only when static credentials are absent.
	AwsProfile string

	// AssumeRole. Wraps whichever base credentials resolved above.
	RoleArn         string
	ExternalID      string
	RoleSessionName string // empty defaults to "synq-athena"

	// AllowDefaultChain opts into the AWS default credential chain (env vars,
	// shared config, EC2/ECS/EKS instance role) when no explicit auth method
	// is configured. Safe to enable on the on-prem agent host — that's the
	// customer's own AWS environment. MUST stay false on the SYNQ cloud
	// backend so a misconfigured customer integration cannot fall through
	// to SYNQ's own AWS identity.
	AllowDefaultChain bool
}

AthenaConf carries everything the executor needs to open a connection.

Authentication is resolved in priority order:

1. Explicit AccessKeyID + SecretAccessKey (+ optional SessionToken).
2. AwsProfile — named profile from ~/.aws/credentials.
3. AWS default credential chain (env vars, shared config, EC2/ECS/EKS instance role) — ONLY when AllowDefaultChain is set. Refused otherwise to prevent a customer configuration with empty fields from accidentally inheriting the host process's identity (a real risk on the SYNQ cloud backend).

If RoleArn is set on top of any of the above, the executor wraps the base credentials in an STS AssumeRole provider.

type AthenaExecutor

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

func NewAthenaExecutor

func NewAthenaExecutor(ctx context.Context, conf *AthenaConf) (*AthenaExecutor, error)

func (*AthenaExecutor) AccountID

func (e *AthenaExecutor) AccountID() string

func (*AthenaExecutor) AthenaClient

func (e *AthenaExecutor) AthenaClient() *awsathena.Client

func (*AthenaExecutor) Catalog

func (e *AthenaExecutor) Catalog() string

func (*AthenaExecutor) Close

func (e *AthenaExecutor) Close() error

func (*AthenaExecutor) Exec

func (e *AthenaExecutor) Exec(ctx context.Context, query string, args ...any) error

func (*AthenaExecutor) GetDb

func (e *AthenaExecutor) GetDb() *sqlx.DB

func (*AthenaExecutor) GlueClient

func (e *AthenaExecutor) GlueClient() *awsglue.Client

func (*AthenaExecutor) Instance

func (e *AthenaExecutor) Instance() string

Instance returns the canonical instance identifier for an Athena endpoint: "<account-id>.<region>". Disambiguates two Athena integrations across AWS accounts that share a Glue database name.

func (*AthenaExecutor) QueryRows

func (e *AthenaExecutor) QueryRows(ctx context.Context, sql string, args ...interface{}) (*sqlx.Rows, error)

func (*AthenaExecutor) Region

func (e *AthenaExecutor) Region() string

func (*AthenaExecutor) Select

func (e *AthenaExecutor) Select(ctx context.Context, dest any, query string, args ...any) error

func (*AthenaExecutor) Workgroup

func (e *AthenaExecutor) Workgroup() string

type Executor

type Executor interface {
	stdsql.StdSqlExecutor
}

type Querier

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

Querier wraps stdsql.StdSqlQuerier so callers can run typed parametric queries against an Athena executor through the shared querier.Querier interface — same shape as the Trino exec package.

func (*Querier[T]) Close

func (q *Querier[T]) Close() error

func (*Querier[T]) Exec

func (q *Querier[T]) Exec(ctx context.Context, sql string) error

func (*Querier[T]) QueryAndProcessMany

func (q *Querier[T]) QueryAndProcessMany(
	ctx context.Context,
	sql string,
	handler func(ctx context.Context, batch []*T) error,
	opts ...exec.QueryManyOpt[T],
) error

func (*Querier[T]) QueryMany

func (q *Querier[T]) QueryMany(ctx context.Context, sql string, opts ...exec.QueryManyOpt[T]) ([]*T, error)

func (*Querier[T]) QueryMaps

func (q *Querier[T]) QueryMaps(ctx context.Context, sql string) ([]exec.QueryMapResult, error)