Documentation
¶
Index ¶
- type BatchSQLSource
- func (s *BatchSQLSource) Ack(ctx context.Context, msg hermod.Message) error
- func (s *BatchSQLSource) Close() error
- func (s *BatchSQLSource) GetState() map[string]string
- func (s *BatchSQLSource) Ping(ctx context.Context) error
- func (s *BatchSQLSource) Read(ctx context.Context) (hermod.Message, error)
- func (s *BatchSQLSource) Sample(ctx context.Context, table string) (hermod.Message, error)
- func (s *BatchSQLSource) SetLogger(logger hermod.Logger)
- func (s *BatchSQLSource) SetState(state map[string]string)
- type Config
- type DBProvider
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type BatchSQLSource ¶
type BatchSQLSource struct {
// contains filtered or unexported fields
}
BatchSQLSource implements the hermod.Source interface for scheduled SQL queries.
func NewBatchSQLSource ¶
func NewBatchSQLSource(dbProvider DBProvider, config Config) *BatchSQLSource
NewBatchSQLSource creates a new BatchSQLSource.
func (*BatchSQLSource) Ack ¶
Ack moves the cursor to the acknowledged row's watermark — before releasing the message, whose metadata carries it.
func (*BatchSQLSource) Close ¶
func (s *BatchSQLSource) Close() error
Close stops the cron scheduler and releases resources.
func (*BatchSQLSource) GetState ¶
func (s *BatchSQLSource) GetState() map[string]string
GetState returns the current state for persistence.
func (*BatchSQLSource) Ping ¶
func (s *BatchSQLSource) Ping(ctx context.Context) error
Ping checks if the schedule is valid.
func (*BatchSQLSource) Sample ¶
Sample fetches a single record for preview.
A batch_sql source is query-driven and owns no table: its config carries `queries`, never `table` or `tables`. Callers that have a table name (the column browser, a sink-side preview) still pass one; the workflow editor has none to pass, and used to send the empty string — which built "SELECT * FROM LIMIT 1" and failed as a syntax error. Downstream nodes build their Available Fields list from this sample, so the failure surfaced only as an empty field list on every node wired to a batch_sql source.
With no table name the first configured query is what the pipeline will actually run, so that is what gets previewed — the same statement, with the same watermark substitution, so the previewed columns are the columns the scheduled run emits.
func (*BatchSQLSource) SetLogger ¶
func (s *BatchSQLSource) SetLogger(logger hermod.Logger)
SetLogger sets the logger for the source.
func (*BatchSQLSource) SetState ¶
func (s *BatchSQLSource) SetState(state map[string]string)
SetState sets the initial state for incremental tracking.
type Config ¶
type Config struct {
SourceID string `json:"source_id"`
Cron string `json:"cron"`
Queries string `json:"queries"`
IncrementalColumn string `json:"incremental_column"`
// Parameters is a JSON object of named values bound into the queries. A
// batch_sql source has no inbound message, so a {{ }} token other than
// {{.last_value}} can only come from here.
Parameters string `json:"parameters"`
}
Config defines the configuration for BatchSQLSource.
Source Files