io

package
v0.20.0 Latest Latest
Warning

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

Go to latest
Published: Jun 15, 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 CompareOverlayLayer added in v0.19.0

func CompareOverlayLayer(got, want *types.OverlayLayer) error

CompareOverlayLayer returns nil when the two layers match across all renderer-facing slots (name, kind, scope, ref, payload, summary). Per-shape payload comparison dispatches through CompareOverlayPayload.

func CompareOverlayLayers added in v0.19.0

func CompareOverlayLayers(got, want []*types.OverlayLayer) error

CompareOverlayLayers returns nil when `got` matches `want` slot-for- slot. Mismatch returns an error describing the first divergence — length, slot index, name / kind / scope, shape, or per-shape payload. Designed for use inside table-driven test bodies; callers wrap the error in t.Errorf for the failure message.

Exported so the per-format integration tests under io/exportoverlay/ share one comparison surface across Arrow / Parquet / Excel / NDJSON round-trips.

func CompareOverlayPayload added in v0.19.0

func CompareOverlayPayload(got, want types.OverlayPayload) error

CompareOverlayPayload dispatches by shape and compares the populated arm. Absent arms (the non-matching shape's sub-payload nil) are untouched — the OverlayPayload union always populates exactly one arm per the Shape discriminator.

func CompareOverlayRef added in v0.19.0

func CompareOverlayRef(got, want types.OverlayRef) error

CompareOverlayRef returns nil when the two refs share the same discriminator state. The structural comparison covers the Margin / Population / Stage / Slot arms — extending this helper when new ref arms land follows the additive-only rule (new arms compare nil-vs- nil before recursing into the new fields).

func CompareOverlaySummary added in v0.19.0

func CompareOverlaySummary(got, want *types.OverlaySummary) error

CompareOverlaySummary returns nil when both summaries match across every populated field. nil-vs-nil counts as match; nil-vs-non-nil is a mismatch.

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
	// IncludeOverlays controls whether the export half embeds
	// Response.Overlays in the format-native sidecar. Identical
	// tri-state semantics to ExportJob.IncludeOverlays — nil defaults
	// to emit, *true forces emit, *false drops overlays. The
	// intermediate .pulse file (when KeepPulseAt is set) is byte-
	// identical regardless of this flag because overlays are an
	// export-side concern only; the .pulse byte format never carries
	// overlay payloads. See ExportJob.IncludeOverlays for the per-
	// format wire shape.
	IncludeOverlays *bool
	// Overlays carries the Response.Overlays layers the convert should
	// embed in the target adapter. Identical semantics to
	// ExportJob.Overlays — the export half of the convert dispatches the
	// slice via SetOverlays on writers satisfying OverlayAwareWriter
	// when IncludeOverlays does not explicitly opt out. The intermediate
	// .pulse file (when KeepPulseAt is set) never carries overlay
	// payloads.
	Overlays []*types.OverlayLayer
}

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) Hash added in v0.19.0

func (j *ConvertJob) Hash() string

Hash returns the canonical content hash of the ConvertJob. Same rules as ExportJob.Hash — IncludeOverlays participates so cache keys differ between overlay-bearing and overlay-stripped converts.

The hash is namespaced under "convert_job" to keep ConvertJob hashes disjoint from ExportJob hashes.

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
	OverlayWarnings []*errors.CodedError
}

ConvertReport summarizes the result of a convert operation.

OverlayWarnings carries the warn-and-skip codes the target Writer surfaced through OverlayWarningEmitter — see ExportReport.OverlayWarnings.

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
	// IncludeOverlays controls whether Response.Overlays land in the
	// exported artefact for adapters that can carry them. The slot is
	// a tri-state pointer:
	//
	//   - nil   ⇒ DEFAULT EMIT — when the host Response carries any
	//             overlay layers the adapter embeds them in the
	//             format-native sidecar (additive-by-default, per PRD
	//             §6 FR-L5). Output for an overlay-free Response is
	//             byte-identical to a pre-overlay ExportJob.
	//   - true  ⇒ explicit emit — same behaviour as nil for downstream
	//             adapters, but the explicit toggle distinguishes the
	//             intent in canonical-hash composition so cache keys
	//             differ between "default" and "explicit yes".
	//   - false ⇒ opt out — overlays are dropped on export even when
	//             the host Response carries layers. Output is byte-
	//             identical to a pre-overlay ExportJob against the
	//             same host result.
	//
	// Per-format semantics (research/export-embedding-shape.md):
	//
	//   - Arrow / Parquet — overlays ride as a top-level
	//     LIST<STRUCT> "overlays" field group emitted once in the
	//     first record-batch / row-group; the host record stream is
	//     byte-identical to the opt-out shape.
	//   - Excel — one sheet per layer named "__overlay_<layer_name>";
	//     the host workbook sheet is unchanged from the opt-out shape.
	//   - NDJSON — single trailing line {"_overlays": [...]} after the
	//     last host-record line; the host stream is byte-identical
	//     until the trailer.
	//   - CSV / TSV — warn-and-skip. The dispatcher emits one
	//     PULSE_OVERLAY_EXPORT_CSV_UNSUPPORTED warning and writes the
	//     host CSV verbatim; no overlay output lands. Setting
	//     IncludeOverlays=false suppresses the warning while keeping
	//     the same CSV body.
	//
	// The pointer shape mirrors the Includes-slot precedent (a nil
	// slice means "default": include every field) — nil here means
	// "default: emit overlays when present" rather than "explicit no"
	// because Go bool zero is false. Marshalling via encoding/json
	// honours `omitempty` so nil pointers do not appear in canonical
	// JSON; explicit *true and *false both serialise as booleans and
	// produce distinct canonical-hash keys via ExportJob.Hash().
	IncludeOverlays *bool
	// Overlays carries the Response.Overlays layers the export should
	// embed in the target adapter's format-native sidecar (Arrow /
	// Parquet column family, Excel sheets, NDJSON trailer) per
	// research/export-embedding-shape.md. ExportJob.Run dispatches the
	// slice to writers satisfying OverlayAwareWriter when IncludeOverlays
	// does not explicitly opt out. The CSV / TSV warn-and-skip family
	// also accepts the slice through SetOverlays but emits a
	// PULSE_OVERLAY_EXPORT_CSV_UNSUPPORTED warning (surfaced on
	// ExportReport.OverlayWarnings) instead of writing the layers into
	// the CSV body. nil / empty leaves the export byte-identical to a
	// pre-overlay job. The slot is intentionally NOT part of the
	// canonical-hash composition — two jobs sharing IncludeOverlays /
	// Source / Includes / Labels but differing in the overlay payload
	// resolve through the SAME cache key because the cache identity is
	// the export REQUEST, not the response.
	Overlays []*types.OverlayLayer
}

ExportJob converts a .pulse file into tabular output.

func NewExportJob

func NewExportJob(source string, target Writer) *ExportJob

NewExportJob creates an ExportJob.

func (*ExportJob) Hash added in v0.19.0

func (j *ExportJob) Hash() string

Hash returns the canonical content hash of the ExportJob. Two jobs with the same logical export shape produce the same hash; two jobs with different IncludeOverlays settings (nil / *true / *false) produce distinct hashes so consumers caching export artefacts by hash key do not return stale results when the overlay opt-out flips.

IncludeOverlays participates in the hash composition per research/export-embedding-shape.md §8: the slot is a request-side input to the adapter, so two jobs differing only in IncludeOverlays produce different export artefacts (overlays emitted in one, dropped in the other) and the cache key MUST distinguish.

The hash is namespaced under "export_job" to keep ExportJob and ConvertJob hashes disjoint even when their shapes overlap.

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
	OverlayWarnings []*errors.CodedError
}

ExportReport summarizes the result of an export operation.

OverlayWarnings carries the warn-and-skip codes the target Writer surfaced through the optional OverlayWarningEmitter contract (currently the CSV / TSV adapters via PULSE_OVERLAY_EXPORT_CSV_UNSUPPORTED). Empty / nil when the target adapter embedded the overlays natively (Arrow / Parquet / Excel / NDJSON), when no overlays were supplied, or when IncludeOverlays explicitly opted out.

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 OverlayAwareWriter added in v0.19.0

type OverlayAwareWriter interface {
	Writer
	SetOverlays(layers []*types.OverlayLayer)
}

OverlayAwareWriter is an optional extension of Writer for targets that can embed Response.Overlays in the exported artefact (Arrow / Parquet / Excel / NDJSON per research/export-embedding-shape.md). The ExportJob dispatch wiring calls SetOverlays before WriteHeader on writers that implement this interface when ExportJob.IncludeOverlays resolves to true; the writer then emits the layers in its format-native sidecar shape at Close time (or earlier where the format allows). Writers that do not implement this interface receive no overlay slice — the layers are dropped, which is the correct behaviour for the CSV / TSV warn- and-skip family.

type OverlayWarningEmitter added in v0.19.0

type OverlayWarningEmitter interface {
	OverlayWarnings() []*errors.CodedError
}

OverlayWarningEmitter is an optional extension a Writer can implement to surface per-format overlay warnings the dispatcher should lift onto the ExportReport / ConvertReport. The canonical user is the CSV / TSV adapter which emits PULSE_OVERLAY_EXPORT_CSV_UNSUPPORTED via this surface whenever the dispatcher hands it a non-empty overlay slate. Writers that embed overlays natively (Arrow / Parquet / Excel / NDJSON) do not implement this interface; their overlay output rides in the format-native sidecar instead.

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 exportoverlay holds the higher-level integration tests for the ExportJob and ConvertJob overlay-embedding wiring (E9-S8).
Package exportoverlay holds the higher-level integration tests for the ExportJob and ConvertJob overlay-embedding wiring (E9-S8).
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