Documentation
¶
Index ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
Types ¶
type InferredDiskBatchHandler ¶
type InferredDiskBatchHandler struct {
// contains filtered or unexported fields
}
InferredDiskBatchHandler buffers a batch of JSON messages to a file and lets DuckDB infer the schema, rather than holding the batch in memory.
It follows the Python handler step for step: the batch file is loaded with read_json_auto into a `batch` table, the user SQL is run through COPY into an output file, and that file is read back as the result. The `batch` table is dropped after every invoke, so the next batch re-creates it against whatever schema its own messages imply.
func NewInferredDiskBatchHandler ¶
func NewInferredDiskBatchHandler( conn adbc.Connection, sql string, cacheDir string, opts ...InferredDiskBatchHandlerOption, ) (*InferredDiskBatchHandler, error)
func (*InferredDiskBatchHandler) Close ¶
func (h *InferredDiskBatchHandler) Close() error
Close removes the staged files. The batch file grows with the batch size, so leaving it behind after shutdown wastes disk until the next run.
func (*InferredDiskBatchHandler) Init ¶
func (h *InferredDiskBatchHandler) Init(ctx context.Context) error
Init truncates the batch file, so a batch never inherits rows from the previous one, and clears any `batch` table left behind on the connection.
func (*InferredDiskBatchHandler) Write ¶
func (h *InferredDiskBatchHandler) Write(r []byte) error
type InferredDiskBatchHandlerOption ¶
type InferredDiskBatchHandlerOption func(*InferredDiskBatchHandler)
func InferredDiskBatchWithLogger ¶
func InferredDiskBatchWithLogger(l *zap.Logger) InferredDiskBatchHandlerOption
type InferredMemBatchHandler ¶
type InferredMemBatchHandler struct {
// contains filtered or unexported fields
}
InferredMemBatchHandler buffers a batch of JSON messages in memory and infers the Arrow schema from the messages themselves, rather than requiring a pre-declared table.
Schema inference follows pyarrow.Table.from_pylist, which the Python engine uses: the column set and their order come from the first message, types are promoted across the batch, and a value that cannot be promoted fails the batch rather than being silently nulled.
func NewInferredMemBatchHandler ¶
func NewInferredMemBatchHandler( conn adbc.Connection, sql string, opts ...InferredMemBatchHandlerOption, ) (*InferredMemBatchHandler, error)
func (*InferredMemBatchHandler) Init ¶
func (h *InferredMemBatchHandler) Init(ctx context.Context) error
func (*InferredMemBatchHandler) Write ¶
func (h *InferredMemBatchHandler) Write(r []byte) error
func (*InferredMemBatchHandler) WriteMessage ¶
func (h *InferredMemBatchHandler) WriteMessage(msg core.Message) error
WriteMessage buffers a message along with its source metadata, which Invoke exposes as kafka_topic / kafka_partition / kafka_offset columns.
type InferredMemBatchHandlerOption ¶
type InferredMemBatchHandlerOption func(*InferredMemBatchHandler)
func InferredMemBatchWithLogger ¶
func InferredMemBatchWithLogger(l *zap.Logger) InferredMemBatchHandlerOption
type StructuredBatchHandler ¶
type StructuredBatchHandler struct {
// contains filtered or unexported fields
}
func NewStructuredBatchHandler ¶
func NewStructuredBatchHandler( conn adbc.Connection, sql string, tableName string, schema *arrow.Schema, opts ...StructuredBatchHandlerOption, ) (*StructuredBatchHandler, error)
func (*StructuredBatchHandler) Init ¶
func (h *StructuredBatchHandler) Init(ctx context.Context) error
func (*StructuredBatchHandler) Write ¶
func (h *StructuredBatchHandler) Write(r []byte) error
type StructuredBatchHandlerOption ¶
type StructuredBatchHandlerOption func(*StructuredBatchHandler)
func StructuredBatchWithLogger ¶
func StructuredBatchWithLogger(l *zap.Logger) StructuredBatchHandlerOption
Source Files