io

package
v0.16.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Jun 10, 2026 License: MIT Imports: 12 Imported by: 0

Documentation

Overview

Package io defines the I/O pipeline framework for Pulse: Reader/Writer interfaces, schema inference, and job types (ImportJob, ExportJob, ConvertJob).

Format-specific adapters live in sub-packages (csv, tsv).

Index

Constants

View Source
const DefaultSetDelimiter = "|"

DefaultSetDelimiter is the delimiter assumed by convertValue when no per-column delimiter has been recorded (explicit-schema imports that skipped the inference pass). Always |.

Variables

This section is empty.

Functions

func ErrStopIteration

func ErrStopIteration() error

ErrStopIteration returns the stop iteration sentinel for use by readers.

Types

type ConvertJob

type ConvertJob struct {
	Source      Reader
	Target      Writer
	Schema      *encoding.Schema
	KeepPulseAt string // optional: also write intermediate .pulse
	SampleRows  int
	FS          afero.Fs
	// Includes restricts the export half to the named schema fields,
	// in schema order. Nil / empty means write every field. The
	// intermediate .pulse file (when KeepPulseAt is set) always
	// carries the full schema — projection is an output-time overlay,
	// not an on-disk schema change. See ExportJob.Includes.
	Includes []string
	// Labels apply to the export phase only — the import side reads
	// raw source bytes and has no use for label translation. See
	// ExportJob.Labels.
	Labels []*types.LabelBinding
	// LabelResolver carries the runtime resolver. See
	// ExportJob.LabelResolver.
	LabelResolver LabelResolver
}

ConvertJob chains import and export with no intermediate file on disk (unless KeepPulseAt is set).

func NewConvertJob

func NewConvertJob(source Reader, target Writer) *ConvertJob

NewConvertJob creates a ConvertJob with default settings.

func (*ConvertJob) Predict

func (j *ConvertJob) Predict(ctx context.Context) (*PredictReport, error)

Predict validates the conversion without writing.

func (*ConvertJob) Run

func (j *ConvertJob) Run(ctx context.Context) (*ConvertReport, error)

Run executes the convert job, streaming from source to target. If KeepPulseAt is set, also writes an intermediate .pulse file.

type ConvertReport

type ConvertReport struct {
	RowsConverted int
	Schema        *encoding.Schema
	RowErrors     []RowError
	LabelWarnings []LabelWarning
}

ConvertReport summarizes the result of a convert operation.

type ExportJob

type ExportJob struct {
	Source string // input .pulse path
	Target Writer
	FS     afero.Fs
	// Includes restricts the export to the named source-schema fields,
	// in source-schema order. Nil / empty means export every field
	// (prior behaviour). Names must match Schema.Fields[i].Name
	// exactly. Unknown names return PULSE_EXPORT_FIELD_UNKNOWN with
	// the offending name + list of known fields. Label augment
	// siblings ("<field>_label") are emitted only for included source
	// fields; replace mode applies to included fields as before.
	Includes []string
	// Labels rewrites or augments categorical column values during
	// export using embedder-registered label tables. Bindings name a
	// categorical field plus a label table; mode=replace overwrites
	// the column value, mode=augment emits a sibling "<field>_label"
	// column. See types.LabelBinding for semantics. Nil/empty means
	// the exporter writes raw resolved categorical values, matching
	// pre-label behaviour.
	Labels []*types.LabelBinding
	// LabelResolver carries the runtime resolver built from Labels +
	// the pulse Service's registered LabelTables. The pulse.Export
	// facade builds the resolver and sets this field; callers using
	// io.ExportJob directly without a Pulse instance must supply a
	// satisfying implementation (typically wrapping
	// processing.BuildLabelResolver). Nil means no label translation.
	LabelResolver LabelResolver
}

ExportJob converts a .pulse file into tabular output.

func NewExportJob

func NewExportJob(source string, target Writer) *ExportJob

NewExportJob creates an ExportJob.

func (*ExportJob) Predict

func (j *ExportJob) Predict(ctx context.Context) (*PredictReport, error)

Predict validates the export without writing.

func (*ExportJob) Run

func (j *ExportJob) Run(ctx context.Context) (*ExportReport, error)

Run executes the export job, converting a .pulse file to a tabular target.

type ExportReport

type ExportReport struct {
	RowsExported  int
	RowErrors     []RowError
	LabelWarnings []LabelWarning
}

ExportReport summarizes the result of an export operation.

type ImportJob

type ImportJob struct {
	Source     Reader
	Target     string // output .pulse path
	Schema     *encoding.Schema
	SampleRows int // default 500, min 50
	FS         afero.Fs
	// SetInferenceMinPct configures the delimited-cell heuristic for
	// inferring set_* field types during the inference pass. A column
	// is classified as set_* when at least this percentage of non-
	// null sampled cells contain the inferred delimiter, the
	// post-split unique token count fits in set_u64 (≤64), and the
	// average post-split cardinality is > 1. Zero is treated as 30%.
	// Ignored when Schema is supplied.
	SetInferenceMinPct int
	// ColumnTypeOverrides bypasses inference for the named columns
	// (keyed by column name). Used by the managed-import sidecar's
	// force_type escape hatch. Ignored when Schema is supplied.
	ColumnTypeOverrides map[string]encoding.FieldType
	// SetDelimiters maps set-typed column name to the delimiter the
	// importer should use when splitting cell strings into tokens
	// for per-row mask packing. Populated by inference; absent
	// entries (explicit-schema imports or non-set columns) fall back
	// to DefaultSetDelimiter ("|").
	SetDelimiters map[string]string
}

ImportJob converts tabular source data into a .pulse file.

func NewImportJob

func NewImportJob(source Reader, target string) *ImportJob

NewImportJob creates an ImportJob with default settings.

func (*ImportJob) Predict

func (j *ImportJob) Predict(ctx context.Context) (*PredictReport, error)

Predict validates the import without writing any output.

func (*ImportJob) Run

func (j *ImportJob) Run(ctx context.Context) (*ImportReport, error)

Run executes the import job, converting tabular source into a .pulse file.

type ImportReport

type ImportReport struct {
	RowsImported int
	Schema       *encoding.Schema
	RowErrors    []RowError
}

ImportReport summarizes the result of an import operation.

type InferOptions added in v0.15.0

type InferOptions struct {
	// SampleRows caps how many rows InferSchema reads before
	// finalizing column types. Defaults to defaultSampleRows when
	// <= 0; min-clamped to minSampleRows.
	SampleRows int
	// SetInferenceMinPct is the minimum percentage of non-null
	// sampled cells that must contain the inferred delimiter for the
	// column to be classified as set_*. Defaults to
	// defaultSetInferenceMinPct (30) when 0.
	SetInferenceMinPct int
	// ColumnTypeOverrides maps column name to forced FieldType,
	// bypassing the inference heuristics entirely for those columns.
	// For set_* / categorical_* types the dictionary is still built
	// from observed sample values during the inference pass; the
	// override only fixes the column's storage type.
	ColumnTypeOverrides map[string]encoding.FieldType
}

InferOptions parameterises InferSchemaWithOptions. The zero value is safe — SampleRows / SetInferenceMinPct fall back to their package defaults and ColumnTypeOverrides is treated as empty.

type InferenceResult added in v0.15.0

type InferenceResult struct {
	Schema     *encoding.Schema
	Warnings   []InferenceWarning
	Delimiters map[string]string
}

InferenceResult bundles the inference output. The Delimiters map is populated for columns classified as set_*; callers (importers) use it to pack per-cell masks during the row pass.

func InferSchemaWithOptions added in v0.15.0

func InferSchemaWithOptions(reader Reader, opts InferOptions) (*InferenceResult, error)

InferSchemaWithOptions is the parameterised inference entry. Returns the schema, accumulated warnings, AND the per-column delimiter map for columns classified as set_*. The delimiter map is used by ImportJob.Run during the row pass to split delimited cells into token lists for bit-packing.

type InferenceWarning

type InferenceWarning struct {
	Column  string
	Message string
}

InferenceWarning records a non-fatal observation during inference.

func InferSchema

func InferSchema(reader Reader, sampleRows int) (*encoding.Schema, []InferenceWarning, error)

InferSchema samples up to sampleRows rows from reader and proposes a Schema. If sampleRows <= 0, defaultSampleRows is used. The minimum is minSampleRows. Wrapper over InferSchemaWithOptions that drops the delimiter map; only CSV/TSV/Excel paths that ignore set-typed import packing should call this entry. Importers should call InferSchemaWithOptions directly.

type LabelResolver added in v0.10.1

type LabelResolver interface {
	// Has reports whether the resolver carries any binding for the
	// named field.
	Has(field string) bool
	// Mode returns the binding mode for the named field. Empty when
	// no binding exists.
	Mode(field string) types.LabelMode
	// Apply translates a raw value through the binding for the named
	// field. See processing.LabelResolver.Apply for semantics.
	Apply(field, raw string) (out string, sibling string, ok bool)
	// FieldsWithAugment returns the field names that should emit a
	// "<field>_label" sibling column.
	FieldsWithAugment() []string
	// Warnings flushes the per-binding collision and miss summaries
	// once the job finishes.
	Warnings() []LabelWarning
}

LabelResolver is the io-package projection of the processing-layer label resolver. ExportJob / ConvertJob consume the interface so io stays free of the processing dependency.

processing.LabelResolver satisfies this interface as-is; the pulse-package facade builds the resolver from the user-facing LabelBinding slice + LabelTable registry and hands the value to the job via ExportJob.LabelResolver.

type LabelWarning added in v0.10.1

type LabelWarning struct {
	Code    string
	Message string
	Details map[string]any
}

LabelWarning is the io-package projection of one resolver diagnostic record. ExportReport surfaces these for the caller; the CLI / MCP envelope wrapper promotes them to envelope warnings.

type PredictReport

type PredictReport struct {
	Schema        *encoding.Schema
	EstimatedRows int
	Warnings      []InferenceWarning
}

PredictReport summarizes a validation-only run.

type Reader

type Reader interface {
	// ReadHeader returns column names from the source.
	ReadHeader() ([]string, error)
	// ReadRows streams rows; calls fn for each row.
	// The context controls cancellation.
	ReadRows(ctx context.Context, fn func(row []string) error) error
	// Close releases underlying resources.
	Close() error
}

Reader reads tabular data from a source format.

type ResetReader

type ResetReader interface {
	Reader
	// Reset rewinds the reader to the beginning.
	Reset() error
}

ResetReader is an optional interface for readers that support rewinding to the beginning. This is needed for schema inference followed by import.

type RowError

type RowError struct {
	Row int
	Err error
}

RowError records a per-row error during import or export.

type SchemaAwareWriter added in v0.2.0

type SchemaAwareWriter interface {
	Writer
	SetPulseSchema(s *encoding.Schema)
}

SchemaAwareWriter is an optional extension of Writer for targets that emit native typed columns (Arrow, Parquet, Excel) and want the source .pulse schema to drive column-type selection. ExportJob calls SetPulseSchema before WriteHeader on writers that implement this interface, then passes typed values through WriteRow:

  • encoding.Decimal128 for decimal128 / nullable_decimal128 columns
  • canonical strings for narrow types (current behavior)

Writers that do not implement SchemaAwareWriter receive only canonical strings, which is the prior text-only export contract.

type Writer

type Writer interface {
	// WriteHeader writes column names to the target.
	WriteHeader(columns []string) error
	// WriteRow writes a single row of values.
	WriteRow(values []any) error
	// Close flushes and releases resources.
	Close() error
}

Writer writes tabular data to a target format.

Directories

Path Synopsis
Package arrow provides Arrow IPC (Feather V2) import and export for the pulse I/O pipeline, plus shared Arrow<->Pulse type-mapping helpers used by both this package and io/parquet.
Package arrow provides Arrow IPC (Feather V2) import and export for the pulse I/O pipeline, plus shared Arrow<->Pulse type-mapping helpers used by both this package and io/parquet.
Package csv provides CSV format adapters for the Pulse I/O pipeline.
Package csv provides CSV format adapters for the Pulse I/O pipeline.
Package excel provides Excel import and export for the pulse I/O pipeline.
Package excel provides Excel import and export for the pulse I/O pipeline.
Package format dispatches tabular Reader construction by format identifier, sitting between the io/ interface definitions and the per-format leaf packages (io/csv, io/tsv, io/ndjson, io/jsonarray, io/parquet, io/arrow, io/excel).
Package format dispatches tabular Reader construction by format identifier, sitting between the io/ interface definitions and the per-format leaf packages (io/csv, io/tsv, io/ndjson, io/jsonarray, io/parquet, io/arrow, io/excel).
Package jsonarray provides JSON-array import and export for the pulse I/O pipeline.
Package jsonarray provides JSON-array import and export for the pulse I/O pipeline.
Package jsonshared holds value coercion helpers shared by the ndjson and jsonarray packages.
Package jsonshared holds value coercion helpers shared by the ndjson and jsonarray packages.
Package ndjson provides NDJSON (newline-delimited JSON) import and export for the pulse I/O pipeline.
Package ndjson provides NDJSON (newline-delimited JSON) import and export for the pulse I/O pipeline.
Package parquet provides Parquet import and export for the pulse I/O pipeline.
Package parquet provides Parquet import and export for the pulse I/O pipeline.
Package tsv provides TSV import and export for the pulse I/O pipeline.
Package tsv provides TSV import and export for the pulse I/O pipeline.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL